Make AnalyzedSeries change points lazy properties

otava/series.py

@@ -114,7 +114,6 @@ def find_by_attribute(self, name: str, value: str) -> List[int]:
     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 @@ def can_append(self, time, new_data, attributes):
         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 @@ def append(self, time, new_data, attributes):
 
         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 @@ def append(self, time, new_data, attributes):
         # 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 @@ def change_points_from_json(change_points_json):
         )
 
         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

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)