diff --git a/geomagio/edge/LegacySNCL.py b/geomagio/edge/LegacySNCL.py
index 1b8cd2169c2ac5b4e45c207bd116e343f2032f44..f80b438a602ce15ec0dc552ff10c7345701c0222 100644
--- a/geomagio/edge/LegacySNCL.py
+++ b/geomagio/edge/LegacySNCL.py
@@ -62,6 +62,8 @@ def get_channel(element: str, interval: str) -> str:
def get_location(element: str, data_type: str) -> str:
+ if len(data_type) == 2:
+ return data_type
return _get_location_start(data_type=data_type) + _get_location_end(element=element)
diff --git a/geomagio/edge/SNCL.py b/geomagio/edge/SNCL.py
index 8dc8a8e252891cf8df60f436fed2fa1f4fc51810..e137b2b163e5f5b44688774b2dd9b62a2c000468 100644
--- a/geomagio/edge/SNCL.py
+++ b/geomagio/edge/SNCL.py
@@ -95,6 +95,8 @@ def get_channel(element: str, interval: str, data_type: str) -> str:
def get_location(element: str, data_type: str) -> str:
+ if len(data_type) == 2:
+ return data_type
return _get_location_start(data_type=data_type) + _get_location_end(element=element)
diff --git a/geomagio/iaga2002/IAGA2002Writer.py b/geomagio/iaga2002/IAGA2002Writer.py
index 2e9bbdfdd35cfc15c92273d25265b9bb4151c660..bd215ef98deae916d2cbdf83610b25b4d5d1adc2 100644
--- a/geomagio/iaga2002/IAGA2002Writer.py
+++ b/geomagio/iaga2002/IAGA2002Writer.py
@@ -117,7 +117,11 @@ class IAGA2002Writer(object):
if (
"declination_base" in stats
and stats.declination_base is not None
- and (stats.data_type == "variation" or stats.data_type == "reported")
+ and (
+ stats.data_type == "variation"
+ or stats.data_type == "reported"
+ or stats.data_type[0] == "R"
+ )
):
comments.append(
"DECBAS {:<8d}"
diff --git a/test/edge_test/EdgeFactory_test.py b/test/edge_test/EdgeFactory_test.py
index 55e94f1266fef85c9f6b0cdafa15b781420d656e..52a4f662f5fbb21aa483ffe34b8819e2d55cb6ce 100644
--- a/test/edge_test/EdgeFactory_test.py
+++ b/test/edge_test/EdgeFactory_test.py
@@ -5,7 +5,7 @@ from geomagio.edge import EdgeFactory
from numpy.testing import assert_equal
-def dont_get_timeseries():
+def test_get_timeseries():
"""edge_test.EdgeFactory_test.test_get_timeseries()"""
# Call get_timeseries, and test stats for comfirmation that it came back.
# TODO, need to pass in host and port from a config file, or manually
@@ -29,6 +29,68 @@ def dont_get_timeseries():
"H",
"Expect timeseries stats channel to be equal to H",
)
+ assert_equal(
+ timeseries.select(channel="H")[0].stats.data_type,
+ "variation",
+ "Expect timeseries stats data_type to be equal to variation",
+ )
+
+
+def test_get_timeseries_by_location():
+ """test.edge_test.EdgeFactory_test.test_get_timeseries_by_location()"""
+ edge_factory = EdgeFactory(host="TODO", port="TODO")
+ timeseries = edge_factory.get_timeseries(
+ UTCDateTime(2015, 3, 1, 0, 0, 0),
+ UTCDateTime(2015, 3, 1, 1, 0, 0),
+ "BOU",
+ ("H"),
+ "R0",
+ "minute",
+ )
+ assert_equal(
+ timeseries.select(channel="H")[0].stats.data_type,
+ "R0",
+ "Expect timeseries stats data_type to be equal to R0",
+ )
+ timeseries = edge_factory.get_timeseries(
+ UTCDateTime(2015, 3, 1, 0, 0, 0),
+ UTCDateTime(2015, 3, 1, 1, 0, 0),
+ "BOU",
+ ("H"),
+ "A0",
+ "minute",
+ )
+ assert_equal(
+ timeseries.select(channel="H")[0].stats.data_type,
+ "A0",
+ "Expect timeseries stats data_type to be equal to A0",
+ )
+ timeseries = edge_factory.get_timeseries(
+ UTCDateTime(2015, 3, 1, 0, 0, 0),
+ UTCDateTime(2015, 3, 1, 1, 0, 0),
+ "BOU",
+ ("X"),
+ "Q0",
+ "minute",
+ )
+ assert_equal(
+ timeseries.select(channel="X")[0].stats.data_type,
+ "Q0",
+ "Expect timeseries stats data_type to be equal to Q0",
+ )
+ timeseries = edge_factory.get_timeseries(
+ UTCDateTime(2015, 3, 1, 0, 0, 0),
+ UTCDateTime(2015, 3, 1, 1, 0, 0),
+ "BOU",
+ ("X"),
+ "D0",
+ "minute",
+ )
+ assert_equal(
+ timeseries.select(channel="X")[0].stats.data_type,
+ "D0",
+ "Expect timeseries stats data_type to be equal to D0",
+ )
def test_add_empty_channels():
diff --git a/test/edge_test/LegacySNCL_test.py b/test/edge_test/LegacySNCL_test.py
index adbcc8a8388ba85a543cd075d2e3bccc2c0e5202..190dc09a9a20376f6a3cd2aeb49c754d37bf41b1 100644
--- a/test/edge_test/LegacySNCL_test.py
+++ b/test/edge_test/LegacySNCL_test.py
@@ -133,6 +133,13 @@ def test_get_location():
assert get_location(element="D", data_type="quasi-definitive") == "Q0"
assert get_location(element="D", data_type="definitive") == "D0"
assert get_location(element="D_Sat", data_type="variation") == "R1"
+ assert get_location(element="D_Sat", data_type="adjusted") == "A1"
+ assert get_location(element="D", data_type="R0") == "R0"
+ assert get_location(element="D", data_type="A0") == "A0"
+ assert get_location(element="D", data_type="Q0") == "Q0"
+ assert get_location(element="D", data_type="D0") == "D0"
+ assert get_location(element="D", data_type="R1") == "R1"
+ assert get_location(element="D", data_type="A1") == "A1"
def test_get_sncl():
@@ -140,6 +147,9 @@ def test_get_sncl():
assert LegacySNCL.get_sncl(
station="BOU", data_type="variation", interval="second", element="H"
) == LegacySNCL(station="BOU", network="NT", channel="SVH", location="R0")
+ assert LegacySNCL.get_sncl(
+ station="BOU", data_type="R0", interval="second", element="H"
+ ) == LegacySNCL(station="BOU", network="NT", channel="SVH", location="R0")
def test_interval():
diff --git a/test/edge_test/MiniSeedFactory_test.py b/test/edge_test/MiniSeedFactory_test.py
index 5689d69c5246c972913963eeac11f9fae9ca7ca2..516850f79ab14bfc8dc98290512f6779b6b5ac00 100644
--- a/test/edge_test/MiniSeedFactory_test.py
+++ b/test/edge_test/MiniSeedFactory_test.py
@@ -4,9 +4,11 @@ import io
import numpy
from numpy.testing import assert_equal
from obspy.core import read, Stats, Stream, Trace, UTCDateTime
+import pytest
from geomagio import TimeseriesUtility
from geomagio.edge import MiniSeedFactory, MiniSeedInputClient
+from .MockMiniSeedClient import MockMiniSeedClient
class MockMiniSeedInputClient(object):
@@ -21,6 +23,14 @@ class MockMiniSeedInputClient(object):
self.last_sent = stream
+@pytest.fixture(scope="class")
+def miniseed_factory() -> MiniSeedFactory:
+ """instance of MiniSeedFactory with MockMiniseedClient"""
+ factory = MiniSeedFactory()
+ factory.client = MockMiniSeedClient()
+ yield factory
+
+
def test__put_timeseries():
"""edge_test.MiniSeedFactory_test.test__put_timeseries()"""
trace1 = __create_trace([0, 1, 2, 3, numpy.nan, 5, 6, 7, 8, 9], channel="H")
@@ -85,14 +95,12 @@ def test__set_metadata():
assert_equal(stream[1].stats["channel"], "H")
-# def test_get_timeseries():
-def dont_get_timeseries():
+def test_get_timeseries(miniseed_factory):
"""edge_test.MiniSeedFactory_test.test_get_timeseries()"""
# Call get_timeseries, and test stats for comfirmation that it came back.
# TODO, need to pass in host and port from a config file, or manually
# change for a single test.
- edge_factory = MiniSeedFactory(host="TODO", port="TODO")
- timeseries = edge_factory.get_timeseries(
+ timeseries = miniseed_factory.get_timeseries(
UTCDateTime(2015, 3, 1, 0, 0, 0),
UTCDateTime(2015, 3, 1, 1, 0, 0),
"BOU",
@@ -110,6 +118,67 @@ def dont_get_timeseries():
"H",
"Expect timeseries stats channel to be equal to H",
)
+ assert_equal(
+ timeseries.select(channel="H")[0].stats.data_type,
+ "variation",
+ "Expect timeseries stats data_type to be equal to variation",
+ )
+
+
+def test_get_timeseries_by_location(miniseed_factory):
+ """test.edge_test.MiniSeedFactory_test.test_get_timeseries_by_location()"""
+ timeseries = miniseed_factory.get_timeseries(
+ UTCDateTime(2015, 3, 1, 0, 0, 0),
+ UTCDateTime(2015, 3, 1, 1, 0, 0),
+ "BOU",
+ ("H"),
+ "R0",
+ "minute",
+ )
+ assert_equal(
+ timeseries.select(channel="H")[0].stats.data_type,
+ "R0",
+ "Expect timeseries stats data_type to be equal to R0",
+ )
+ timeseries = miniseed_factory.get_timeseries(
+ UTCDateTime(2015, 3, 1, 0, 0, 0),
+ UTCDateTime(2015, 3, 1, 1, 0, 0),
+ "BOU",
+ ("H"),
+ "A0",
+ "minute",
+ )
+ assert_equal(
+ timeseries.select(channel="H")[0].stats.data_type,
+ "A0",
+ "Expect timeseries stats data_type to be equal to A0",
+ )
+ timeseries = miniseed_factory.get_timeseries(
+ UTCDateTime(2015, 3, 1, 0, 0, 0),
+ UTCDateTime(2015, 3, 1, 1, 0, 0),
+ "BOU",
+ ("X"),
+ "Q0",
+ "minute",
+ )
+ assert_equal(
+ timeseries.select(channel="X")[0].stats.data_type,
+ "Q0",
+ "Expect timeseries stats data_type to be equal to Q0",
+ )
+ timeseries = miniseed_factory.get_timeseries(
+ UTCDateTime(2015, 3, 1, 0, 0, 0),
+ UTCDateTime(2015, 3, 1, 1, 0, 0),
+ "BOU",
+ ("X"),
+ "D0",
+ "minute",
+ )
+ assert_equal(
+ timeseries.select(channel="X")[0].stats.data_type,
+ "D0",
+ "Expect timeseries stats data_type to be equal to D0",
+ )
def __create_trace(
diff --git a/test/edge_test/MockMiniSeedClient.py b/test/edge_test/MockMiniSeedClient.py
new file mode 100644
index 0000000000000000000000000000000000000000..88c6a4b612b5382e4eafb070501f83779c1a4dc1
--- /dev/null
+++ b/test/edge_test/MockMiniSeedClient.py
@@ -0,0 +1,39 @@
+import numpy
+from obspy import Stream, UTCDateTime
+from obspy.clients.neic.client import Client
+
+from geomagio import TimeseriesUtility
+from geomagio.edge import SNCL
+
+
+class MockMiniSeedClient(Client):
+ """replaces default obspy miniseed client's get_waveforms method to return trace of ones"""
+
+ def get_waveforms(
+ self,
+ network: str,
+ station: str,
+ location: str,
+ channel: str,
+ starttime: UTCDateTime,
+ endtime: UTCDateTime,
+ ):
+ sncl = SNCL(
+ station=station,
+ network=network,
+ channel=channel,
+ location=location,
+ )
+ trace = TimeseriesUtility.create_empty_trace(
+ starttime=starttime,
+ endtime=endtime,
+ observatory=station,
+ channel=channel,
+ type=sncl.data_type,
+ interval=sncl.interval,
+ network=network,
+ station=station,
+ location=location,
+ )
+ trace.data = numpy.ones(trace.stats.npts)
+ return Stream([trace])
diff --git a/test/edge_test/SNCL_test.py b/test/edge_test/SNCL_test.py
index 51d5b73c3709777e9dd8ad37a065d46895fce94f..0ce923683a3dae72e4fdbd9d9df1e6dafe1659a0 100644
--- a/test/edge_test/SNCL_test.py
+++ b/test/edge_test/SNCL_test.py
@@ -115,16 +115,22 @@ def test_get_channel():
== "BYU"
)
assert get_channel(element="D", interval="second", data_type="variation") == "LFD"
+ assert get_channel(element="D", interval="second", data_type="R0") == "LFD"
assert get_channel(element="F", interval="minute", data_type="variation") == "UFF"
assert get_channel(element="U", interval="hour", data_type="variation") == "RFU"
assert get_channel(element="V", interval="hour", data_type="variation") == "RFV"
assert get_channel(element="W", interval="hour", data_type="variation") == "RFW"
assert get_channel(element="H", interval="hour", data_type="variation") == "RFU"
+ assert get_channel(element="H", interval="hour", data_type="R0") == "RFH"
assert get_channel(element="E", interval="hour", data_type="variation") == "RFV"
+ assert get_channel(element="E", interval="hour", data_type="R0") == "RFE"
assert get_channel(element="Z", interval="hour", data_type="variation") == "RFW"
+ assert get_channel(element="Z", interval="hour", data_type="R0") == "RFZ"
# not variation data_type, test that H,Z is not converted to U,V
assert get_channel(element="H", interval="hour", data_type="adjusted") == "RFH"
+ assert get_channel(element="H", interval="hour", data_type="A0") == "RFH"
assert get_channel(element="Z", interval="hour", data_type="adjusted") == "RFZ"
+ assert get_channel(element="Z", interval="hour", data_type="A0") == "RFZ"
assert get_channel(element="Dst4", interval="day", data_type="variation") == "PX4"
assert (
get_channel(element="Dst3", interval="minute", data_type="variation") == "UX3"
@@ -135,12 +141,23 @@ def test_get_channel():
assert (
get_channel(element="U_Dist", interval="minute", data_type="variation") == "UFU"
)
+ assert get_channel(element="U", interval="minute", data_type="RD") == "UFU"
assert (
get_channel(element="U_SQ", interval="minute", data_type="variation") == "UFU"
)
+ assert get_channel(element="U", interval="minute", data_type="RQ") == "UFU"
assert (
get_channel(element="U_SV", interval="minute", data_type="variation") == "UFU"
)
+ assert get_channel(element="U", interval="minute", data_type="RV") == "UFU"
+ assert (
+ get_channel(element="U_Dist", interval="minute", data_type="adjusted") == "UFU"
+ )
+ assert get_channel(element="U", interval="minute", data_type="AD") == "UFU"
+ assert get_channel(element="U_SQ", interval="minute", data_type="adjusted") == "UFU"
+ assert get_channel(element="U", interval="minute", data_type="AQ") == "UFU"
+ assert get_channel(element="U_SV", interval="minute", data_type="adjusted") == "UFU"
+ assert get_channel(element="U", interval="minute", data_type="AV") == "UFU"
assert (
get_channel(element="UK1.R0", interval="minute", data_type="variation") == "UK1"
)
@@ -149,13 +166,21 @@ def test_get_channel():
def test_get_location():
"""edge_test.SNCL_test.test_get_location()"""
assert get_location(element="D", data_type="variation") == "R0"
+ assert get_location(element="D", data_type="R0") == "R0"
assert get_location(element="D", data_type="adjusted") == "A0"
+ assert get_location(element="D", data_type="A0") == "A0"
assert get_location(element="D", data_type="quasi-definitive") == "Q0"
+ assert get_location(element="D", data_type="Q0") == "Q0"
assert get_location(element="D", data_type="definitive") == "D0"
+ assert get_location(element="D", data_type="D0") == "D0"
assert get_location(element="D_Sat", data_type="variation") == "R1"
+ assert get_location(element="D", data_type="R1") == "R1"
assert get_location(element="D_Dist", data_type="variation") == "RD"
+ assert get_location(element="D", data_type="RD") == "RD"
assert get_location(element="D_SQ", data_type="variation") == "RQ"
+ assert get_location(element="D", data_type="RQ") == "RQ"
assert get_location(element="D_SV", data_type="variation") == "RV"
+ assert get_location(element="D", data_type="RV") == "RV"
def test_get_sncl():
@@ -163,6 +188,15 @@ def test_get_sncl():
assert SNCL.get_sncl(
station="BOU", data_type="variation", interval="second", element="U"
) == SNCL(station="BOU", network="NT", channel="LFU", location="R0")
+ assert SNCL.get_sncl(
+ station="BOU", data_type="variation", interval="second", element="H"
+ ) == SNCL(station="BOU", network="NT", channel="LFU", location="R0")
+ assert SNCL.get_sncl(
+ station="BOU", data_type="R0", interval="second", element="H"
+ ) == SNCL(station="BOU", network="NT", channel="LFH", location="R0")
+ assert SNCL.get_sncl(
+ station="BOU", data_type="A0", interval="second", element="H"
+ ) == SNCL(station="BOU", network="NT", channel="LFH", location="A0")
def test_interval():