Split persistence and report JSON serialization (+introduce Pydantic)

otava/change_point_divisive/base.py

@@ -201,9 +201,8 @@ def to_candidate(self) -> CandidateChangePoint:
         data = {f.name: getattr(self, f.name) for f in fields(CandidateChangePoint)}
         return CandidateChangePoint(**data)
 
-    def to_json(self, rounded=True):
-        cps = ChangePointSerializer(self)
-        return cps.to_json(rounded)
+    def to_json(self):
+        return ChangePointSerializer(self).to_json()
 
 
 class ChangePointSerializer(ChangePoint):
@@ -217,6 +216,7 @@ class ChangePointSerializer(ChangePoint):
     def __init__(self, cp: ChangePoint[GenericStats]):
         self.stats = cp.stats
         self.index = cp.index
+        self.qhat = cp.qhat
         self.metric = cp.metric
 
     def forward_change_percent(self) -> float:
@@ -243,33 +243,19 @@ def stddev_after(self):
     def pvalue(self):
         return self.stats.pvalue
 
-    def to_json(self, rounded=True):
-        if rounded:
-            return {
-                "metric": self.metric,
-                "index": int(self.index),
-                "forward_change_percent": f"{self.forward_change_percent():.0f}",
-                "backward_change_percent": f"{self.backward_change_percent():.0f}",
-                "magnitude": f"{self.magnitude():-0f}",
-                "mean_before": f"{self.mean_before():-0f}",
-                "stddev_before": f"{self.stddev_before():-0f}",
-                "mean_after": f"{self.mean_after():-0f}",
-                "stddev_after": f"{self.stddev_after():-0f}",
-                "pvalue": f"{self.pvalue():-0f}",
-            }
-
-        else:
-            return {
-                "metric": self.metric,
-                "index": int(self.index),
-                "forward_change_percent": self.forward_change_percent(),
-                "magnitude": self.magnitude(),
-                "mean_before": self.mean_before(),
-                "stddev_before": self.stddev_before(),
-                "mean_after": self.mean_after(),
-                "stddev_after": self.stddev_after(),
-                "pvalue": self.pvalue(),
-            }
+    def to_json(self):
+        return {
+            "metric": self.metric,
+            "index": int(self.index),
+            "qhat": self.qhat,
+            "forward_change_percent": self.forward_change_percent(),
+            "magnitude": self.magnitude(),
+            "mean_before": self.mean_before(),
+            "stddev_before": self.stddev_before(),
+            "mean_after": self.mean_after(),
+            "stddev_after": self.stddev_after(),
+            "pvalue": self.pvalue(),
+        }
 
 
 @dataclass
@@ -288,15 +274,16 @@ class ChangePointGroup:
     :param attributes: The attributes of the test result at this timestamp. Commit and test metadata.
     :param changes: For each metric that has a change point at this time(stamp), the ChangePoint object.
     """
+
     time: float
     attributes: Dict[str, str]
     # ChangePointGroup.changes.keys() stores the set of metrics that were used at this ChangePointGroup.time.
     changes: Dict[str, ChangePoint]
 
-    def to_json(self, rounded=False):
+    def to_json(self):
         changes = []
         for metric, cp in self.changes.items():
-            changes.append(cp.to_json(rounded=rounded))
+            changes.append(cp.to_json())
 
         return {
             "time": self.time,
@@ -365,9 +352,7 @@ def __init__(self):
     def from_list(cls, cps: list[ChangePointGroup]):
         """Build from a list of ChangePointGroup objects, ordered by time."""
         if not isinstance(cps, list):
-            raise TypeError(
-                f"from_list() argument must be a list. Got {type(cps)}."
-            )
+            raise TypeError(f"from_list() argument must be a list. Got {type(cps)}.")
         for cpg in cps:
             if not isinstance(cpg, ChangePointGroup):
                 raise TypeError(
@@ -396,12 +381,12 @@ def from_dict(cls, cps: dict):
         for metric, cpglist in cps.items():
             for cpg in cpglist:
                 if not isinstance(cpg, ChangePointGroup):
-                    raise TypeError(
-                        "from_dict() takes a dict[str, list[ChangePointGroup]]."
-                    )
+                    raise TypeError("from_dict() takes a dict[str, list[ChangePointGroup]].")
                 for cp in cpg:
                     if cp.metric and cp.metric != metric:
-                        raise ValueError(f"metric field is not internally consistent. {cp.metric} != {metric} at {cpg.time}")
+                        raise ValueError(
+                            f"metric field is not internally consistent. {cp.metric} != {metric} at {cpg.time}"
+                        )
             # store each metric's groups sorted by timestamp
             obj._change_points[metric] = sorted(cpglist, key=lambda cpg: cpg.time)
         return obj
@@ -522,7 +507,9 @@ def get_change_points_for_metric(self, m: str):
         for cpg in single_metric._change_points:
             for metric, cp in cpg.changes.items():
                 if cp.metric and cp.metric != metric:
-                    raise ValueError(f"metric field is not internally consistent. {cp.metric} != {metric} at {cpg.time}")
+                    raise ValueError(
+                        f"metric field is not internally consistent. {cp.metric} != {metric} at {cpg.time}"
+                    )
                 metric_change_points.append(cp)
         return metric_change_points
 
@@ -536,7 +523,7 @@ def at_timestamp(self, t: float):
 
     def at_commit(self, sha: str):
         for row in self:
-            if row.attributes['commit'] == sha:
+            if row.attributes["commit"] == sha:
                 return row
         raise LookupError(sha)
 
@@ -555,17 +542,18 @@ def pivot(self):
 
 class ChangePointsByTime(ChangePoints):
     """
