_get_empty_channels method for ts factories

geomagio/DerivedTimeseriesFactory.py

@@ -54,45 +54,18 @@ class DerivedTimeseriesFactory(TimeseriesFactory):
             )
         missing = get_missing(timeseries, channels)
         if missing and add_empty_channels:
-            timeseries += self._get_empty_channels(
+            timeseries += self.factory._get_empty_channels(
                 starttime=starttime,
                 endtime=endtime,
                 observatory=observatory,
                 channels=channels,
                 data_type=type,
                 interval=interval,
-                location=timeseries[0].stats.location,
             )
         # file-based factories return all channels found in file
         timeseries = Stream([t for t in timeseries if t.stats.channel in channels])
         return timeseries
 
-    def _get_empty_channels(
-        self,
-        starttime: UTCDateTime,
-        endtime: UTCDateTime,
-        observatory: str,
-        channels: List[str],
-        data_type: str,
-        interval: str,
-        location: str,
-    ) -> Stream:
-        """create empty channels"""
-        output_stream = Stream()
-        for channel in channels:
-            output_stream += TimeseriesUtility.create_empty_trace(
-                starttime=starttime,
-                endtime=endtime,
-                observatory=observatory,
-                channel=channel,
-                type=data_type,
-                interval=interval,
-                network="NT",
-                station=observatory,
-                location=location,
-            )
-        return output_stream
-
     def _get_derived_channels(
         self,
         starttime: UTCDateTime,

geomagio/TimeseriesFactory.py

 """Abstract Timeseries Factory Interface."""
 from __future__ import absolute_import, print_function
+from typing import List
 
 import numpy
 import obspy.core
@@ -297,6 +298,32 @@ class TimeseriesFactory(object):
         """
         raise NotImplementedError('"write_file" not implemented')
 
+    def _get_empty_channels(
+        self,
+        starttime: obspy.core.UTCDateTime,
+        endtime: obspy.core.UTCDateTime,
+        observatory: str,
+        channels: List[str],
+        data_type: str,
+        interval: str,
+        location: str = "",
+    ) -> obspy.core.Stream:
+        """creates stream with empty channels"""
+        output_stream = obspy.core.Stream()
+        for channel in channels:
+            output_stream += TimeseriesUtility.create_empty_trace(
+                starttime=starttime,
+                endtime=endtime,
+                observatory=observatory,
+                channel=channel,
+                type=data_type,
+                interval=interval,
+                network="NT",
+                station=observatory,
+                location=location,
+            )
+        return output_stream
+
     def _get_file_from_url(self, url):
         """Get a file for writing.
 

geomagio/edge/EdgeFactory.py

@@ -9,6 +9,7 @@ to take advantage of it's newer realtime abilities.
 Edge is the USGS earthquake hazard centers replacement for earthworm.
 """
 from __future__ import absolute_import
+from typing import List, Optional
 
 import sys
 import numpy
@@ -22,7 +23,7 @@ from ..TimeseriesFactory import TimeseriesFactory
 from ..TimeseriesFactoryException import TimeseriesFactoryException
 from ..ObservatoryMetadata import ObservatoryMetadata
 from .RawInputClient import RawInputClient
-from .LegacySNCL import LegacySNCL
+from .LegacySNCL import LegacySNCL, get_location
 
 
 class EdgeFactory(TimeseriesFactory):
@@ -288,6 +289,42 @@ class EdgeFactory(TimeseriesFactory):
             trace.data = numpy.ma.masked_invalid(trace.data)
         return stream
 
+    def _get_empty_channels(
+        self,
+        starttime: obspy.core.UTCDateTime,
+        endtime: obspy.core.UTCDateTime,
+        observatory: str,
+        channels: List[str],
+        data_type: str,
+        interval: str,
+        location: Optional[str] = None,
+    ) -> obspy.core.Stream:
+        """creates stream with empty channels"""
+        output_stream = obspy.core.Stream()
+        for channel in channels:
+            trace = super()._get_empty_channels(
+                starttime=starttime,
+                endtime=endtime,
+                observatory=observatory,
+                channels=(channel,),
+                data_type=data_type,
+                interval=interval,
+                location=location
+                or get_location(
+                    element=channel,
+                    data_type=data_type,
+                ),
+            )
+            self._set_metadata(
+                stream=trace,
+                observatory=observatory,
+                channel=channel,
+                type=data_type,
+                interval=interval,
+            )
+            output_stream += trace
+        return output_stream
+
     def _get_timeseries(
         self,
         starttime,
@@ -346,19 +383,17 @@ class EdgeFactory(TimeseriesFactory):
         for trace in data:
             trace.data = trace.data.astype("i4")
         data.merge()
+        self._set_metadata(data, observatory, channel, type, interval)
         if data.count() == 0 and add_empty_channels:
-            data += TimeseriesUtility.create_empty_trace(
-                starttime,
-                endtime,
-                observatory,
-                channel,
-                type,
-                interval,
-                sncl.network,
-                sncl.station,
-                sncl.location,
+            data += self._get_empty_channels(
+                starttime=starttime,
+                endtime=endtime,
+                observatory=observatory,
+                channels=(channel,),
+                data_type=type,
+                interval=interval,
+                location=sncl.location,
             )
-        self._set_metadata(data, observatory, channel, type, interval)
         return data
 
     def _post_process(self, timeseries, starttime, endtime, channels):

geomagio/edge/MiniSeedFactory.py

@@ -11,6 +11,7 @@ Edge is the USGS earthquake hazard centers replacement for earthworm.
 from __future__ import absolute_import
 
 import sys
+from typing import List, Optional
 import numpy
 import numpy.ma
 
@@ -23,7 +24,7 @@ from ..TimeseriesFactory import TimeseriesFactory
 from ..TimeseriesFactoryException import TimeseriesFactoryException
 from ..ObservatoryMetadata import ObservatoryMetadata
 from .MiniSeedInputClient import MiniSeedInputClient
-from .SNCL import SNCL
+from .SNCL import SNCL, get_location
 
 
 class MiniSeedFactory(TimeseriesFactory):
@@ -307,6 +308,42 @@ class MiniSeedFactory(TimeseriesFactory):
             trace.data = numpy.ma.masked_invalid(trace.data)
         return stream
 
+    def _get_empty_channels(
+        self,
+        starttime: obspy.core.UTCDateTime,
+        endtime: obspy.core.UTCDateTime,
+        observatory: str,
+        channels: List[str],
+        data_type: str,
+        interval: str,
+        location: Optional[str] = None,
+    ) -> obspy.core.Stream:
+        """creates stream with empty channels"""
+        output_stream = obspy.core.Stream()
+        for channel in channels:
+            trace = super()._get_empty_channels(
+                starttime=starttime,
+                endtime=endtime,
+                observatory=observatory,
+                channels=(channel,),
+                data_type=data_type,
+                interval=interval,
+                location=location
+                or get_location(
+                    element=channel,
+                    data_type=data_type,
+                ),
+            )
+            self._set_metadata(
+                stream=trace,
+                observatory=observatory,
+                channel=channel,
+                type=data_type,
+                interval=interval,
+            )
+            output_stream += trace
+        return output_stream
+
     def _get_timeseries(
         self,
         starttime,
@@ -352,19 +389,17 @@ class MiniSeedFactory(TimeseriesFactory):
             sncl.network, sncl.station, sncl.location, sncl.channel, starttime, endtime
         )
         data.merge()
+        self._set_metadata(data, observatory, channel, type, interval)
         if data.count() == 0 and add_empty_channels:
-            data += TimeseriesUtility.create_empty_trace(
-                starttime,
-                endtime,
-                observatory,
-                channel,
-                type,
-                interval,
-                sncl.network,
-                sncl.station,
-                sncl.location,
+            data += self._get_empty_channels(
+                starttime=starttime,
+                endtime=endtime,
+                observatory=observatory,
+                channels=(channel,),
+                data_type=type,
+                interval=interval,
+                location=sncl.location,
             )
-        self._set_metadata(data, observatory, channel, type, interval)
         return data
 
     def _convert_timeseries(