internal mseed HEZ->UVW conversions

geomagio/edge/SNCL.py

@@ -35,7 +35,9 @@ class SNCL(BaseModel):
         return SNCL(
             station=station,
             network=network,
-            channel=get_channel(element=element, interval=interval),
+            channel=get_channel(
+                element=element, interval=interval, data_type=data_type
+            ),
             location=location or get_location(element=element, data_type=data_type),
         )
 
@@ -85,9 +87,10 @@ class SNCL(BaseModel):
         raise ValueError(f"Unexcepted interval code: {channel_start}")
 
 
-def get_channel(element: str, interval: str) -> str:
+def get_channel(element: str, interval: str, data_type: str) -> str:
     return _check_predefined_channel(element=element, interval=interval) or (
-        _get_channel_start(interval=interval) + _get_channel_end(element=element)
+        _get_channel_start(interval=interval)
+        + _get_channel_end(element=element, data_type=data_type)
     )
 
 
@@ -154,7 +157,7 @@ def _check_predefined_channel(element: str, interval: str) -> Optional[str]:
         return None
 
 
-def _get_channel_end(element: str) -> str:
+def _get_channel_end(element: str, data_type: str) -> str:
     channel_middle = "F"
     if "_Volt" in element:
         channel_middle = "E"
@@ -163,6 +166,13 @@ def _get_channel_end(element: str) -> str:
     elif "_Temp" in element:
         channel_middle = "K"
     channel_end = element.split("_")[0]
+    if data_type == "variation":
+        if channel_end == "H":
+            channel_end = "U"
+        elif channel_end == "E":
+            channel_end = "V"
+        elif channel_end == "Z":
+            channel_end = "W"
     return channel_middle + channel_end
 
 

test/edge_test/MiniSeedFactory_test.py

@@ -34,11 +34,11 @@ def test__put_timeseries():
     sent = client.last_sent
     assert_equal(len(sent), 2)
     # first trace includes [0...4]
-    assert_equal(sent[0].stats.channel, "LFH")
+    assert_equal(sent[0].stats.channel, "LFU")
     assert_equal(len(sent[0]), 4)
     assert_equal(sent[0].stats.endtime, trace1.stats.starttime + 3)
     # second trace includes [5...9]
-    assert_equal(sent[1].stats.channel, "LFH")
+    assert_equal(sent[1].stats.channel, "LFU")
     assert_equal(len(sent[1]), 5)
     assert_equal(sent[1].stats.starttime, trace1.stats.starttime + 5)
     assert_equal(sent[1].stats.endtime, trace1.stats.endtime)

test/edge_test/SNCL_test.py

@@ -106,20 +106,42 @@ def test_element():
 
 def test_get_channel():
     """edge_test.SNCL_test.test_get_channel()"""
-    assert get_channel(element="U_Volt", interval="tenhertz") == "BEU"
-    assert get_channel(element="U_Bin", interval="tenhertz") == "BYU"
-    assert get_channel(element="D", interval="second") == "LFD"
-    assert get_channel(element="F", interval="minute") == "UFF"
-    assert get_channel(element="H", interval="hour") == "RFH"
-    assert get_channel(element="Dst4", interval="day") == "PX4"
-    assert get_channel(element="Dst3", interval="minute") == "UX3"
-    assert get_channel(element="E-E", interval="minute") == "UQE"
-    assert get_channel(element="E-N", interval="minute") == "UQN"
-    assert get_channel(element="UK1", interval="minute") == "UK1"
-    assert get_channel(element="U_Dist", interval="minute") == "UFU"
-    assert get_channel(element="U_SQ", interval="minute") == "UFU"
-    assert get_channel(element="U_SV", interval="minute") == "UFU"
-    assert get_channel(element="UK1.R0", interval="minute") == "UK1"
+    assert (
+        get_channel(element="U_Volt", interval="tenhertz", data_type="variation")
+        == "BEU"
+    )
+    assert (
+        get_channel(element="U_Bin", interval="tenhertz", data_type="variation")
+        == "BYU"
+    )
+    assert get_channel(element="D", interval="second", data_type="variation") == "LFD"
+    assert get_channel(element="F", interval="minute", data_type="variation") == "UFF"
+    assert get_channel(element="U", interval="hour", data_type="variation") == "RFU"
+    assert get_channel(element="V", interval="hour", data_type="variation") == "RFV"
+    assert get_channel(element="W", interval="hour", data_type="variation") == "RFW"
+    assert get_channel(element="H", interval="hour", data_type="variation") == "RFU"
+    assert get_channel(element="E", interval="hour", data_type="variation") == "RFV"
+    assert get_channel(element="Z", interval="hour", data_type="variation") == "RFW"
+    assert get_channel(element="Z", interval="hour", data_type="adjusted") == "RFZ"
+    assert get_channel(element="Dst4", interval="day", data_type="variation") == "PX4"
+    assert (
+        get_channel(element="Dst3", interval="minute", data_type="variation") == "UX3"
+    )
+    assert get_channel(element="E-E", interval="minute", data_type="variation") == "UQE"
+    assert get_channel(element="E-N", interval="minute", data_type="variation") == "UQN"
+    assert get_channel(element="UK1", interval="minute", data_type="variation") == "UK1"
+    assert (
+        get_channel(element="U_Dist", interval="minute", data_type="variation") == "UFU"
+    )
+    assert (
+        get_channel(element="U_SQ", interval="minute", data_type="variation") == "UFU"
+    )
+    assert (
+        get_channel(element="U_SV", interval="minute", data_type="variation") == "UFU"
+    )
+    assert (
+        get_channel(element="UK1.R0", interval="minute", data_type="variation") == "UK1"
+    )
 
 
 def test_get_location():