implement SNCLFactory in edge/miniseed factories

geomagio/edge/EdgeFactory.py

@@ -22,6 +22,7 @@ from ..TimeseriesFactory import TimeseriesFactory
 from ..TimeseriesFactoryException import TimeseriesFactoryException
 from ..ObservatoryMetadata import ObservatoryMetadata
 from .RawInputClient import RawInputClient
+from .SNCLFactory import SNCLFactory
 
 
 class EdgeFactory(TimeseriesFactory):
@@ -482,13 +483,17 @@ class EdgeFactory(TimeseriesFactory):
         obspy.core.trace
             timeseries trace of the requested channel data
         """
-        station = self._get_edge_station(observatory, channel, type, interval)
-        location = self._get_edge_location(observatory, channel, type, interval)
-        network = self._get_edge_network(observatory, channel, type, interval)
-        edge_channel = self._get_edge_channel(observatory, channel, type, interval)
+        sncl = SNCLFactory(data_format="legacy").get_sncl(
+            station=observatory, data_type=type, element=channel, interval=interval
+        )
         try:
             data = self.client.get_waveforms(
-                network, station, location, edge_channel, starttime, endtime
+                sncl.network,
+                sncl.station,
+                sncl.location,
+                sncl.channel,
+                starttime,
+                endtime,
             )
         except TypeError:
             # get_waveforms() fails if no data is returned from Edge
@@ -506,9 +511,9 @@ class EdgeFactory(TimeseriesFactory):
                 channel,
                 type,
                 interval,
-                network,
-                station,
-                location,
+                sncl.network,
+                sncl.station,
+                sncl.location,
             )
         self._set_metadata(data, observatory, channel, type, interval)
         return data
@@ -576,10 +581,9 @@ class EdgeFactory(TimeseriesFactory):
         -----
         RawInputClient seems to only work when sockets are
         """
-        station = self._get_edge_station(observatory, channel, type, interval)
-        location = self._get_edge_location(observatory, channel, type, interval)
-        network = self._get_edge_network(observatory, channel, type, interval)
-        edge_channel = self._get_edge_channel(observatory, channel, type, interval)
+        sncl = SNCLFactory(data_format="legacy").get_sncl(
+            station=observatory, data_type=type, element=channel, interval=interval
+        )
 
         now = obspy.core.UTCDateTime(datetime.utcnow())
         if ((now - endtime) > 864000) and (self.cwbport > 0):
@@ -590,7 +594,13 @@ class EdgeFactory(TimeseriesFactory):
             port = self.write_port
 
         ric = RawInputClient(
-            self.tag, host, port, station, edge_channel, location, network
+            self.tag,
+            host,
+            port,
+            sncl.station,
+            sncl.channel,
+            sncl.location,
+            sncl.network,
         )
 
         stream = self._convert_stream_to_masked(timeseries=timeseries, channel=channel)

geomagio/edge/MiniSeedFactory.py

@@ -23,6 +23,7 @@ from ..TimeseriesFactory import TimeseriesFactory
 from ..TimeseriesFactoryException import TimeseriesFactoryException
 from ..ObservatoryMetadata import ObservatoryMetadata
 from .MiniSeedInputClient import MiniSeedInputClient
+from .SNCLFactory import SNCLFactory
 
 
 class MiniSeedFactory(TimeseriesFactory):
