diff --git a/geomagio/TimeseriesUtility.py b/geomagio/TimeseriesUtility.py
index 459287dbe69add0b6dc98da98cd56d8395e18bd2..e0729c21b81796707a06cff1184765aeab5be29c 100644
--- a/geomagio/TimeseriesUtility.py
+++ b/geomagio/TimeseriesUtility.py
@@ -4,7 +4,9 @@ from datetime import datetime
import math
import sys
import numpy
-import obspy.core
+from obspy.core import Stats, Stream, Trace, UTCDateTime
+
+from .Util import get_intervals
def create_empty_trace(
@@ -14,9 +16,9 @@ def create_empty_trace(
Parameters
----------
- starttime: obspy.core.UTCDateTime
+ starttime: UTCDateTime
the starttime of the requested data
- endtime: obspy.core.UTCDateTime
+ endtime: UTCDateTime
the endtime of the requested data
observatory : str
observatory code
@@ -34,19 +36,17 @@ def create_empty_trace(
the location code
Returns
-------
- obspy.core.Trace
+ Trace:
trace for the requested channel
"""
delta = get_delta_from_interval(interval)
- stats = obspy.core.Stats()
+ stats = Stats()
stats.network = network
stats.station = station
stats.location = location
stats.channel = channel
# Calculate first valid sample time based on interval
- trace_starttime = obspy.core.UTCDateTime(
- numpy.ceil(starttime.timestamp / delta) * delta
- )
+ trace_starttime = UTCDateTime(numpy.ceil(starttime.timestamp / delta) * delta)
if delta > 60.0:
trace_starttime += (delta - 60) / 2
if trace_starttime > endtime:
@@ -60,7 +60,30 @@ def create_empty_trace(
length = int((endtime - trace_starttime) / delta)
stats.npts = length + 1
data = numpy.full(stats.npts, numpy.nan, dtype=numpy.float64)
- return obspy.core.Trace(data, stats)
+ return Trace(data, stats)
+
+
+def encode_stream(stream: Stream, encoding: str) -> Stream:
+ """Ensures that factory encoding matches output data encoding
+
+ Parameters:
+ -----------
+ stream: Stream
+ stream of input data
+
+ Returns:
+ --------
+ out_stream: Stream
+ stream with matching data encoding to factory specification
+
+ """
+ out_stream = Stream()
+ for trace in stream:
+ trace_out = trace.copy()
+ if trace_out.data.dtype != encoding:
+ trace_out.data = trace_out.data.astype(encoding)
+ out_stream += trace_out
+ return out_stream
def get_delta_from_interval(data_interval):
@@ -125,20 +148,20 @@ def get_stream_start_end_times(timeseries, without_gaps=False):
the latest endtime.
Parameters
----------
- timeseries: obspy.core.stream
+ timeseries: Stream
The timeseries stream
Returns
-------
tuple: (starttime, endtime)
- starttime: obspy.core.UTCDateTime
- endtime: obspy.core.UTCDateTime
+ starttime: UTCDateTime
+ endtime: UTCDateTime
NOTE: when the entire timeseries is a gap, and without_gaps is True,
the returned endtime will be one delta earlier than starttime.
"""
- starttime = obspy.core.UTCDateTime(datetime.now())
- endtime = obspy.core.UTCDateTime(0)
+ starttime = UTCDateTime(datetime.now())
+ endtime = UTCDateTime(0)
for trace in timeseries:
if trace.stats.starttime < starttime:
starttime = trace.stats.starttime
@@ -160,7 +183,7 @@ def get_stream_gaps(stream, channels=None):
"""Get gaps in a given stream
Parameters
----------
- stream: obspy.core.Stream
+ stream: Stream
the stream to check for gaps
channels: array_like
list of channels to check for gaps
@@ -188,7 +211,7 @@ def get_trace_gaps(trace):
"""Gets gaps in a trace representing a single channel
Parameters
----------
- trace: obspy.core.Trace
+ trace: Trace
a stream containing a single channel of data.
Returns
@@ -270,7 +293,7 @@ def get_channels(stream):
Parameters
----------
- stream : obspy.core.Stream
+ stream : Stream
Returns
-------
@@ -289,8 +312,8 @@ def get_trace_value(traces, time, default=None):
Parameters
----------
- trace : obspy.core.Trace
- time : obspy.core.UTCDateTime
+ trace : Trace
+ time : UTCDateTime
Returns
-------
@@ -316,7 +339,7 @@ def has_all_channels(stream, channels, starttime, endtime):
Parameters
----------
- stream: obspy.core.Stream
+ stream: 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
@@ -346,7 +369,7 @@ def has_any_channels(stream, channels, starttime, endtime):
Parameters
----------
- stream: obspy.core.Stream
+ stream: 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
@@ -381,17 +404,17 @@ def mask_stream(stream):
Parameters
----------
- stream : obspy.core.Stream
+ stream : Stream
stream to mask
Returns
-------
- obspy.core.Stream
+ Stream
stream with new Trace objects with numpy masked array data.
"""
- masked = obspy.core.Stream()
+ masked = Stream()
for trace in stream:
- masked += obspy.core.Trace(numpy.ma.masked_invalid(trace.data), trace.stats)
+ masked += Trace(numpy.ma.masked_invalid(trace.data), trace.stats)
return masked
@@ -400,18 +423,18 @@ def unmask_stream(stream):
Parameters
----------
- stream : obspy.core.Stream
+ stream : Stream
stream to unmask
Returns
-------
- obspy.core.Stream
+ Stream
stream with new Trace objects with numpy array data, with numpy.nan
as a fill value in a filled array.
"""
- unmasked = obspy.core.Stream()
+ unmasked = Stream()
for trace in stream:
- unmasked += obspy.core.Trace(
+ unmasked += Trace(
trace.data.filled(fill_value=numpy.nan)
if isinstance(trace.data, numpy.ma.MaskedArray)
else trace.data,
@@ -425,15 +448,15 @@ def merge_streams(*streams):
Parameters
----------
- *streams : obspy.core.Stream
+ *streams : Stream
one or more streams to merge
Returns
-------
- obspy.core.Stream
+ Stream
stream with contiguous traces merged, and gaps filled with numpy.nan
"""
- merged = obspy.core.Stream()
+ merged = Stream()
# sort out empty
for stream in streams:
@@ -445,7 +468,7 @@ def merge_streams(*streams):
split = split.split()
# Re-add any empty traces that were removed by split()
- readd = obspy.core.Stream()
+ readd = Stream()
for trace in merged:
stats = trace.stats
split_stream = split.select(
@@ -480,11 +503,11 @@ def pad_timeseries(timeseries, starttime, endtime):
Parameters
----------
- timeseries: obspy.core.stream
+ timeseries: Stream
The timeseries stream as returned by the call to getWaveform
- starttime: obspy.core.UTCDateTime
+ starttime: UTCDateTime
the starttime of the requested data
- endtime: obspy.core.UTCDateTime
+ endtime: UTCDateTime
the endtime of the requested data
Notes: the original timeseries object is changed.
@@ -501,17 +524,17 @@ def pad_and_trim_trace(trace, starttime, endtime):
Parameters
----------
- trace: obspy.core.Trace
+ trace: Trace
One trace to be processed
- starttime: obspy.core.UTCDateTime
+ starttime: UTCDateTime
the starttime of the requested data
- endtime: obspy.core.UTCDateTime
+ endtime: UTCDateTime
the endtime of the requested data
Notes: the original timeseries object is changed.
"""
- trace_starttime = obspy.core.UTCDateTime(trace.stats.starttime)
- trace_endtime = obspy.core.UTCDateTime(trace.stats.endtime)
+ trace_starttime = UTCDateTime(trace.stats.starttime)
+ trace_endtime = UTCDateTime(trace.stats.endtime)
trace_delta = trace.stats.delta
if trace_starttime < starttime:
# trim to starttime
@@ -568,3 +591,28 @@ def round_usecs(time):
if rounded_usecs != usecs:
time = time.replace(microsecond=rounded_usecs)
return time
+
+
+def split_stream(stream: Stream, size: int = 86400) -> Stream:
+ out_stream = Stream()
+ for trace in stream:
+ out_stream += split_trace(trace, size)
+ return out_stream
+
+
+def split_trace(trace: Trace, size: int = 86400) -> Stream:
+ # copy in case original trace changes later
+ stream = Stream()
+ out_trace = trace.copy()
+ for interval in get_intervals(
+ starttime=out_trace.stats.starttime,
+ endtime=out_trace.stats.endtime,
+ size=size,
+ trim=True,
+ ):
+ stream += out_trace.slice(
+ starttime=interval["start"],
+ endtime=interval["end"] - out_trace.stats.delta,
+ nearest_sample=False,
+ )
+ return stream
diff --git a/geomagio/edge/MiniSeedInputClient.py b/geomagio/edge/MiniSeedInputClient.py
index 9d83701871b524f5d4cff7eb9dc4684e392d8fc8..a25d31ce89741d8483aa9a1bf55220b1d2c3358e 100644
--- a/geomagio/edge/MiniSeedInputClient.py
+++ b/geomagio/edge/MiniSeedInputClient.py
@@ -2,6 +2,11 @@ from __future__ import absolute_import, print_function
import io
import socket
import sys
+from typing import BinaryIO
+
+from obspy.core import Stream
+
+from ..TimeseriesUtility import encode_stream, split_stream
class MiniSeedInputClient(object):
@@ -20,7 +25,7 @@ class MiniSeedInputClient(object):
Floating point precision for output data
"""
- def __init__(self, host, port=2061, encoding="FLOAT32"):
+ def __init__(self, host, port=2061, encoding="float32"):
self.host = host
self.port = port
self.encoding = encoding
@@ -66,14 +71,45 @@ class MiniSeedInputClient(object):
Parameters
----------
- stream: obspy.core.Stream
+ stream: Stream
stream with trace(s) to send.
"""
# connect if needed
if self.socket is None:
self.connect()
- # convert stream to miniseed
buf = io.BytesIO()
- stream.write(buf, encoding=self.encoding, format="MSEED", reclen=512)
+ self._format_miniseed(stream=stream, buf=buf)
# send data
self.socket.sendall(buf.getvalue())
+
+ def _format_miniseed(self, stream: Stream, buf: BinaryIO) -> io.BytesIO:
+ """Processes and writes stream to buffer as miniseed
+
+ Parameters:
+ -----------
+ stream: Stream
+ stream with data to write
+ buf: BinaryIO
+ memory buffer for output data
+ """
+ processed = self._pre_process(stream=stream)
+ for trace in processed:
+ # convert stream to miniseed
+ trace.write(buf, format="MSEED", reclen=512)
+
+ def _pre_process(self, stream: Stream) -> Stream:
+ """Encodes and splits streams at daily intervals
+
+ Paramters:
+ ----------
+ stream: Stream
+ stream of input data
+
+ Returns:
+ --------
+ stream: Stream
+ list of encoded trace split at daily intervals
+ """
+ stream = encode_stream(stream=stream, encoding=self.encoding)
+ stream = split_stream(stream=stream, size=86400)
+ return stream
diff --git a/geomagio/edge/__init__.py b/geomagio/edge/__init__.py
index 93888daa3356a5d4f8b5d3fed1034772106e6f13..2a53748a195983a3dc737b87731330584f10c739 100644
--- a/geomagio/edge/__init__.py
+++ b/geomagio/edge/__init__.py
@@ -5,6 +5,13 @@ from __future__ import absolute_import
from .EdgeFactory import EdgeFactory
from .LocationCode import LocationCode
from .MiniSeedFactory import MiniSeedFactory
+from .MiniSeedInputClient import MiniSeedInputClient
from .RawInputClient import RawInputClient
-__all__ = ["EdgeFactory", "LocationCode", "MiniSeedFactory", "RawInputClient"]
+__all__ = [
+ "EdgeFactory",
+ "LocationCode",
+ "MiniSeedFactory",
+ "MiniSeedInputClient",
+ "RawInputClient",
+]
diff --git a/test/edge_test/MiniSeedFactory_test.py b/test/edge_test/MiniSeedFactory_test.py
index 24397dbcbde8e20781418f133a23696609b6ec9e..a0c57869e7ff0ffcea37f7c6f5c98842347e2492 100644
--- a/test/edge_test/MiniSeedFactory_test.py
+++ b/test/edge_test/MiniSeedFactory_test.py
@@ -1,10 +1,12 @@
"""Tests for MiniSeedFactory.py"""
+import io
import numpy
from numpy.testing import assert_equal
-from obspy.core import Stats, Stream, Trace, UTCDateTime
+from obspy.core import read, Stats, Stream, Trace, UTCDateTime
+
from geomagio import TimeseriesUtility
-from geomagio.edge import MiniSeedFactory
+from geomagio.edge import MiniSeedFactory, MiniSeedInputClient
def test__get_edge_network():
@@ -90,6 +92,35 @@ def test__put_timeseries():
assert_equal(sent[1].stats.endtime, trace1.stats.endtime)
+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 trace.data.dtype == "float32"
+ stats = trace.stats
+ assert stats.npts == 86400
+ assert stats.starttime.timestamp % 86400 == 0
+ assert stats.endtime.timestamp % 86400 != 0
+
+
+def test__format_miniseed():
+ """edge_test.MiniseedFactory_test.test__format_miniseed()"""
+ buf = io.BytesIO()
+ trace = __create_trace(numpy.arange((86400 * 2) + 1), channel="H")
+ MiniSeedInputClient(host=None)._format_miniseed(stream=Stream(trace), buf=buf)
+ 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)
+ block_start = 758 * block_size
+ block = data[block_start : block_start + block_size]
+ out_stream = read(io.BytesIO(block))
+ assert out_stream[0].stats.starttime.timestamp % 86400 == 0
+
+
def test__set_metadata():
"""edge_test.MiniSeedFactory_test.test__set_metadata()"""
# Call _set_metadata with 2 traces, and make certain the stats get