diff --git a/geomagio/WebService.py b/geomagio/WebService.py
index f5ac8ba19a5713cc9e9bd44acfcb15f048eee230..5d880eb83036a0d03d25a520c5b672030b27fb0c 100644
--- a/geomagio/WebService.py
+++ b/geomagio/WebService.py
@@ -2,7 +2,7 @@
"""
from __future__ import print_function
-from cgi import parse_qs, escape
+from cgi import escape, parse_qs
from datetime import datetime
import sys
@@ -121,7 +121,8 @@ class WebService(object):
Parameters
----------
- query : dictionary of parsed query parameters
+ query : dict
+ parsed query parameters
Returns
-------
diff --git a/geomagio/imfjson/IMFJSONWriter.py b/geomagio/imfjson/IMFJSONWriter.py
index 6639816b58dab800e7f81b6e585545fdbed12b12..6c01e41c23a4b871e2acf475cf801709c0f86246 100644
--- a/geomagio/imfjson/IMFJSONWriter.py
+++ b/geomagio/imfjson/IMFJSONWriter.py
@@ -77,6 +77,13 @@ class IMFJSONWriter(object):
metadata['station'] = stats.station
edge_channel = trace.stats.channel
metadata['channel'] = edge_channel
+ if stats.location == "":
+ if stats.data_type == 'variation':
+ stats.location = 'R0'
+ elif stats.data_type == 'quasi-definitive':
+ stats.location = 'Q0'
+ elif stats.data_type == 'definitive':
+ stats.location = 'D0'
metadata['location'] = stats.location
values += [value_dict]
series = np.copy(trace.data)
@@ -133,8 +140,9 @@ class IMFJSONWriter(object):
else:
rate = 1
intermag['sampling_period'] = rate
+ # 1/sampling_rate to output in seconds rather than hertz
if 'sensor_sampling_rate' in stats:
- sampling = stats.sensor_sampling_rate
+ sampling = 1 / stats.sensor_sampling_rate
intermag['digital_sampling_rate'] = sampling
metadata_dict['intermagnet'] = intermag
metadata_dict['status'] = 200
diff --git a/test/imfjson_test/IMFJSONWriter_test.py b/test/imfjson_test/IMFJSONWriter_test.py
index a697c09ef00d0cf0cb58089b61a213862ce53f5e..d3e6b79ec231a6c1ef201731492a3d8fd5d97df0 100644
--- a/test/imfjson_test/IMFJSONWriter_test.py
+++ b/test/imfjson_test/IMFJSONWriter_test.py
@@ -1,20 +1,17 @@
"""Tests for the IMFJSON Writer class."""
from nose.tools import assert_equals
-from geomagio import edge
+from geomagio.iaga2002 import IAGA2002Factory
from geomagio.imfjson import IMFJSONWriter
+import numpy as np
-from obspy.core import UTCDateTime
-EXAMPLE_INPUT_FACTORY = edge.EdgeFactory()
-EXAMPLE_CHANNELS = ('H', 'E', 'Z')
-EXAMPLE_DATA = EXAMPLE_INPUT_FACTORY.get_timeseries(
- observatory='BOU',
- channels=EXAMPLE_CHANNELS,
- type='variation',
- interval='minute',
- starttime=UTCDateTime('2018-01-02'),
- endtime=UTCDateTime('2018-01-02T00:10:00'))
+EXAMPLE_INPUT_FACTORY = IAGA2002Factory()
+EXAMPLE_CHANNELS = ('H', 'D', 'Z', 'F')
+EXAMPLE_FILE = "etc/iaga2002/BOU/OneMinute/bou20141101vmin.min"
+with open(EXAMPLE_FILE, "r") as input_file:
+ data = input_file.read()
+EXAMPLE_DATA = EXAMPLE_INPUT_FACTORY.parse_string(data)
EXAMPLE_STATS = EXAMPLE_DATA[0].stats
@@ -30,15 +27,16 @@ def test_metadata():
assert_equals(metadata['status'], 200)
# Test intermagnet parameters
intermag = metadata['intermagnet']
- assert_equals(intermag['reported_orientation'], "HEZ")
+ assert_equals(intermag['reported_orientation'], "HDZF")
assert_equals(intermag['sensor_orientation'], "HDZF")
+ assert_equals(intermag['data_type'], "variation")
assert_equals(intermag['sampling_period'], 60)
assert_equals(intermag['digital_sampling_rate'], 0.01)
# Test intermagnet-imo parameters
imo = metadata['intermagnet']['imo']
assert_equals(imo['iaga_code'], "BOU")
assert_equals(imo['name'], "Boulder")
- assert_equals(imo['coordinates'], [254.763, 40.137, 1682])
+ assert_equals(imo['coordinates'], [254.764, 40.137, 1682])
def test_times():
@@ -50,19 +48,12 @@ def test_times():
"""
writer = IMFJSONWriter()
times = writer._format_times(EXAMPLE_DATA, EXAMPLE_CHANNELS)
- assert_equals(times, [
- "2018-01-02T00:00:00.000Z",
- "2018-01-02T00:01:00.000Z",
- "2018-01-02T00:02:00.000Z",
- "2018-01-02T00:03:00.000Z",
- "2018-01-02T00:04:00.000Z",
- "2018-01-02T00:05:00.000Z",
- "2018-01-02T00:06:00.000Z",
- "2018-01-02T00:07:00.000Z",
- "2018-01-02T00:08:00.000Z",
- "2018-01-02T00:09:00.000Z",
- "2018-01-02T00:10:00.000Z"
- ])
+ test_day, test_time = np.loadtxt(EXAMPLE_FILE, skiprows=25,
+ usecols=(0, 1), unpack=True, dtype=str)
+ test_date_times = []
+ for idx in range(test_day.shape[0]):
+ test_date_times += [test_day[idx] + "T" + test_time[idx] + "Z"]
+ assert_equals(times, test_date_times)
def test_values():
@@ -81,8 +72,9 @@ def test_values():
for key, test in zip(val, test_val_keys):
assert_equals(key, test)
assert_equals(values[0]['id'], "H")
- assert_equals(values[1]['id'], "E")
+ assert_equals(values[1]['id'], "D")
assert_equals(values[2]['id'], "Z")
+ assert_equals(values[3]['id'], "F")
# Test values-metadata (need to add flags)
metadata = values[0]['metadata']
test_metadata_keys = ["element", "network", "station",
@@ -96,17 +88,11 @@ def test_values():
# assert_equals(metadata['channel'], "MVH")
assert_equals(metadata['location'], "R0")
# Test values-values
- vals = values[0]['values']
- assert_equals(vals, [
- 20827.89,
- 20827.982,
- 20827.7,
- 20827.542,
- 20827.245,
- 20826.802,
- 20827.007,
- 20826.774,
- 20826.784,
- 20826.946,
- 20827.088
- ])
+ # Round to match iaga format
+ vals_H = np.around(values[0]['values'], 2)
+ vals_D = np.around(values[1]['values'], 2)
+ test_val_H, test_val_D = np.loadtxt(EXAMPLE_FILE, skiprows=25,
+ usecols=(3, 4), unpack=True, dtype=float)
+ # tolist required to prevent ValueError in comparison
+ assert_equals(vals_H.tolist(), test_val_H.tolist())
+ assert_equals(vals_D.tolist(), test_val_D.tolist())