Merge pull request #276 from jmfee-usgs/channel-foreach

Make filter algorithm process any channels

geomagio/TimeseriesUtility.py

@@ -250,6 +250,72 @@ def get_channels(stream):
     return [ch for ch in channels]
 
 
+def has_all_channels(stream, channels, starttime, endtime):
+    """Check whether all channels have any data within time range.
+
+    Parameters
+    ----------
+    stream: obspy.core.Stream
+        The input stream we want to make certain has data
+    channels: array_like
+        The list of channels that we want to have concurrent data
+    starttime: UTCDateTime
+        start time of requested output
+    end : UTCDateTime
+        end time of requested output
+
+    Returns
+    -------
+    bool: True if data found across all channels between starttime/endtime
+    """
+    input_gaps = get_merged_gaps(
+            get_stream_gaps(stream=stream, channels=channels))
+    for input_gap in input_gaps:
+        # Check for gaps that include the entire range
+        if (starttime >= input_gap[0] and
+                starttime <= input_gap[1] and
+                endtime < input_gap[2]):
+            return False
+    return True
+
+
+def has_any_channels(stream, channels, starttime, endtime):
+    """Check whether any channel has data within time range.
+
+    Parameters
+    ----------
+    stream: obspy.core.Stream
+        The input stream we want to make certain has data
+    channels: array_like
+        The list of channels that we want to have concurrent data
+    starttime: UTCDateTime
+        start time of requested output
+    end : UTCDateTime
+        end time of requested output
+
+    Returns
+    -------
+    bool: True if any data found between starttime/endtime
+    """
+    # process if any channels have data not covering the time range
+    input_gaps = get_stream_gaps(stream=stream, channels=channels)
+    for channel in channels:
+        if channel not in input_gaps:
+            continue
+        channel_gaps = input_gaps[channel]
+        if len(channel_gaps) == 0:
+            # no gaps in channel
+            return True
+        for gap in channel_gaps:
+            if not (starttime >= gap[0] and
+                    starttime <= gap[1] and
+                    endtime < gap[2]):
+                # gap doesn't span channel
+                return True
+    # didn't find any data
+    return False
+
+
 def mask_stream(stream):
     """Convert stream traces to masked arrays.
 

geomagio/algorithm/Algorithm.py

@@ -88,7 +88,9 @@ class Algorithm(object):
         return (start, end)
 
     def can_produce_data(self, starttime, endtime, stream):
-        """Can Product data
+        """Can Produce data
+
+        By default require all channels to have data at the same time.
 
         Parameters
         ----------
@@ -99,17 +101,11 @@ class Algorithm(object):
         stream: obspy.core.Stream
             The input stream we want to make certain has data for the algorithm
         """
-        input_gaps = TimeseriesUtility.get_merged_gaps(
-                TimeseriesUtility.get_stream_gaps(
-                    stream=stream,
-                    channels=self.get_required_channels()))
-        for input_gap in input_gaps:
-            # Check for gaps that include the entire range
-            if (starttime >= input_gap[0] and
-                    starttime <= input_gap[1] and
-                    endtime < input_gap[2]):
-                return False
-        return True
+        return TimeseriesUtility.has_all_channels(
+                stream,
+                self.get_required_channels(),
+                starttime,
+                endtime)
 
     def get_next_starttime(self):
         """Check whether algorithm has a stateful start time.

geomagio/algorithm/FilterAlgorithm.py

@@ -5,8 +5,7 @@ import scipy.signal as sps
 from numpy.lib import stride_tricks as npls
 from obspy.core import Stream, Stats
 import json
-from ..TimeseriesUtility import get_delta_from_interval
-
+from .. import TimeseriesUtility
 
 STEPS = [
     {  # 10 Hz to one second filter
@@ -101,6 +100,26 @@ class FilterAlgorithm(Algorithm):
                     steps.append(step)
         return steps
 
+    def can_produce_data(self, starttime, endtime, stream):
+        """Can Produce data
+
+        The FilterAlgorithm can produce data for each channel independently.
+
+        Parameters
+        ----------
+        starttime: UTCDateTime
+            start time of requested output
+        end : UTCDateTime
+            end time of requested output
+        stream: obspy.core.Stream
+            The input stream we want to make certain has data for the algorithm
+        """
+        return TimeseriesUtility.has_any_channels(
+                stream,
+                self.get_required_channels(),
+                starttime,
+                endtime)
+
     def create_trace(self, channel, stats, data):
         """Utility to create a new trace object.
         This may be necessary for more sophisticated metadata
