diff --git a/geomagio/edge/EdgeFactory.py b/geomagio/edge/EdgeFactory.py
index e0f1091c59cab35a47b07ed2b5150641cf847440..1190dbb167356bed32e28dcb11bc9e4bbbc33c6e 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 .LegacySNCL import LegacySNCL
class EdgeFactory(TimeseriesFactory):
@@ -277,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.
@@ -482,13 +301,20 @@ 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 = LegacySNCL.get_sncl(
+ station=observatory,
+ data_type=type,
+ interval=interval,
+ element=channel,
+ )
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 +332,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 +402,12 @@ 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 = LegacySNCL.get_sncl(
+ station=observatory,
+ data_type=type,
+ interval=interval,
+ element=channel,
+ )
now = obspy.core.UTCDateTime(datetime.utcnow())
if ((now - endtime) > 864000) and (self.cwbport > 0):
@@ -590,7 +418,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/LegacySNCL.py b/geomagio/edge/LegacySNCL.py
new file mode 100644
index 0000000000000000000000000000000000000000..a2478b599a145f2e95b1b37f99269b7b8932eb8a
--- /dev/null
+++ b/geomagio/edge/LegacySNCL.py
@@ -0,0 +1,136 @@
+from typing import Optional
+
+from .SNCL import SNCL, _get_location_start
+
+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):
+ @classmethod
+ def get_sncl(
+ cls,
+ data_type: str,
+ element: str,
+ interval: str,
+ station: str,
+ network: str = "NT",
+ ) -> "LegacySNCL":
+ return LegacySNCL(
+ station=station,
+ network=network,
+ channel=get_channel(element=element, interval=interval),
+ location=get_location(element=element, data_type=data_type),
+ )
+
+ @property
+ def element(self) -> str:
+ return _check_predefined_element(channel=self.channel) or _get_element(
+ channel=self.channel, location=self.location
+ )
+
+ @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_channel(element: str, interval: str) -> str:
+ 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:
+ 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:
+ 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 _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]
+ 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 99c3b9eec31e63a125131b6c0f65933220b13822..ec8b54692693e6f033326274e39deab0d51e719f 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 .SNCL import SNCL
class MiniSeedFactory(TimeseriesFactory):
@@ -295,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.
@@ -524,12 +319,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 = SNCL.get_sncl(
+ station=observatory, data_type=type, interval=interval, element=channel
+ )
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 +334,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 +460,17 @@ 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 = SNCL.get_sncl(
+ station=observatory,
+ data_type=type,
+ interval=interval,
+ element=channel,
+ )
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/geomagio/edge/SNCL.py b/geomagio/edge/SNCL.py
new file mode 100644
index 0000000000000000000000000000000000000000..de292587c098ea3866784dcc149784be5c37f046
--- /dev/null
+++ b/geomagio/edge/SNCL.py
@@ -0,0 +1,191 @@
+from typing import Dict, Optional
+
+from pydantic import BaseModel
+
+ELEMENT_CONVERSIONS = {
+ # e-field
+ "E-E": "QE",
+ "E-N": "QN",
+ # derived indicies
+ "Dst3": "X3",
+ "Dst4": "X4",
+}
+
+CHANNEL_CONVERSIONS = {
+ ELEMENT_CONVERSIONS[key]: key for key in ELEMENT_CONVERSIONS.keys()
+}
+
+
+class SNCL(BaseModel):
+ station: str
+ network: str = "NT"
+ channel: str
+ location: str
+
+ @classmethod
+ def get_sncl(
+ cls,
+ data_type: str,
+ element: str,
+ interval: str,
+ station: str,
+ network: str = "NT",
+ ) -> "SNCL":
+ return SNCL(
+ station=station,
+ network=network,
+ channel=get_channel(element=element, interval=interval),
+ location=get_location(element=element, data_type=data_type),
+ )
+
+ def parse_sncl(self) -> Dict:
+ return {
+ "station": self.station,
+ "network": self.network,
+ "data_type": self.data_type,
+ "element": self.element,
+ "interval": self.interval,
+ }
+
+ @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"
+ 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:
+ return _check_predefined_element(channel=self.channel) or _get_element(
+ channel=self.channel, location=self.location
+ )
+
+ @property
+ def interval(self) -> str:
+ """Translates beginning of channel to interval"""
+ channel_start = self.channel[0]
+ 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_channel(element: str, interval: str) -> str:
+ 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:
+ 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:
+ 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 _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]
+ 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/__init__.py b/geomagio/edge/__init__.py
index 2a53748a195983a3dc737b87731330584f10c739..45b3d49c3ac29b3dd213d0f83cf1017ea5d95549 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 .LegacySNCL import LegacySNCL
__all__ = [
"EdgeFactory",
@@ -14,4 +16,6 @@ __all__ = [
"MiniSeedFactory",
"MiniSeedInputClient",
"RawInputClient",
+ "LegacySNCL",
+ "SNCL",
]
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/EdgeFactory_test.py b/test/edge_test/EdgeFactory_test.py
index fca38a8c8df6fb8cd08a7717ff2114f5e7f5daf3..6b524fb318b3fad25afdee0613255d4551ff8aba 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/LegacySNCL_test.py b/test/edge_test/LegacySNCL_test.py
new file mode 100644
index 0000000000000000000000000000000000000000..adbcc8a8388ba85a543cd075d2e3bccc2c0e5202
--- /dev/null
+++ b/test/edge_test/LegacySNCL_test.py
@@ -0,0 +1,193 @@
+from geomagio.edge.LegacySNCL import LegacySNCL, get_channel, get_location
+
+
+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_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(
+ 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/MiniSeedFactory_test.py b/test/edge_test/MiniSeedFactory_test.py
index a0c57869e7ff0ffcea37f7c6f5c98842347e2492..bd1367d37ad3487522ddad7f5df879c3c37ad5f5 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.
diff --git a/test/edge_test/SNCL_test.py b/test/edge_test/SNCL_test.py
new file mode 100644
index 0000000000000000000000000000000000000000..ea92aa524602cff70dd8688cdd2ed14ebf758c04
--- /dev/null
+++ b/test/edge_test/SNCL_test.py
@@ -0,0 +1,196 @@
+from geomagio.edge.SNCL import SNCL, get_channel, get_location
+
+
+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 (
+ SNCL(station="BOU", channel="LFU", location="Q0").data_type
+ == "quasi-definitive"
+ )
+ assert SNCL(station="BOU", channel="LFU", location="D0").data_type == "definitive"
+
+
+def test_element():
+ """edge_test.SNCL_test.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"
+ )
+
+
+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(
+ 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="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"
+ )
+
+
+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",
+ }