Skip to content
Snippets Groups Projects
Commit c307eb7c authored by Cain, Payton David's avatar Cain, Payton David
Browse files

separate sncl models

parent 39b9512b
No related branches found
No related tags found
2 merge requests!146Release CMO metadata to production,!96SNCL/SNCLFactory
...@@ -22,7 +22,7 @@ from ..TimeseriesFactory import TimeseriesFactory ...@@ -22,7 +22,7 @@ from ..TimeseriesFactory import TimeseriesFactory
from ..TimeseriesFactoryException import TimeseriesFactoryException from ..TimeseriesFactoryException import TimeseriesFactoryException
from ..ObservatoryMetadata import ObservatoryMetadata from ..ObservatoryMetadata import ObservatoryMetadata
from .RawInputClient import RawInputClient from .RawInputClient import RawInputClient
from .SNCLFactory import SNCLFactory from .LegacySNCL import LegacySNCL
class EdgeFactory(TimeseriesFactory): class EdgeFactory(TimeseriesFactory):
...@@ -301,8 +301,11 @@ class EdgeFactory(TimeseriesFactory): ...@@ -301,8 +301,11 @@ class EdgeFactory(TimeseriesFactory):
obspy.core.trace obspy.core.trace
timeseries trace of the requested channel data timeseries trace of the requested channel data
""" """
sncl = SNCLFactory(data_format="legacy").get_sncl( sncl = LegacySNCL().get_sncl(
station=observatory, data_type=type, element=channel, interval=interval station=observatory,
data_type=type,
interval=interval,
element=channel,
) )
try: try:
data = self.client.get_waveforms( data = self.client.get_waveforms(
...@@ -399,8 +402,8 @@ class EdgeFactory(TimeseriesFactory): ...@@ -399,8 +402,8 @@ class EdgeFactory(TimeseriesFactory):
----- -----
RawInputClient seems to only work when sockets are RawInputClient seems to only work when sockets are
""" """
sncl = SNCLFactory(data_format="legacy").get_sncl( sncl = LegacySNCL().get_sncl(
station=observatory, data_type=type, element=channel, interval=interval station=observatory, data_type=type, interval=interval, element=channel
) )
now = obspy.core.UTCDateTime(datetime.utcnow()) now = obspy.core.UTCDateTime(datetime.utcnow())
......
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
...@@ -23,7 +23,7 @@ from ..TimeseriesFactory import TimeseriesFactory ...@@ -23,7 +23,7 @@ from ..TimeseriesFactory import TimeseriesFactory
from ..TimeseriesFactoryException import TimeseriesFactoryException from ..TimeseriesFactoryException import TimeseriesFactoryException
from ..ObservatoryMetadata import ObservatoryMetadata from ..ObservatoryMetadata import ObservatoryMetadata
from .MiniSeedInputClient import MiniSeedInputClient from .MiniSeedInputClient import MiniSeedInputClient
from .SNCLFactory import SNCLFactory from .SNCL import SNCL
class MiniSeedFactory(TimeseriesFactory): class MiniSeedFactory(TimeseriesFactory):
...@@ -319,8 +319,8 @@ class MiniSeedFactory(TimeseriesFactory): ...@@ -319,8 +319,8 @@ class MiniSeedFactory(TimeseriesFactory):
obspy.core.trace obspy.core.trace
timeseries trace of the requested channel data timeseries trace of the requested channel data
""" """
sncl = SNCLFactory().get_sncl( sncl = SNCL().get_sncl(
station=observatory, data_type=type, element=channel, interval=interval station=observatory, data_type=type, interval=interval, element=channel
) )
data = self.client.get_waveforms( data = self.client.get_waveforms(
sncl.network, sncl.station, sncl.location, sncl.channel, starttime, endtime sncl.network, sncl.station, sncl.location, sncl.channel, starttime, endtime
...@@ -460,8 +460,11 @@ class MiniSeedFactory(TimeseriesFactory): ...@@ -460,8 +460,11 @@ class MiniSeedFactory(TimeseriesFactory):
to_write = to_write.split() to_write = to_write.split()
to_write = TimeseriesUtility.unmask_stream(to_write) to_write = TimeseriesUtility.unmask_stream(to_write)
# relabel channels from internal to edge conventions # relabel channels from internal to edge conventions
sncl = SNCLFactory().get_sncl( sncl = SNCL().get_sncl(
station=observatory, data_type=type, element=channel, interval=interval station=observatory,
data_type=type,
interval=interval,
element=channel,
) )
for trace in to_write: for trace in to_write:
trace.stats.station = sncl.station trace.stats.station = sncl.station
......
from typing import Dict, Optional, Set from __future__ import annotations
from typing import Dict, Optional
from pydantic import BaseModel 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 = { ELEMENT_CONVERSIONS = {
# e-field # e-field
"E-E": "QE", "E-E": "QE",
...@@ -24,10 +10,6 @@ ELEMENT_CONVERSIONS = { ...@@ -24,10 +10,6 @@ ELEMENT_CONVERSIONS = {
# derived indicies # derived indicies
"Dst3": "X3", "Dst3": "X3",
"Dst4": "X4", "Dst4": "X4",
"SQ": "SQ",
"SV": "SV",
"DIST": "DT",
"DST": "GD",
} }
CHANNEL_CONVERSIONS = { CHANNEL_CONVERSIONS = {
...@@ -36,11 +18,27 @@ CHANNEL_CONVERSIONS = { ...@@ -36,11 +18,27 @@ CHANNEL_CONVERSIONS = {
class SNCL(BaseModel): class SNCL(BaseModel):
station: str station: str = None
network: str = "NT" network: str = "NT"
channel: str channel: str = None
location: str location: str = None
data_format: str = "miniseed"
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: def parse_sncl(self) -> Dict:
return { return {
...@@ -51,16 +49,6 @@ class SNCL(BaseModel): ...@@ -51,16 +49,6 @@ class SNCL(BaseModel):
"interval": self.interval, "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 @property
def data_type(self) -> str: def data_type(self) -> str:
"""Translates beginning of location code to data type""" """Translates beginning of location code to data type"""
...@@ -84,15 +72,18 @@ class SNCL(BaseModel): ...@@ -84,15 +72,18 @@ class SNCL(BaseModel):
@property @property
def interval(self) -> str: def interval(self) -> str:
"""Translates beginning of channel to interval""" """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] channel_start = self.channel[0]
try: if channel_start == "B":
return interval_code_conversions[channel_start] return "tenhertz"
except: elif channel_start == "L":
raise ValueError(f"Unexcepted interval code: {channel_start}") 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): def __get_element(self):
"""Translates channel/location to element""" """Translates channel/location to element"""
...@@ -100,7 +91,7 @@ class SNCL(BaseModel): ...@@ -100,7 +91,7 @@ class SNCL(BaseModel):
channel = self.channel channel = self.channel
channel_middle = channel[1] channel_middle = channel[1]
location_end = self.location[1] location_end = self.location[1]
if channel_middle in ["Q", "E"]: if channel_middle == "E":
element_end = "_Volt" element_end = "_Volt"
elif channel_middle == "Y": elif channel_middle == "Y":
element_end = "_Bin" element_end = "_Bin"
......
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): class SNCLFactory(object):
def __init__(self, data_format: str = "miniseed"): def __init__(self, data_format: Literal["miniseed", "legacy"] = "miniseed"):
self.data_format = data_format self.data_format = data_format
def get_sncl( def get_sncl(
self, self,
station: str, station: str,
data_type: str, network: str,
element: str, channel: str,
interval: str, location: str,
network: str = "NT", ) -> Union[SNCL, LegacySNCL]:
) -> SNCL: sncl_params = {
return SNCL( "station": station,
station=station, "network": network,
network=network, "channel": channel,
channel=self.get_channel(element=element, interval=interval), "location": location,
location=self.get_location(element=element, data_type=data_type), }
return (
SNCL(**sncl_params)
if self.data_format == "miniseed"
else LegacySNCL(**sncl_params)
) )
def get_channel(self, element: str, interval: str) -> str: def get_channel(self, element: str, interval: str) -> str:
...@@ -36,16 +57,23 @@ class SNCLFactory(object): ...@@ -36,16 +57,23 @@ class SNCLFactory(object):
return location_start + location_end return location_start + location_end
def __get_channel_start(self, interval: str) -> str: def __get_channel_start(self, interval: str) -> str:
interval_conversions = INTERVAL_CONVERSIONS[self.data_format]
try: try:
return INTERVAL_CONVERSIONS[self.data_format][interval] return interval_conversions[interval]
except: except:
raise ValueError(f"Unexpected interval: {interval}") raise ValueError(f"Unexpected interval: {interval}")
def __check_predefined_channel(self, element: str, interval: str) -> Optional[str]: 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 ( return (
self.__get_channel_start(interval=interval) self.__get_channel_start(interval=interval)
+ ELEMENT_CONVERSIONS[element] + channel_conversions[element]
) )
elif len(element) == 3: elif len(element) == 3:
return element return element
...@@ -85,10 +113,11 @@ class SNCLFactory(object): ...@@ -85,10 +113,11 @@ class SNCLFactory(object):
"""Translates element suffix to end of location code""" """Translates element suffix to end of location code"""
if "_Sat" in element: if "_Sat" in element:
return "1" return "1"
if "_Dist" in element: if self.data_format == "miniseed":
return "D" if "_Dist" in element:
if "_SQ" in element: return "D"
return "Q" if "_SQ" in element:
if "_SV" in element: return "Q"
return "V" if "_SV" in element:
return "V"
return "0" return "0"
...@@ -9,6 +9,7 @@ from .MiniSeedInputClient import MiniSeedInputClient ...@@ -9,6 +9,7 @@ from .MiniSeedInputClient import MiniSeedInputClient
from .RawInputClient import RawInputClient from .RawInputClient import RawInputClient
from .SNCL import SNCL from .SNCL import SNCL
from .SNCLFactory import SNCLFactory from .SNCLFactory import SNCLFactory
from .LegacySNCL import LegacySNCL
__all__ = [ __all__ = [
"EdgeFactory", "EdgeFactory",
...@@ -20,4 +21,5 @@ __all__ = [ ...@@ -20,4 +21,5 @@ __all__ = [
"LegacySNIDE", "LegacySNIDE",
"SNCL", "SNCL",
"SNCLFactory", "SNCLFactory",
"LegacySNCL",
] ]
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",
}
from geomagio.edge import SNCL, SNCLFactory from geomagio.edge import SNCL, SNCLFactory, LegacySNCL
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(): def test_get_channel():
# test miniseed format """edge_test.SNCLFactory_test.test_get_channel()"""
factory = SNCLFactory(data_format="miniseed") 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="F", interval="minute") == "UFF"
assert factory.get_channel(element="H", interval="minute") == "UFH" assert factory.get_channel(element="H", interval="hour") == "RFH"
assert factory.get_channel(element="Dst4", interval="minute") == "UX4" assert factory.get_channel(element="Dst4", interval="day") == "PX4"
assert factory.get_channel(element="Dst3", interval="minute") == "UX3" 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-E", interval="minute") == "UQE"
assert factory.get_channel(element="E-N", interval="minute") == "UQN" 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="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_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" assert factory.get_channel(element="UK1.R0", interval="minute") == "UK1"
# test legacy format # test legacy format
factory = SNCLFactory(data_format="legacy") 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="F", interval="minute") == "MSF"
assert factory.get_channel(element="H", interval="minute") == "MVH" assert factory.get_channel(element="H", interval="hour") == "HVH"
assert factory.get_channel(element="Dst4", interval="minute") == "MX4" assert factory.get_channel(element="E-E", interval="day") == "DQE"
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="E-N", interval="minute") == "MQN"
assert factory.get_channel(element="SQ", interval="minute") == "MSQ" assert factory.get_channel(element="SQ", interval="minute") == "MSQ"
assert factory.get_channel(element="SV", interval="minute") == "MSV" assert factory.get_channel(element="SV", interval="minute") == "MSV"
assert factory.get_channel(element="UK1", interval="minute") == "UK1" 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="DIST", interval="minute") == "MDT"
assert factory.get_channel(element="DST", interval="minute") == "MGD" 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" assert factory.get_channel(element="UK1.R0", interval="minute") == "UK1"
def test_get_location(): def test_get_location():
"""edge_test.SNCLFactory_test.test_get_location()"""
factory = SNCLFactory(data_format="miniseed") factory = SNCLFactory(data_format="miniseed")
assert factory.get_location(element="D", data_type="variation") == "R0" 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="adjusted") == "A0"
...@@ -74,3 +45,20 @@ def test_get_location(): ...@@ -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_Dist", data_type="variation") == "RD"
assert factory.get_location(element="D_SQ", data_type="variation") == "RQ" assert factory.get_location(element="D_SQ", data_type="variation") == "RQ"
assert factory.get_location(element="D_SV", data_type="variation") == "RV" 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")
...@@ -2,6 +2,7 @@ from geomagio.edge import SNCL ...@@ -2,6 +2,7 @@ from geomagio.edge import SNCL
def test_data_type(): 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="R0").data_type == "variation"
assert SNCL(station="BOU", channel="LFU", location="A0").data_type == "adjusted" assert SNCL(station="BOU", channel="LFU", location="A0").data_type == "adjusted"
assert ( assert (
...@@ -11,88 +12,8 @@ def test_data_type(): ...@@ -11,88 +12,8 @@ def test_data_type():
assert SNCL(station="BOU", channel="LFU", location="D0").data_type == "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(): def test_element():
"""edge_test.SNCL_test.test_element()"""
assert ( assert (
SNCL( SNCL(
station="BOU", station="BOU",
...@@ -182,147 +103,64 @@ def test_element(): ...@@ -182,147 +103,64 @@ def test_element():
== "U_Sat" == "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 ( assert (
SNCL( SNCL(
station="BOU", station="BOU",
channel="MVD", channel="BEU",
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", location="R0",
data_format="legacy", ).interval
).element == "tenhertz"
== "Dst3"
) )
assert ( assert (
SNCL( SNCL(
station="BOU", station="BOU",
channel="MQE", channel="LEU",
location="R0", location="R0",
data_format="legacy", ).interval
).element == "second"
== "E-E"
) )
assert ( assert (
SNCL( SNCL(
station="BOU", station="BOU",
channel="MQN", channel="UEU",
location="R0", location="R0",
data_format="legacy", ).interval
).element == "minute"
== "E-N"
) )
assert ( assert (
SNCL( SNCL(
station="BOU", station="BOU",
channel="MEH", channel="REU",
location="R0", location="R0",
data_format="legacy", ).interval
).element == "hour"
== "H_Volt"
) )
assert ( assert (
SNCL( SNCL(
station="BOU", station="BOU",
channel="MYH", channel="PEU",
location="R0", location="R0",
data_format="legacy", ).interval
).element == "day"
== "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"
) )
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",
}
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment