This is an automated email from the ASF dual-hosted git repository.
Gerrrr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/otava.git
The following commit(s) were added to refs/heads/master by this push:
new f15bb29 Make AnalyzedSeries change points lazy properties (#169)
f15bb29 is described below
commit f15bb29ec2bd047fff33b32b4cbec5715fb345d4
Author: DanaerLee <[email protected]>
AuthorDate: Wed Aug 26 05:06:00 2026 +0800
Make AnalyzedSeries change points lazy properties (#169)
* Make AnalyzedSeries change point computation lazy
* Cover lazy change point computation in series tests
---
otava/series.py | 79 +++++++++++++-------
tests/series_test.py | 204 ++++++++++++++++++++++++++++++++++++++++++++++-----
2 files changed, 241 insertions(+), 42 deletions(-)
diff --git a/otava/series.py b/otava/series.py
index cb10d7b..4ddb492 100644
--- a/otava/series.py
+++ b/otava/series.py
@@ -114,7 +114,6 @@ class Series:
def analyze(self, options: Optional[AnalysisOptions] = None) ->
"AnalyzedSeries":
if options is None:
options = AnalysisOptions()
- logging.info(f"Computing change points for test {self.test_name}...")
return AnalyzedSeries(self, options)
@@ -125,25 +124,50 @@ class AnalyzedSeries:
__series: Series
options: AnalysisOptions
- change_points: ChangePointsByMetric
- change_points_by_time: ChangePointsByTime
- change_points_timestamp: datetime
def __init__(
self, series: Series, options: AnalysisOptions, change_points:
Dict[str, ChangePoint] = None
):
self.__series = series
self.options = options
- # record when these change points were calculated
- self.change_points_timestamp = datetime.now(timezone.utc)
- self.change_points = None
- if change_points is not None:
- self.change_points = change_points
- else:
- cp, weak_cps = self.__compute_change_points(series, options)
- self.change_points = cp
- self.weak_change_points = weak_cps
- self.change_points_by_time =
self.__group_change_points_by_time(series, self.change_points)
+ self.__change_points = change_points
+ self.__weak_change_points = ChangePointsByMetric() if change_points is
not None else None
+ self.__change_points_by_time = None
+ # records when the change points were calculated
+ self.__change_points_timestamp = (
+ datetime.now(timezone.utc) if change_points is not None else None
+ )
+
+ def __ensure_change_points_computed(self):
+ if self.__change_points is None:
+ logging.info(f"Computing change points for test
{self.__series.test_name}...")
+ cp, weak_cps = self.__compute_change_points(self.__series,
self.options)
+ self.__change_points = cp
+ self.__weak_change_points = weak_cps
+ self.__change_points_timestamp = datetime.now(timezone.utc)
+
+ @property
+ def change_points(self) -> ChangePointsByMetric:
+ self.__ensure_change_points_computed()
+ return self.__change_points
+
+ @property
+ def weak_change_points(self) -> ChangePointsByMetric:
+ self.__ensure_change_points_computed()
+ return self.__weak_change_points
+
+ @property
+ def change_points_timestamp(self) -> datetime:
+ self.__ensure_change_points_computed()
+ return self.__change_points_timestamp
+
+ @property
+ def change_points_by_time(self) -> ChangePointsByTime:
+ if self.__change_points_by_time is None:
+ self.__change_points_by_time = self.__group_change_points_by_time(
+ self.__series, self.change_points
+ )
+ return self.__change_points_by_time
@staticmethod
def __compute_change_points(
@@ -227,8 +251,8 @@ class AnalyzedSeries:
return self._validate_append(time, new_data, attributes) is None
def _validate_append(self, time, new_data, attributes):
- if not self.change_points:
- return RuntimeError("You must use __compute_change_points() once
first.")
+ # appending updates the cached results, so they must exist first
+ self.__ensure_change_points_computed()
if not isinstance(time, list):
return ValueError("time argument must be an array.")
if not isinstance(new_data, dict):
@@ -273,13 +297,19 @@ class AnalyzedSeries:
for metric in self.__series.data.keys():
if metric not in new_data:
- weak_change_points[metric] =
self.weak_change_points.select_metrics(metric)
+ if metric in self.weak_change_points:
+ weak_change_points[metric] =
self.weak_change_points.select_metrics(metric)
continue
new_data_len = len(new_data[metric])
+ previous_weak_cp = (
+ self.weak_change_points.get_change_points_for_metric(metric)
+ if metric in self.weak_change_points
+ else []
+ )
old_weak_cp = [
cp
- for cp in
self.weak_change_points.get_change_points_for_metric(metric)
+ for cp in previous_weak_cp
if cp.index < len(self.__series.data[metric]) - new_data_len -
1
]
change_points, weak_cps = compute_change_points(
@@ -320,8 +350,10 @@ class AnalyzedSeries:
# r has a subset of all metrics, so can't just set change_points to r
for metric, cpglist in r.items():
self.change_points[metric] = cpglist
- self.weak_change_points = w
- self.change_points_by_time = self.change_points.by_time()
+ self.__weak_change_points = w
+ # invalidate rather than rebuild: the property recomputes it on first
read
+ self.__change_points_by_time = None
+ self.__change_points_timestamp = datetime.now(timezone.utc)
return r, w
def test_name(self) -> str:
@@ -466,14 +498,11 @@ class AnalyzedSeries:
)
analyzed_series = cls(new_series, new_options, new_change_points)
- analyzed_series.weak_change_points = new_weak_change_points
+ analyzed_series.__weak_change_points = new_weak_change_points
if "change_points_timestamp" in analyzed_json.keys():
- analyzed_series.change_points_timestamp =
_datetime_adapter.validate_python(
+ analyzed_series.__change_points_timestamp =
_datetime_adapter.validate_python(
analyzed_json["change_points_timestamp"]
)
- analyzed_series.change_points_by_time =
AnalyzedSeries.__group_change_points_by_time(
- analyzed_series.__series, analyzed_series.change_points
- )
return analyzed_series
diff --git a/tests/series_test.py b/tests/series_test.py
index 0497380..cd54966 100644
--- a/tests/series_test.py
+++ b/tests/series_test.py
@@ -188,7 +188,8 @@ def test_change_point_detection_performance():
data={"series": series},
attributes={},
)
- test.analyze()
+ # access the results so the timing covers detection, not just
construction
+ test.analyze().change_points_by_time
end_time = time.process_time()
assert (end_time - start_time) < 0.5
@@ -472,22 +473,6 @@ def test_validate():
data={"series1": series_1, "series2": series_2},
attributes={},
)
- test_fail = Series(
- "test",
- branch=None,
- time=time,
- metrics={"series1": Metric(1, 1.0), "series2": Metric(1, 1.0)},
- data={"series1": series_1, "series2": series_2},
- attributes={},
- )
-
- analyzed_series_fail = test_fail.analyze()
- analyzed_series_fail.change_points = None
- err = analyzed_series_fail._validate_append(
- time=[len(time)], new_data={"series1": [0.51]}, attributes={}
- )
- assert isinstance(err, RuntimeError)
-
analyzed_series = test.analyze()
analyzed_series.append(
time=[len(time)], new_data={"series1": [0.5], "series2": [1.97]},
attributes={}
@@ -599,3 +584,188 @@ def test_series_raw_initialization():
assert len(series.time) == 3
assert series.data["throughput"] == [10.0, 12.0, 11.5]
+
+
+def test_change_points_computed_lazily_and_cached(monkeypatch):
+ from otava import series as series_module
+
+ calls = {"count": 0}
+ real_compute = series_module.compute_change_points
+
+ def counting_compute(*args, **kwargs):
+ calls["count"] += 1
+ return real_compute(*args, **kwargs)
+
+ monkeypatch.setattr(series_module, "compute_change_points",
counting_compute)
+
+ data = [1.0] * 10 + [5.0] * 10
+ test = Series(
+ "lazy_test",
+ branch=None,
+ time=list(range(len(data))),
+ metrics={"m1": Metric(1, 1.0), "m2": Metric(1, 1.0)},
+ data={"m1": data, "m2": data.copy()},
+ attributes={},
+ )
+
+ analyzed = test.analyze()
+ assert calls["count"] == 0
+
+ change_points = analyzed.change_points
+ assert calls["count"] == 2 # one computation per metric
+ assert [c.index for c in change_points.get_change_points_for_metric("m1")]
== [10]
+
+ assert analyzed.change_points is change_points
+ assert len(analyzed.change_points_by_time) == 1
+ assert analyzed.weak_change_points is not None
+ assert analyzed.change_points_timestamp is not None
+ assert calls["count"] == 2
+
+
+def test_append_on_stable_series(monkeypatch):
+ from otava import series as series_module
+
+ calls = {"count": 0}
+ real_compute = series_module.compute_change_points
+
+ def counting_compute(*args, **kwargs):
+ calls["count"] += 1
+ return real_compute(*args, **kwargs)
+
+ monkeypatch.setattr(series_module, "compute_change_points",
counting_compute)
+
+ stable = [1.0] * 20
+ test = Series(
+ "stable_test",
+ branch=None,
+ time=list(range(len(stable))),
+ metrics={"m1": Metric(1, 1.0)},
+ data={"m1": stable},
+ attributes={},
+ )
+
+ analyzed = test.analyze()
+ assert calls["count"] == 0
+
+ # a stable series has no change points, which must not be mistaken for
"not computed yet"
+ assert analyzed.can_append(time=[len(stable)], new_data={"m1": [1.0]},
attributes={})
+ assert calls["count"] == 1
+ assert len(list(analyzed.change_points)) == 0
+
+ analyzed.append(time=[len(stable)], new_data={"m1": [1.0]}, attributes={})
+ assert len(list(analyzed.change_points)) == 0
+
+
+def test_append_invalidates_by_time_view(monkeypatch):
+ data = [1.0] * 10 + [5.0] * 10
+ test = Series(
+ "invalidation_test",
+ branch=None,
+ time=list(range(len(data))),
+ metrics={"m1": Metric(1, 1.0)},
+ data={"m1": data},
+ attributes={},
+ )
+
+ analyzed = test.analyze()
+ assert [cpg.time for cpg in analyzed.change_points_by_time] == [10]
+
+ # a second shift, so the by-time view must change after the append
+ analyzed.append(
+ time=list(range(20, 32)), new_data={"m1": [50.0] * 12}, attributes={}
+ )
+
+ assert [cpg.time for cpg in analyzed.change_points_by_time] == [10, 20]
+
+
+def test_append_does_not_rebuild_unread_by_time_view(monkeypatch):
+ from otava.change_point_divisive import base as base_module
+
+ data = [1.0] * 10 + [5.0] * 10
+ test = Series(
+ "no_rebuild_test",
+ branch=None,
+ time=list(range(len(data))),
+ metrics={"m1": Metric(1, 1.0)},
+ data={"m1": data},
+ attributes={},
+ )
+
+ analyzed = test.analyze()
+ calls = {"count": 0}
+ real_by_time = base_module.ChangePointsByMetric.by_time
+
+ def counting_by_time(self, *args, **kwargs):
+ calls["count"] += 1
+ return real_by_time(self, *args, **kwargs)
+
+ monkeypatch.setattr(base_module.ChangePointsByMetric, "by_time",
counting_by_time)
+
+ analyzed.append(time=[len(data)], new_data={"m1": [5.0]}, attributes={})
+ assert calls["count"] == 0
+
+ _ = analyzed.change_points_by_time
+ assert calls["count"] == 1
+
+
+def test_append_refreshes_change_points_timestamp():
+ data = [1.0] * 10 + [5.0] * 10
+ test = Series(
+ "timestamp_test",
+ branch=None,
+ time=list(range(len(data))),
+ metrics={"m1": Metric(1, 1.0)},
+ data={"m1": data},
+ attributes={},
+ )
+
+ analyzed = test.analyze()
+ before = analyzed.change_points_timestamp
+
+ analyzed.append(time=[len(data)], new_data={"m1": [5.0]}, attributes={})
+
+ assert analyzed.change_points_timestamp > before
+
+
+def test_append_computes_change_points_first():
+ data = [1.0] * 10 + [5.0] * 10
+ test = Series(
+ "append_lazy_test",
+ branch=None,
+ time=list(range(len(data))),
+ metrics={"m1": Metric(1, 1.0)},
+ data={"m1": data},
+ attributes={},
+ )
+
+ analyzed = test.analyze()
+ analyzed.append(time=[len(data)], new_data={"m1": [5.0]}, attributes={})
+
+ assert [c.index for c in
analyzed.change_points.get_change_points_for_metric("m1")] == [10]
+ assert len(analyzed.change_points_by_time) == 1
+
+
+def test_from_json_does_not_recompute(monkeypatch):
+ from otava import series as series_module
+
+ data = [1.0] * 10 + [5.0] * 10
+ test = Series(
+ "roundtrip_lazy_test",
+ branch=None,
+ time=list(range(len(data))),
+ metrics={"m1": Metric(1, 1.0)},
+ data={"m1": data},
+ attributes={},
+ )
+ analyzed = test.analyze()
+ payload = analyzed.to_json()
+
+ def fail_compute(*args, **kwargs):
+ raise AssertionError("a deserialized series must not recompute change
points")
+
+ monkeypatch.setattr(series_module, "compute_change_points", fail_compute)
+
+ restored = AnalyzedSeries.from_json(payload)
+ assert restored.change_points_timestamp == analyzed.change_points_timestamp
+ assert [c.index for c in
restored.change_points.get_change_points_for_metric("m1")] == [10]
+ assert len(restored.change_points_by_time) ==
len(analyzed.change_points_by_time)
