From 68d5d5ab6a66f044939ac936da3f7832f37e4835 Mon Sep 17 00:00:00 2001
From: pcain-usgs <pcain@usgs.gov>
Date: Thu, 13 May 2021 17:33:56 -0600
Subject: [PATCH 1/7] SNCL/SNCLFactory
---
geomagio/edge/SNCL.py | 114 ++++++++++
geomagio/edge/SNCLFactory.py | 83 ++++++++
geomagio/edge/__init__.py | 6 +
geomagio/edge/sncl.py | 258 -----------------------
test/edge_test/SNCLFactory_test.py | 66 ++++++
test/edge_test/SNCL_test.py | 328 +++++++++++++++++++++++++++++
6 files changed, 597 insertions(+), 258 deletions(-)
create mode 100644 geomagio/edge/SNCL.py
create mode 100644 geomagio/edge/SNCLFactory.py
delete mode 100644 geomagio/edge/sncl.py
create mode 100644 test/edge_test/SNCLFactory_test.py
create mode 100644 test/edge_test/SNCL_test.py
diff --git a/geomagio/edge/SNCL.py b/geomagio/edge/SNCL.py
new file mode 100644
index 000000000..20a0a885f
--- /dev/null
+++ b/geomagio/edge/SNCL.py
@@ -0,0 +1,114 @@
+from typing import Dict, Optional, Set
+
+from pydantic import BaseModel
+
+INTERVAL_CONVERSIONS = {
+ "legacy": {
+ "second": "S",
+ "minute": "M",
+ "hour": "H",
+ "day": "D",
+ },
+ "miniseed": {
+ "tenhertz": "B",
+ "second": "L",
+ "minute": "U",
+ "hour": "R",
+ "day": "P",
+ },
+}
+ELEMENT_CONVERSIONS = {
+ # e-field
+ "E-E": "QE",
+ "E-N": "QN",
+ # derived indicies
+ "Dst3": "X3",
+ "Dst4": "X4",
+ "SQ": "SQ",
+ "SV": "SV",
+ "DIST": "DT",
+ "DST": "GD",
+}
+
+CHANNEL_CONVERSIONS = {
+ ELEMENT_CONVERSIONS[key]: key for key in ELEMENT_CONVERSIONS.keys()
+}
+
+
+class SNCL(BaseModel):
+ station: str
+ network: str = "NT"
+ channel: str
+ location: str
+ data_format: str = "miniseed"
+
+ def dict(self, exclude: Set = {"data_format"}) -> dict:
+ return super().dict(
+ exclude=exclude,
+ )
+
+ def json(self, exclude: Set = {"data_format"}) -> str:
+ return super().json(
+ exclude=exclude,
+ )
+
+ @property
+ def data_type(self) -> str:
+ location_start = self.location[0]
+ if location_start == "R":
+ return "variation"
+ elif location_start == "A":
+ return "adjusted"
+ elif location_start == "Q":
+ return "quasi-definitive"
+ elif location_start == "D":
+ return "definitive"
+ raise ValueError(f"Unexpected location start: {location_start}")
+
+ @property
+ def element(self) -> str:
+ element = self.__get_predefined_element()
+ element = element or self.__get_element()
+ return element
+
+ @property
+ def interval(self) -> str:
+ interval_conversions = INTERVAL_CONVERSIONS[self.data_format]
+ interval_code_conversions = {
+ interval_conversions[key]: key for key in interval_conversions.keys()
+ }
+ channel_start = self.channel[0]
+ try:
+ return interval_code_conversions[channel_start]
+ except:
+ raise ValueError(f"Unexcepted interval code: {channel_start}")
+
+ def __get_element(self):
+ element_start = self.channel[2]
+ channel = self.channel
+ channel_middle = channel[1]
+ location_end = self.location[1]
+ if channel_middle in ["Q", "E"]:
+ element_end = "_Volt"
+ elif channel_middle == "Y":
+ element_end = "_Bin"
+ elif channel_middle == "K":
+ element_end = "_Temp"
+ elif location_end == "1":
+ element_end = "_Sat"
+ elif location_end == "D":
+ element_end = "_Dist"
+ elif location_end == "Q":
+ element_end = "_SQ"
+ elif location_end == "V":
+ element_end = "_SV"
+ else:
+ element_end = ""
+ return element_start + element_end
+
+ def __get_predefined_element(self) -> Optional[str]:
+ channel = self.channel
+ channel_end = channel[1:]
+ if channel_end in CHANNEL_CONVERSIONS:
+ return CHANNEL_CONVERSIONS[channel_end]
+ return None
diff --git a/geomagio/edge/SNCLFactory.py b/geomagio/edge/SNCLFactory.py
new file mode 100644
index 000000000..4e677c0af
--- /dev/null
+++ b/geomagio/edge/SNCLFactory.py
@@ -0,0 +1,83 @@
+from typing import Optional
+
+from .SNCL import SNCL, INTERVAL_CONVERSIONS, ELEMENT_CONVERSIONS
+
+
+class SNCLFactory(object):
+ def __init__(self, data_format: str = "miniseed"):
+ self.data_format = data_format
+
+ def get_sncl(
+ self,
+ station: str,
+ data_type: str,
+ element: str,
+ interval: str,
+ network: str = "NT",
+ ) -> SNCL:
+ return SNCL(
+ station=station,
+ network=network,
+ channel=self.get_channel(element=element, interval=interval),
+ location=self.get_location(element=element, data_type=data_type),
+ )
+
+ def get_channel(self, element: str, interval: str) -> str:
+ channel_start = self.__get_channel_start(interval=interval)
+ channel_end = self.__get_predefined_channel(element=element)
+ channel_end = channel_end or self.__get_channel_end(element=element)
+ return 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:
+ try:
+ return INTERVAL_CONVERSIONS[self.data_format][interval]
+ except:
+ raise ValueError(f"Unexpected interval: {interval}")
+
+ def __get_predefined_channel(self, element: str) -> Optional[str]:
+ if len(element) == 3 and "-" not in element and element != "DST":
+ return element[1:]
+ elif element in ELEMENT_CONVERSIONS:
+ return ELEMENT_CONVERSIONS[element]
+ 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:
+ 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:
+ 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/__init__.py b/geomagio/edge/__init__.py
index 2a53748a1..bf808a25a 100644
--- a/geomagio/edge/__init__.py
+++ b/geomagio/edge/__init__.py
@@ -7,6 +7,8 @@ from .LocationCode import LocationCode
from .MiniSeedFactory import MiniSeedFactory
from .MiniSeedInputClient import MiniSeedInputClient
from .RawInputClient import RawInputClient
+from .SNCL import SNCL
+from .SNCLFactory import SNCLFactory
__all__ = [
"EdgeFactory",
@@ -14,4 +16,8 @@ __all__ = [
"MiniSeedFactory",
"MiniSeedInputClient",
"RawInputClient",
+ "LegacySNCL",
+ "LegacySNIDE",
+ "SNCL",
+ "SNCLFactory",
]
diff --git a/geomagio/edge/sncl.py b/geomagio/edge/sncl.py
deleted file mode 100644
index 5bfa4ff2e..000000000
--- a/geomagio/edge/sncl.py
+++ /dev/null
@@ -1,258 +0,0 @@
-"""SNCL utilities.
-
-Station
-Network
-Channel
-Location
-"""
-
-# components that map directly to channel suffixes
-CHANNEL_FROM_COMPONENT = {
- # e-field
- "E-E": "QY",
- "E-N": "QX",
- "E-U": "QU",
- "E-V": "QV",
- # derived indicies
- "AE": "XA",
- "DST3": "X3",
- "DST": "X4",
- "K": "XK",
-}
-# reverse lookup of component from channel
-COMPONENT_FROM_CHANNEL = dict((v, k) for (k, v) in CHANNEL_FROM_COMPONENT.iteritems())
-
-
-class SNCLException(Exception):
- pass
-
-
-def get_scnl(
- observatory,
- component=None,
- channel=None,
- data_type="variation",
- interval="second",
- location=None,
- network="NT",
-):
- """Generate a SNCL code from data attributes.
-
- Parameters
- ----------
- observatory : str
- observatory code.
- component : str
- geomag component name.
- channel : str
- default None.
- use a specific channel code, instead of generating.
- data_type : str
- default 'variation'
- 'variation', 'adjusted', 'quasi-definitive', or 'definitive'.
- interval: str|float
- default 'second'
- 'tenhertz', 'second', 'minute', 'hour', 'day',
- or equivalent interval in seconds
- location : str
- default None
- use a specific location code, instead of generating.
- network : str
- default 'NT'
- network `observatory` is a part of.
-
- Raises
- ------
- SNCLException : when unable to generate a SNCL
-
- Returns
- -------
- dict : dictionary containing the following keys
- 'station' : observatory code
- 'network' : network code
- 'channel' : channel code
- 'location' : location code
- """
- # use explicit channel/location if specified
- channel = channel or __get_channel(component, interval)
- location = location or __get_location(component, data_type)
- return {
- "station": observatory,
- "network": network,
- "channel": channel,
- "location": location,
- }
-
-
-def parse_sncl(sncl):
- """Parse a SNCL code into data attributes.
-
- Parameters
- ----------
- sncl : dict
- dictionary object with the following keys
- 'station' : observatory code
- 'network' : network code
- 'channel' : channel code
- 'location' : location code
-
- Raises
- ------
- SNCLException : when unable to parse a SNCL
-
- Returns
- -------
- dict : dictionary containing the following keys
- 'observatory' : observatory code
- 'network' : network code
- 'component' : geomag component name
- 'data_type' : geomag data type (e.g. 'variation')
- 'interval' : data interval in seconds (e.g. 1)
- """
- network = sncl["network"]
- station = sncl["station"]
- channel = sncl["channel"]
- location = sncl["location"]
- return {
- "observatory": station,
- "network": network,
- "component": __parse_component(channel, location),
- "data_type": __parse_data_type(location),
- "interval": __parse_interval(channel),
- }
-
-
-def __get_channel(component, interval):
- channel_start = __get_channel_start(interval)
- # check for direct component mappings
- if component in CHANNEL_FROM_COMPONENT:
- channel_end = CHANNEL_FROM_COMPONENT[component]
- else:
- channel_end = __get_channel_end(component)
- return channel_start + channel_end
-
-
-def __get_channel_start(interval):
- if interval == "tenhertz" or interval == 0.1:
- return "B"
- if interval == "second" or interval == 1:
- return "L"
- if interval == "minute" or interval == 60:
- return "U"
- if interval == "hour" or interval == 3600:
- return "R"
- if interval == "day" or interval == 86400:
- return "P"
- raise SNCLException("Unexpected interval {}".format(interval))
-
-
-def __get_channel_end(component):
- # default to engineering units
- channel_middle = "F"
- # check for suffix that may override
- component_parts = component.split("-")
- channel_end = component_parts[0]
- if len(component_parts) > 1:
- component_suffix = component_parts[1]
- if component_suffix == "-Bin":
- channel_middle = "Y"
- elif component_suffix == "-Temp":
- channel_middle = "K"
- elif component_suffix == "-Volt":
- channel_middle = "E"
- else:
- raise SNCLException("Unexpected component {}".format(component))
- return channel_middle + channel_end
-
-
-def __get_location(component, data_type):
- location_start = __get_location_start(data_type)
- location_end = __get_location_end(component)
- return location_start + location_end
-
-
-def __get_location_start(data_type):
- 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 SNCLException("Unexpected data type {}".format(data_type))
-
-
-def __get_location_end(component):
- if component.endswith("-Sat"):
- return "1"
- if component.endswith("-Dist"):
- return "D"
- if component.endswith("-SQ"):
- return "Q"
- if component.endswith("-SV"):
- return "V"
- return "0"
-
-
-def __parse_component(channel, location):
- channel_end = channel[1:]
- if channel_end in COMPONENT_FROM_CHANNEL:
- return COMPONENT_FROM_CHANNEL[channel_end]
- channel_middle = channel[1]
- component = channel[2]
- component_end = ""
- if channel_middle == "E":
- component_end = "-Volt"
- elif channel_middle == "K":
- component_end = "-Temp"
- elif channel_middle == "Y":
- component_end = "-Bin"
- elif channel_middle == "F":
- component_end = __parse_component_end(location)
- else:
- raise SNCLException("Unexpected channel middle {}".format(channel))
- return component + component_end
-
-
-def __parse_component_end(location):
- location_end = location[1]
- if location_end == "0":
- return ""
- if location_end == "1":
- return "-Sat"
- if location_end == "D":
- return "-Dist"
- if location_end == "Q":
- return "-SQ"
- if location_end == "V":
- return "-SV"
- raise SNCLException("Unexpected location end {}".format(location_end))
-
-
-def __parse_data_type(location):
- location_start = location[0]
- if location_start == "R":
- return "variation"
- if location_start == "A":
- return "adjusted"
- if location_start == "Q":
- return "quasi-definitive"
- if location_start == "D":
- return "definitive"
- raise SNCLException("Unexpected location start {}".format(location_start))
-
-
-def __parse_interval(channel):
- channel_start = channel[0]
- if channel_start == "B":
- return 0.1
- if channel_start == "L":
- return 1
- if channel_start == "U":
- return 60
- if channel_start == "R":
- return 3600
- if channel_start == "P":
- return 86400
- raise SNCLException("Unexpected channel {}".format(channel))
diff --git a/test/edge_test/SNCLFactory_test.py b/test/edge_test/SNCLFactory_test.py
new file mode 100644
index 000000000..d12ab5f47
--- /dev/null
+++ b/test/edge_test/SNCLFactory_test.py
@@ -0,0 +1,66 @@
+from geomagio.edge import SNCL, SNCLFactory
+
+
+def test_get_sncl():
+ assert (
+ SNCLFactory(data_format="miniseed").get_sncl(
+ station="BOU",
+ data_type="variation",
+ element="UFU",
+ interval="minute",
+ )
+ == SNCL(station="BOU", network="NT", channel="UFU", location="R0")
+ )
+ assert (
+ SNCLFactory(data_format="legacy").get_sncl(
+ station="BOU",
+ data_type="variation",
+ element="MVH",
+ interval="minute",
+ )
+ == SNCL(station="BOU", network="NT", channel="MVH", location="R0")
+ )
+
+
+def test_get_channel():
+ # test miniseed format
+ factory = SNCLFactory(data_format="miniseed")
+ assert factory.get_channel(element="D", interval="minute") == "UFD"
+ assert factory.get_channel(element="F", interval="minute") == "UFF"
+ assert factory.get_channel(element="H", interval="minute") == "UFH"
+ assert factory.get_channel(element="Dst4", interval="minute") == "UX4"
+ 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="SQ", interval="minute") == "USQ"
+ assert factory.get_channel(element="SV", interval="minute") == "USV"
+ assert factory.get_channel(element="DIST", interval="minute") == "UDT"
+ assert factory.get_channel(element="DST", interval="minute") == "UGD"
+ assert factory.get_channel(element="U_Dist", interval="minute") == "UFU"
+
+ # test legacy format
+ factory = SNCLFactory(data_format="legacy")
+ assert factory.get_channel(element="D", interval="minute") == "MVD"
+ assert factory.get_channel(element="F", interval="minute") == "MSF"
+ assert factory.get_channel(element="H", interval="minute") == "MVH"
+ assert factory.get_channel(element="Dst4", interval="minute") == "MX4"
+ assert factory.get_channel(element="Dst3", interval="minute") == "MX3"
+ assert factory.get_channel(element="E-E", interval="minute") == "MQE"
+ 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="DIST", interval="minute") == "MDT"
+ assert factory.get_channel(element="DST", interval="minute") == "MGD"
+ assert factory.get_channel(element="H_Dist", interval="minute") == "MVH"
+
+
+def 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"
diff --git a/test/edge_test/SNCL_test.py b/test/edge_test/SNCL_test.py
new file mode 100644
index 000000000..7f59107d9
--- /dev/null
+++ b/test/edge_test/SNCL_test.py
@@ -0,0 +1,328 @@
+from geomagio.edge import SNCL
+
+
+def test_data_type():
+ assert SNCL(station="BOU", channel="LFU", location="R0").data_type == "variation"
+ assert SNCL(station="BOU", channel="LFU", location="A0").data_type == "adjusted"
+ assert (
+ SNCL(station="BOU", channel="LFU", location="Q0").data_type
+ == "quasi-definitive"
+ )
+ assert SNCL(station="BOU", channel="LFU", location="D0").data_type == "definitive"
+
+
+def test_interval():
+ # miniseed format
+ assert (
+ SNCL(
+ station="BOU",
+ channel="BEU",
+ location="R0",
+ ).interval
+ == "tenhertz"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="LEU",
+ location="R0",
+ ).interval
+ == "second"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="UEU",
+ location="R0",
+ ).interval
+ == "minute"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="REU",
+ location="R0",
+ ).interval
+ == "hour"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="PEU",
+ location="R0",
+ ).interval
+ == "day"
+ )
+ # legacy format
+ assert (
+ SNCL(
+ station="BOU",
+ channel="SVH",
+ location="R0",
+ data_format="legacy",
+ ).interval
+ == "second"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MVH",
+ location="R0",
+ data_format="legacy",
+ ).interval
+ == "minute"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="HVH",
+ location="R0",
+ data_format="legacy",
+ ).interval
+ == "hour"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="DVH",
+ location="R0",
+ data_format="legacy",
+ ).interval
+ == "day"
+ )
+
+
+def test_element():
+ assert (
+ SNCL(
+ station="BOU",
+ channel="UFD",
+ location="R0",
+ ).element
+ == "D"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="UFU",
+ location="R0",
+ ).element
+ == "U"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="UFF",
+ location="R0",
+ ).element
+ == "F"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="UFH",
+ location="R0",
+ ).element
+ == "H"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="UX4",
+ location="R0",
+ ).element
+ == "Dst4"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="UX3",
+ location="R0",
+ ).element
+ == "Dst3"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="UQE",
+ location="R0",
+ ).element
+ == "E-E"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="UQN",
+ location="R0",
+ ).element
+ == "E-N"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="BEU",
+ location="R0",
+ ).element
+ == "U_Volt"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="BYU",
+ location="R0",
+ ).element
+ == "U_Bin"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="UFU",
+ location="R1",
+ ).element
+ == "U_Sat"
+ )
+
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MVD",
+ location="R0",
+ data_format="legacy",
+ ).element
+ == "D"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MVU",
+ location="R0",
+ data_format="legacy",
+ ).element
+ == "U"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MSF",
+ location="R0",
+ data_format="legacy",
+ ).element
+ == "F"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MVH",
+ location="R0",
+ data_format="legacy",
+ ).element
+ == "H"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MX4",
+ location="R0",
+ data_format="legacy",
+ ).element
+ == "Dst4"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MX3",
+ location="R0",
+ data_format="legacy",
+ ).element
+ == "Dst3"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MQE",
+ location="R0",
+ data_format="legacy",
+ ).element
+ == "E-E"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MQN",
+ location="R0",
+ data_format="legacy",
+ ).element
+ == "E-N"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MEH",
+ location="R0",
+ data_format="legacy",
+ ).element
+ == "H_Volt"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MYH",
+ location="R0",
+ data_format="legacy",
+ ).element
+ == "H_Bin"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MVH",
+ location="R1",
+ data_format="legacy",
+ ).element
+ == "H_Sat"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MVH",
+ location="RD",
+ data_format="legacy",
+ ).element
+ == "H_Dist"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MVH",
+ location="RQ",
+ data_format="legacy",
+ ).element
+ == "H_SQ"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MVH",
+ location="RV",
+ data_format="legacy",
+ ).element
+ == "H_SV"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MDT",
+ location="RV",
+ data_format="legacy",
+ ).element
+ == "DIST"
+ )
+ assert (
+ SNCL(
+ station="BOU",
+ channel="MGD",
+ location="RV",
+ data_format="legacy",
+ ).element
+ == "DST"
+ )
--
GitLab
From a2f7aed49b4960dbaed6509c7e14bb31a25e8236 Mon Sep 17 00:00:00 2001
From: pcain-usgs <pcain@usgs.gov>
Date: Thu, 13 May 2021 17:43:34 -0600
Subject: [PATCH 2/7] implement parse sncl method
---
geomagio/edge/SNCL.py | 9 +++++++++
1 file changed, 9 insertions(+)
diff --git a/geomagio/edge/SNCL.py b/geomagio/edge/SNCL.py
index 20a0a885f..9e9e92b55 100644
--- a/geomagio/edge/SNCL.py
+++ b/geomagio/edge/SNCL.py
@@ -42,6 +42,15 @@ class SNCL(BaseModel):
location: str
data_format: str = "miniseed"
+ def parse_sncl(self) -> Dict:
+ return {
+ "station": self.station,
+ "network": self.network,
+ "data_type": self.data_type,
+ "element": self.element,
+ "interval": self.interval,
+ }
+
def dict(self, exclude: Set = {"data_format"}) -> dict:
return super().dict(
exclude=exclude,
--
GitLab
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 3/7] 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
From 39b9512b3c7a82140fefb5cae9cd99ce0dc3aceb Mon Sep 17 00:00:00 2001
From: pcain-usgs <pcain@usgs.gov>
Date: Fri, 14 May 2021 07:29:34 -0600
Subject: [PATCH 4/7] Remove old sncl methods from factories, fix interval code
for predefined channels, support chan.loc format
---
geomagio/edge/EdgeFactory.py | 182 -------------------------
geomagio/edge/MiniSeedFactory.py | 206 -----------------------------
geomagio/edge/SNCL.py | 11 +-
geomagio/edge/SNCLFactory.py | 27 ++--
test/edge_test/SNCLFactory_test.py | 10 ++
5 files changed, 36 insertions(+), 400 deletions(-)
diff --git a/geomagio/edge/EdgeFactory.py b/geomagio/edge/EdgeFactory.py
index bed9e5ec8..a4294c79b 100644
--- a/geomagio/edge/EdgeFactory.py
+++ b/geomagio/edge/EdgeFactory.py
@@ -278,188 +278,6 @@ class EdgeFactory(TimeseriesFactory):
trace.data = numpy.ma.masked_invalid(trace.data)
return stream
- def _get_edge_channel(self, observatory, channel, type, interval):
- """get edge channel.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F, X, Y, G} or
- any appropriate edge channel, ie MSD, MGD, HGD.
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {minute, second}
-
- Returns
- -------
- edge_channel
- {MVH, MVE, MVD, MGD etc}
- """
- edge_interval_code = self._get_interval_code(interval)
- edge_channel = None
-
- # If form is chan.loc, return chan (left) portion.
- # Allows specific chan/loc selection.
- if channel.find(".") >= 0:
- tmplist = channel.split(".")
- return tmplist[0].strip()
-
- if channel == "D":
- edge_channel = edge_interval_code + "VD"
- elif channel == "E":
- edge_channel = edge_interval_code + "VE"
- elif channel == "F":
- edge_channel = edge_interval_code + "SF"
- elif channel == "H":
- edge_channel = edge_interval_code + "VH"
- elif channel == "Z":
- edge_channel = edge_interval_code + "VZ"
- elif channel == "G":
- edge_channel = edge_interval_code + "SG"
- elif channel == "X":
- edge_channel = edge_interval_code + "VX"
- elif channel == "Y":
- edge_channel = edge_interval_code + "VY"
- elif channel == "E-E":
- edge_channel = edge_interval_code + "QE"
- elif channel == "E-N":
- edge_channel = edge_interval_code + "QN"
- elif channel == "DIST":
- edge_channel = edge_interval_code + "DT"
- elif channel == "DST":
- edge_channel = edge_interval_code + "GD"
- elif channel == "SQ":
- edge_channel = edge_interval_code + "SQ"
- elif channel == "SV":
- edge_channel = edge_interval_code + "SV"
- else:
- edge_channel = channel
- return edge_channel
-
- def _get_edge_location(self, observatory, channel, type, interval):
- """get edge location.
-
- The edge location code is currently determined by the type
- passed in.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F}
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {minute, second}
-
- Returns
- -------
- location
- returns an edge location code
- """
- location = None
-
- # If form is chan.loc, return loc (right) portion
- # Allows specific chan/loc selection.
- if channel.find(".") >= 0:
- tmplist = channel.split(".")
- return tmplist[1].strip()
-
- if self.locationCode is not None:
- location = self.locationCode
- else:
- if type == "variation" or type == "reported":
- location = "R0"
- elif type == "adjusted" or type == "provisional":
- location = "A0"
- elif type == "quasi-definitive":
- location = "Q0"
- elif type == "definitive":
- location = "D0"
- elif len(type) == 2:
- location = type
- return location
-
- def _get_edge_network(self, observatory, channel, type, interval):
- """get edge network code.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F}
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {minute, second}
-
- Returns
- -------
- network
- always NT
- """
- return "NT"
-
- def _get_edge_station(self, observatory, channel, type, interval):
- """get edge station.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F}
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {minute, second}
-
- Returns
- -------
- station
- the observatory is returned as the station
- """
- return observatory
-
- def _get_interval_code(self, interval):
- """get edge interval code.
-
- Converts the metadata interval string, into an edge single character
- edge code.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F}
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {minute, second}
-
- Returns
- -------
- interval type
- """
- interval_code = None
- if interval == "day":
- interval_code = "D"
- elif interval == "hour":
- interval_code = "H"
- elif interval == "minute":
- interval_code = "M"
- elif interval == "second":
- interval_code = "S"
- else:
- raise TimeseriesFactoryException('Unexpected interval "%s"' % interval)
- return interval_code
-
def _get_timeseries(self, starttime, endtime, observatory, channel, type, interval):
"""get timeseries data for a single channel.
diff --git a/geomagio/edge/MiniSeedFactory.py b/geomagio/edge/MiniSeedFactory.py
index 211ccf349..133edeeae 100644
--- a/geomagio/edge/MiniSeedFactory.py
+++ b/geomagio/edge/MiniSeedFactory.py
@@ -296,212 +296,6 @@ class MiniSeedFactory(TimeseriesFactory):
trace.data = numpy.ma.masked_invalid(trace.data)
return stream
- def _get_edge_channel(self, observatory, channel, type, interval):
- """get edge channel.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F, X, Y, G} or
- any appropriate edge channel, ie MSD, MGD, HGD.
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {'day', 'hour', 'minute', 'second', 'tenhertz'}
-
- Returns
- -------
- edge_channel
- {MVH, MVE, MVD, MGD etc}
- """
- edge_interval_code = self._get_interval_code(interval)
- edge_channel = None
-
- # If form is chan.loc, return chan (left) portion.
- # Allows specific chan/loc selection.
- if channel.find(".") >= 0:
- tmplist = channel.split(".")
- return tmplist[0].strip()
-
- # see if channel name uses _ for ELEMENT_SUFFIX
- element = None
- suffix = None
- if channel.find("_") >= 0:
- element, suffix = channel.split("_")
-
- # 10Hz should be bin/volt
- if interval == "tenhertz":
- middle = None
- if suffix == "Bin":
- middle = "Y"
- elif suffix == "Volt":
- middle = "E"
- elif suffix is not None:
- raise TimeseriesFactoryException(
- 'bad channel suffix "%s", wanted "Bin" or "Volt"' % suffix
- )
- # check for expected channels
- if element in ("U", "V", "W") and middle is not None:
- return edge_interval_code + middle + element
- else:
- # unknown, assume advanced user
- return channel
-
- if suffix is not None:
- if suffix == "Dist" or suffix == "SQ" or suffix == "SV" or suffix == "DT":
- # these suffixes modify location code, but use element channel
- channel = element
- else:
- raise TimeseriesFactoryException(
- 'bad channel suffix "%s", wanted "Dist", "SQ", or "SV"' % suffix
- )
- if channel in ("D", "F", "G", "H", "U", "V", "W", "X", "Y", "Z"):
- # normal elements
- edge_channel = edge_interval_code + "F" + channel
- elif channel == "E-E":
- edge_channel = edge_interval_code + "QE"
- elif channel == "E-N":
- edge_channel = edge_interval_code + "QN"
- elif channel == "Dst4":
- edge_channel = edge_interval_code + "X4"
- elif channel == "Dst3":
- edge_channel = edge_interval_code + "X3"
- else:
- edge_channel = channel
- return edge_channel
-
- def _get_edge_location(self, observatory, channel, data_type, interval):
- """get edge location.
-
- The edge location code is currently determined by the type
- passed in.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F}
- data_type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {'day', 'hour', 'minute', 'second', 'tenhertz'}
-
- Returns
- -------
- location
- returns an edge location code
- """
- # If form is chan.loc, return loc (right) portion
- # Allows specific chan/loc selection.
- if channel.find(".") >= 0:
- tmplist = channel.split(".")
- return tmplist[1].strip()
- # factory override
- if self.locationCode is not None:
- return self.locationCode
- # determine prefix
- location_prefix = "R"
- if data_type == "variation" or data_type == "reported":
- location_prefix = "R"
- elif data_type == "adjusted" or data_type == "provisional":
- location_prefix = "A"
- elif data_type == "quasi-definitive":
- location_prefix = "Q"
- elif data_type == "definitive":
- location_prefix = "D"
- # determine suffix
- location_suffix = "0"
- if channel.find("_") >= 0:
- _, suffix = channel.split("_")
- if suffix == "Dist":
- location_suffix = "D"
- elif suffix == "SQ":
- location_suffix = "Q"
- elif suffix == "SV":
- location_suffix = "V"
- elif suffix == "DT":
- location_suffix = "R"
- elif suffix not in ("Bin", "Volt"):
- raise TimeseriesFactoryException(
- 'bad channel suffix "%s", wanted "Dist", "SQ", or "SV"' % suffix
- )
- return location_prefix + location_suffix
-
- def _get_edge_network(self, observatory, channel, type, interval):
- """get edge network code.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F}
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {'day', 'hour', 'minute', 'second', 'tenhertz'}
-
- Returns
- -------
- network
- always NT
- """
- return "NT"
-
- def _get_edge_station(self, observatory, channel, type, interval):
- """get edge station.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F}
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {'day', 'hour', 'minute', 'second', 'tenhertz'}
-
- Returns
- -------
- station
- the observatory is returned as the station
- """
- return observatory
-
- def _get_interval_code(self, interval):
- """get edge interval code.
-
- Converts the metadata interval string, into an edge single character
- edge code.
-
- Parameters
- ----------
- interval : str
- interval length {'day', 'hour', 'minute', 'second', 'tenhertz'}
-
- Returns
- -------
- interval type
- """
- interval_code = None
- if interval == "day":
- interval_code = "P"
- elif interval == "hour":
- interval_code = "R"
- elif interval == "minute":
- interval_code = "U"
- elif interval == "second":
- interval_code = "L"
- elif interval == "tenhertz":
- interval_code = "B"
- else:
- raise TimeseriesFactoryException('Unexpected interval "%s"' % interval)
- return interval_code
-
def _get_timeseries(self, starttime, endtime, observatory, channel, type, interval):
"""get timeseries data for a single channel.
diff --git a/geomagio/edge/SNCL.py b/geomagio/edge/SNCL.py
index 9e9e92b55..df75fa3a4 100644
--- a/geomagio/edge/SNCL.py
+++ b/geomagio/edge/SNCL.py
@@ -63,6 +63,7 @@ class SNCL(BaseModel):
@property
def data_type(self) -> str:
+ """Translates beginning of location code to data type"""
location_start = self.location[0]
if location_start == "R":
return "variation"
@@ -76,12 +77,13 @@ class SNCL(BaseModel):
@property
def element(self) -> str:
- element = self.__get_predefined_element()
- element = element or self.__get_element()
- return element
+ predefined_element = self.__check_predefined_element()
+ element = self.__get_element()
+ return predefined_element or element
@property
def interval(self) -> str:
+ """Translates beginning of channel to interval"""
interval_conversions = INTERVAL_CONVERSIONS[self.data_format]
interval_code_conversions = {
interval_conversions[key]: key for key in interval_conversions.keys()
@@ -93,6 +95,7 @@ class SNCL(BaseModel):
raise ValueError(f"Unexcepted interval code: {channel_start}")
def __get_element(self):
+ """Translates channel/location to element"""
element_start = self.channel[2]
channel = self.channel
channel_middle = channel[1]
@@ -115,7 +118,7 @@ class SNCL(BaseModel):
element_end = ""
return element_start + element_end
- def __get_predefined_element(self) -> Optional[str]:
+ def __check_predefined_element(self) -> Optional[str]:
channel = self.channel
channel_end = channel[1:]
if channel_end in CHANNEL_CONVERSIONS:
diff --git a/geomagio/edge/SNCLFactory.py b/geomagio/edge/SNCLFactory.py
index 4e677c0af..4b76fa157 100644
--- a/geomagio/edge/SNCLFactory.py
+++ b/geomagio/edge/SNCLFactory.py
@@ -23,10 +23,12 @@ class SNCLFactory(object):
)
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_predefined_channel(element=element)
- channel_end = channel_end or self.__get_channel_end(element=element)
- return channel_start + channel_end
+ 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)
@@ -39,11 +41,18 @@ class SNCLFactory(object):
except:
raise ValueError(f"Unexpected interval: {interval}")
- def __get_predefined_channel(self, element: str) -> Optional[str]:
- if len(element) == 3 and "-" not in element and element != "DST":
- return element[1:]
- elif element in ELEMENT_CONVERSIONS:
- return ELEMENT_CONVERSIONS[element]
+ def __check_predefined_channel(self, element: str, interval: str) -> Optional[str]:
+ if element in ELEMENT_CONVERSIONS:
+ return (
+ self.__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
@@ -61,6 +70,7 @@ class SNCLFactory(object):
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":
@@ -72,6 +82,7 @@ class SNCLFactory(object):
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 "_Dist" in element:
diff --git a/test/edge_test/SNCLFactory_test.py b/test/edge_test/SNCLFactory_test.py
index d12ab5f47..44f2d0de0 100644
--- a/test/edge_test/SNCLFactory_test.py
+++ b/test/edge_test/SNCLFactory_test.py
@@ -34,9 +34,14 @@ def test_get_channel():
assert factory.get_channel(element="E-N", interval="minute") == "UQN"
assert factory.get_channel(element="SQ", interval="minute") == "USQ"
assert factory.get_channel(element="SV", interval="minute") == "USV"
+ assert factory.get_channel(element="UK1", interval="minute") == "UK1"
+ assert factory.get_channel(element="UK2", interval="minute") == "UK2"
+ assert factory.get_channel(element="UK3", interval="minute") == "UK3"
+ assert factory.get_channel(element="UK4", interval="minute") == "UK4"
assert factory.get_channel(element="DIST", interval="minute") == "UDT"
assert factory.get_channel(element="DST", interval="minute") == "UGD"
assert factory.get_channel(element="U_Dist", interval="minute") == "UFU"
+ assert factory.get_channel(element="UK1.R0", interval="minute") == "UK1"
# test legacy format
factory = SNCLFactory(data_format="legacy")
@@ -49,9 +54,14 @@ def test_get_channel():
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="UK2", interval="minute") == "UK2"
+ assert factory.get_channel(element="UK3", interval="minute") == "UK3"
+ assert factory.get_channel(element="UK4", interval="minute") == "UK4"
assert factory.get_channel(element="DIST", interval="minute") == "MDT"
assert factory.get_channel(element="DST", interval="minute") == "MGD"
assert factory.get_channel(element="H_Dist", interval="minute") == "MVH"
+ assert factory.get_channel(element="UK1.R0", interval="minute") == "UK1"
def test_get_location():
--
GitLab
From c307eb7ca63a59bf6c06705da0fdadb36b6b22b0 Mon Sep 17 00:00:00 2001
From: pcain-usgs <pcain@usgs.gov>
Date: Mon, 17 May 2021 15:33:37 -0600
Subject: [PATCH 5/7] separate sncl models
---
geomagio/edge/EdgeFactory.py | 13 +-
geomagio/edge/LegacySNCL.py | 81 ++++++++++
geomagio/edge/MiniSeedFactory.py | 13 +-
geomagio/edge/SNCL.py | 77 +++++-----
geomagio/edge/SNCLFactory.py | 73 ++++++---
geomagio/edge/__init__.py | 2 +
test/edge_test/LegacySNCL_test.py | 169 ++++++++++++++++++++
test/edge_test/SNCLFactory_test.py | 72 ++++-----
test/edge_test/SNCL_test.py | 238 +++++------------------------
9 files changed, 421 insertions(+), 317 deletions(-)
create mode 100644 geomagio/edge/LegacySNCL.py
create mode 100644 test/edge_test/LegacySNCL_test.py
diff --git a/geomagio/edge/EdgeFactory.py b/geomagio/edge/EdgeFactory.py
index a4294c79b..af226a63a 100644
--- a/geomagio/edge/EdgeFactory.py
+++ b/geomagio/edge/EdgeFactory.py
@@ -22,7 +22,7 @@ from ..TimeseriesFactory import TimeseriesFactory
from ..TimeseriesFactoryException import TimeseriesFactoryException
from ..ObservatoryMetadata import ObservatoryMetadata
from .RawInputClient import RawInputClient
-from .SNCLFactory import SNCLFactory
+from .LegacySNCL import LegacySNCL
class EdgeFactory(TimeseriesFactory):
@@ -301,8 +301,11 @@ class EdgeFactory(TimeseriesFactory):
obspy.core.trace
timeseries trace of the requested channel data
"""
- sncl = SNCLFactory(data_format="legacy").get_sncl(
- station=observatory, data_type=type, element=channel, interval=interval
+ sncl = LegacySNCL().get_sncl(
+ station=observatory,
+ data_type=type,
+ interval=interval,
+ element=channel,
)
try:
data = self.client.get_waveforms(
@@ -399,8 +402,8 @@ class EdgeFactory(TimeseriesFactory):
-----
RawInputClient seems to only work when sockets are
"""
- sncl = SNCLFactory(data_format="legacy").get_sncl(
- station=observatory, data_type=type, element=channel, interval=interval
+ 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
new file mode 100644
index 000000000..be1c3300d
--- /dev/null
+++ b/geomagio/edge/LegacySNCL.py
@@ -0,0 +1,81 @@
+from __future__ import annotations
+from typing import Optional
+
+from .SNCL import SNCL
+
+ELEMENT_CONVERSIONS = {
+ # e-field
+ "E-E": "QE",
+ "E-N": "QN",
+ # derived indicies
+ "SQ": "SQ",
+ "SV": "SV",
+ "DIST": "DT",
+ "DST": "GD",
+}
+CHANNEL_CONVERSIONS = {
+ ELEMENT_CONVERSIONS[key]: key for key in ELEMENT_CONVERSIONS.keys()
+}
+
+
+class LegacySNCL(SNCL):
+ def get_sncl(
+ self,
+ station: str,
+ data_type: str,
+ interval: str,
+ element: str,
+ ) -> LegacySNCL:
+ from .SNCLFactory import SNCLFactory
+
+ factory = SNCLFactory(data_format="legacy")
+ 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),
+ )
+
+ @property
+ def element(self) -> str:
+ predefined_element = self.__check_predefined_element()
+ element = self.__get_element()
+ return predefined_element or element
+
+ @property
+ def interval(self) -> str:
+ channel_start = self.channel[0]
+ if channel_start == "S":
+ return "second"
+ elif channel_start == "M":
+ return "minute"
+ elif channel_start == "H":
+ return "hour"
+ elif channel_start == "D":
+ return "day"
+ raise ValueError(f"Unexcepted interval code: {channel_start}")
+
+ def __get_element(self):
+ """Translates channel/location to element"""
+ element_start = self.channel[2]
+ channel = self.channel
+ channel_middle = channel[1]
+ location_end = self.location[1]
+ if channel_middle in ["Q", "E"]:
+ element_end = "_Volt"
+ elif channel_middle == "Y":
+ element_end = "_Bin"
+ elif channel_middle == "K":
+ element_end = "_Temp"
+ elif location_end == "1":
+ element_end = "_Sat"
+ else:
+ element_end = ""
+ return element_start + element_end
+
+ def __check_predefined_element(self) -> Optional[str]:
+ channel = self.channel
+ channel_end = channel[1:]
+ if channel_end in CHANNEL_CONVERSIONS:
+ return CHANNEL_CONVERSIONS[channel_end]
+ return None
diff --git a/geomagio/edge/MiniSeedFactory.py b/geomagio/edge/MiniSeedFactory.py
index 133edeeae..f3530cdd6 100644
--- a/geomagio/edge/MiniSeedFactory.py
+++ b/geomagio/edge/MiniSeedFactory.py
@@ -23,7 +23,7 @@ from ..TimeseriesFactory import TimeseriesFactory
from ..TimeseriesFactoryException import TimeseriesFactoryException
from ..ObservatoryMetadata import ObservatoryMetadata
from .MiniSeedInputClient import MiniSeedInputClient
-from .SNCLFactory import SNCLFactory
+from .SNCL import SNCL
class MiniSeedFactory(TimeseriesFactory):
@@ -319,8 +319,8 @@ class MiniSeedFactory(TimeseriesFactory):
obspy.core.trace
timeseries trace of the requested channel data
"""
- sncl = SNCLFactory().get_sncl(
- station=observatory, data_type=type, element=channel, interval=interval
+ sncl = SNCL().get_sncl(
+ station=observatory, data_type=type, interval=interval, element=channel
)
data = self.client.get_waveforms(
sncl.network, sncl.station, sncl.location, sncl.channel, starttime, endtime
@@ -460,8 +460,11 @@ class MiniSeedFactory(TimeseriesFactory):
to_write = to_write.split()
to_write = TimeseriesUtility.unmask_stream(to_write)
# relabel channels from internal to edge conventions
- sncl = SNCLFactory().get_sncl(
- station=observatory, data_type=type, element=channel, interval=interval
+ sncl = SNCL().get_sncl(
+ station=observatory,
+ data_type=type,
+ interval=interval,
+ element=channel,
)
for trace in to_write:
trace.stats.station = sncl.station
diff --git a/geomagio/edge/SNCL.py b/geomagio/edge/SNCL.py
index df75fa3a4..395bfefe5 100644
--- a/geomagio/edge/SNCL.py
+++ b/geomagio/edge/SNCL.py
@@ -1,22 +1,8 @@
-from typing import Dict, Optional, Set
+from __future__ import annotations
+from typing import Dict, Optional
from pydantic import BaseModel
-INTERVAL_CONVERSIONS = {
- "legacy": {
- "second": "S",
- "minute": "M",
- "hour": "H",
- "day": "D",
- },
- "miniseed": {
- "tenhertz": "B",
- "second": "L",
- "minute": "U",
- "hour": "R",
- "day": "P",
- },
-}
ELEMENT_CONVERSIONS = {
# e-field
"E-E": "QE",
@@ -24,10 +10,6 @@ ELEMENT_CONVERSIONS = {
# derived indicies
"Dst3": "X3",
"Dst4": "X4",
- "SQ": "SQ",
- "SV": "SV",
- "DIST": "DT",
- "DST": "GD",
}
CHANNEL_CONVERSIONS = {
@@ -36,11 +18,27 @@ CHANNEL_CONVERSIONS = {
class SNCL(BaseModel):
- station: str
+ station: str = None
network: str = "NT"
- channel: str
- location: str
- data_format: str = "miniseed"
+ channel: str = None
+ location: str = None
+
+ def get_sncl(
+ self,
+ station: str,
+ data_type: str,
+ interval: str,
+ element: str,
+ ) -> SNCL:
+ from .SNCLFactory import SNCLFactory
+
+ factory = SNCLFactory(data_format="miniseed")
+ 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),
+ )
def parse_sncl(self) -> Dict:
return {
@@ -51,16 +49,6 @@ class SNCL(BaseModel):
"interval": self.interval,
}
- def dict(self, exclude: Set = {"data_format"}) -> dict:
- return super().dict(
- exclude=exclude,
- )
-
- def json(self, exclude: Set = {"data_format"}) -> str:
- return super().json(
- exclude=exclude,
- )
-
@property
def data_type(self) -> str:
"""Translates beginning of location code to data type"""
@@ -84,15 +72,18 @@ class SNCL(BaseModel):
@property
def interval(self) -> str:
"""Translates beginning of channel to interval"""
- interval_conversions = INTERVAL_CONVERSIONS[self.data_format]
- interval_code_conversions = {
- interval_conversions[key]: key for key in interval_conversions.keys()
- }
channel_start = self.channel[0]
- try:
- return interval_code_conversions[channel_start]
- except:
- raise ValueError(f"Unexcepted interval code: {channel_start}")
+ if channel_start == "B":
+ return "tenhertz"
+ elif channel_start == "L":
+ return "second"
+ elif channel_start == "U":
+ return "minute"
+ elif channel_start == "R":
+ return "hour"
+ elif channel_start == "P":
+ return "day"
+ raise ValueError(f"Unexcepted interval code: {channel_start}")
def __get_element(self):
"""Translates channel/location to element"""
@@ -100,7 +91,7 @@ class SNCL(BaseModel):
channel = self.channel
channel_middle = channel[1]
location_end = self.location[1]
- if channel_middle in ["Q", "E"]:
+ if channel_middle == "E":
element_end = "_Volt"
elif channel_middle == "Y":
element_end = "_Bin"
diff --git a/geomagio/edge/SNCLFactory.py b/geomagio/edge/SNCLFactory.py
index 4b76fa157..c4be5e899 100644
--- a/geomagio/edge/SNCLFactory.py
+++ b/geomagio/edge/SNCLFactory.py
@@ -1,25 +1,46 @@
-from typing import Optional
+from typing import Literal, Optional, Union
-from .SNCL import SNCL, INTERVAL_CONVERSIONS, ELEMENT_CONVERSIONS
+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: str = "miniseed"):
+ def __init__(self, data_format: Literal["miniseed", "legacy"] = "miniseed"):
self.data_format = data_format
def get_sncl(
self,
station: str,
- data_type: str,
- element: str,
- interval: str,
- network: str = "NT",
- ) -> SNCL:
- return SNCL(
- station=station,
- network=network,
- channel=self.get_channel(element=element, interval=interval),
- location=self.get_location(element=element, data_type=data_type),
+ 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:
@@ -36,16 +57,23 @@ class SNCLFactory(object):
return location_start + location_end
def __get_channel_start(self, interval: str) -> str:
+ interval_conversions = INTERVAL_CONVERSIONS[self.data_format]
try:
- return INTERVAL_CONVERSIONS[self.data_format][interval]
+ return interval_conversions[interval]
except:
raise ValueError(f"Unexpected interval: {interval}")
def __check_predefined_channel(self, element: str, interval: str) -> Optional[str]:
- if element in ELEMENT_CONVERSIONS:
+ channel_conversions = (
+ MINISEED_CONVERSIONS
+ if self.data_format == "miniseed"
+ else LEGACY_CONVERSIONS
+ )
+
+ if element in channel_conversions:
return (
self.__get_channel_start(interval=interval)
- + ELEMENT_CONVERSIONS[element]
+ + channel_conversions[element]
)
elif len(element) == 3:
return element
@@ -85,10 +113,11 @@ class SNCLFactory(object):
"""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"
+ 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 bf808a25a..77d5fee30 100644
--- a/geomagio/edge/__init__.py
+++ b/geomagio/edge/__init__.py
@@ -9,6 +9,7 @@ from .MiniSeedInputClient import MiniSeedInputClient
from .RawInputClient import RawInputClient
from .SNCL import SNCL
from .SNCLFactory import SNCLFactory
+from .LegacySNCL import LegacySNCL
__all__ = [
"EdgeFactory",
@@ -20,4 +21,5 @@ __all__ = [
"LegacySNIDE",
"SNCL",
"SNCLFactory",
+ "LegacySNCL",
]
diff --git a/test/edge_test/LegacySNCL_test.py b/test/edge_test/LegacySNCL_test.py
new file mode 100644
index 000000000..8d7233ffc
--- /dev/null
+++ b/test/edge_test/LegacySNCL_test.py
@@ -0,0 +1,169 @@
+from geomagio.edge import LegacySNCL
+
+
+def test_data_type():
+ """edge_test.LegacySNCL_test.test_data_type()"""
+ assert (
+ LegacySNCL(station="BOU", channel="LFU", location="R0").data_type == "variation"
+ )
+ assert (
+ LegacySNCL(station="BOU", channel="LFU", location="A0").data_type == "adjusted"
+ )
+ assert (
+ LegacySNCL(station="BOU", channel="LFU", location="Q0").data_type
+ == "quasi-definitive"
+ )
+ assert (
+ LegacySNCL(station="BOU", channel="LFU", location="D0").data_type
+ == "definitive"
+ )
+
+
+def test_element():
+ """edge_test.LegacySNCL_test.test_element()"""
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="MVD",
+ location="R0",
+ ).element
+ == "D"
+ )
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="MVU",
+ location="R0",
+ ).element
+ == "U"
+ )
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="MSF",
+ location="R0",
+ ).element
+ == "F"
+ )
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="MVH",
+ location="R0",
+ ).element
+ == "H"
+ )
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="MQE",
+ location="R0",
+ ).element
+ == "E-E"
+ )
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="MQN",
+ location="R0",
+ ).element
+ == "E-N"
+ )
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="MEH",
+ location="R0",
+ ).element
+ == "H_Volt"
+ )
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="MYH",
+ location="R0",
+ ).element
+ == "H_Bin"
+ )
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="MVH",
+ location="R1",
+ ).element
+ == "H_Sat"
+ )
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="MDT",
+ location="R0",
+ ).element
+ == "DIST"
+ )
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="MGD",
+ location="R0",
+ ).element
+ == "DST"
+ )
+
+
+def test_get_sncl():
+ """edge_test.LegacySNCL_test.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")
+
+
+def test_interval():
+ """edge_test.LegacySNCL_test.test_interval()"""
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="SVH",
+ location="R0",
+ data_format="legacy",
+ ).interval
+ == "second"
+ )
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="MVH",
+ location="R0",
+ data_format="legacy",
+ ).interval
+ == "minute"
+ )
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="HVH",
+ location="R0",
+ data_format="legacy",
+ ).interval
+ == "hour"
+ )
+ assert (
+ LegacySNCL(
+ station="BOU",
+ channel="DVH",
+ location="R0",
+ data_format="legacy",
+ ).interval
+ == "day"
+ )
+
+
+def test_parse_sncl():
+ """edge_test.LegacySNCL_test.test_parse_sncl()"""
+ assert LegacySNCL(station="BOU", channel="MVH", location="R0").parse_sncl() == {
+ "station": "BOU",
+ "network": "NT",
+ "data_type": "variation",
+ "element": "H",
+ "interval": "minute",
+ }
diff --git a/test/edge_test/SNCLFactory_test.py b/test/edge_test/SNCLFactory_test.py
index 44f2d0de0..0920c804c 100644
--- a/test/edge_test/SNCLFactory_test.py
+++ b/test/edge_test/SNCLFactory_test.py
@@ -1,70 +1,41 @@
-from geomagio.edge import SNCL, SNCLFactory
-
-
-def test_get_sncl():
- assert (
- SNCLFactory(data_format="miniseed").get_sncl(
- station="BOU",
- data_type="variation",
- element="UFU",
- interval="minute",
- )
- == SNCL(station="BOU", network="NT", channel="UFU", location="R0")
- )
- assert (
- SNCLFactory(data_format="legacy").get_sncl(
- station="BOU",
- data_type="variation",
- element="MVH",
- interval="minute",
- )
- == SNCL(station="BOU", network="NT", channel="MVH", location="R0")
- )
+from geomagio.edge import SNCL, SNCLFactory, LegacySNCL
def test_get_channel():
- # test miniseed format
+ """edge_test.SNCLFactory_test.test_get_channel()"""
factory = SNCLFactory(data_format="miniseed")
- assert factory.get_channel(element="D", interval="minute") == "UFD"
+ 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="minute") == "UFH"
- assert factory.get_channel(element="Dst4", interval="minute") == "UX4"
+ 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="SQ", interval="minute") == "USQ"
- assert factory.get_channel(element="SV", interval="minute") == "USV"
assert factory.get_channel(element="UK1", interval="minute") == "UK1"
- assert factory.get_channel(element="UK2", interval="minute") == "UK2"
- assert factory.get_channel(element="UK3", interval="minute") == "UK3"
- assert factory.get_channel(element="UK4", interval="minute") == "UK4"
- assert factory.get_channel(element="DIST", interval="minute") == "UDT"
- assert factory.get_channel(element="DST", interval="minute") == "UGD"
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="minute") == "MVD"
+ 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="minute") == "MVH"
- assert factory.get_channel(element="Dst4", interval="minute") == "MX4"
- assert factory.get_channel(element="Dst3", interval="minute") == "MX3"
- assert factory.get_channel(element="E-E", interval="minute") == "MQE"
+ 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="UK2", interval="minute") == "UK2"
- assert factory.get_channel(element="UK3", interval="minute") == "UK3"
- assert factory.get_channel(element="UK4", interval="minute") == "UK4"
assert factory.get_channel(element="DIST", interval="minute") == "MDT"
assert factory.get_channel(element="DST", interval="minute") == "MGD"
- assert factory.get_channel(element="H_Dist", interval="minute") == "MVH"
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"
@@ -74,3 +45,20 @@ def test_get_location():
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 7f59107d9..b2ca4a1be 100644
--- a/test/edge_test/SNCL_test.py
+++ b/test/edge_test/SNCL_test.py
@@ -2,6 +2,7 @@ from geomagio.edge import SNCL
def test_data_type():
+ """edge_test.SNCL_test.test_data_type()"""
assert SNCL(station="BOU", channel="LFU", location="R0").data_type == "variation"
assert SNCL(station="BOU", channel="LFU", location="A0").data_type == "adjusted"
assert (
@@ -11,88 +12,8 @@ def test_data_type():
assert SNCL(station="BOU", channel="LFU", location="D0").data_type == "definitive"
-def test_interval():
- # miniseed format
- assert (
- SNCL(
- station="BOU",
- channel="BEU",
- location="R0",
- ).interval
- == "tenhertz"
- )
- assert (
- SNCL(
- station="BOU",
- channel="LEU",
- location="R0",
- ).interval
- == "second"
- )
- assert (
- SNCL(
- station="BOU",
- channel="UEU",
- location="R0",
- ).interval
- == "minute"
- )
- assert (
- SNCL(
- station="BOU",
- channel="REU",
- location="R0",
- ).interval
- == "hour"
- )
- assert (
- SNCL(
- station="BOU",
- channel="PEU",
- location="R0",
- ).interval
- == "day"
- )
- # legacy format
- assert (
- SNCL(
- station="BOU",
- channel="SVH",
- location="R0",
- data_format="legacy",
- ).interval
- == "second"
- )
- assert (
- SNCL(
- station="BOU",
- channel="MVH",
- location="R0",
- data_format="legacy",
- ).interval
- == "minute"
- )
- assert (
- SNCL(
- station="BOU",
- channel="HVH",
- location="R0",
- data_format="legacy",
- ).interval
- == "hour"
- )
- assert (
- SNCL(
- station="BOU",
- channel="DVH",
- location="R0",
- data_format="legacy",
- ).interval
- == "day"
- )
-
-
def test_element():
+ """edge_test.SNCL_test.test_element()"""
assert (
SNCL(
station="BOU",
@@ -182,147 +103,64 @@ def test_element():
== "U_Sat"
)
+
+def test_get_sncl():
+ """edge_test.SNCL_test.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")
+
+
+def test_interval():
+ """edge_test.SNCL_test.test_interval()"""
assert (
SNCL(
station="BOU",
- channel="MVD",
- location="R0",
- data_format="legacy",
- ).element
- == "D"
- )
- assert (
- SNCL(
- station="BOU",
- channel="MVU",
- location="R0",
- data_format="legacy",
- ).element
- == "U"
- )
- assert (
- SNCL(
- station="BOU",
- channel="MSF",
- location="R0",
- data_format="legacy",
- ).element
- == "F"
- )
- assert (
- SNCL(
- station="BOU",
- channel="MVH",
- location="R0",
- data_format="legacy",
- ).element
- == "H"
- )
- assert (
- SNCL(
- station="BOU",
- channel="MX4",
- location="R0",
- data_format="legacy",
- ).element
- == "Dst4"
- )
- assert (
- SNCL(
- station="BOU",
- channel="MX3",
+ channel="BEU",
location="R0",
- data_format="legacy",
- ).element
- == "Dst3"
+ ).interval
+ == "tenhertz"
)
assert (
SNCL(
station="BOU",
- channel="MQE",
+ channel="LEU",
location="R0",
- data_format="legacy",
- ).element
- == "E-E"
+ ).interval
+ == "second"
)
assert (
SNCL(
station="BOU",
- channel="MQN",
+ channel="UEU",
location="R0",
- data_format="legacy",
- ).element
- == "E-N"
+ ).interval
+ == "minute"
)
assert (
SNCL(
station="BOU",
- channel="MEH",
+ channel="REU",
location="R0",
- data_format="legacy",
- ).element
- == "H_Volt"
+ ).interval
+ == "hour"
)
assert (
SNCL(
station="BOU",
- channel="MYH",
+ channel="PEU",
location="R0",
- data_format="legacy",
- ).element
- == "H_Bin"
- )
- assert (
- SNCL(
- station="BOU",
- channel="MVH",
- location="R1",
- data_format="legacy",
- ).element
- == "H_Sat"
- )
- assert (
- SNCL(
- station="BOU",
- channel="MVH",
- location="RD",
- data_format="legacy",
- ).element
- == "H_Dist"
- )
- assert (
- SNCL(
- station="BOU",
- channel="MVH",
- location="RQ",
- data_format="legacy",
- ).element
- == "H_SQ"
- )
- assert (
- SNCL(
- station="BOU",
- channel="MVH",
- location="RV",
- data_format="legacy",
- ).element
- == "H_SV"
- )
- assert (
- SNCL(
- station="BOU",
- channel="MDT",
- location="RV",
- data_format="legacy",
- ).element
- == "DIST"
- )
- assert (
- SNCL(
- station="BOU",
- channel="MGD",
- location="RV",
- data_format="legacy",
- ).element
- == "DST"
+ ).interval
+ == "day"
)
+
+
+def test_parse_sncl():
+ """edge_test.SNCL_test.test_parse_sncl()"""
+ assert SNCL(station="BOU", channel="UFU", location="R0").parse_sncl() == {
+ "station": "BOU",
+ "network": "NT",
+ "data_type": "variation",
+ "element": "U",
+ "interval": "minute",
+ }
--
GitLab
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 6/7] 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
From 712f5c99e79ba6abad39143c5b31808e39377aef Mon Sep 17 00:00:00 2001
From: pcain-usgs <pcain@usgs.gov>
Date: Wed, 16 Jun 2021 15:59:58 -0600
Subject: [PATCH 7/7] move _get_element outside of classes, chain commands
---
geomagio/edge/LegacySNCL.py | 80 ++++++++++++++++----------------
geomagio/edge/SNCL.py | 92 ++++++++++++++++++-------------------
2 files changed, 84 insertions(+), 88 deletions(-)
diff --git a/geomagio/edge/LegacySNCL.py b/geomagio/edge/LegacySNCL.py
index 4c747df3a..a2478b599 100644
--- a/geomagio/edge/LegacySNCL.py
+++ b/geomagio/edge/LegacySNCL.py
@@ -1,6 +1,6 @@
from typing import Optional
-from .SNCL import SNCL, __get_location_start
+from .SNCL import SNCL, _get_location_start
ELEMENT_CONVERSIONS = {
# e-field
@@ -36,9 +36,9 @@ class LegacySNCL(SNCL):
@property
def element(self) -> str:
- predefined_element = self.__check_predefined_element()
- element = self.__get_element()
- return predefined_element or element
+ return _check_predefined_element(channel=self.channel) or _get_element(
+ channel=self.channel, location=self.location
+ )
@property
def interval(self) -> str:
@@ -53,46 +53,25 @@ class LegacySNCL(SNCL):
return "day"
raise ValueError(f"Unexcepted interval code: {channel_start}")
- def __get_element(self):
- """Translates channel/location to element"""
- element_start = self.channel[2]
- channel = self.channel
- channel_middle = channel[1]
- location_end = self.location[1]
- if channel_middle in ["Q", "E"]:
- element_end = "_Volt"
- elif channel_middle == "Y":
- element_end = "_Bin"
- elif channel_middle == "K":
- element_end = "_Temp"
- elif location_end == "1":
- element_end = "_Sat"
- else:
- element_end = ""
- return element_start + element_end
-
- def __check_predefined_element(self) -> Optional[str]:
- channel = self.channel
- channel_end = channel[1:]
- 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)
+ return _check_predefined_channel(element=element, interval=interval) or (
+ _get_channel_start(interval=interval) + _get_channel_end(element=element)
+ )
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
+ return _get_location_start(data_type=data_type) + _get_location_end(element=element)
+
+def _check_predefined_element(channel: str) -> Optional[str]:
+ channel_end = channel[1:]
+ if channel_end in CHANNEL_CONVERSIONS:
+ return CHANNEL_CONVERSIONS[channel_end]
+ return None
-def __get_channel_start(interval: str) -> str:
+
+def _get_channel_start(interval: str) -> str:
if interval == "second":
return "S"
elif interval == "minute":
@@ -104,9 +83,28 @@ def __get_channel_start(interval: str) -> str:
raise ValueError(f" Unexcepted interval: {interval}")
-def __check_predefined_channel(element: str, interval: str) -> Optional[str]:
+def _get_element(channel: str, location: str) -> str:
+ """Translates channel/location to element"""
+ element_start = channel[2]
+ channel = channel
+ channel_middle = channel[1]
+ location_end = location[1]
+ if channel_middle in ["Q", "E"]:
+ element_end = "_Volt"
+ elif channel_middle == "Y":
+ element_end = "_Bin"
+ elif channel_middle == "K":
+ element_end = "_Temp"
+ elif location_end == "1":
+ element_end = "_Sat"
+ else:
+ element_end = ""
+ return element_start + element_end
+
+
+def _check_predefined_channel(element: str, interval: str) -> Optional[str]:
if element in ELEMENT_CONVERSIONS:
- return __get_channel_start(interval=interval) + ELEMENT_CONVERSIONS[element]
+ return _get_channel_start(interval=interval) + ELEMENT_CONVERSIONS[element]
elif len(element) == 3:
return element
# chan.loc format
@@ -117,7 +115,7 @@ def __check_predefined_channel(element: str, interval: str) -> Optional[str]:
return None
-def __get_channel_end(element: str) -> str:
+def _get_channel_end(element: str) -> str:
channel_middle = "V"
if "_Volt" in element:
channel_middle = "E"
@@ -131,7 +129,7 @@ def __get_channel_end(element: str) -> str:
return channel_middle + channel_end
-def __get_location_end(element: str) -> str:
+def _get_location_end(element: str) -> str:
"""Translates element suffix to end of location code"""
if "_Sat" in element:
return "1"
diff --git a/geomagio/edge/SNCL.py b/geomagio/edge/SNCL.py
index 3dc4b9413..de292587c 100644
--- a/geomagio/edge/SNCL.py
+++ b/geomagio/edge/SNCL.py
@@ -63,9 +63,9 @@ class SNCL(BaseModel):
@property
def element(self) -> str:
- predefined_element = self.__check_predefined_element()
- element = self.__get_element()
- return predefined_element or element
+ return _check_predefined_element(channel=self.channel) or _get_element(
+ channel=self.channel, location=self.location
+ )
@property
def interval(self) -> str:
@@ -83,52 +83,25 @@ class SNCL(BaseModel):
return "day"
raise ValueError(f"Unexcepted interval code: {channel_start}")
- def __get_element(self):
- """Translates channel/location to element"""
- element_start = self.channel[2]
- channel = self.channel
- channel_middle = channel[1]
- location_end = self.location[1]
- if channel_middle == "E":
- element_end = "_Volt"
- elif channel_middle == "Y":
- element_end = "_Bin"
- elif channel_middle == "K":
- element_end = "_Temp"
- elif location_end == "1":
- element_end = "_Sat"
- elif location_end == "D":
- element_end = "_Dist"
- elif location_end == "Q":
- element_end = "_SQ"
- elif location_end == "V":
- element_end = "_SV"
- else:
- element_end = ""
- return element_start + element_end
-
- def __check_predefined_element(self) -> Optional[str]:
- channel = self.channel
- channel_end = channel[1:]
- 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)
+ return _check_predefined_channel(element=element, interval=interval) or (
+ _get_channel_start(interval=interval) + _get_channel_end(element=element)
+ )
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
+ return _get_location_start(data_type=data_type) + _get_location_end(element=element)
+
+def _check_predefined_element(channel: str) -> Optional[str]:
+ channel_end = channel[1:]
+ if channel_end in CHANNEL_CONVERSIONS:
+ return CHANNEL_CONVERSIONS[channel_end]
+ return None
-def __get_channel_start(interval: str) -> str:
+
+def _get_channel_start(interval: str) -> str:
if interval == "tenhertz":
return "B"
if interval == "second":
@@ -142,9 +115,34 @@ def __get_channel_start(interval: str) -> str:
raise ValueError(f" Unexcepted interval: {interval}")
-def __check_predefined_channel(element: str, interval: str) -> Optional[str]:
+def _get_element(channel: str, location: str) -> str:
+ """Translates channel/location to element"""
+ element_start = channel[2]
+ channel = channel
+ channel_middle = channel[1]
+ location_end = location[1]
+ if channel_middle == "E":
+ element_end = "_Volt"
+ elif channel_middle == "Y":
+ element_end = "_Bin"
+ elif channel_middle == "K":
+ element_end = "_Temp"
+ elif location_end == "1":
+ element_end = "_Sat"
+ elif location_end == "D":
+ element_end = "_Dist"
+ elif location_end == "Q":
+ element_end = "_SQ"
+ elif location_end == "V":
+ element_end = "_SV"
+ else:
+ element_end = ""
+ return element_start + element_end
+
+
+def _check_predefined_channel(element: str, interval: str) -> Optional[str]:
if element in ELEMENT_CONVERSIONS:
- return __get_channel_start(interval=interval) + ELEMENT_CONVERSIONS[element]
+ return _get_channel_start(interval=interval) + ELEMENT_CONVERSIONS[element]
elif len(element) == 3:
return element
# chan.loc format
@@ -155,7 +153,7 @@ def __check_predefined_channel(element: str, interval: str) -> Optional[str]:
return None
-def __get_channel_end(element: str) -> str:
+def _get_channel_end(element: str) -> str:
channel_middle = "F"
if "_Volt" in element:
channel_middle = "E"
@@ -167,7 +165,7 @@ def __get_channel_end(element: str) -> str:
return channel_middle + channel_end
-def __get_location_start(data_type: str) -> str:
+def _get_location_start(data_type: str) -> str:
"""Translates data type to beginning of location code"""
if data_type == "variation":
return "R"
@@ -180,7 +178,7 @@ def __get_location_start(data_type: str) -> str:
raise ValueError(f"Unexpected data type: {data_type}")
-def __get_location_end(element: str) -> str:
+def _get_location_end(element: str) -> str:
"""Translates element suffix to end of location code"""
if "_Sat" in element:
return "1"
--
GitLab