@@ -282,8 +301,8 @@ class FilterAlgorithm(Algorithm):
         Algorithm.configure(self, arguments)
         # intialize filter with command line arguments
         self.coeff_filename = arguments.filter_coefficients
-        input_interval = arguments.input_interval
-        output_interval = arguments.output_interval
-        self.input_sample_period = get_delta_from_interval(input_interval)
-        self.output_sample_period = get_delta_from_interval(output_interval)
+        self.input_sample_period = TimeseriesUtility.get_delta_from_interval(
+                arguments.input_interval or arguments.interval)
+        self.output_sample_period = TimeseriesUtility.get_delta_from_interval(
+                arguments.output_interval or arguments.interval)
         self.load_state()

test/TimeseriesUtility_test.py

@@ -90,7 +90,6 @@ def test_get_stream_gaps_channels():
 
     test that gaps are only checked in specified channels.
     """
-    stream = Stream
     stream = Stream([
         __create_trace('H', [numpy.nan, 1, 1, numpy.nan, numpy.nan]),
         __create_trace('Z', [0, 0, 0, 1, 1, 1])
@@ -163,6 +162,58 @@ def test_get_merged_gaps():
     assert_equal(gap[1], UTCDateTime('2015-01-01T00:00:07Z'))
 
 
+def test_has_all_channels():
+    """TimeseriesUtility_test.test_has_all_channels():
+    """
+    nan = numpy.nan
+    stream = Stream([
+        __create_trace('H', [nan, 1, 1, nan, nan]),
+        __create_trace('Z', [0, 0, 0, 1, 1]),
+        __create_trace('E', [nan, nan, nan, nan, nan])
+    ])
+    for trace in stream:
+        # set time of first sample
+        trace.stats.starttime = UTCDateTime('2015-01-01T00:00:00Z')
+        # set sample rate to 1 second
+        trace.stats.delta = 1
+        trace.stats.npts = len(trace.data)
+    # check for channels
+    starttime = stream[0].stats.starttime
+    endtime = stream[0].stats.endtime
+    assert_equal(TimeseriesUtility.has_all_channels(
+            stream, ['H', 'Z'], starttime, endtime), True)
+    assert_equal(TimeseriesUtility.has_all_channels(
+            stream, ['H', 'Z', 'E'], starttime, endtime), False)
+    assert_equal(TimeseriesUtility.has_all_channels(
+            stream, ['E'], starttime, endtime), False)
+
+
+def test_has_any_channels():
+    """TimeseriesUtility_test.test_has_any_channels():
+    """
+    nan = numpy.nan
+    stream = Stream([
+        __create_trace('H', [nan, 1, 1, nan, nan]),
+        __create_trace('Z', [0, 0, 0, 1, 1, 1]),
+        __create_trace('E', [nan, nan, nan, nan, nan])
+    ])
+    for trace in stream:
+        # set time of first sample
+        trace.stats.starttime = UTCDateTime('2015-01-01T00:00:00Z')
+        # set sample rate to 1 second
+        trace.stats.delta = 1
+        trace.stats.npts = len(trace.data)
+    # check for channels
+    starttime = stream[0].stats.starttime
+    endtime = stream[0].stats.endtime
+    assert_equal(TimeseriesUtility.has_any_channels(
+            stream, ['H', 'Z'], starttime, endtime), True)
+    assert_equal(TimeseriesUtility.has_any_channels(
+            stream, ['H', 'Z', 'E'], starttime, endtime), True)
+    assert_equal(TimeseriesUtility.has_any_channels(
+            stream, ['E'], starttime, endtime), False)
+
+
 def test_merge_streams():
     """TimeseriesUtility_test.test_merge_streams()