From da02469466af361b856b02089d4697c4930add41 Mon Sep 17 00:00:00 2001
From: Heather Schovanec <hschovanec@usgs.gov>
Date: Tue, 16 Jan 2018 15:00:27 -0700
Subject: [PATCH] Add unit tests and fix PR comments
---
geomagio/imfjson/IMFJSONWriter.py | 58 +++++------
test/imfjson_test/IMFJSONWriter_test.py | 126 ++++++++++++++++++++++++
test/imfjson_test/__init__.py | 0
3 files changed, 151 insertions(+), 33 deletions(-)
create mode 100644 test/imfjson_test/IMFJSONWriter_test.py
create mode 100644 test/imfjson_test/__init__.py
diff --git a/geomagio/imfjson/IMFJSONWriter.py b/geomagio/imfjson/IMFJSONWriter.py
index 252966178..6639816b5 100644
--- a/geomagio/imfjson/IMFJSONWriter.py
+++ b/geomagio/imfjson/IMFJSONWriter.py
@@ -13,7 +13,7 @@ class IMFJSONWriter(object):
"""JSON writer.
"""
- def write(self, out, timeseries, channels, **kwargs):
+ def write(self, out, timeseries, channels, url=None):
"""Write timeseries to json file.
Parameters
@@ -24,8 +24,8 @@ class IMFJSONWriter(object):
timeseries object with data to be written
channels: array_like
channels to be written from timeseries object
- kwargs
- request : query string
+ url: str
+ string with the requested url
Raises
------
@@ -33,7 +33,6 @@ class IMFJSONWriter(object):
if there is a missing channel.
"""
file_dict = OrderedDict()
- request = kwargs.get('request')
for channel in channels:
if timeseries.select(channel=channel).count() == 0:
raise TimeseriesFactoryException(
@@ -42,13 +41,11 @@ class IMFJSONWriter(object):
stats = timeseries[0].stats
file_dict['type'] = 'Timeseries'
file_dict['metadata'] = self._format_metadata(stats, channels)
- if request:
- base = 'http://geomag.usgs.gov/ws/edge/?'
- file_dict['metadata']['url'] = base + request
+ file_dict['metadata']['url'] = url
file_dict['times'] = self._format_times(timeseries, channels)
file_dict['values'] = self._format_data(timeseries, channels, stats)
formatted_timeseries = json.dumps(file_dict,
- ensure_ascii=True).encode('utf8')
+ ensure_ascii=True, separators=(',', ':')).encode('utf8')
out.write(str(formatted_timeseries))
def _format_data(self, timeseries, channels, stats):
@@ -73,19 +70,14 @@ class IMFJSONWriter(object):
value_dict = OrderedDict()
trace = timeseries.select(channel=c)[0]
value_dict['id'] = c
- value_dict = OrderedDict()
- value_dict['element'] = c
- if 'network' in stats:
- value_dict['network'] = stats.network
- value_dict['station'] = stats.station
- if 'channel' in stats:
- edge_channel = trace.stats.channel
- else:
- edge_channel = c
- value_dict['channel'] = edge_channel
- if 'location' in stats:
- value_dict['location'] = stats.location
- # TODO: Add flag metadata
+ value_dict['metadata'] = OrderedDict()
+ metadata = value_dict['metadata']
+ metadata['element'] = c
+ metadata['network'] = stats.network
+ metadata['station'] = stats.station
+ edge_channel = trace.stats.channel
+ metadata['channel'] = edge_channel
+ metadata['location'] = stats.location
values += [value_dict]
series = np.copy(trace.data)
if c == 'D':
@@ -93,6 +85,7 @@ class IMFJSONWriter(object):
series[np.isnan(series)] = None
# Converting numpy array to list required for JSON serialization
value_dict['values'] = series.tolist()
+ # TODO: Add flag metadata
return values
def _format_metadata(self, stats, channels):
@@ -118,11 +111,11 @@ class IMFJSONWriter(object):
imo['name'] = stats.station_name
coords = [None] * 3
if 'geodetic_longitude' in stats:
- coords[0] = str(stats.geodetic_longitude)
+ coords[0] = float(stats.geodetic_longitude)
if 'geodetic_latitude' in stats:
- coords[1] = str(stats.geodetic_latitude)
+ coords[1] = float(stats.geodetic_latitude)
if 'elevation' in stats:
- coords[2] = str(stats.elevation)
+ coords[2] = float(stats.elevation)
imo['coordinates'] = coords
intermag['imo'] = imo
intermag['reported_orientation'] = ''.join(channels)
@@ -139,10 +132,10 @@ class IMFJSONWriter(object):
rate = 86400
else:
rate = 1
- intermag['sampling_period'] = str(rate)
+ intermag['sampling_period'] = rate
if 'sensor_sampling_rate' in stats:
- sampling = 1 / stats.sensor_sampling_rate
- intermag['digital_sampling_rate'] = str(sampling)
+ sampling = stats.sensor_sampling_rate
+ intermag['digital_sampling_rate'] = sampling
metadata_dict['intermagnet'] = intermag
metadata_dict['status'] = 200
generated = datetime.utcnow()
@@ -192,7 +185,7 @@ class IMFJSONWriter(object):
''.format(tt, int(time.microsecond / 1000))
@classmethod
- def format(self, timeseries, channels, **kwargs):
+ def format(self, timeseries, channels, url=None):
"""Get a json formatted string.
Calls write() with a BytesIO, and returns the output.
@@ -202,17 +195,16 @@ class IMFJSONWriter(object):
timeseries : obspy.core.Stream
stream containing traces with channel listed in channels
channels: array_like
- channels to be written from timeseries object
- kwargs
- request : query string
+ channels to be written from timeseries
+ url: str
+ string with the requested url
Returns
-------
unicode
json formatted string.
"""
- request = kwargs.get('request')
out = BytesIO()
writer = IMFJSONWriter()
- writer.write(out, timeseries, channels, request=request)
+ writer.write(out, timeseries, channels, url=url)
return out.getvalue()
diff --git a/test/imfjson_test/IMFJSONWriter_test.py b/test/imfjson_test/IMFJSONWriter_test.py
new file mode 100644
index 000000000..387c442e3
--- /dev/null
+++ b/test/imfjson_test/IMFJSONWriter_test.py
@@ -0,0 +1,126 @@
+"""Tests for the IMFJSON Writer class."""
+
+from nose.tools import assert_equals
+from geomagio import edge
+from geomagio.imfjson import IMFJSONWriter
+
+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_STATS = EXAMPLE_DATA[0].stats
+
+
+def test_output_type():
+ """imfjson.IMFJSONWriter_test.test_output_type()
+
+ Call the format method with the test
+ stats and channels.
+ Verify that the output type is correct.
+ """
+ writer = IMFJSONWriter()
+ output = writer.format(EXAMPLE_DATA, EXAMPLE_CHANNELS)
+ assert_equals(type(output), str)
+
+
+def test_metadata():
+ """imfjson.IMFJSONWriter_test.test_metadata()
+
+ Call the _format_metadata method with the test
+ stats and channels.
+ Verify, the set metadata.
+ """
+ writer = IMFJSONWriter()
+ metadata = writer._format_metadata(EXAMPLE_STATS, EXAMPLE_CHANNELS)
+ assert_equals(metadata['status'], 200)
+ # Test intermagnet parameters
+ intermag = metadata['intermagnet']
+ assert_equals(intermag['reported_orientation'], "HEZ")
+ assert_equals(intermag['sensor_orientation'], "HDZF")
+ 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])
+
+
+def test_times():
+ """imfjson.IMFJSONWriter_test.test_times()
+
+ Call the _format_times method with the test
+ data and channels.
+ Verify, the times are the correct value and format.
+ """
+ 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"
+ ])
+
+
+def test_values():
+ """imfjson.IMFJSONWriter_test.test_values()
+
+ Call the _format_data method with the test
+ data, channels, and stats.
+ Verify, the values and associated metadata
+ are the correct value and format.
+ """
+ writer = IMFJSONWriter()
+ values = writer._format_data(EXAMPLE_DATA, EXAMPLE_CHANNELS,
+ EXAMPLE_STATS)
+ assert_equals(values[0].keys(), ["id", "metadata", "values"])
+ assert_equals(values[0]['id'], "H")
+ assert_equals(values[1].keys(), ["id", "metadata", "values"])
+ assert_equals(values[1]['id'], "E")
+ assert_equals(values[2].keys(), ["id", "metadata", "values"])
+ assert_equals(values[2]['id'], "Z")
+ # Test values-metadata (need to add flags)
+ metadata = values[0]['metadata']
+ assert_equals(metadata.keys(), [
+ "element",
+ "network",
+ "station",
+ "channel",
+ "location"
+ ])
+ assert_equals(metadata['element'], "H")
+ assert_equals(metadata['network'], "NT")
+ assert_equals(metadata['station'], "BOU")
+ # channels do not match H v MVH
+ # 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
+ ])
diff --git a/test/imfjson_test/__init__.py b/test/imfjson_test/__init__.py
new file mode 100644
index 000000000..e69de29bb
--
GitLab