-        Implementation of ChangePoints class where the internal structure is ordered by time/commit.
+    Implementation of ChangePoints class where the internal structure is ordered by time/commit.
 
-        In fact, this is the default way, and therefore all of this class' implementation is already
-        in its parent class ChangePoints. However, you can still create instances from this class
-        to make it explicit that your code at that point explicitly wanted a collection of ChangePoints
-        ordered by time.
+    In fact, this is the default way, and therefore all of this class' implementation is already
+    in its parent class ChangePoints. However, you can still create instances from this class
+    to make it explicit that your code at that point explicitly wanted a collection of ChangePoints
+    ordered by time.
 
-        The pivot() method returns a new view holding the same ChangePoint objects, but ordered by metrics
-        as the primary and optimized axis. The method by_metric() can be used for the same purpose. Use
-        copy().pivot() for an isolated copy. The method by_time() is a no-op and returns self.
+    The pivot() method returns a new view holding the same ChangePoint objects, but ordered by metrics
+    as the primary and optimized axis. The method by_metric() can be used for the same purpose. Use
+    copy().pivot() for an isolated copy. The method by_time() is a no-op and returns self.
     """
+
     @classmethod
     def from_dict(cls, cps: dict):
         """
@@ -579,14 +567,14 @@ def from_dict(cls, cps: dict):
 
 class ChangePointsByMetric(ChangePoints):
     """
-        Provides same interface as ChangePointsByTime, but internally stores with metric first.
+    Provides same interface as ChangePointsByTime, but internally stores with metric first.
 
-        You can create empty instances of this class, or you can also use the factory method
-        `ChangePoints.from_dict()` to get an instance of this type.
+    You can create empty instances of this class, or you can also use the factory method
+    `ChangePoints.from_dict()` to get an instance of this type.
 
-        The pivot() method returns a new view holding the same ChangePoint objects, but ordered by time
-        as the primary and optimized axis. The method by_time() can be used for the same purpose. Use
-        copy().pivot() for an isolated copy. The method by_metric() is a no-op and returns self.
+    The pivot() method returns a new view holding the same ChangePoint objects, but ordered by time
+    as the primary and optimized axis. The method by_time() can be used for the same purpose. Use
+    copy().pivot() for an isolated copy. The method by_metric() is a no-op and returns self.
     """
 
     def __init__(self):
@@ -636,7 +624,9 @@ def append(self, cpg: ChangePointGroup):
                 self._change_points[metric] = []
             for metric1, cp in cpg.changes.items():
                 if metric1 != cp.metric:
-                    raise ValueError(f"metric field is not internally consistent. {metric1} != {cp.metric} at {cpg.time}")
+                    raise ValueError(
+                        f"metric field is not internally consistent. {metric1} != {cp.metric} at {cpg.time}"
+                    )
             if (not self._change_points[metric]) or cpg.time > self._change_points[metric][-1].time:
                 self._change_points[metric].append(cpg.select_metrics(metric))
             else:
@@ -653,10 +643,14 @@ def extend(self, cps: list[ChangePointGroup]):
                 raise TypeError(errmsg)
             for metric1, cp in cpg.changes.items():
                 if metric1 != cp.metric:
-                    raise ValueError(f"metric field is not internally consistent. {metric1} != {cp.metric} at {cpg.time}")
+                    raise ValueError(
+                        f"metric field is not internally consistent. {metric1} != {cp.metric} at {cpg.time}"
+                    )
                 if metric1 not in self._change_points:
                     self._change_points[metric1] = []
