diff --git a/geomagio/edge/EdgeFactory.py b/geomagio/edge/EdgeFactory.py
index bed9e5ec81482e83bb6c4649bee29516ab5f9e28..a4294c79b553aa34c677afb6e48192efcb0a146e 100644
--- a/geomagio/edge/EdgeFactory.py
+++ b/geomagio/edge/EdgeFactory.py
@@ -278,188 +278,6 @@ class EdgeFactory(TimeseriesFactory):
trace.data = numpy.ma.masked_invalid(trace.data)
return stream
- def _get_edge_channel(self, observatory, channel, type, interval):
- """get edge channel.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F, X, Y, G} or
- any appropriate edge channel, ie MSD, MGD, HGD.
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {minute, second}
-
- Returns
- -------
- edge_channel
- {MVH, MVE, MVD, MGD etc}
- """
- edge_interval_code = self._get_interval_code(interval)
- edge_channel = None
-
- # If form is chan.loc, return chan (left) portion.
- # Allows specific chan/loc selection.
- if channel.find(".") >= 0:
- tmplist = channel.split(".")
- return tmplist[0].strip()
-
- if channel == "D":
- edge_channel = edge_interval_code + "VD"
- elif channel == "E":
- edge_channel = edge_interval_code + "VE"
- elif channel == "F":
- edge_channel = edge_interval_code + "SF"
- elif channel == "H":
- edge_channel = edge_interval_code + "VH"
- elif channel == "Z":
- edge_channel = edge_interval_code + "VZ"
- elif channel == "G":
- edge_channel = edge_interval_code + "SG"
- elif channel == "X":
- edge_channel = edge_interval_code + "VX"
- elif channel == "Y":
- edge_channel = edge_interval_code + "VY"
- elif channel == "E-E":
- edge_channel = edge_interval_code + "QE"
- elif channel == "E-N":
- edge_channel = edge_interval_code + "QN"
- elif channel == "DIST":
- edge_channel = edge_interval_code + "DT"
- elif channel == "DST":
- edge_channel = edge_interval_code + "GD"
- elif channel == "SQ":
- edge_channel = edge_interval_code + "SQ"
- elif channel == "SV":
- edge_channel = edge_interval_code + "SV"
- else:
- edge_channel = channel
- return edge_channel
-
- def _get_edge_location(self, observatory, channel, type, interval):
- """get edge location.
-
- The edge location code is currently determined by the type
- passed in.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F}
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {minute, second}
-
- Returns
- -------
- location
- returns an edge location code
- """
- location = None
-
- # If form is chan.loc, return loc (right) portion
- # Allows specific chan/loc selection.
- if channel.find(".") >= 0:
- tmplist = channel.split(".")
- return tmplist[1].strip()
-
- if self.locationCode is not None:
- location = self.locationCode
- else:
- if type == "variation" or type == "reported":
- location = "R0"
- elif type == "adjusted" or type == "provisional":
- location = "A0"
- elif type == "quasi-definitive":
- location = "Q0"
- elif type == "definitive":
- location = "D0"
- elif len(type) == 2:
- location = type
- return location
-
- def _get_edge_network(self, observatory, channel, type, interval):
- """get edge network code.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F}
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {minute, second}
-
- Returns
- -------
- network
- always NT
- """
- return "NT"
-
- def _get_edge_station(self, observatory, channel, type, interval):
- """get edge station.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F}
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {minute, second}
-
- Returns
- -------
- station
- the observatory is returned as the station
- """
- return observatory
-
- def _get_interval_code(self, interval):
- """get edge interval code.
-
- Converts the metadata interval string, into an edge single character
- edge code.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F}
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {minute, second}
-
- Returns
- -------
- interval type
- """
- interval_code = None
- if interval == "day":
- interval_code = "D"
- elif interval == "hour":
- interval_code = "H"
- elif interval == "minute":
- interval_code = "M"
- elif interval == "second":
- interval_code = "S"
- else:
- raise TimeseriesFactoryException('Unexpected interval "%s"' % interval)
- return interval_code
-
def _get_timeseries(self, starttime, endtime, observatory, channel, type, interval):
"""get timeseries data for a single channel.
diff --git a/geomagio/edge/MiniSeedFactory.py b/geomagio/edge/MiniSeedFactory.py
index 211ccf349d8cd5006c011b437fb2fc576f31f2a7..133edeeaeca8237add1547613c660ba4325058b0 100644
--- a/geomagio/edge/MiniSeedFactory.py
+++ b/geomagio/edge/MiniSeedFactory.py
@@ -296,212 +296,6 @@ class MiniSeedFactory(TimeseriesFactory):
trace.data = numpy.ma.masked_invalid(trace.data)
return stream
- def _get_edge_channel(self, observatory, channel, type, interval):
- """get edge channel.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F, X, Y, G} or
- any appropriate edge channel, ie MSD, MGD, HGD.
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {'day', 'hour', 'minute', 'second', 'tenhertz'}
-
- Returns
- -------
- edge_channel
- {MVH, MVE, MVD, MGD etc}
- """
- edge_interval_code = self._get_interval_code(interval)
- edge_channel = None
-
- # If form is chan.loc, return chan (left) portion.
- # Allows specific chan/loc selection.
- if channel.find(".") >= 0:
- tmplist = channel.split(".")
- return tmplist[0].strip()
-
- # see if channel name uses _ for ELEMENT_SUFFIX
- element = None
- suffix = None
- if channel.find("_") >= 0:
- element, suffix = channel.split("_")
-
- # 10Hz should be bin/volt
- if interval == "tenhertz":
- middle = None
- if suffix == "Bin":
- middle = "Y"
- elif suffix == "Volt":
- middle = "E"
- elif suffix is not None:
- raise TimeseriesFactoryException(
- 'bad channel suffix "%s", wanted "Bin" or "Volt"' % suffix
- )
- # check for expected channels
- if element in ("U", "V", "W") and middle is not None:
- return edge_interval_code + middle + element
- else:
- # unknown, assume advanced user
- return channel
-
- if suffix is not None:
- if suffix == "Dist" or suffix == "SQ" or suffix == "SV" or suffix == "DT":
- # these suffixes modify location code, but use element channel
- channel = element
- else:
- raise TimeseriesFactoryException(
- 'bad channel suffix "%s", wanted "Dist", "SQ", or "SV"' % suffix
- )
- if channel in ("D", "F", "G", "H", "U", "V", "W", "X", "Y", "Z"):
- # normal elements
- edge_channel = edge_interval_code + "F" + channel
- elif channel == "E-E":
- edge_channel = edge_interval_code + "QE"
- elif channel == "E-N":
- edge_channel = edge_interval_code + "QN"
- elif channel == "Dst4":
- edge_channel = edge_interval_code + "X4"
- elif channel == "Dst3":
- edge_channel = edge_interval_code + "X3"
- else:
- edge_channel = channel
- return edge_channel
-
- def _get_edge_location(self, observatory, channel, data_type, interval):
- """get edge location.
-
- The edge location code is currently determined by the type
- passed in.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F}
- data_type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {'day', 'hour', 'minute', 'second', 'tenhertz'}
-
- Returns
- -------
- location
- returns an edge location code
- """
- # If form is chan.loc, return loc (right) portion
- # Allows specific chan/loc selection.
- if channel.find(".") >= 0:
- tmplist = channel.split(".")
- return tmplist[1].strip()
- # factory override
- if self.locationCode is not None:
- return self.locationCode
- # determine prefix
- location_prefix = "R"
- if data_type == "variation" or data_type == "reported":
- location_prefix = "R"
- elif data_type == "adjusted" or data_type == "provisional":
- location_prefix = "A"
- elif data_type == "quasi-definitive":
- location_prefix = "Q"
- elif data_type == "definitive":
- location_prefix = "D"
- # determine suffix
- location_suffix = "0"
- if channel.find("_") >= 0:
- _, suffix = channel.split("_")
- if suffix == "Dist":
- location_suffix = "D"
- elif suffix == "SQ":
- location_suffix = "Q"
- elif suffix == "SV":
- location_suffix = "V"
- elif suffix == "DT":
- location_suffix = "R"
- elif suffix not in ("Bin", "Volt"):
- raise TimeseriesFactoryException(
- 'bad channel suffix "%s", wanted "Dist", "SQ", or "SV"' % suffix
- )
- return location_prefix + location_suffix
-
- def _get_edge_network(self, observatory, channel, type, interval):
- """get edge network code.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F}
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {'day', 'hour', 'minute', 'second', 'tenhertz'}
-
- Returns
- -------
- network
- always NT
- """
- return "NT"
-
- def _get_edge_station(self, observatory, channel, type, interval):
- """get edge station.
-
- Parameters
- ----------
- observatory : str
- observatory code
- channel : str
- single character channel {H, E, D, Z, F}
- type : str
- data type {definitive, quasi-definitive, variation}
- interval : str
- interval length {'day', 'hour', 'minute', 'second', 'tenhertz'}
-
- Returns
- -------
- station
- the observatory is returned as the station
- """
- return observatory
-
- def _get_interval_code(self, interval):
- """get edge interval code.
-
- Converts the metadata interval string, into an edge single character
- edge code.
-
- Parameters
- ----------
- interval : str
- interval length {'day', 'hour', 'minute', 'second', 'tenhertz'}
-
- Returns
- -------
- interval type
- """
- interval_code = None
- if interval == "day":
- interval_code = "P"
- elif interval == "hour":
- interval_code = "R"
- elif interval == "minute":
- interval_code = "U"
- elif interval == "second":
- interval_code = "L"
- elif interval == "tenhertz":
- interval_code = "B"
- else:
- raise TimeseriesFactoryException('Unexpected interval "%s"' % interval)
- return interval_code
-
def _get_timeseries(self, starttime, endtime, observatory, channel, type, interval):
"""get timeseries data for a single channel.
diff --git a/geomagio/edge/SNCL.py b/geomagio/edge/SNCL.py
index 9e9e92b559231cfc41df72bab37e92a092cd7cba..df75fa3a45f35abc8f4f433b2142d8e23e86fb59 100644
--- a/geomagio/edge/SNCL.py
+++ b/geomagio/edge/SNCL.py
@@ -63,6 +63,7 @@ class SNCL(BaseModel):
@property
def data_type(self) -> str:
+ """Translates beginning of location code to data type"""
location_start = self.location[0]
if location_start == "R":
return "variation"
@@ -76,12 +77,13 @@ class SNCL(BaseModel):
@property
def element(self) -> str:
- element = self.__get_predefined_element()
- element = element or self.__get_element()
- return element
+ predefined_element = self.__check_predefined_element()
+ element = self.__get_element()
+ return predefined_element or element
@property
def interval(self) -> str:
+ """Translates beginning of channel to interval"""
interval_conversions = INTERVAL_CONVERSIONS[self.data_format]
interval_code_conversions = {
interval_conversions[key]: key for key in interval_conversions.keys()
@@ -93,6 +95,7 @@ class SNCL(BaseModel):
raise ValueError(f"Unexcepted interval code: {channel_start}")
def __get_element(self):
+ """Translates channel/location to element"""
element_start = self.channel[2]
channel = self.channel
channel_middle = channel[1]
@@ -115,7 +118,7 @@ class SNCL(BaseModel):
element_end = ""
return element_start + element_end
- def __get_predefined_element(self) -> Optional[str]:
+ def __check_predefined_element(self) -> Optional[str]:
channel = self.channel
channel_end = channel[1:]
if channel_end in CHANNEL_CONVERSIONS:
diff --git a/geomagio/edge/SNCLFactory.py b/geomagio/edge/SNCLFactory.py
index 4e677c0afb97504d0be369733e8afba41ee1482f..4b76fa1572207084cdeecf105b7a52b7efb5777a 100644
--- a/geomagio/edge/SNCLFactory.py
+++ b/geomagio/edge/SNCLFactory.py
@@ -23,10 +23,12 @@ class SNCLFactory(object):
)
def get_channel(self, element: str, interval: str) -> str:
+ predefined_channel = self.__check_predefined_channel(
+ element=element, interval=interval
+ )
channel_start = self.__get_channel_start(interval=interval)
- channel_end = self.__get_predefined_channel(element=element)
- channel_end = channel_end or self.__get_channel_end(element=element)
- return channel_start + channel_end
+ channel_end = self.__get_channel_end(element=element)
+ return predefined_channel or (channel_start + channel_end)
def get_location(self, element: str, data_type: str) -> str:
location_start = self.__get_location_start(data_type=data_type)
@@ -39,11 +41,18 @@ class SNCLFactory(object):
except:
raise ValueError(f"Unexpected interval: {interval}")
- def __get_predefined_channel(self, element: str) -> Optional[str]:
- if len(element) == 3 and "-" not in element and element != "DST":
- return element[1:]
- elif element in ELEMENT_CONVERSIONS:
- return ELEMENT_CONVERSIONS[element]
+ def __check_predefined_channel(self, element: str, interval: str) -> Optional[str]:
+ if element in ELEMENT_CONVERSIONS:
+ return (
+ self.__get_channel_start(interval=interval)
+ + ELEMENT_CONVERSIONS[element]
+ )
+ elif len(element) == 3:
+ return element
+ # chan.loc format
+ elif "." in element:
+ channel = element.split(".")[0]
+ return channel.strip()
else:
return None
@@ -61,6 +70,7 @@ class SNCLFactory(object):
return channel_middle + channel_end
def __get_location_start(self, data_type: str) -> str:
+ """Translates data type to beginning of location code"""
if data_type == "variation":
return "R"
elif data_type == "adjusted":
@@ -72,6 +82,7 @@ class SNCLFactory(object):
raise ValueError(f"Unexpected data type: {data_type}")
def __get_location_end(self, element: str) -> str:
+ """Translates element suffix to end of location code"""
if "_Sat" in element:
return "1"
if "_Dist" in element:
diff --git a/test/edge_test/SNCLFactory_test.py b/test/edge_test/SNCLFactory_test.py
index d12ab5f4776672cabdd35c198dde441db6888ddf..44f2d0de01856690aafcd9a496b8f9ce3edf4c57 100644
--- a/test/edge_test/SNCLFactory_test.py
+++ b/test/edge_test/SNCLFactory_test.py
@@ -34,9 +34,14 @@ def test_get_channel():
assert factory.get_channel(element="E-N", interval="minute") == "UQN"
assert factory.get_channel(element="SQ", interval="minute") == "USQ"
assert factory.get_channel(element="SV", interval="minute") == "USV"
+ assert factory.get_channel(element="UK1", interval="minute") == "UK1"
+ assert factory.get_channel(element="UK2", interval="minute") == "UK2"
+ assert factory.get_channel(element="UK3", interval="minute") == "UK3"
+ assert factory.get_channel(element="UK4", interval="minute") == "UK4"
assert factory.get_channel(element="DIST", interval="minute") == "UDT"
assert factory.get_channel(element="DST", interval="minute") == "UGD"
assert factory.get_channel(element="U_Dist", interval="minute") == "UFU"
+ assert factory.get_channel(element="UK1.R0", interval="minute") == "UK1"
# test legacy format
factory = SNCLFactory(data_format="legacy")
@@ -49,9 +54,14 @@ def test_get_channel():
assert factory.get_channel(element="E-N", interval="minute") == "MQN"
assert factory.get_channel(element="SQ", interval="minute") == "MSQ"
assert factory.get_channel(element="SV", interval="minute") == "MSV"
+ assert factory.get_channel(element="UK1", interval="minute") == "UK1"
+ assert factory.get_channel(element="UK2", interval="minute") == "UK2"
+ assert factory.get_channel(element="UK3", interval="minute") == "UK3"
+ assert factory.get_channel(element="UK4", interval="minute") == "UK4"
assert factory.get_channel(element="DIST", interval="minute") == "MDT"
assert factory.get_channel(element="DST", interval="minute") == "MGD"
assert factory.get_channel(element="H_Dist", interval="minute") == "MVH"
+ assert factory.get_channel(element="UK1.R0", interval="minute") == "UK1"
def test_get_location():