From 486864887e569bd95e0d3d23b0852512f057966d Mon Sep 17 00:00:00 2001
From: pcain-usgs <pcain@usgs.gov>
Date: Tue, 18 May 2021 12:00:06 -0600
Subject: [PATCH] remove SNCLFactory, get channel/location from models
---
geomagio/edge/EdgeFactory.py | 9 ++-
geomagio/edge/LegacySNCL.py | 81 ++++++++++++++++---
geomagio/edge/MiniSeedFactory.py | 4 +-
geomagio/edge/SNCL.py | 104 ++++++++++++++++++++----
geomagio/edge/SNCLFactory.py | 123 -----------------------------
geomagio/edge/__init__.py | 4 -
test/edge_test/LegacySNCL_test.py | 28 ++++++-
test/edge_test/SNCLFactory_test.py | 64 ---------------
test/edge_test/SNCL_test.py | 34 +++++++-
9 files changed, 225 insertions(+), 226 deletions(-)
delete mode 100644 geomagio/edge/SNCLFactory.py
delete mode 100644 test/edge_test/SNCLFactory_test.py
diff --git a/geomagio/edge/EdgeFactory.py b/geomagio/edge/EdgeFactory.py
index af226a63a..1190dbb16 100644
--- a/geomagio/edge/EdgeFactory.py
+++ b/geomagio/edge/EdgeFactory.py
@@ -301,7 +301,7 @@ class EdgeFactory(TimeseriesFactory):
obspy.core.trace
timeseries trace of the requested channel data
"""
- sncl = LegacySNCL().get_sncl(
+ sncl = LegacySNCL.get_sncl(
station=observatory,
data_type=type,
interval=interval,
@@ -402,8 +402,11 @@ class EdgeFactory(TimeseriesFactory):
-----
RawInputClient seems to only work when sockets are
"""
- sncl = LegacySNCL().get_sncl(
- station=observatory, data_type=type, interval=interval, element=channel
+ sncl = LegacySNCL.get_sncl(
+ station=observatory,
+ data_type=type,
+ interval=interval,
+ element=channel,
)
now = obspy.core.UTCDateTime(datetime.utcnow())
diff --git a/geomagio/edge/LegacySNCL.py b/geomagio/edge/LegacySNCL.py
index be1c3300d..4c747df3a 100644
--- a/geomagio/edge/LegacySNCL.py
+++ b/geomagio/edge/LegacySNCL.py
@@ -1,7 +1,6 @@
-from __future__ import annotations
from typing import Optional
-from .SNCL import SNCL
+from .SNCL import SNCL, __get_location_start
ELEMENT_CONVERSIONS = {
# e-field
@@ -19,21 +18,20 @@ CHANNEL_CONVERSIONS = {
class LegacySNCL(SNCL):
+ @classmethod
def get_sncl(
- self,
- station: str,
+ cls,
data_type: str,
- interval: str,
element: str,
- ) -> LegacySNCL:
- from .SNCLFactory import SNCLFactory
-
- factory = SNCLFactory(data_format="legacy")
+ interval: str,
+ station: str,
+ network: str = "NT",
+ ) -> "LegacySNCL":
return LegacySNCL(
station=station,
- network=self.network,
- channel=factory.get_channel(element=element, interval=interval),
- location=factory.get_location(element=element, data_type=data_type),
+ network=network,
+ channel=get_channel(element=element, interval=interval),
+ location=get_location(element=element, data_type=data_type),
)
@property
@@ -79,3 +77,62 @@ class LegacySNCL(SNCL):
if channel_end in CHANNEL_CONVERSIONS:
return CHANNEL_CONVERSIONS[channel_end]
return None
+
+
+def get_channel(element: str, interval: str) -> str:
+ predefined_channel = __check_predefined_channel(element=element, interval=interval)
+ channel_start = __get_channel_start(interval=interval)
+ channel_end = __get_channel_end(element=element)
+ return predefined_channel or (channel_start + channel_end)
+
+
+def get_location(element: str, data_type: str) -> str:
+ location_start = __get_location_start(data_type=data_type)
+ location_end = __get_location_end(element=element)
+ return location_start + location_end
+
+
+def __get_channel_start(interval: str) -> str:
+ if interval == "second":
+ return "S"
+ elif interval == "minute":
+ return "M"
+ elif interval == "hour":
+ return "H"
+ elif interval == "day":
+ return "D"
+ raise ValueError(f" Unexcepted interval: {interval}")
+
+
+def __check_predefined_channel(element: str, interval: str) -> Optional[str]:
+ if element in ELEMENT_CONVERSIONS:
+ return __get_channel_start(interval=interval) + ELEMENT_CONVERSIONS[element]
+ elif len(element) == 3:
+ return element
+ # chan.loc format
+ elif "." in element:
+ channel = element.split(".")[0]
+ return channel.strip()
+ else:
+ return None
+
+
+def __get_channel_end(element: str) -> str:
+ channel_middle = "V"
+ if "_Volt" in element:
+ channel_middle = "E"
+ elif "_Bin" in element:
+ channel_middle = "Y"
+ elif "_Temp" in element:
+ channel_middle = "K"
+ elif element in ["F", "G"]:
+ channel_middle = "S"
+ channel_end = element.split("_")[0]
+ return channel_middle + channel_end
+
+
+def __get_location_end(element: str) -> str:
+ """Translates element suffix to end of location code"""
+ if "_Sat" in element:
+ return "1"
+ return "0"
diff --git a/geomagio/edge/MiniSeedFactory.py b/geomagio/edge/MiniSeedFactory.py
index f3530cdd6..ec8b54692 100644
--- a/geomagio/edge/MiniSeedFactory.py
+++ b/geomagio/edge/MiniSeedFactory.py
@@ -319,7 +319,7 @@ class MiniSeedFactory(TimeseriesFactory):
obspy.core.trace
timeseries trace of the requested channel data
"""
- sncl = SNCL().get_sncl(
+ sncl = SNCL.get_sncl(
station=observatory, data_type=type, interval=interval, element=channel
)
data = self.client.get_waveforms(
@@ -460,7 +460,7 @@ class MiniSeedFactory(TimeseriesFactory):
to_write = to_write.split()
to_write = TimeseriesUtility.unmask_stream(to_write)
# relabel channels from internal to edge conventions
- sncl = SNCL().get_sncl(
+ sncl = SNCL.get_sncl(
station=observatory,
data_type=type,
interval=interval,
diff --git a/geomagio/edge/SNCL.py b/geomagio/edge/SNCL.py
index 395bfefe5..3dc4b9413 100644
--- a/geomagio/edge/SNCL.py
+++ b/geomagio/edge/SNCL.py
@@ -1,4 +1,3 @@
-from __future__ import annotations
from typing import Dict, Optional
from pydantic import BaseModel
@@ -18,26 +17,25 @@ CHANNEL_CONVERSIONS = {
class SNCL(BaseModel):
- station: str = None
+ station: str
network: str = "NT"
- channel: str = None
- location: str = None
+ channel: str
+ location: str
+ @classmethod
def get_sncl(
- self,
- station: str,
+ cls,
data_type: str,
- interval: str,
element: str,
- ) -> SNCL:
- from .SNCLFactory import SNCLFactory
-
- factory = SNCLFactory(data_format="miniseed")
+ interval: str,
+ station: str,
+ network: str = "NT",
+ ) -> "SNCL":
return SNCL(
station=station,
- network=self.network,
- channel=factory.get_channel(element=element, interval=interval),
- location=factory.get_location(element=element, data_type=data_type),
+ network=network,
+ channel=get_channel(element=element, interval=interval),
+ location=get_location(element=element, data_type=data_type),
)
def parse_sncl(self) -> Dict:
@@ -115,3 +113,81 @@ class SNCL(BaseModel):
if channel_end in CHANNEL_CONVERSIONS:
return CHANNEL_CONVERSIONS[channel_end]
return None
+
+
+def get_channel(element: str, interval: str) -> str:
+ predefined_channel = __check_predefined_channel(element=element, interval=interval)
+ channel_start = __get_channel_start(interval=interval)
+ channel_end = __get_channel_end(element=element)
+ return predefined_channel or (channel_start + channel_end)
+
+
+def get_location(element: str, data_type: str) -> str:
+ location_start = __get_location_start(data_type=data_type)
+ location_end = __get_location_end(element=element)
+ return location_start + location_end
+
+
+def __get_channel_start(interval: str) -> str:
+ if interval == "tenhertz":
+ return "B"
+ if interval == "second":
+ return "L"
+ elif interval == "minute":
+ return "U"
+ elif interval == "hour":
+ return "R"
+ elif interval == "day":
+ return "P"
+ raise ValueError(f" Unexcepted interval: {interval}")
+
+
+def __check_predefined_channel(element: str, interval: str) -> Optional[str]:
+ if element in ELEMENT_CONVERSIONS:
+ return __get_channel_start(interval=interval) + ELEMENT_CONVERSIONS[element]
+ elif len(element) == 3:
+ return element
+ # chan.loc format
+ elif "." in element:
+ channel = element.split(".")[0]
+ return channel.strip()
+ else:
+ return None
+
+
+def __get_channel_end(element: str) -> str:
+ channel_middle = "F"
+ if "_Volt" in element:
+ channel_middle = "E"
+ elif "_Bin" in element:
+ channel_middle = "Y"
+ elif "_Temp" in element:
+ channel_middle = "K"
+ channel_end = element.split("_")[0]
+ return channel_middle + channel_end
+
+
+def __get_location_start(data_type: str) -> str:
+ """Translates data type to beginning of location code"""
+ if data_type == "variation":
+ return "R"
+ elif data_type == "adjusted":
+ return "A"
+ elif data_type == "quasi-definitive":
+ return "Q"
+ elif data_type == "definitive":
+ return "D"
+ raise ValueError(f"Unexpected data type: {data_type}")
+
+
+def __get_location_end(element: str) -> str:
+ """Translates element suffix to end of location code"""
+ if "_Sat" in element:
+ return "1"
+ if "_Dist" in element:
+ return "D"
+ if "_SQ" in element:
+ return "Q"
+ if "_SV" in element:
+ return "V"
+ return "0"
diff --git a/geomagio/edge/SNCLFactory.py b/geomagio/edge/SNCLFactory.py
deleted file mode 100644
index c4be5e899..000000000
--- a/geomagio/edge/SNCLFactory.py
+++ /dev/null
@@ -1,123 +0,0 @@
-from typing import Literal, Optional, Union
-
-from .SNCL import SNCL, ELEMENT_CONVERSIONS as MINISEED_CONVERSIONS
-from .LegacySNCL import LegacySNCL, ELEMENT_CONVERSIONS as LEGACY_CONVERSIONS
-
-INTERVAL_CONVERSIONS = {
- "miniseed": {
- "tenhertz": "B",
- "second": "L",
- "minute": "U",
- "hour": "R",
- "day": "P",
- },
- "legacy": {
- "second": "S",
- "minute": "M",
- "hour": "H",
- "day": "D",
- },
-}
-
-
-class SNCLFactory(object):
- def __init__(self, data_format: Literal["miniseed", "legacy"] = "miniseed"):
- self.data_format = data_format
-
- def get_sncl(
- self,
- station: str,
- network: str,
- channel: str,
- location: str,
- ) -> Union[SNCL, LegacySNCL]:
- sncl_params = {
- "station": station,
- "network": network,
- "channel": channel,
- "location": location,
- }
- return (
- SNCL(**sncl_params)
- if self.data_format == "miniseed"
- else LegacySNCL(**sncl_params)
- )
-
- def get_channel(self, element: str, interval: str) -> str:
- predefined_channel = self.__check_predefined_channel(
- element=element, interval=interval
- )
- channel_start = self.__get_channel_start(interval=interval)
- channel_end = self.__get_channel_end(element=element)
- return predefined_channel or (channel_start + channel_end)
-
- def get_location(self, element: str, data_type: str) -> str:
- location_start = self.__get_location_start(data_type=data_type)
- location_end = self.__get_location_end(element=element)
- return location_start + location_end
-
- def __get_channel_start(self, interval: str) -> str:
- interval_conversions = INTERVAL_CONVERSIONS[self.data_format]
- try:
- return interval_conversions[interval]
- except:
- raise ValueError(f"Unexpected interval: {interval}")
-
- def __check_predefined_channel(self, element: str, interval: str) -> Optional[str]:
- channel_conversions = (
- MINISEED_CONVERSIONS
- if self.data_format == "miniseed"
- else LEGACY_CONVERSIONS
- )
-
- if element in channel_conversions:
- return (
- self.__get_channel_start(interval=interval)
- + channel_conversions[element]
- )
- elif len(element) == 3:
- return element
- # chan.loc format
- elif "." in element:
- channel = element.split(".")[0]
- return channel.strip()
- else:
- return None
-
- def __get_channel_end(self, element: str) -> str:
- channel_middle = "F" if self.data_format == "miniseed" else "V"
- if "_Volt" in element:
- channel_middle = "E"
- elif "_Bin" in element:
- channel_middle = "Y"
- elif "_Temp" in element:
- channel_middle = "K"
- elif element in ["F", "G"] and self.data_format == "legacy":
- channel_middle = "S"
- channel_end = element.split("_")[0]
- return channel_middle + channel_end
-
- def __get_location_start(self, data_type: str) -> str:
- """Translates data type to beginning of location code"""
- if data_type == "variation":
- return "R"
- elif data_type == "adjusted":
- return "A"
- elif data_type == "quasi-definitive":
- return "Q"
- elif data_type == "definitive":
- return "D"
- raise ValueError(f"Unexpected data type: {data_type}")
-
- def __get_location_end(self, element: str) -> str:
- """Translates element suffix to end of location code"""
- if "_Sat" in element:
- return "1"
- if self.data_format == "miniseed":
- if "_Dist" in element:
- return "D"
- if "_SQ" in element:
- return "Q"
- if "_SV" in element:
- return "V"
- return "0"
diff --git a/geomagio/edge/__init__.py b/geomagio/edge/__init__.py
index 77d5fee30..45b3d49c3 100644
--- a/geomagio/edge/__init__.py
+++ b/geomagio/edge/__init__.py
@@ -8,7 +8,6 @@ from .MiniSeedFactory import MiniSeedFactory
from .MiniSeedInputClient import MiniSeedInputClient
from .RawInputClient import RawInputClient
from .SNCL import SNCL
-from .SNCLFactory import SNCLFactory
from .LegacySNCL import LegacySNCL
__all__ = [
@@ -18,8 +17,5 @@ __all__ = [
"MiniSeedInputClient",
"RawInputClient",
"LegacySNCL",
- "LegacySNIDE",
"SNCL",
- "SNCLFactory",
- "LegacySNCL",
]
diff --git a/test/edge_test/LegacySNCL_test.py b/test/edge_test/LegacySNCL_test.py
index 8d7233ffc..adbcc8a83 100644
--- a/test/edge_test/LegacySNCL_test.py
+++ b/test/edge_test/LegacySNCL_test.py
@@ -1,4 +1,4 @@
-from geomagio.edge import LegacySNCL
+from geomagio.edge.LegacySNCL import LegacySNCL, get_channel, get_location
def test_data_type():
@@ -111,9 +111,33 @@ def test_element():
)
+def test_get_channel():
+ """edge_test.LegacySNCL_test.test_get_channel()"""
+ assert get_channel(element="D", interval="second") == "SVD"
+ assert get_channel(element="F", interval="minute") == "MSF"
+ assert get_channel(element="H", interval="hour") == "HVH"
+ assert get_channel(element="E-E", interval="day") == "DQE"
+ assert get_channel(element="E-N", interval="minute") == "MQN"
+ assert get_channel(element="SQ", interval="minute") == "MSQ"
+ assert get_channel(element="SV", interval="minute") == "MSV"
+ assert get_channel(element="UK1", interval="minute") == "UK1"
+ assert get_channel(element="DIST", interval="minute") == "MDT"
+ assert get_channel(element="DST", interval="minute") == "MGD"
+ assert get_channel(element="UK1.R0", interval="minute") == "UK1"
+
+
+def test_get_location():
+ """edge_test.LegacySNCL_test.test_get_location()"""
+ assert get_location(element="D", data_type="variation") == "R0"
+ assert get_location(element="D", data_type="adjusted") == "A0"
+ 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"
+
+
def test_get_sncl():
"""edge_test.LegacySNCL_test.test_get_sncl()"""
- assert LegacySNCL().get_sncl(
+ assert LegacySNCL.get_sncl(
station="BOU", data_type="variation", interval="second", element="H"
) == LegacySNCL(station="BOU", network="NT", channel="SVH", location="R0")
diff --git a/test/edge_test/SNCLFactory_test.py b/test/edge_test/SNCLFactory_test.py
deleted file mode 100644
index 0920c804c..000000000
--- a/test/edge_test/SNCLFactory_test.py
+++ /dev/null
@@ -1,64 +0,0 @@
-from geomagio.edge import SNCL, SNCLFactory, LegacySNCL
-
-
-def test_get_channel():
- """edge_test.SNCLFactory_test.test_get_channel()"""
- factory = SNCLFactory(data_format="miniseed")
- assert factory.get_channel(element="U_Volt", interval="tenhertz") == "BEU"
- assert factory.get_channel(element="U_Bin", interval="tenhertz") == "BYU"
- assert factory.get_channel(element="D", interval="second") == "LFD"
- assert factory.get_channel(element="F", interval="minute") == "UFF"
- assert factory.get_channel(element="H", interval="hour") == "RFH"
- assert factory.get_channel(element="Dst4", interval="day") == "PX4"
- assert factory.get_channel(element="Dst3", interval="minute") == "UX3"
- assert factory.get_channel(element="E-E", interval="minute") == "UQE"
- assert factory.get_channel(element="E-N", interval="minute") == "UQN"
- assert factory.get_channel(element="UK1", interval="minute") == "UK1"
- assert factory.get_channel(element="U_Dist", interval="minute") == "UFU"
- assert factory.get_channel(element="U_SQ", interval="minute") == "UFU"
- assert factory.get_channel(element="U_SV", interval="minute") == "UFU"
- assert factory.get_channel(element="UK1.R0", interval="minute") == "UK1"
-
- # test legacy format
- factory = SNCLFactory(data_format="legacy")
- assert factory.get_channel(element="D", interval="second") == "SVD"
- assert factory.get_channel(element="F", interval="minute") == "MSF"
- assert factory.get_channel(element="H", interval="hour") == "HVH"
- assert factory.get_channel(element="E-E", interval="day") == "DQE"
- assert factory.get_channel(element="E-N", interval="minute") == "MQN"
- assert factory.get_channel(element="SQ", interval="minute") == "MSQ"
- assert factory.get_channel(element="SV", interval="minute") == "MSV"
- assert factory.get_channel(element="UK1", interval="minute") == "UK1"
- assert factory.get_channel(element="DIST", interval="minute") == "MDT"
- assert factory.get_channel(element="DST", interval="minute") == "MGD"
- assert factory.get_channel(element="UK1.R0", interval="minute") == "UK1"
-
-
-def test_get_location():
- """edge_test.SNCLFactory_test.test_get_location()"""
- factory = SNCLFactory(data_format="miniseed")
- assert factory.get_location(element="D", data_type="variation") == "R0"
- assert factory.get_location(element="D", data_type="adjusted") == "A0"
- assert factory.get_location(element="D", data_type="quasi-definitive") == "Q0"
- assert factory.get_location(element="D", data_type="definitive") == "D0"
- assert factory.get_location(element="D_Sat", data_type="variation") == "R1"
- assert factory.get_location(element="D_Dist", data_type="variation") == "RD"
- assert factory.get_location(element="D_SQ", data_type="variation") == "RQ"
- assert factory.get_location(element="D_SV", data_type="variation") == "RV"
-
- factory = SNCLFactory(data_format="legacy")
- assert factory.get_location(element="D", data_type="variation") == "R0"
- assert factory.get_location(element="D", data_type="adjusted") == "A0"
- assert factory.get_location(element="D", data_type="quasi-definitive") == "Q0"
- assert factory.get_location(element="D", data_type="definitive") == "D0"
- assert factory.get_location(element="D_Sat", data_type="variation") == "R1"
-
-
-def test_get_sncl():
- """edge_test.SNCLFactory_test.test_get_sncl()"""
- assert SNCLFactory(data_format="miniseed").get_sncl(
- station="BOU", network="NT", channel="UFU", location="R0"
- ) == SNCL(station="BOU", network="NT", channel="UFU", location="R0")
- assert SNCLFactory(data_format="legacy").get_sncl(
- station="BOU", network="NT", channel="MVH", location="R0"
- ) == LegacySNCL(station="BOU", network="NT", channel="MVH", location="R0")
diff --git a/test/edge_test/SNCL_test.py b/test/edge_test/SNCL_test.py
index b2ca4a1be..ea92aa524 100644
--- a/test/edge_test/SNCL_test.py
+++ b/test/edge_test/SNCL_test.py
@@ -1,4 +1,4 @@
-from geomagio.edge import SNCL
+from geomagio.edge.SNCL import SNCL, get_channel, get_location
def test_data_type():
@@ -104,9 +104,39 @@ def test_element():
)
+def test_get_channel():
+ """edge_test.SNCL_test.test_get_channel()"""
+ assert get_channel(element="U_Volt", interval="tenhertz") == "BEU"
+ assert get_channel(element="U_Bin", interval="tenhertz") == "BYU"
+ assert get_channel(element="D", interval="second") == "LFD"
+ assert get_channel(element="F", interval="minute") == "UFF"
+ assert get_channel(element="H", interval="hour") == "RFH"
+ assert get_channel(element="Dst4", interval="day") == "PX4"
+ assert get_channel(element="Dst3", interval="minute") == "UX3"
+ assert get_channel(element="E-E", interval="minute") == "UQE"
+ assert get_channel(element="E-N", interval="minute") == "UQN"
+ assert get_channel(element="UK1", interval="minute") == "UK1"
+ assert get_channel(element="U_Dist", interval="minute") == "UFU"
+ assert get_channel(element="U_SQ", interval="minute") == "UFU"
+ assert get_channel(element="U_SV", interval="minute") == "UFU"
+ assert get_channel(element="UK1.R0", interval="minute") == "UK1"
+
+
+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="adjusted") == "A0"
+ 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_Dist", data_type="variation") == "RD"
+ assert get_location(element="D_SQ", data_type="variation") == "RQ"
+ assert get_location(element="D_SV", data_type="variation") == "RV"
+
+
def test_get_sncl():
"""edge_test.SNCL_test.test_get_sncl()"""
- assert SNCL().get_sncl(
+ assert SNCL.get_sncl(
station="BOU", data_type="variation", interval="second", element="U"
) == SNCL(station="BOU", network="NT", channel="LFU", location="R0")
--
GitLab