From 89022f8bf6e4b887de909aefeea6a0d9117815be Mon Sep 17 00:00:00 2001
From: pcain-usgs <pcain@usgs.gov>
Date: Thu, 13 May 2021 17:56:27 -0600
Subject: [PATCH] implement SNCLFactory in edge/miniseed factories
---
geomagio/edge/EdgeFactory.py | 36 +++++++++++------
geomagio/edge/MiniSeedFactory.py | 31 +++++++-------
test/edge_test/EdgeFactory_test.py | 56 --------------------------
test/edge_test/MiniSeedFactory_test.py | 50 +----------------------
4 files changed, 39 insertions(+), 134 deletions(-)
diff --git a/geomagio/edge/EdgeFactory.py b/geomagio/edge/EdgeFactory.py
index e0f1091c5..bed9e5ec8 100644
--- a/geomagio/edge/EdgeFactory.py
+++ b/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)
diff --git a/geomagio/edge/MiniSeedFactory.py b/geomagio/edge/MiniSeedFactory.py
index 99c3b9eec..211ccf349 100644
--- a/geomagio/edge/MiniSeedFactory.py
+++ b/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)
diff --git a/test/edge_test/EdgeFactory_test.py b/test/edge_test/EdgeFactory_test.py
index fca38a8c8..6b524fb31 100644
--- a/test/edge_test/EdgeFactory_test.py
+++ b/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.
diff --git a/test/edge_test/MiniSeedFactory_test.py b/test/edge_test/MiniSeedFactory_test.py
index a0c57869e..bd1367d37 100644
--- a/test/edge_test/MiniSeedFactory_test.py
+++ b/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.
--
GitLab