diff --git a/docs/BASICS.md b/docs/BASICS.md index 352fdc52..8368bcde 100644 --- a/docs/BASICS.md +++ b/docs/BASICS.md @@ -19,6 +19,21 @@ # Basics +## Python persistence API + +`Series`, `AnalyzedSeries`, metrics, statistics, and change-point groups are Pydantic v2 +domain models. Persist an analysis with `analysis.model_dump(mode="json")`, and restore it +with `AnalyzedSeries.model_validate(payload)`. The JSON document remains compatible with +previous releases, including legacy string metric units and older flat change-point entries. + +`to_json()` and `from_json()` remain as deprecated wrappers for the current release and +emit `DeprecationWarning`. Migrate callers before the next breaking release, when those +wrappers and the `ChangePointSerializer` compatibility shim will be removed. + +The old `otava.serialization.*Model` imports also remain available for this release as +flat-persistence compatibility models. New code should use the domain models and +`AnalyzedSeries.model_validate()` instead. + ## Listing Available Tests ``` diff --git a/otava/analysis.py b/otava/analysis.py index 54cfdb9a..fdbfae56 100644 --- a/otava/analysis.py +++ b/otava/analysis.py @@ -16,7 +16,6 @@ # under the License. import copy -from dataclasses import dataclass, replace from typing import List, Optional, Sequence, SupportsFloat, Tuple from scipy.stats import ttest_ind_from_stats @@ -36,7 +35,6 @@ ) -@dataclass class TTestStats(BaseStats): """ Statistics related to the calculation of a two-sided Student's T-test. @@ -47,8 +45,7 @@ class TTestStats(BaseStats): degrees_of_freedom: int = 0 def copy(self): - # replace() preserves the subclass - return replace(self) + return self.model_copy(deep=True) def to_json(self): obj = super().to_json() diff --git a/otava/bigquery.py b/otava/bigquery.py index 47088639..f5f8ed71 100644 --- a/otava/bigquery.py +++ b/otava/bigquery.py @@ -28,7 +28,7 @@ ScalarQueryParameter = Any from otava._optional import import_optional_dependency -from otava.change_point_divisive.base import ChangePointGroup, ChangePointSerializer +from otava.change_point_divisive.base import ChangePointGroup from otava.test_config import BigQueryTestConfig @@ -106,7 +106,7 @@ def insert_change_point( attributes: Dict, change_point_group: ChangePointGroup, ): - change_point = ChangePointSerializer(change_point_group[metric_name]) + change_point = change_point_group[metric_name] kwargs = {**attributes, **{test.time_column: datetime.utcfromtimestamp(change_point_group.time)}} update_stmt = test.update_stmt.format( metric=metric_name, diff --git a/otava/change_point_divisive/base.py b/otava/change_point_divisive/base.py index b2a088e3..1f5a2d0f 100644 --- a/otava/change_point_divisive/base.py +++ b/otava/change_point_divisive/base.py @@ -45,24 +45,31 @@ .pivot() < - - > .pivot() """ -from dataclasses import dataclass, fields, replace from datetime import datetime, timezone from typing import Dict, Generic, List, Optional, Sequence, SupportsFloat, TypeVar +from warnings import warn import numpy as np from numpy.typing import NDArray +from pydantic import BaseModel, ConfigDict, field_validator, model_validator +JsonScalar = str | int | float | bool | None -@dataclass -class CandidateChangePoint: + +class DomainModel(BaseModel): + """Shared validation and assignment semantics for domain value objects.""" + + model_config = ConfigDict(validate_assignment=True, arbitrary_types_allowed=True) + + +class CandidateChangePoint(DomainModel): """Candidate for a change point. The point that maximizes Q-hat function on [start:end+1] slice""" index: int qhat: float -@dataclass -class BaseStats: +class BaseStats(DomainModel): """Abstract statistics class for change point. Implementation depends on the statistical test.""" # The pvalue for this change point. Exact value depends on the algorithm that was used. @@ -99,7 +106,7 @@ def calculate(left: Sequence[SupportsFloat], right: Sequence[SupportsFloat], pva def copy(self): """Return a copy of these statistics, preserving the concrete (sub)class.""" - return replace(self) + return self.model_copy(deep=True) def forward_rel_change(self, value_if_nan=0): """Relative change from left to right""" @@ -138,6 +145,7 @@ def stddev_after(self): return self.std_2 def to_json(self): + warn("BaseStats.to_json() is deprecated; use model_dump(mode='json')", DeprecationWarning, stacklevel=2) return { "forward_change_percent": f"{self.forward_change_percent():-0f}", "magnitude": f"{self.change_magnitude():-0f}", @@ -153,7 +161,6 @@ def to_json(self): GenericStats = TypeVar("GenericStats", bound=BaseStats) -@dataclass class ChangePoint(CandidateChangePoint, Generic[GenericStats]): """ ChangePoint class. @@ -182,7 +189,7 @@ def copy(self): :return: A deep copy of self, recursively calls also stats.copy(). """ - return ChangePoint( + return self.__class__( index=self.index, qhat=self.qhat, stats=self.stats.copy(), metric=self.metric ) @@ -198,11 +205,35 @@ def from_candidate( def to_candidate(self) -> CandidateChangePoint: """Downgrades Change Point to a Candidate Change Point. Used to recompute stats for Weak Change Points.""" - data = {f.name: getattr(self, f.name) for f in fields(CandidateChangePoint)} - return CandidateChangePoint(**data) + return CandidateChangePoint(index=self.index, qhat=self.qhat) def to_json(self): - return ChangePointSerializer(self).to_json() + warn("ChangePoint.to_json() is deprecated; use model_dump(mode='json')", DeprecationWarning, stacklevel=2) + return self.persistence_dict() + + def forward_change_percent(self) -> float: + return self.stats.forward_change_percent() + + def backward_change_percent(self) -> float: + return self.stats.backward_change_percent() + + def magnitude(self) -> float: + return self.stats.change_magnitude() + + def persistence_dict(self) -> dict: + """The historical flat change-point document used inside persisted analyses.""" + return { + "metric": self.metric, + "index": int(self.index), + "qhat": self.qhat, + "forward_change_percent": self.forward_change_percent(), + "magnitude": self.magnitude(), + "mean_before": self.stats.mean_before(), + "stddev_before": self.stats.stddev_before(), + "mean_after": self.stats.mean_after(), + "stddev_after": self.stats.stddev_after(), + "pvalue": self.stats.pvalue, + } class ChangePointSerializer(ChangePoint): @@ -214,10 +245,12 @@ 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 + warn( + "ChangePointSerializer is deprecated; use ChangePoint's model-derived values instead", + DeprecationWarning, + stacklevel=2, + ) + super().__init__(stats=cp.stats, index=cp.index, qhat=cp.qhat, metric=cp.metric) def forward_change_percent(self) -> float: return self.stats.forward_rel_change() * 100.0 @@ -244,22 +277,10 @@ def pvalue(self): return self.stats.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(), - } + return self.persistence_dict() -@dataclass -class ChangePointGroup: +class ChangePointGroup(DomainModel): """ A group of change points on multiple metrics, at the same point in time. @@ -275,15 +296,30 @@ class ChangePointGroup: :param changes: For each metric that has a change point at this time(stamp), the ChangePoint object. """ - time: float - attributes: Dict[str, str] + time: int | float + attributes: Dict[str, JsonScalar] # ChangePointGroup.changes.keys() stores the set of metrics that were used at this ChangePointGroup.time. changes: Dict[str, ChangePoint] + @field_validator("time") + @classmethod + def validate_time(cls, value): + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError("time must be numeric") + return value + + @model_validator(mode="after") + def validate_metric_mapping(self): + for metric, change in self.changes.items(): + if change.metric is None: + change.metric = metric + elif change.metric != metric: + raise ValueError(f"metric field is not internally consistent. {metric} != {change.metric}") + return self + def to_json(self): - changes = [] - for metric, cp in self.changes.items(): - changes.append(cp.to_json()) + warn("ChangePointGroup.to_json() is deprecated; use model_dump(mode='json')", DeprecationWarning, stacklevel=2) + changes = [cp.persistence_dict() for cp in self.changes.values()] return { "time": self.time, @@ -294,8 +330,7 @@ def to_json(self): def copy(self): new_attributes = {k: v for k, v in self.attributes.items()} new_changes = {metric: cp.copy() for metric, cp in self.changes.items()} - new_obj = ChangePointGroup(time=self.time, attributes=new_attributes, changes=new_changes) - return new_obj + return ChangePointGroup(time=self.time, attributes=new_attributes, changes=new_changes) def __getitem__(self, metric): return self.changes[metric] diff --git a/otava/change_point_divisive/significance_test.py b/otava/change_point_divisive/significance_test.py index 61ba0635..be86a1ed 100644 --- a/otava/change_point_divisive/significance_test.py +++ b/otava/change_point_divisive/significance_test.py @@ -15,11 +15,11 @@ # specific language governing permissions and limitations # under the License. -from dataclasses import dataclass, replace from typing import List, Optional, Type import numpy as np from numpy.typing import NDArray +from pydantic import field_validator from otava.change_point_divisive.base import ( BaseStats, @@ -30,17 +30,23 @@ ) -@dataclass class PermutationStats(BaseStats): '''Statistics for permutation significance test''' permuted_qhats: NDArray extreme_qhat_perm: int n_perm: int + @field_validator("permuted_qhats", mode="before") + @classmethod + def unwrap_single_array_tuple(cls, value): + # Older callers occasionally passed ``np.array(...),`` by accident; + # dataclasses accepted it, so retain that input shape during migration. + if isinstance(value, tuple) and len(value) == 1 and isinstance(value[0], np.ndarray): + return value[0] + return value + def copy(self): - c = replace(self) - c.permuted_qhats = self.permuted_qhats.copy() - return c + return self.model_copy(deep=True) def to_json(self): obj = super().to_json() diff --git a/otava/postgres.py b/otava/postgres.py index d00631d2..3bfc31f2 100644 --- a/otava/postgres.py +++ b/otava/postgres.py @@ -27,7 +27,7 @@ Connection = Any from otava._optional import import_optional_dependency -from otava.change_point_divisive.base import ChangePointGroup, ChangePointSerializer +from otava.change_point_divisive.base import ChangePointGroup from otava.test_config import PostgresTestConfig @@ -102,7 +102,7 @@ def insert_change_point( change_point_group: ChangePointGroup, ): cursor = self.__get_conn().cursor() - change_point = ChangePointSerializer(change_point_group[metric_name]) + change_point = change_point_group[metric_name] kwargs = {**attributes, **{test.time_column: datetime.utcfromtimestamp(change_point_group.time)}} update_stmt = test.update_stmt.format(metric=metric_name, **kwargs) cursor.execute( diff --git a/otava/report.py b/otava/report.py index 1f1e3dc8..4fdd39cd 100644 --- a/otava/report.py +++ b/otava/report.py @@ -21,7 +21,7 @@ from tabulate import tabulate -from otava.change_point_divisive.base import ChangePoints, ChangePointSerializer +from otava.change_point_divisive.base import ChangePoints from otava.series import Series from otava.util import format_timestamp, insert_multiple, remove_common_prefix @@ -87,8 +87,7 @@ def __format_log_annotated(self, test_name: str) -> str: col_width = col_widths[col_index] change = [c for m, c in cpg.changes.items() if m == col_name] if change: - change = ChangePointSerializer(change[0]) - change_percent = change.forward_change_percent() + change_percent = change[0].forward_change_percent() separator += "ยท" * col_width + " " info += f"{change_percent:+.1f}%".rjust(col_width) + " " else: @@ -121,18 +120,17 @@ def __format_change_point_group_json(cpg): @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}", + "mean_before": f"{cp.stats.mean_before():.6f}", + "stddev_before": f"{cp.stats.stddev_before():.6f}", + "mean_after": f"{cp.stats.mean_after():.6f}", + "stddev_after": f"{cp.stats.stddev_after():.6f}", + "pvalue": f"{cp.stats.pvalue:.6f}", } def __format_regressions_only(self, test_name: str) -> str: @@ -140,7 +138,6 @@ def __format_regressions_only(self, test_name: str) -> str: for cpg in self.__change_points: regressions = [] for metric_name, cp in cpg.changes.items(): - cp = ChangePointSerializer(cp) metric = self.__series.metrics[metric_name] if metric.direction * cp.forward_change_percent() < 0: regressions.append( diff --git a/otava/serialization.py b/otava/serialization.py index 1b95ad8b..7fa293dc 100644 --- a/otava/serialization.py +++ b/otava/serialization.py @@ -15,6 +15,8 @@ # specific language governing permissions and limitations # under the License. +"""Deprecated compatibility models for the pre-#78 persistence API.""" + from datetime import datetime from typing import Dict, List, Optional @@ -22,6 +24,15 @@ JsonScalar = str | int | float | bool | None +__all__ = [ + "AnalysisOptionsModel", + "AnalyzedSeriesModel", + "ChangePointGroupModel", + "ChangePointModel", + "JsonScalar", + "MetricModel", +] + class AnalysisOptionsModel(BaseModel): model_config = ConfigDict(extra="forbid", validate_assignment=True) diff --git a/otava/series.py b/otava/series.py index 897502db..8a79b43f 100644 --- a/otava/series.py +++ b/otava/series.py @@ -16,14 +16,21 @@ # under the License. import logging -from dataclasses import dataclass from datetime import datetime, timezone from typing import Dict, Iterable, List, Optional - -from pydantic import TypeAdapter +from warnings import warn + +from pydantic import ( + BaseModel, + ConfigDict, + PrivateAttr, + TypeAdapter, + field_validator, + model_serializer, + model_validator, +) from otava.analysis import ( - TTestStats, compute_change_points, compute_change_points_orig, ) @@ -34,31 +41,36 @@ ChangePointsByMetric, ChangePointsByTime, ) -from otava.serialization import AnalysisOptionsModel, AnalyzedSeriesModel, JsonScalar +JsonScalar = str | int | float | bool | None _datetime_adapter = TypeAdapter(datetime) -class AnalysisOptions(AnalysisOptionsModel): - pass +class DomainModel(BaseModel): + model_config = ConfigDict(validate_assignment=True, arbitrary_types_allowed=True, extra="forbid") + +class AnalysisOptions(DomainModel): + window_len: int = 50 + max_pvalue: float = 0.001 + min_magnitude: float = 0.0 + orig_edivisive: bool = False -@dataclass -class Metric: - direction: int - scale: float + +class Metric(DomainModel): + direction: Optional[int] = 1 + scale: Optional[float] = 1.0 unit: str def __init__(self, direction: int = 1, scale: float = 1.0, unit: str = ""): - self.direction = direction - self.scale = scale - self.unit = unit + super().__init__(direction=direction, scale=scale, unit=unit) def to_json(self): - return {"direction": self.direction, "scale": self.scale, "unit": self.unit} + warn("Metric.to_json() is deprecated; use model_dump(mode='json')", DeprecationWarning, stacklevel=2) + return self.model_dump(mode="json") -class Series: +class Series(DomainModel): """ Stores values of interesting metrics of all runs of a fallout test indexed by a single time variable. @@ -70,25 +82,48 @@ class Series: time: List[int | float] metrics: Dict[str, Metric] attributes: Dict[str, List[JsonScalar]] - data: Dict[str, List[float]] + data: Dict[str, List[Optional[float]]] def __init__( self, test_name: str, - branch: Optional[str], - time: List[int | float], - metrics: Dict[str, Metric], - data: Dict[str, List[float]], - attributes: Dict[str, List[JsonScalar]], + branch: Optional[str] = None, + time: Optional[List[int | float]] = None, + metrics: Optional[Dict[str, Metric]] = None, + data: Optional[Dict[str, List[Optional[float]]]] = None, + attributes: Optional[Dict[str, List[JsonScalar]]] = None, ): - self.test_name = test_name - self.branch = branch - self.time = time - self.metrics = metrics - self.attributes = attributes if attributes else {} - self.data = data - assert all(len(x) == len(time) for x in data.values()) - assert all(len(x) == len(time) for x in attributes.values()) + super().__init__( + test_name=test_name, + branch=branch, + time=[] if time is None else time, + metrics={} if metrics is None else metrics, + data={} if data is None else data, + attributes={} if attributes is None else attributes, + ) + # append() historically extends the caller-provided timeline in place. + # Keep that mutation behaviour while validation continues to happen at construction/assignment. + if time is not None: + time[:] = self.time + object.__setattr__(self, "time", time) + + @field_validator("time") + @classmethod + def validate_timestamps(cls, value): + if any(isinstance(timestamp, bool) or not isinstance(timestamp, (int, float)) for timestamp in value): + raise ValueError("time values must be numeric") + return value + + @model_validator(mode="after") + def validate_structure(self): + expected_length = len(self.time) + if self.metrics and set(self.data) != set(self.metrics): + raise ValueError("data and metrics must have the same metric names") + if any(len(values) != expected_length for values in self.data.values()): + raise ValueError("all data series must align with time") + if any(len(values) != expected_length for values in self.attributes.values()): + raise ValueError("all attribute series must align with time") + return self def attributes_at(self, index: int) -> Dict[str, JsonScalar]: result = {} @@ -117,60 +152,143 @@ def analyze(self, options: Optional[AnalysisOptions] = None) -> "AnalyzedSeries" return AnalyzedSeries(self, options) -class AnalyzedSeries: +class AnalyzedSeries(DomainModel): """ Time series data with computed change points. """ - __series: Series + series: Series options: AnalysisOptions + _change_points: Optional[ChangePointsByMetric] = PrivateAttr(default=None) + _weak_change_points: Optional[ChangePointsByMetric] = PrivateAttr(default=None) + _change_points_by_time: Optional[ChangePointsByTime] = PrivateAttr(default=None) + _change_points_timestamp: Optional[datetime] = PrivateAttr(default=None) def __init__( self, series: Series, - options: AnalysisOptions, + options: Optional[AnalysisOptions] = None, change_points: Optional[ChangePointsByMetric] = None, + weak_change_points: Optional[ChangePointsByMetric] = None, + change_points_timestamp: Optional[datetime] = None, ): - self.__series = series - self.options = options - 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 + super().__init__(series=series, options=options or AnalysisOptions()) + self._change_points = change_points + self._weak_change_points = ( + weak_change_points if weak_change_points is not None else ChangePointsByMetric() + ) if change_points is not None else None + self._change_points_timestamp = ( + change_points_timestamp or 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) + if self._change_points is None: + logging.info(f"Computing change points for test {self.series.test_name}...") + self._change_points, self._weak_change_points = self.__compute_change_points( + self.series, self.options + ) + self._change_points_timestamp = datetime.now(timezone.utc) @property def change_points(self) -> ChangePointsByMetric: self.__ensure_change_points_computed() - return self.__change_points + return self._change_points + + @change_points.setter + def change_points(self, value: Optional[ChangePointsByMetric]): + self._change_points = value + self._change_points_by_time = None @property def weak_change_points(self) -> ChangePointsByMetric: self.__ensure_change_points_computed() - return self.__weak_change_points + return self._weak_change_points + + @weak_change_points.setter + def weak_change_points(self, value: ChangePointsByMetric): + self._weak_change_points = value @property def change_points_timestamp(self) -> datetime: self.__ensure_change_points_computed() - return self.__change_points_timestamp + 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 + 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 + + @classmethod + def _parse_persistence_document(cls, value): + def parse_changes(by_metric): + result = {} + for metric, groups in by_metric.items(): + parsed_groups = [] + for group in groups: + changes = {} + for raw_change in group.get("changes", [group]): + change_metric = raw_change.get("metric") or metric + stats = raw_change.get("stats") or { + "mean_1": raw_change["mean_before"], + "mean_2": raw_change["mean_after"], + "std_1": raw_change["stddev_before"], + "std_2": raw_change["stddev_after"], + "pvalue": raw_change["pvalue"], + } + changes[change_metric] = ChangePoint( + index=raw_change["index"], + qhat=raw_change.get("qhat", 0.0), + metric=change_metric, + stats=stats, + ) + parsed_groups.append( + ChangePointGroup( + time=group["time"], attributes=group.get("attributes", {}), changes=changes + ) + ) + result[metric] = parsed_groups + return ChangePointsByMetric.from_dict(result) + + metrics = { + name: Metric(unit=metric) if isinstance(metric, str) else Metric.model_validate(metric) + for name, metric in value["metrics"].items() + } + return { + "series": Series( + value["test_name"], + value.get("branch_name"), + value["time"], + metrics, + value["data"], + value.get("attributes", {}), + ), + "options": value.get("options", {}), + "change_points": parse_changes(value.get("change_points", {})), + "weak_change_points": parse_changes(value.get("weak_change_points", {})), + "change_points_timestamp": _datetime_adapter.validate_python( + value.get("change_points_timestamp", datetime.now(timezone.utc)) + ), + } + + @model_validator(mode="wrap") + @classmethod + def validate_persistence_document(cls, value, handler): + """Accept both the domain representation and the historical flat document.""" + if isinstance(value, dict) and "test_name" in value: + parsed = cls._parse_persistence_document(value) + return cls( + parsed["series"], + AnalysisOptions.model_validate(parsed["options"]), + parsed["change_points"], + parsed["weak_change_points"], + parsed["change_points_timestamp"], ) - return self.__change_points_by_time + return handler(value) @staticmethod def __compute_change_points( @@ -265,7 +383,7 @@ def _validate_append(self, time, new_data, attributes): if not isinstance(attributes, dict): return ValueError("attributes must be a dict.") - max_time = max(self.__series.time) + max_time = max(self.series.time) for t in time: if t <= max_time: return ValueError( @@ -288,17 +406,17 @@ def append(self, time, new_data, attributes): raise err for t in time: - self.__series.time.append(t) - for m in self.__series.metrics.keys(): + self.series.time.append(t) + for m in self.series.metrics.keys(): if m in new_data.keys(): - self.__series.data[m] += new_data[m] + self.series.data[m] += new_data[m] for k, v in attributes.items(): - self.__series.attributes[k].append(v) + self.series.attributes[k].append(v) result = {} weak_change_points = {} - for metric in self.__series.data.keys(): + for metric in self.series.data.keys(): if metric not in new_data: if metric in self.weak_change_points: weak_change_points[metric] = self.weak_change_points.select_metrics(metric) @@ -313,10 +431,10 @@ def append(self, time, new_data, attributes): old_weak_cp = [ cp for cp in previous_weak_cp - if cp.index < len(self.__series.data[metric]) - new_data_len - 1 + if cp.index < len(self.series.data[metric]) - new_data_len - 1 ] change_points, weak_cps = compute_change_points( - self.__series.data[metric], + self.series.data[metric], window_len=self.options.window_len, max_pvalue=self.options.max_pvalue, min_magnitude=self.options.min_magnitude, @@ -330,9 +448,9 @@ def append(self, time, new_data, attributes): cp.metric = metric result[metric].append( ChangePointGroup( - time=self.__series.time[cp.index], + time=self.series.time[cp.index], changes={metric: cp}, - attributes=self.__series.attributes_at(cp.index), + attributes=self.series.attributes_at(cp.index), ) ) if metric not in weak_change_points: @@ -342,9 +460,9 @@ def append(self, time, new_data, attributes): cp.metric = metric weak_change_points[metric].append( ChangePointGroup( - time=self.__series.time[cp.index], + time=self.series.time[cp.index], changes={metric: cp}, - attributes=self.__series.attributes_at(cp.index), + attributes=self.series.attributes_at(cp.index), ) ) @@ -353,159 +471,101 @@ 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._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) + self._change_points_by_time = None + self._change_points_timestamp = datetime.now(timezone.utc) return r, w def test_name(self) -> str: - return self.__series.test_name + return self.series.test_name def branch_name(self) -> Optional[str]: - return self.__series.branch + return self.series.branch def len(self) -> int: - return len(self.__series.time) + return len(self.series.time) def time(self) -> List[int | float]: - return list(self.__series.time) + return list(self.series.time) - def data(self, metric: str) -> List[float]: - return [float(d) for d in self.__series.data[metric]] + def data(self, metric: str) -> List[Optional[float]]: + return [float(d) if d is not None else None for d in self.series.data[metric]] def attributes(self) -> Iterable[str]: - return self.__series.attributes.keys() + return self.series.attributes.keys() def attributes_at(self, index: int) -> Dict[str, JsonScalar]: - return self.__series.attributes_at(index) + return self.series.attributes_at(index) def attribute_values(self, attribute: str) -> List[JsonScalar]: - return self.__series.attributes[attribute] + return self.series.attributes[attribute] def metric_names(self) -> Iterable[str]: - return self.__series.metrics.keys() + return self.series.metrics.keys() def metric(self, name: str) -> Metric: - return self.__series.metrics[name] + return self.series.metrics[name] - def to_json(self): + @model_serializer(mode="plain") + def serialize_persistence_document(self, info): change_points_json = {} - cpbm = self.change_points.by_metric() - for metric_name in self.change_points.metrics(): + cpbm = self.change_points.by_metric() if self.change_points else ChangePointsByMetric() + for metric_name in cpbm.metrics(): change_points_json[metric_name] = [] for cp in cpbm.select_metrics(metric_name): - change_points_json[metric_name].append(cp.to_json()) + change_points_json[metric_name].append( + { + "time": cp.time, + "attributes": cp.attributes, + "changes": [change.persistence_dict() for change in cp.changes.values()], + } + ) weak_change_points_json = {} wcpbm = self.weak_change_points.by_metric() for metric_name in self.weak_change_points.metrics(): weak_change_points_json[metric_name] = [] for cp in wcpbm.select_metrics(metric_name): - weak_change_points_json[metric_name].append(cp.to_json()) + weak_change_points_json[metric_name].append( + { + "time": cp.time, + "attributes": cp.attributes, + "changes": [change.persistence_dict() for change in cp.changes.values()], + } + ) data_json = {} - for metric, datapoints in self.__series.data.items(): + for metric, datapoints in self.series.data.items(): data_json[metric] = [float(d) if d is not None else None for d in datapoints] metrics_json = {} - for metric, unit in self.__series.metrics.items(): - metrics_json[metric] = unit.to_json() + for metric, unit in self.series.metrics.items(): + metrics_json[metric] = unit.model_dump(mode="json") payload = { "test_name": self.test_name(), "time": self.time(), - "change_points_timestamp": self.change_points_timestamp, + "change_points_timestamp": _datetime_adapter.dump_python(self.change_points_timestamp, mode=info.mode), "branch_name": self.branch_name(), "options": self.options.model_dump(mode="json"), "metrics": metrics_json, - "attributes": self.__series.attributes, + "attributes": self.series.attributes, "data": data_json, "change_points": change_points_json, "weak_change_points": weak_change_points_json, } - return AnalyzedSeriesModel.model_validate(payload).model_dump(mode="json") + if info.include is not None: + payload = {key: value for key, value in payload.items() if key in info.include} + if info.exclude is not None: + payload = {key: value for key, value in payload.items() if key not in info.exclude} + return payload + + def to_json(self): + warn("AnalyzedSeries.to_json() is deprecated; use model_dump(mode='json')", DeprecationWarning, stacklevel=2) + return self.model_dump(mode="json") @classmethod def from_json(cls, analyzed_json): - def stats_from_json(cp_json): - return TTestStats( - mean_1=cp_json["mean_before"], - mean_2=cp_json["mean_after"], - std_1=cp_json["stddev_before"], - std_2=cp_json["stddev_after"], - pvalue=cp_json["pvalue"], - ) - - def change_point_from_json(metric, cp_json): - return ChangePoint( - index=cp_json["index"], - qhat=cp_json.get("qhat", 0.0), - metric=cp_json.get("metric") or metric, - stats=stats_from_json(cp_json), - ) - - def change_points_from_json(change_points_json): - new_change_points = {} - for metric, groups in change_points_json.items(): - new_change_points[metric] = [] - for group in groups: - if "changes" in group: - changes = { - cp_json.get("metric") or metric: change_point_from_json(metric, cp_json) - for cp_json in group["changes"] - } - new_change_points[metric].append( - ChangePointGroup( - time=group["time"], - attributes=group["attributes"], - changes=changes, - ) - ) - else: - new_change_points[metric].append( - ChangePointGroup( - time=group["time"], - attributes=group.get("attributes", {}), - changes={metric: change_point_from_json(metric, group)}, - ) - ) - return ChangePointsByMetric.from_dict(new_change_points) - - new_metrics = {} - - for metric_name, metric_json in analyzed_json["metrics"].items(): - if isinstance(metric_json, dict): - new_metrics[metric_name] = Metric( - metric_json.get("direction"), - metric_json.get("scale"), - metric_json.get("unit", ""), - ) - else: - new_metrics[metric_name] = Metric(None, None, metric_json) - - new_series = Series( - analyzed_json["test_name"], - analyzed_json["branch_name"], - analyzed_json["time"], - new_metrics, - analyzed_json["data"], - analyzed_json["attributes"], - ) - - new_options = AnalysisOptions.model_validate(analyzed_json["options"]) - - new_change_points = change_points_from_json(analyzed_json["change_points"]) - new_weak_change_points = change_points_from_json( - analyzed_json.get("weak_change_points", {}) - ) - - analyzed_series = cls(new_series, new_options, new_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_json["change_points_timestamp"] - ) - - return analyzed_series + warn("AnalyzedSeries.from_json() is deprecated; use model_validate()", DeprecationWarning, stacklevel=2) + return cls.model_validate(analyzed_json) diff --git a/otava/slack.py b/otava/slack.py index 22ee8b5e..27e1155e 100644 --- a/otava/slack.py +++ b/otava/slack.py @@ -30,7 +30,7 @@ WebClient = Any from otava._optional import import_optional_dependency -from otava.change_point_divisive.base import ChangePointGroup, ChangePointSerializer +from otava.change_point_divisive.base import ChangePointGroup from otava.data_selector import DataSelector from otava.series import AnalyzedSeries @@ -211,8 +211,7 @@ def __dates_change_points_summary(self, test_changes: Dict[str, ChangePointGroup fields.append(f"*{test_name}*") summary = "" for metric, change in group.changes.items(): - c = ChangePointSerializer(change) - change_percent = c.forward_change_percent() + change_percent = change.forward_change_percent() change_emoji = self.__get_change_emoji(test_name, change) if isinf(change_percent): report_percent = change_percent @@ -237,8 +236,7 @@ def __dates_change_points_summary(self, test_changes: Dict[str, ChangePointGroup def __get_change_emoji(self, test_name, change): metric_direction = self.test_analyzed_series[test_name].metric(change.metric).direction - c = ChangePointSerializer(change) - regression = metric_direction * c.forward_change_percent() + regression = metric_direction * change.forward_change_percent() if regression >= 0: return ":large_blue_circle:" else: diff --git a/tests/series_test.py b/tests/series_test.py index cd549662..09acd718 100644 --- a/tests/series_test.py +++ b/tests/series_test.py @@ -17,14 +17,19 @@ import json import time +import warnings from datetime import datetime from random import random import pytest -from pydantic import ValidationError +from pydantic import TypeAdapter, ValidationError from otava.change_point_divisive.base import ChangePointSerializer -from otava.serialization import AnalysisOptionsModel, AnalyzedSeriesModel +from otava.serialization import ( + AnalyzedSeriesModel, + ChangePointGroupModel, + ChangePointModel, +) from otava.series import AnalysisOptions, AnalyzedSeries, Metric, Series @@ -57,7 +62,6 @@ def test_analysis_options_is_pydantic_model(): orig_edivisive=True, ) - assert isinstance(options, AnalysisOptionsModel) assert options.model_dump(mode="json") == { "window_len": 25, "max_pvalue": 0.05, @@ -168,7 +172,7 @@ def test_div_by_zero(): analyzed_series = test.analyze() change_points = analyzed_series.change_points_by_time - cpjson = analyzed_series.to_json() + cpjson = analyzed_series.model_dump(mode="json") assert cpjson assert len(change_points) == 2 assert change_points[0].time == 3 @@ -295,6 +299,87 @@ def test_analyzed_series_json_round_trip(): assert restored.to_json()["weak_change_points"] == payload["weak_change_points"] +def test_pydantic_persistence_api_and_deprecation_wrappers(): + series = Series( + "test", + None, + [1, 2], + {"latency": Metric(1, 1.0, "ms")}, + {"latency": [1.0, 2.0]}, + {"commit": [None, "abc"]}, + ) + analysis = series.analyze() + payload = analysis.model_dump(mode="json") + + assert json.loads(json.dumps(payload)) == payload + assert AnalyzedSeries.model_validate(payload).model_dump(mode="json") == payload + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + assert analysis.to_json() == payload + assert AnalyzedSeries.from_json(payload).model_dump(mode="json") == payload + assert [warning.category for warning in caught] == [DeprecationWarning, DeprecationWarning] + + +def test_persistence_validation_uses_every_pydantic_entry_point(): + analysis = Series( + "test", + time=[1, 2], + metrics={"latency": Metric(unit="ms")}, + data={"latency": [1.0, None]}, + ).analyze() + payload = analysis.model_dump(mode="json") + + restored = [ + AnalyzedSeries.model_validate(payload), + AnalyzedSeries.model_validate_json(json.dumps(payload)), + TypeAdapter(AnalyzedSeries).validate_python(payload), + ] + + assert [item.model_dump(mode="json") for item in restored] == [payload, payload, payload] + + +def test_persistence_serializer_is_used_by_json_and_honors_top_level_filters(): + analysis = Series( + "test", + time=[1, 2], + metrics={"latency": Metric(unit="ms")}, + data={"latency": [1.0, 2.0]}, + ).analyze() + + assert "data" not in analysis.model_dump(exclude={"data"}) + assert set(analysis.model_dump(include={"test_name", "data"})) == {"test_name", "data"} + assert "test_name" in json.loads(analysis.model_dump_json()) + + +def test_series_preserves_validated_timeline_list_and_empty_weak_change_points(): + timeline = ["1", 2] + series = Series("test", time=timeline) + + assert series.time is timeline + assert timeline == [1, 2] + assert AnalyzedSeries(series, change_points={}).model_dump()["weak_change_points"] == {} + + +def test_legacy_serialization_models_accept_their_original_flat_shape(): + point = ChangePointModel.model_validate( + { + "metric": "latency", + "index": 1, + "qhat": 0.1, + "forward_change_percent": 2.0, + "magnitude": 0.02, + "mean_before": 10.0, + "stddev_before": 1.0, + "mean_after": 10.2, + "stddev_after": 1.1, + "pvalue": 0.01, + } + ) + group = ChangePointGroupModel(time=1, attributes={}, changes=[point]) + + assert group.model_dump()["changes"][0]["mean_before"] == 10.0 + + def test_analyzed_series_json_round_trip_through_json_module(): series_1 = [1.02, 0.95, 0.99, 1.00, 1.12, 0.90, 0.50, 0.51, 0.48, 0.48, 0.55] series_2 = [2.02, 2.03, 2.01, 2.04, 1.82, 1.85, 1.79, 1.81, 1.80, 1.76, 1.78]