From 654364279d1342fdac1d53cca49b117a1848a8c1 Mon Sep 17 00:00:00 2001
From: Jeremy Fee <jmfee@usgs.gov>
Date: Thu, 11 Aug 2022 21:32:01 -0600
Subject: [PATCH] Add get_previous_interval method and test
---
geomagio/Controller.py | 30 ++++++++++++++++++++++++++----
test/Controller_test.py | 22 +++++++++++++++++++++-
2 files changed, 47 insertions(+), 5 deletions(-)
diff --git a/geomagio/Controller.py b/geomagio/Controller.py
index 6d37c112..458e74e4 100644
--- a/geomagio/Controller.py
+++ b/geomagio/Controller.py
@@ -426,11 +426,10 @@ class Controller(object):
# check for fillable gap at start
if output_gap[0] == starttime:
# found fillable gap at start, recurse to previous interval
- delta = TimeseriesUtility.get_delta_from_interval(output_interval)
- recurse_starttime = (
- starttime - (endtime - starttime) - delta * bool(update_count)
+ recurse_starttime, recurse_endtime = get_previous_interval(
+ start=starttime,
+ end=endtime,
)
- recurse_endtime = starttime - delta
self.run_as_update(
algorithm=algorithm,
observatory=observatory,
@@ -638,6 +637,29 @@ def get_output_factory(args):
return output_factory
+def get_previous_interval(
+ start: UTCDateTime,
+ end: UTCDateTime,
+) -> Tuple[UTCDateTime, UTCDateTime]:
+ """Get previous interval for given interval.
+
+ Parameters
+ ----------
+ start
+ start of interval
+ end
+ end of interval
+
+ Returns
+ -------
+ Previous interval of approximately the same size.
+ Interval is rounded to nearest second, and ends one microsecond earlier.
+ """
+ # round to nearest second to recover removed microsecond from repeated calls
+ interval_size = round(end - start)
+ return (start - interval_size, start - 1e-6)
+
+
def get_realtime_interval(interval_seconds: int) -> Tuple[UTCDateTime, UTCDateTime]:
# calculate endtime/starttime
now = UTCDateTime()
diff --git a/test/Controller_test.py b/test/Controller_test.py
index 2e51338a..9acab443 100644
--- a/test/Controller_test.py
+++ b/test/Controller_test.py
@@ -6,7 +6,7 @@ from geomagio.algorithm import Algorithm
from geomagio.iaga2002 import IAGA2002Factory
# needed to emulate geomag.py script
-from geomagio.Controller import _main, parse_args
+from geomagio.Controller import _main, get_previous_interval, parse_args
# needed to copy SqDistAlgorithm statefile
from shutil import copy
@@ -290,3 +290,23 @@ def test_controller_update_sqdist():
)
expected = expected_factory.get_timeseries(starttime=starttime1, endtime=endtime6)
assert_allclose(actual, expected)
+
+
+def test_get_previous_interval():
+ """Test get_previous_interval produces stable interval sizes"""
+ # initial interval is one hour
+ start = UTCDateTime("2022-01-05T00:00:00Z")
+ end = UTCDateTime("2022-01-05T01:00:00Z")
+ # previous interval starts at beginning of previous hour
+ previous = get_previous_interval(start=start, end=end)
+ assert previous == (
+ UTCDateTime("2022-01-04T23:00:00"),
+ UTCDateTime("2022-01-04T23:59:59.999999Z"),
+ )
+ # previous interval still starts at beginning of previous hour
+ # even though interval is one microsecond smaller
+ previous = get_previous_interval(*previous)
+ assert previous == (
+ UTCDateTime("2022-01-04T22:00:00"),
+ UTCDateTime("2022-01-04T22:59:59.999999Z"),
+ )
--
GitLab