Merge branch 'miniseed-file-factory-mvp' into 'master'

MiniSeedFactory - File Implementations

Closes #133

See merge request !381

geomagio/Controller.py

@@ -494,9 +494,13 @@ def get_input_factory(args):
     input_factory_args["interval"] = args.input_interval or args.interval
     input_factory_args["observatory"] = args.observatory
     input_factory_args["type"] = args.type
+    input_type = args.input
     # stream/url arguments
     if args.input_file is not None:
-        input_stream = open(args.input_file, "r")
+        if input_type == "miniseed":
+            input_stream = open(args.input_file, "rb")
+        else:
+            input_stream = open(args.input_file, "r")
     elif args.input_stdin:
         input_stream = sys.stdin
     elif args.input_url is not None:
@@ -505,7 +509,6 @@ def get_input_factory(args):
             input_factory_args["urlTemplate"] = args.input_url
         else:
             input_stream = StringIO(Util.read_url(args.input_url))
-    input_type = args.input
     if input_type == "edge":
         input_factory = edge.EdgeFactory(
             host=args.input_host,
@@ -513,14 +516,6 @@ def get_input_factory(args):
             locationCode=args.locationcode,
             **input_factory_args,
         )
-    elif input_type == "miniseed":
-        input_factory = edge.MiniSeedFactory(
-            host=args.input_host,
-            port=args.input_port,
-            locationCode=args.locationcode,
-            convert_channels=args.convert_voltbin,
-            **input_factory_args,
-        )
     elif input_type == "goes":
         # TODO: deal with other goes arguments
         input_factory = imfv283.GOESIMFV283Factory(
@@ -556,6 +551,14 @@ def get_input_factory(args):
             input_factory = imfv283.IMFV283Factory(**input_factory_args)
         elif input_type == "pcdcp":
             input_factory = pcdcp.PCDCPFactory(**input_factory_args)
+        elif input_type == "miniseed":
+            input_factory = edge.MiniSeedFactory(
+                host=args.input_host,
+                port=args.input_port,
+                locationCode=args.locationcode,
+                convert_channels=args.convert_voltbin,
+                **input_factory_args,
+            )
         # wrap stream
         if input_stream is not None:
             input_factory = StreamTimeseriesFactory(
@@ -615,16 +618,6 @@ def get_output_factory(args):
             forceout=args.output_edge_forceout,
             **output_factory_args,
         )
-    elif output_type == "miniseed":
-        # TODO: deal with other miniseed arguments
-        locationcode = args.outlocationcode or args.locationcode or None
-        output_factory = edge.MiniSeedFactory(
-            host=args.output_host,
-            port=args.output_read_port,
-            write_port=args.output_port,
-            locationCode=locationcode,
-            **output_factory_args,
-        )
     elif output_type == "plot":
         output_factory = PlotTimeseriesFactory()
     else:
@@ -641,6 +634,16 @@ def get_output_factory(args):
             output_factory = temperature.TEMPFactory(**output_factory_args)
         elif output_type == "vbf":
             output_factory = vbf.VBFFactory(**output_factory_args)
+        elif output_type == "miniseed":
+            # TODO: deal with other miniseed arguments
+            locationcode = args.outlocationcode or args.locationcode or None
+            output_factory = edge.MiniSeedFactory(
+                host=args.output_host,
+                port=args.output_read_port,
+                write_port=args.output_port,
+                locationCode=locationcode,
+                **output_factory_args,
+            )
         # wrap stream
         if output_stream is not None:
             output_factory = StreamTimeseriesFactory(

geomagio/edge/MiniSeedFactory.py

@@ -10,11 +10,13 @@ Edge is the USGS earthquake hazard centers replacement for earthworm.
 """
 
 from __future__ import absolute_import
+import io
 import sys
 from typing import List, Optional
 
 import numpy
 import numpy.ma
+from obspy import read
 from obspy.clients.neic import client as miniseed
 from obspy.core import Stats, Stream, Trace, UTCDateTime
 
@@ -596,3 +598,52 @@ class MiniSeedFactory(TimeseriesFactory):
             self.observatoryMetadata.set_metadata(
                 trace.stats, observatory, channel, type, interval
             )
+
+    def parse_string(self, data: str, **kwargs) -> Stream:
+        """Parse a MiniSEED byte string into an ObsPy Stream.
+
+        Parameters
+        ----------
+        data : str
+            The MiniSEED content (binary) in string form.
+
+        Returns
+        -------
+        Stream
+            An ObsPy Stream object containing the parsed MiniSEED data.
+        """
+        try:
+            # If data is already bytes, this is a no-op; if data is a Python string
+            # containing binary content, encode to bytes.
+            if not isinstance(data, bytes):
+                data = data.encode("utf-8", errors="ignore")
+
+            with io.BytesIO(data) as bio:
+                st = read(bio, format="MSEED")
+            return st
+
+        except Exception as e:
+            print(f"Failed to parse MiniSEED data: {str(e)}")
+            return Stream()
+
+    def write_file(self, fh, timeseries: Stream, channels: list):
+        """Write an ObsPy Stream to a file handle in MiniSEED format.
+
+        Parameters
+        ----------
+        fh : file-like (writable)
+            A file handle where data is written. Could be an open file or BytesIO.
+        timeseries : Stream
+            Stream containing traces to store.
+        """
+        try:
+            # Write to MiniSEED
+            ts = Stream()
+            for channel in channels:
+                ts += timeseries.select(channel=channel)
+            ts.write(fh, format="MSEED")
+
+        except Exception as e:
+            raise TimeseriesFactoryException(
+                f"Error writing MiniSEED to file: {str(e)}"
+            )