From b3829a0f232ae8de885bc7e7bc19e3d9b29b5f06 Mon Sep 17 00:00:00 2001
From: pcain <pcain@usgs.gov>
Date: Tue, 24 Aug 2021 13:07:47 -0600
Subject: [PATCH] _set_metadata and _get_empty for derived factory
---
geomagio/DerivedTimeseriesFactory.py | 46 ++++++++++++++++++++++++++--
1 file changed, 43 insertions(+), 3 deletions(-)
diff --git a/geomagio/DerivedTimeseriesFactory.py b/geomagio/DerivedTimeseriesFactory.py
index c722bf31..648a33f9 100644
--- a/geomagio/DerivedTimeseriesFactory.py
+++ b/geomagio/DerivedTimeseriesFactory.py
@@ -1,6 +1,6 @@
from typing import List, Optional
-from obspy import Stream, UTCDateTime
+from obspy import Stream, Trace, UTCDateTime
from .algorithm import Algorithm, DeltaFAlgorithm, XYZAlgorithm
from .TimeseriesFactory import TimeseriesFactory, TimeseriesUtility
@@ -55,7 +55,7 @@ class DerivedTimeseriesFactory(TimeseriesFactory):
missing = get_missing(timeseries, channels)
if missing and add_empty_channels:
for channel in missing:
- timeseries += self.factory._get_empty_trace(
+ timeseries += self._get_empty_trace(
starttime=starttime,
endtime=endtime,
observatory=observatory,
@@ -66,7 +66,7 @@ class DerivedTimeseriesFactory(TimeseriesFactory):
# file-based factories return all channels found in file
timeseries = Stream([t for t in timeseries if t.stats.channel in channels])
for channel in channels:
- self.factory._set_metadata(
+ self._set_metadata(
stream=timeseries.select(channel=channel),
observatory=observatory,
channel=channel,
@@ -160,6 +160,46 @@ class DerivedTimeseriesFactory(TimeseriesFactory):
)
return Stream()
+ def _get_empty_trace(
+ self,
+ starttime: UTCDateTime,
+ endtime: UTCDateTime,
+ observatory: str,
+ channel: str,
+ data_type: str,
+ interval: str,
+ network: str = "NT",
+ location: str = "",
+ ) -> Trace:
+ """creates empty trace"""
+ return self.factory._get_empty_trace(
+ starttime,
+ endtime,
+ observatory,
+ channel,
+ data_type,
+ interval,
+ network=network,
+ location=location,
+ )
+
+ def _set_metadata(
+ self, stream: Stream, observatory: str, channel: str, type: str, interval: str
+ ):
+ """set metadata for a given stream/channel
+ Parameters
+ ----------
+ observatory
+ observatory code
+ channel
+ edge channel code {MVH, MVE, MVD, ...}
+ type
+ data type {definitive, quasi-definitive, variation}
+ interval
+ interval length {minute, second}
+ """
+ return self.factory._set_metadata(stream, observatory, channel, type, interval)
+
def get_missing(input: Stream, desired: List[str]) -> List[str]:
"""Return missing channels from input"""
--
GitLab