-                if (not self._change_points[metric1]) or cpg.time > self._change_points[metric1][-1].time:
+                if (not self._change_points[metric1]) or cpg.time > self._change_points[metric1][
+                    -1
+                ].time:
                     self._change_points[metric1].append(cpg.select_metrics(metric1))
                 else:
                     raise ValueError(
@@ -728,7 +722,9 @@ def get_change_points_for_metric(self, m: str):
             for cpg in cpglist:
                 for metric2, cp in cpg.changes.items():
                     if metric1 != metric2 or cp.metric and metric1 != cp.metric:
-                        raise ValueError(f"metric field is not internally consistent. {metric1} != {cp.metric} at {cpg.time}")
+                        raise ValueError(
+                            f"metric field is not internally consistent. {metric1} != {cp.metric} at {cpg.time}"
+                        )
                     metric_change_points.append(cp)
 
         return metric_change_points
@@ -752,13 +748,18 @@ class SignificanceTester(Generic[GenericStats]):
     def __init__(self, max_pvalue: float):
         self.max_pvalue = max_pvalue
 
-    def compare(self, left: Sequence[SupportsFloat], right: Sequence[SupportsFloat], p: float = None) -> GenericStats:
+    def compare(
+        self, left: Sequence[SupportsFloat], right: Sequence[SupportsFloat], p: float = None
+    ) -> GenericStats:
         if len(left) == 0 or len(right) == 0:
             raise ValueError
         return BaseStats.calculate(left, right, p)
 
     def get_sides(
-        self, candidate: CandidateChangePoint, series: Sequence[SupportsFloat], intervals: List[slice]
+        self,
+        candidate: CandidateChangePoint,
+        series: Sequence[SupportsFloat],
+        intervals: List[slice],
     ) -> (Sequence, Sequence):
         """
         Computes properties of the change point if the Candidate Change Point based on the provided intervals.
@@ -779,7 +780,9 @@ def get_sides(
                 left_interval = interval
                 right_interval = intervals[i + 1]
                 break
-            elif (interval.start is None or interval.start < candidate.index) and (interval.stop is None or candidate.index < interval.stop):
+            elif (interval.start is None or interval.start < candidate.index) and (
+                interval.stop is None or candidate.index < interval.stop
+            ):
                 # Split step
                 # Note: handles slices with omitted indexes:
                 #

otava/report.py

@@ -103,7 +103,37 @@ def __format_log_annotated(self, test_name: str) -> str:
     def __format_json(self, test_name: str) -> str:
         import json
 
-        return json.dumps({test_name: [cpg.to_json(rounded=True) for cpg in self.__change_points]})
+        return json.dumps(
+            {
+                test_name: [
+                    self.__format_change_point_group_json(cpg) for cpg in self.__change_points
+                ]
+            }
+        )
+
+    @staticmethod
+    def __format_change_point_group_json(cpg):
+        return {
+            "time": cpg.time,
+            "attributes": cpg.attributes,
+            "changes": [Report.__format_change_point_json(cp) for cp in cpg.changes.values()],
+        }
+
+    @staticmethod
+    def __format_change_point_json(cp):
+        cp = ChangePointSerializer(cp)
+        return {
+            "metric": cp.metric,
+            "index": int(cp.index),
+            "forward_change_percent": f"{cp.forward_change_percent():.0f}",
+            "backward_change_percent": f"{cp.backward_change_percent():.0f}",
+            "magnitude": f"{cp.magnitude():.6f}",
+            "mean_before": f"{cp.mean_before():.6f}",
+            "stddev_before": f"{cp.stddev_before():.6f}",
+            "mean_after": f"{cp.mean_after():.6f}",
+            "stddev_after": f"{cp.stddev_after():.6f}",
+            "pvalue": f"{cp.pvalue():.6f}",
+        }
 
     def __format_regressions_only(self, test_name: str) -> str:
         output = []

otava/serialization.py

@@ -0,0 +1,72 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from datetime import datetime
+from typing import Dict, List, Optional
+
+from pydantic import BaseModel, ConfigDict
+
+JsonScalar = str | int | float | bool | None
+
+
+class AnalysisOptionsModel(BaseModel):
+    model_config = ConfigDict(extra="forbid", validate_assignment=True)
+
+    window_len: int = 50
+    max_pvalue: float = 0.001
+    min_magnitude: float = 0.0
+    orig_edivisive: bool = False
+
+
+class MetricModel(BaseModel):
+    direction: Optional[int] = None
+    scale: Optional[float] = None
+    unit: str = ""
+
+
+class ChangePointModel(BaseModel):
+    metric: Optional[str] = None
+    index: int
+    qhat: float
+    forward_change_percent: float
+    magnitude: float
+    mean_before: float
+    stddev_before: float
+    mean_after: float
+    stddev_after: float
+    pvalue: float
+
+
+class ChangePointGroupModel(BaseModel):
+    time: int | float
+    attributes: Dict[str, JsonScalar]
+    changes: List[ChangePointModel]
+
+
+class AnalyzedSeriesModel(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=False)
+
+    test_name: str
+    time: List[int | float]
+    change_points_timestamp: datetime
+    branch_name: Optional[str] = None
+    options: AnalysisOptionsModel
+    metrics: Dict[str, MetricModel]
+    attributes: Dict[str, List[JsonScalar]]
+    data: Dict[str, List[Optional[float]]]
+    change_points: Dict[str, List[ChangePointGroupModel]]
+    weak_change_points: Dict[str, List[ChangePointGroupModel]]