Add channels parameter to TimeseriesUtility.get_stream_gaps

geomagio/TimeseriesUtility.py

@@ -4,7 +4,7 @@ import numpy
 import obspy.core
 
 
-def get_stream_gaps(stream):
+def get_stream_gaps(stream, channels=None):
     """Get gaps in a given stream
     Parameters
     ----------
@@ -12,6 +12,7 @@ def get_stream_gaps(stream):
         the stream to check for gaps
     channels: array_like
         list of channels to check for gaps
+        Default is None (check all channels).
 
     Returns
     -------
@@ -25,6 +26,8 @@ def get_stream_gaps(stream):
     gaps = {}
     for trace in stream:
         channel = trace.stats.channel
+        if channels is not None and channel not in channels:
+            continue
         gaps[channel] = get_trace_gaps(trace)
     return gaps
 

test/TimeseriesUtility_test.py

@@ -36,6 +36,25 @@ def test_get_stream_gaps():
     # no gaps in Z channel
     assert_equals(len(gaps['Z']), 0)
 
+def test_get_stream_gaps_channels():
+    """TimeseriesUtility_test.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])
+    ])
+    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
+    # find gaps
+    gaps = TimeseriesUtility.get_stream_gaps(stream, ['Z'])
+    assert_equals('H' in gaps, False)
+    assert_equals(len(gaps['Z']), 0)
 
 def test_get_trace_gaps():
     """TimeseriesUtility_test.test_get_trace_gaps()