@@ -524,12 +525,11 @@ class MiniSeedFactory(TimeseriesFactory):
         obspy.core.trace
             timeseries trace of the requested channel data
         """
-        station = self._get_edge_station(observatory, channel, type, interval)
-        location = self._get_edge_location(observatory, channel, type, interval)
-        network = self._get_edge_network(observatory, channel, type, interval)
-        edge_channel = self._get_edge_channel(observatory, channel, type, interval)
+        sncl = SNCLFactory().get_sncl(
+            station=observatory, data_type=type, element=channel, interval=interval
+        )
         data = self.client.get_waveforms(
-            network, station, location, edge_channel, starttime, endtime
+            sncl.network, sncl.station, sncl.location, sncl.channel, starttime, endtime
         )
         data.merge()
         if data.count() == 0:
@@ -540,9 +540,9 @@ class MiniSeedFactory(TimeseriesFactory):
                 channel,
                 type,
                 interval,
-                network,
-                station,
-                location,
+                sncl.network,
+                sncl.station,
+                sncl.location,
             )
         self._set_metadata(data, observatory, channel, type, interval)
         return data
@@ -666,15 +666,14 @@ class MiniSeedFactory(TimeseriesFactory):
         to_write = to_write.split()
         to_write = TimeseriesUtility.unmask_stream(to_write)
         # relabel channels from internal to edge conventions
-        station = self._get_edge_station(observatory, channel, type, interval)
-        location = self._get_edge_location(observatory, channel, type, interval)
-        network = self._get_edge_network(observatory, channel, type, interval)
-        edge_channel = self._get_edge_channel(observatory, channel, type, interval)
+        sncl = SNCLFactory().get_sncl(
+            station=observatory, data_type=type, element=channel, interval=interval
+        )
         for trace in to_write:
-            trace.stats.station = station
-            trace.stats.location = location
-            trace.stats.network = network
-            trace.stats.channel = edge_channel
+            trace.stats.station = sncl.station
+            trace.stats.location = sncl.location
+            trace.stats.network = sncl.network
+            trace.stats.channel = sncl.channel
         # finally, send to edge
         self.write_client.send(to_write)
 

test/edge_test/EdgeFactory_test.py

@@ -5,62 +5,6 @@ from geomagio.edge import EdgeFactory
 from numpy.testing import assert_equal
 
 
-def test__get_edge_network():
-    """edge_test.EdgeFactory_test.test__get_edge_network()"""
-    # _get_edge_network should always return NT for use by USGS geomag
-    assert_equal(EdgeFactory()._get_edge_network(" ", " ", " ", " "), "NT")
-
-
-def test__get_edge_station():
-    """edge_test.EdgeFactory_test.test__get_edge_station()"""
-    # _get_edge_station will return the observatory code passed in.
-    assert_equal(EdgeFactory()._get_edge_station("BOU", " ", " ", " "), "BOU")
-
-
-def test__get_edge_channel():
-    """edge_test.EdgeFactory_test.test__get_edge_channel()"""
-    # Call private function _get_edge_channel, make certain
-    # it gets back the appropriate 2 character code.
-    assert_equal(EdgeFactory()._get_edge_channel("", "D", "", "minute"), "MVD")
-    assert_equal(EdgeFactory()._get_edge_channel("", "E", "", "minute"), "MVE")
-    assert_equal(EdgeFactory()._get_edge_channel("", "F", "", "minute"), "MSF")
-    assert_equal(EdgeFactory()._get_edge_channel("", "H", "", "minute"), "MVH")
-    assert_equal(EdgeFactory()._get_edge_channel("", "DIST", "", "minute"), "MDT")
-    assert_equal(EdgeFactory()._get_edge_channel("", "DST", "", "minute"), "MGD")
-    assert_equal(EdgeFactory()._get_edge_channel("", "E-E", "", "minute"), "MQE")
-    assert_equal(EdgeFactory()._get_edge_channel("", "E-N", "", "minute"), "MQN")
-
-
-def test__get_edge_location():
-    """edge_test.EdgeFactory_test.test__get_edge_location()"""
-    # Call _get_edge_location, make certain it returns the correct edge
-    # location code.
-    assert_equal(EdgeFactory()._get_edge_location("", "", "variation", ""), "R0")
-    assert_equal(EdgeFactory()._get_edge_location("", "", "quasi-definitive", ""), "Q0")
-    assert_equal(EdgeFactory()._get_edge_location("", "", "definitive", ""), "D0")
-
-
-def test__get_interval_code():
-    """edge_test.EdgeFactory_test.test__get_interval_code()"""
-    assert_equal(EdgeFactory()._get_interval_code("day"), "D")
-    assert_equal(EdgeFactory()._get_interval_code("hour"), "H")
-    assert_equal(EdgeFactory()._get_interval_code("minute"), "M")
-    assert_equal(EdgeFactory()._get_interval_code("second"), "S")
-
-
-def test__set_metadata():
-    """edge_test.EdgeFactory_test.test__set_metadata()"""
-    # Call _set_metadata with 2 traces,  and make certain the stats get
-    # set for both traces.
-    trace1 = Trace()
-    trace2 = Trace()
-    stream = Stream(traces=[trace1, trace2])
-    EdgeFactory()._set_metadata(stream, "BOU", "H", "variation", "minute")
-    assert_equal(stream[0].stats["channel"], "H")
-    assert_equal(stream[1].stats["channel"], "H")
-
-
-# def test_get_timeseries():
 def dont_get_timeseries():
     """edge_test.EdgeFactory_test.test_get_timeseries()"""
     # Call get_timeseries, and test stats for comfirmation that it came back.

test/edge_test/MiniSeedFactory_test.py

@@ -9,54 +9,6 @@ from geomagio import TimeseriesUtility
 from geomagio.edge import MiniSeedFactory, MiniSeedInputClient
 
 
-def test__get_edge_network():
-    """edge_test.MiniSeedFactory_test.test__get_edge_network()"""
-    # _get_edge_network should always return NT for use by USGS geomag
-    assert_equal(MiniSeedFactory()._get_edge_network(" ", " ", " ", " "), "NT")
-
-
-def test__get_edge_station():
-    """edge_test.MiniSeedFactory_test.test__get_edge_station()"""
-    # _get_edge_station will return the observatory code passed in.
-    assert_equal(MiniSeedFactory()._get_edge_station("BOU", " ", " ", " "), "BOU")
-
-
-def test__get_edge_channel():
-    """edge_test.MiniSeedFactory_test.test__get_edge_channel()"""
-    # Call private function _get_edge_channel, make certain
-    # it gets back the appropriate 2 character code.
-    factory = MiniSeedFactory()
-    assert_equal(factory._get_edge_channel("", "D", "", "minute"), "UFD")
-    assert_equal(factory._get_edge_channel("", "U", "", "minute"), "UFU")
-    assert_equal(factory._get_edge_channel("", "F", "", "minute"), "UFF")
-    assert_equal(factory._get_edge_channel("", "H", "", "minute"), "UFH")
-    assert_equal(factory._get_edge_channel("", "BEU", "", "minute"), "BEU")
-    assert_equal(factory._get_edge_channel("", "Dst4", "", "minute"), "UX4")
-    assert_equal(factory._get_edge_channel("", "Dst3", "", "minute"), "UX3")
-    assert_equal(factory._get_edge_channel("", "E-E", "", "minute"), "UQE")
-    assert_equal(factory._get_edge_channel("", "E-N", "", "minute"), "UQN")
-
-
-def test__get_edge_location():
-    """edge_test.MiniSeedFactory_test.test__get_edge_location()"""
-    # Call _get_edge_location, make certain it returns the correct edge
-    # location code.
-    assert_equal(MiniSeedFactory()._get_edge_location("", "", "variation", ""), "R0")
-    assert_equal(
-        MiniSeedFactory()._get_edge_location("", "", "quasi-definitive", ""), "Q0"
-    )
-    assert_equal(MiniSeedFactory()._get_edge_location("", "", "definitive", ""), "D0")
-
-
-def test__get_interval_code():
-    """edge_test.MiniSeedFactory_test.test__get_interval_code()"""
-    assert_equal(MiniSeedFactory()._get_interval_code("day"), "P")
-    assert_equal(MiniSeedFactory()._get_interval_code("hour"), "R")
-    assert_equal(MiniSeedFactory()._get_interval_code("minute"), "U")
-    assert_equal(MiniSeedFactory()._get_interval_code("second"), "L")
-    assert_equal(MiniSeedFactory()._get_interval_code("tenhertz"), "B")
-
-
 class MockMiniSeedInputClient(object):
     def __init__(self):
         self.close_called = False
@@ -167,7 +119,7 @@ def __create_trace(
     channel="H",
     location="R0",
     data_interval="second",
-    data_type="interval",
+    data_type="variation",
 ):
     """
     Utility to create a trace containing the given numpy array.