From 07bfe69d1ec492296aae4ff5529d6945d374d7f1 Mon Sep 17 00:00:00 2001
From: pcain <pcain@usgs.gov>
Date: Thu, 21 Oct 2021 18:47:52 +0000
Subject: [PATCH] add IRISFactory
---
geomagio/edge/IRISFactory.py | 154 +++++++++++++++++++++++++++++++++++
1 file changed, 154 insertions(+)
create mode 100644 geomagio/edge/IRISFactory.py
diff --git a/geomagio/edge/IRISFactory.py b/geomagio/edge/IRISFactory.py
new file mode 100644
index 000000000..950fb61f5
--- /dev/null
+++ b/geomagio/edge/IRISFactory.py
@@ -0,0 +1,154 @@
+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
--
GitLab