diff --git a/geomagio/edge/SNCL.py b/geomagio/edge/SNCL.py
new file mode 100644
index 0000000000000000000000000000000000000000..20a0a885fcdbf411d7cd5f5f4e0c9ae40d92e546
--- /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 0000000000000000000000000000000000000000..4e677c0afb97504d0be369733e8afba41ee1482f
--- /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 2a53748a195983a3dc737b87731330584f10c739..bf808a25a139d19e517045c80229950d7a8c34e1 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 5bfa4ff2ee186cd832a70623fd6b8604ceee3aea..0000000000000000000000000000000000000000
--- 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 0000000000000000000000000000000000000000..d12ab5f4776672cabdd35c198dde441db6888ddf
--- /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 0000000000000000000000000000000000000000..7f59107d9b9d49a2d096c583a8e28ef4c39496bb
--- /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"
+ )