Merge branch 'fix-miniseed-writer-dropping-last-sample' into 'master'

Modify TimeseriesUtility.py's split_trace()

See merge request !226

geomagio/TimeseriesUtility.py

@@ -621,13 +621,25 @@ def split_trace(trace: Trace, size: int = 86400) -> Stream:
         interval_start = interval["start"]
         interval_end = interval["end"]
         delta = out_trace.stats.delta
-        # accounts for trace containing one sample
         if interval_end - delta < interval_start:
+            # trace contains one sample
             stream += out_trace
             continue
-        stream += out_trace.slice(
-            starttime=interval_start,
-            endtime=interval_end - delta,
-            nearest_sample=False,
-        )
+        if interval_end.timestamp % size:
+            # trace does NOT contain first sample in next interval
+            stream += out_trace.slice(
+                starttime=interval_start, endtime=interval_end, nearest_sample=False
+            )
+        else:
+            # trace DOES contain first sample in next interval
+            stream += out_trace.slice(
+                starttime=interval_start,
+                endtime=interval_end - delta,
+                nearest_sample=False,
+            )
+            if interval_end == out_trace.stats.endtime:
+                # ONLY if it is the last interval
+                stream += out_trace.slice(
+                    starttime=interval_end, endtime=interval_end, nearest_sample=False
+                )
     return stream

test/edge_test/MiniSeedFactory_test.py

@@ -151,13 +151,19 @@ def test__pre_process():
     """edge_test.MiniSeedFactory_test.test__pre_process()"""
     trace = __create_trace(numpy.arange((86400 * 2) + 1), channel="H")
     processed = MiniSeedInputClient(host=None)._pre_process(stream=Stream(trace))
-    assert len(processed) == 2
-    for trace in processed:
+    assert len(processed) == 3
+    for trace in processed[0:2]:
         assert trace.data.dtype == "float32"
         stats = trace.stats
         assert stats.npts == 86400
         assert stats.starttime.timestamp % 86400 == 0
         assert stats.endtime.timestamp % 86400 != 0
+    for trace in processed[-1:]:
+        assert trace.data.dtype == "float32"
+        stats = trace.stats
+        assert stats.npts == 1
+        assert stats.starttime.timestamp % 86400 == 0
+        assert stats.starttime == stats.endtime
 
 
 def test__format_miniseed():
@@ -168,8 +174,10 @@ def test__format_miniseed():
     block_size = 512
     data = buf.getvalue()
     n_blocks = int(len(data) / block_size)
-    assert n_blocks == 1516
-    # 759th block is start of second day(758 blocks per day for 1Hz data)
+    assert n_blocks == 1517
+    # 759th block is start of second day
+    # (758 blocks per day for 1Hz data, which implies 56-byte,
+    #  NOT 64-byte, MiniSeed headers...something to investigate)
     block_start = 758 * block_size
     block = data[block_start : block_start + block_size]
     out_stream = read(io.BytesIO(block))