add IRISFactory

geomagio/edge/IRISFactory.py

+from typing import List, Optional
+
+from obspy import Trace, UTCDateTime
+from obspy.clients.iris.client import Client
+
+from ..geomag_types import DataInterval, DataType
+from ..ObservatoryMetadata import ObservatoryMetadata
+from .. import TimeseriesUtility
+from .IRISSNCL import IRISSNCL
+from .MiniSeedFactory import MiniSeedFactory
+
+
+class IRISFactory(MiniSeedFactory):
+    """MiniSeedFactory for IRIS related data.
+
+    Parameters
+    ----------
+    base_url: str
+        iris web service url
+    host: str
+        a string representing the IP number of the host to connect to
+    write_port: int
+        the port number the miniseed client writes data
+    network: str
+        data network
+    observatory: str
+        observatory id
+    channels: array
+        list of channels
+    type: {"adjusted", "definitive", "quasi-definitive", "variation"}
+        data type
+    interval: {"tenhertz", "second", "minute", "hour", "day"}
+        data interval
+    observatoryMetadata: ObservatoryMetadata
+        observatory metadata to replace the default ObservatoryMetadata
+    locationCode: str
+        the location code for the given edge server
+    convert_channels: array
+        list of channels to convert from volt/bin to nT
+
+    See Also
+    --------
+    MiniSeedFactory
+
+    Notes
+    -----
+    This factory is designed to receive data from IRIS and write to edge.
+    convert_channels should only be used with tenhertz NT data (U, V, W).
+    Non-NT data conversions are automatically handled by the IRIS web service.
+
+
+    """
+
+    def __init__(
+        self,
+        base_url: str = "http://service.iris.edu/irisws",
+        host: str = "cwbpub.cr.usgs.gov",
+        write_port: int = 7974,
+        network: str = "NT",
+        observatory: Optional[str] = None,
+        channels: Optional[List[str]] = None,
+        type: Optional[DataType] = None,
+        interval: Optional[DataInterval] = None,
+        observatoryMetadata: Optional[ObservatoryMetadata] = None,
+        locationCode: Optional[str] = None,
+        convert_channels: Optional[List[str]] = None,
+    ):
+        super().__init__(
+            host=host,
+            write_port=write_port,
+            observatory=observatory,
+            channels=channels,
+            type=type,
+            interval=interval,
+            observatoryMetadata=observatoryMetadata or ObservatoryMetadata(),
+            locationCode=locationCode,
+            convert_channels=convert_channels,
+        )
+        self.base_url = base_url
+        self.network = network
+        self.client = Client(base_url=base_url)
+
+    def _get_timeseries(
+        self,
+        starttime: UTCDateTime,
+        endtime: UTCDateTime,
+        observatory: str,
+        channel: str,
+        type: DataType,
+        interval: DataInterval,
+        add_empty_channels: bool = True,
+    ) -> Trace:
+        """get timeseries data for a single channel
+
+        Parameters
+        ----------
+        starttime: UTCDateTime
+            the starttime of requested data
+        endtime: UTCDateTime
+            the endtime of requested data
+        observatory: str
+            observatory id
+        channel: str
+            channel name
+        type: {"adjusted", "definitive", "quasi-definitive", "variation"}
+            data type
+        interval: {"tenhertz", "second", "minute", "hour", "day"}
+            data interval
+        add_empty_channels: bool
+            if True, returns channels without data as empty traces
+
+        Returns
+        -------
+        data: Trace
+            timeseries trace of the requested channel data
+        """
+        sncl = IRISSNCL.get_sncl(
+            data_type=type,
+            element=channel,
+            interval=interval,
+            station=observatory,
+            network=self.network,
+            location=self.locationCode,
+        )
+        filter = ["correct"]
+        if sncl.channel[1] in ["E", "Y"]:
+            filter = []
+        data = self.client.timeseries(
+            network=sncl.network,
+            station=sncl.station,
+            location=sncl.location,
+            channel=sncl.channel,
+            starttime=starttime,
+            endtime=endtime,
+            filter=filter,
+        )
+        data.merge()
+        if data.count() == 0 and add_empty_channels:
+            data += self._get_empty_trace(
+                starttime=starttime,
+                endtime=endtime,
+                observatory=observatory,
+                channel=channel,
+                data_type=type,
+                interval=interval,
+                network=sncl.network,
+                location=sncl.location,
+            )
+        if data.count() != 0:
+            TimeseriesUtility.pad_and_trim_trace(
+                trace=data[0], starttime=starttime, endtime=endtime
+            )
+        self._set_metadata(data, observatory, channel, type, interval)
+        return data