This is an automated email from the ASF dual-hosted git repository. henrikingo pushed a commit to branch UnifyChangePointClasses in repository https://gitbox.apache.org/repos/asf/otava.git
commit ae3c01926c3a66c83a882c698727c99c63becca0 Author: Henrik Ingo <[email protected]> AuthorDate: Sat May 30 00:32:12 2026 +0300 Unify the two ChangePoint classes and add container classes * Unify the ChangePoint_ class in hunter code and the new ChangePoint introduced by the new edivisive implementation Then it got out of hand a bit ... * Separate index and timestamp into different domains. cp.index is used in the context of a single metric and its history of results. Time and commit otoh are on the ChangePointGroup level (essentially a "row"). Note that different metrics can now have different cp.index for the same cpg.time or cpg.attributes['commit'], if they have a different history. * Introduce a ChangePoints class which is just a list of ChangePointGroups but actually comes with 2 different implementations. The last one is supposed to become the class you are left holding once all the change points are computed. Until now we had lots of nice classes for each step of computation, but in the end you were left holding a dict[str, ChangePointGroup]. The new class now encapsulates that dict, --- otava/analysis.py | 23 +- otava/bigquery.py | 4 +- otava/change_point_divisive/base.py | 494 ++++++++++++++++++++++- otava/change_point_divisive/detector.py | 4 +- otava/change_point_divisive/significance_test.py | 6 +- otava/main.py | 5 +- otava/postgres.py | 7 +- otava/report.py | 15 +- otava/series.py | 240 ++++------- otava/slack.py | 11 +- tests/change_point_divisive_test.py | 14 +- tests/report_test.py | 9 +- tests/series_test.py | 119 ++++-- 13 files changed, 706 insertions(+), 245 deletions(-) diff --git a/otava/analysis.py b/otava/analysis.py index e57453b..84c69c2 100644 --- a/otava/analysis.py +++ b/otava/analysis.py @@ -25,7 +25,7 @@ from scipy.stats import ttest_ind_from_stats from otava.change_point_divisive.base import ( BaseStats, CandidateChangePoint, - ChangePointOtava, + ChangePoint, GenericStats, SignificanceTester, ) @@ -98,11 +98,11 @@ class TTestStats(BaseStats): # Generic Change Point List -GenCPList = List[ChangePointOtava[GenericStats]] +GenCPList = List[ChangePoint[GenericStats]] # Permutation Change Point List -PermCPList = List[ChangePointOtava[PermutationStats]] +PermCPList = List[ChangePoint[PermutationStats]] # T-test Change Point List -TtestCPList = List[ChangePointOtava[TTestStats]] +TtestCPList = List[ChangePoint[TTestStats]] class TTestSignificanceTester(SignificanceTester): @@ -112,6 +112,7 @@ class TTestSignificanceTester(SignificanceTester): This test is good if the data between the change points have normal distribution. It works well even with tiny numbers of points (<10). """ + def compare(self, left: Sequence[SupportsFloat], right: Sequence[SupportsFloat]) -> TTestStats: if len(left) == 0 or len(right) == 0: raise ValueError @@ -130,8 +131,11 @@ class TTestSignificanceTester(SignificanceTester): return TTestStats(mean_1=mean_l, mean_2=mean_r, std_1=std_l, std_2=std_r, pvalue=p) def change_point( - self, candidate: CandidateChangePoint, series: Sequence[SupportsFloat], intervals: List[slice] - ) -> ChangePointOtava[TTestStats]: + self, + candidate: CandidateChangePoint, + series: Sequence[SupportsFloat], + intervals: List[slice], + ) -> ChangePoint[TTestStats]: """ Computes properties of the change point if the Candidate Change Point based on the provided intervals. @@ -164,11 +168,13 @@ class TTestSignificanceTester(SignificanceTester): right_interval = slice(candidate.index, interval.stop) break else: - raise ValueError(f"Candidate Change Point at index={candidate.index} doesn't correspond to any interval in {intervals}.") + raise ValueError( + f"Candidate Change Point at index={candidate.index} doesn't correspond to any interval in {intervals}." + ) left = series[left_interval] right = series[right_interval] stats = self.compare(left, right) - return ChangePointOtava.from_candidate(candidate, stats) + return ChangePoint.from_candidate(candidate, stats) def fill_missing(data: Sequence[SupportsFloat]): @@ -202,7 +208,6 @@ def merge( """ tester = TTestSignificanceTester(max_pvalue) while change_points: - # Select the change point with weakest unacceptable P-value # If all points have acceptable P-values, select the change-point with # the least relative change: diff --git a/otava/bigquery.py b/otava/bigquery.py index 33816b9..34f329d 100644 --- a/otava/bigquery.py +++ b/otava/bigquery.py @@ -22,7 +22,7 @@ from typing import Dict, List, Optional from google.cloud import bigquery from google.oauth2 import service_account -from otava.analysis import ChangePointOtava +from otava.analysis import ChangePoint from otava.test_config import BigQueryTestConfig @@ -87,7 +87,7 @@ class BigQuery: test: BigQueryTestConfig, metric_name: str, attributes: Dict, - change_point: ChangePointOtava, + change_point: ChangePoint, ): kwargs = {**attributes, **{test.time_column: datetime.utcfromtimestamp(change_point.time)}} update_stmt = test.update_stmt.format( diff --git a/otava/change_point_divisive/base.py b/otava/change_point_divisive/base.py index e50d2d4..275a61f 100644 --- a/otava/change_point_divisive/base.py +++ b/otava/change_point_divisive/base.py @@ -14,23 +14,58 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +""" +Hierarchy of ChangePoint classes: + CandidateChangePoint <--> ChangePoint --> ChangePointSerializer + .index .index .to_json() + , .stats .get_this_or_that() + ^ + / `------BaseStats + ^ .pvalue + / ` + `GenericStats + / TTestStats + PermutationStats + / + ChangePointGroup + .time + .attributes.commit + .changes[metric, ChangePoint] + # Essentially a row: One or more ChangePoint at the same commit/time + + / + ChangePoints + # Typically all change points for a given test / run / etc + ^ + | + ^ + ChangePointsByTime `ChangePointsByMetric + .change_points: list(ChangePointGroup) .change_points: dict[metric, list(ChangePointGroup)] + .pivot() < - - > .pivot() +""" + +from collections import OrderedDict from dataclasses import dataclass, fields -from typing import Generic, List, Optional, TypeVar +from datetime import UTC, datetime +from typing import Dict, Generic, List, Optional, TypeVar from numpy.typing import NDArray @dataclass class CandidateChangePoint: - '''Candidate for a change point. The point that maximizes Q-hat function on [start:end+1] slice''' + """Candidate for a change point. The point that maximizes Q-hat function on [start:end+1] slice""" + index: int qhat: float @dataclass class BaseStats: - '''Abstract statistics class for change point. Implementation depends on the statistical test.''' + """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. pvalue: float @@ -39,16 +74,36 @@ GenericStats = TypeVar("GenericStats", bound=BaseStats) @dataclass -class ChangePointOtava(CandidateChangePoint, Generic[GenericStats]): - '''Change point class, defined by index and signigicance test statistic.''' +class ChangePoint(CandidateChangePoint, Generic[GenericStats]): + """ + ChangePoint class. + + Defined by index and signigicance test statistic. + This class is the basic change point that is used during computation + and returned as a result. This class does not however carry additional + attributes like metric, time, or commit sha. Those are in ChangePointGroup + and ChangePoints. + Note that while in theory the index, commit sha, an the time(stamp) should + all be the same, in practice they aren't always. For example if at some point + during a tests lifetime, more metrics are added to the output, then different + metrics will have different histories and therefore their indexes start from + different locations. + To use time(stamp), metric name or timestamp, to access change points, please + use the ChangePointGroup and ChangePoints classes. + """ + stats: GenericStats + # Which metric this change point belongs to. (This is redundant and for convenience.) + metric: Optional[str] = None def __eq__(self, other): - '''Helpful to identify new Change Points during divisive algorithm''' + """Helpful to identify new Change Points during divisive algorithm""" return isinstance(other, self.__class__) and self.index == other.index @classmethod - def from_candidate(cls, candidate: CandidateChangePoint, stats: GenericStats) -> 'ChangePointOtava[GenericStats]': + def from_candidate( + cls, candidate: CandidateChangePoint, stats: GenericStats + ) -> "ChangePoint[GenericStats]": return cls( index=candidate.index, qhat=candidate.qhat, @@ -56,19 +111,417 @@ class ChangePointOtava(CandidateChangePoint, Generic[GenericStats]): ) def to_candidate(self) -> CandidateChangePoint: - '''Downgrades Change Point to a Candidate Change Point. Used to recompute stats for Weak Change Points.''' + """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) + def to_json(self, rounded=True): + cps = ChangePointSerializer(self) + return cps.to_json(rounded) + + +class ChangePointSerializer(ChangePoint): + """ + Utility class with getters and json serialization for a ChangePoint. + + TODO: Maintaining this is tedious. We should replace it with pydantic or some + other standard solution that provides json serialization. + """ + + def __init__(self, cp: ChangePoint[GenericStats]): + self.stats = cp.stats + self.index = cp.index + self.metric = cp.metric + + def forward_change_percent(self) -> float: + return self.stats.forward_rel_change() * 100.0 + + def backward_change_percent(self) -> float: + return self.stats.backward_rel_change() * 100.0 + + def magnitude(self): + return self.stats.change_magnitude() + + def mean_before(self): + return self.stats.mean_1 + + def mean_after(self): + return self.stats.mean_2 + + def stddev_before(self): + return self.stats.std_1 + + def stddev_after(self): + return self.stats.std_2 + + 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}", + "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(), + } + + +@dataclass +class ChangePointGroup: + """A group of change points on multiple metrics, at the same time""" + + 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): + changes = [] + for metric, cp in self.changes.items(): + changes.append(cp.to_json(rounded=rounded)) + + return { + "time": self.time, + "attributes": self.attributes, + "changes": changes, + } + + def __getitem__(self, metric): + return self.changes[metric] + + def metrics(self): + return self.changes.keys() + + def commit(self, idx: int): + return self.attribute_at(idx).get("commit") + + def datetime(self): + return datetime.fromtimestamp(self.time, UTC) + + def select_metrics(self, m: list[str] | str): + if not isinstance(m, list): + m = [m] + filtered = ChangePointGroup(time=self.time, attributes=self.attributes, changes={}) + for metric, cp in self.changes.items(): + if metric in m: + filtered.changes[metric] = cp + return filtered + + def set(self, metric: str, cp: ChangePoint): + self.changes[metric] = cp + + def __iter__(self): + return iter([v for v in list(self.changes.values())]) + + +class ChangePoints: + """ + A list of ChangePointGroup objects. + + Typical usage of this would be to hold all the change points over a history of a single test, + the test producing one or more metrics. Note that this is a sparse structure: It is NOT + guaranteed that each row (each GhangePointGroup) has each metric. Similarly it is not guaranteed + that a given metric will hold the full sequence. + + Companion class ChangePointsByMetric is expected to provide functionally equivalent interface, but + storing each series separately by metric, which is used in parts of the code base, in particular, what + Series.analyze() returns. + Subclass ChangePointsByTime is this same class, but can be used if you explicitly want to mark the ordering. + """ + + def __init__(self, cps=None): + if isinstance(cps, dict) and not isinstance(cps, OrderedDict): + raise TypeError( + "ChangePointsByTime doesn't accept a dict() as constructor input. Did you want ChangePointsByMetric()?" + ) + if isinstance(cps, OrderedDict): + for k, v in cps.items(): + if not isinstance(k, float): + raise TypeError( + "ChangePointsByTime with OrderedDict() as constructor input requires the keys to be float (timestamps)?" + ) + if not isinstance(v, ChangePointGroup): + raise TypeError( + "ChangePointsByTime input must be an OrderedDict() of ChangePointGroup objects as values." + ) + self.change_points = sorted(cps, key=lambda cpg: cpg.time) + return + if cps is None: + self.change_points = [] + return + + if isinstance(cps, ChangePointsByTime): + self.change_points = sorted(cps.change_points, key=lambda cpg: cpg.time) + return + if isinstance(cps, ChangePointsByMetric): + self.change_points = cps.pivot().change_points + return + + if not isinstance(cps, list): + cps = [cps] + for obj in sorted(cps, key=lambda cpg: cpg.time): + if not isinstance(obj, ChangePointGroup): + t = type(obj) + raise TypeError( + f"ChangePoints() takes as argument one or more ChangePointGroup objects. Got {t}." + ) + self.change_points = cps + + def append(self, cpg: ChangePointGroup): + if not isinstance(cpg, ChangePointGroup): + raise TypeError("ChangePoints.append() takes as argument one ChangePointGroup.") + + if (not self.change_points) or cpg.time > self.change_points[-1].time: + self.change_points.append(cpg) + elif self.change_points and cpg.time == self.change_points[-1].time: + for metric, cp in cpg.changes.items(): + if metric in self.change_points[-1].changes: + raise KeyError("Duplicate keys. Shouldn't happen.") + self.change_points[-1].changes[metric] = cp + else: + # TODO: logging + # print(self.change_points) + # print(cpg) + raise ValueError( + "ChangePoints.append() can only be used such that time is monotonically increasing" + ) + + def extend(self, cps): + errmsg = "ChangePoints.extend() takes as argument a list of ChangePointGroup objects." + if not isinstance(cps, list): + raise TypeError(errmsg) + for obj in cps: + if not isinstance(obj, ChangePointGroup): + raise TypeError(errmsg) + if (not self.change_points) or obj.time > self.change_points[-1].time: + self.change_points.append(obj) + else: + raise ValueError( + "ChangePoints.extend() can only be used such that time is monotonically increasing" + ) + + def items(self): + return self.pivot().items() + + def __iter__(self): + return iter(self.change_points) + + def __len__(self): + return len(self.change_points) + + def __getitem__(self, n): + return self.change_points[n] + + def metrics(self) -> set: + all_metrics = set() + for row in self.change_points: + all_metrics.add(row.metrics()) + return all_metrics + + def select_metrics(self, m: list[str] | str): + """ + Get a new ChangePoints object holding only the given metric(s). + + If you think of a ChangePoints object as timestamps being rows, and + the metrics being columns, then this returns a single column. + + Note: The internal data structure doesn't do anything to make this + request efficient. This will loop over all ChangePointGroups. + Use ChangePointsByMetric if you need this to be fast. + """ + filtered = ChangePoints() + for cpg in self.change_points: + filtered.append(cpg.select_metrics(m)) + return filtered + + def get_change_points_for_metric(self, m: str): + single_metric = self.select_metrics(m) + return [list(cpg.changes.values())[0] for cpg in single_metric.change_points] + + def at_timestamp(self, t: float): + for cpg in self.change_points: + if cpg.time == t: + return cpg + if abs(cpg.time - t) < 0.0001: + return cpg + raise LookupError(t) + + def at_commit(self, sha: str): + for row in self: + if row.attributes['commit'] == sha: + return row + raise LookupError(sha) + + def pivot(self): + """ + Return the same object as ChangePointsByMetric. + """ + by_metric = ChangePointsByMetric() + + for row in sorted(self.change_points, key=lambda cpg: cpg.time): + assert isinstance(row, ChangePointGroup) + # append() does the necessary shuffling into separate columns + by_metric.append(row) + + +class ChangePointsByTime(ChangePoints): + pass + + +class ChangePointsByMetric(ChangePoints): + """ + Provides same interface as ChangePoints, but internally stores with metric first. + """ + + def __init__(self, cps=None): + if isinstance(cps, ChangePointsByMetric): + self.change_points = cps.change_points + if isinstance(cps, ChangePointsByTime): + self.change_points = cps.pivot().change_points + + if cps is None: + self.change_points = OrderedDict() + return + if isinstance(cps, list): + cpm = [] + for obj in sorted(cps, key=lambda cpg: cpg.time): + if not isinstance(obj, ChangePointGroup): + t = type(obj) + raise TypeError( + f"ChangePointsByMetric() takes as input a list of ChangePointGroup objects. Got {t}." + ) + cpm.append(obj) + + self.change_points = cpm + if isinstance(cps, dict): + # We actually don't need the ordering in this case, but we want the type to match the other class + self.change_points = OrderedDict() + for metric, cpglist in cps.items(): + for cpg in cpglist: + if not isinstance(cpg, ChangePointGroup): + raise TypeError( + "ChangePointsByMetric takes as constructor argument a dict of ChangePointGroup objects: dict[str, list[ChangePointGroup]]" + ) + self.change_points[metric] = sorted(cpglist, key=lambda cpg: cpg.time) + + def pivot(self): + # Now we pivot (metric,time) to (time,metric) so that we return ChangePoints() objects + intermediate = [] + for metric, points in self.change_points.items(): + for cpg in sorted(points, key=lambda cpg: cpg.time): + assert isinstance(cpg, ChangePointGroup) + intermediate.append(cpg) + cp_by_time = ChangePointsByTime() + for cpg in sorted(intermediate, key=lambda cpg: cpg.time): + cp_by_time.append(cpg) + return cp_by_time + + def append(self, cpg: ChangePointGroup): + if not isinstance(cpg, ChangePointGroup): + raise TypeError("ChangePoints.append() takes as argument one ChangePointGroup.") + for metric in cpg.metrics(): + self.change_points[metric].append(cpg.select_metrics(metric)) + + def extend(self, cps): + errmsg = "ChangePoints.extend() takes as argument a list of ChangePointGroup objects." + if not isinstance(cps, list): + raise TypeError(errmsg) + for obj in cps: + if not isinstance(obj, ChangePointGroup): + raise TypeError(errmsg) + self.append(obj) + + def items(self): + return self.change_points.items() + + def __iter__(self): + return self.pivot().__iter__() + + def __len__(self): + return max([len(cpg) for metric, cpg in self.change_points.items()]) + + def __getitem__(self, n): + if not isinstance(n, int): + raise KeyError("n must be integer index") + cpgrow = [cpg[n] for metric, cpg in self.change_points.items() if len(cpg) > n] + if not cpgrow: + KeyError(f"Nice try {n}. This ChangePoints object only has {len(self)} items.") + cpg = None + for c in cpgrow: + if cpg is None: + # First element + cpg = c + continue + for metric in cpg.metrics(): + cpg[metric] = c + + return cpg + + def metrics(self): + return set(self.change_points.keys()) + + def select_metrics(self, m: list[str] | str): + """ + Get a new ChangePoints object holding only the given metric(s). + """ + if not isinstance(m, list): + if not isinstance(m, str): + TypeError("ChangePoints.select_metrics() takes as argument a str or a list of str.") + m = [m] + + filtered = ChangePointsByMetric() + for metric in m: + filtered.change_points[metric] = self.change_points[metric] + return filtered + + def get_change_points_for_metric(self, m: str): + single_metric = self.select_metrics(m) + metric_change_points = [] + for metric, cpg in single_metric.change_points.items(): + for c in cpg: + metric_change_points.append(c.changes[metric]) + return metric_change_points + + def at_timestamp(self, t: float): + """ + This is slow, please consider using ChangePointsByTime instead. + """ + return self.pivot().at_timestamp(t) + + def at_commit(self, sha: str): + """ + This is slow, please consider using ChangePointsByTime instead. + """ + return self.pivot().at_commit(sha) + class SignificanceTester(Generic[GenericStats]): - '''Abstract class for significance tester''' + """Abstract class for significance tester""" def __init__(self, max_pvalue: float): self.max_pvalue = max_pvalue - def get_intervals(self, change_points: List[ChangePointOtava[GenericStats]]) -> List[slice]: - '''Returns list of slices of the series. Change points must be sorted by index.''' + def get_intervals(self, change_points: List[ChangePoint[GenericStats]]) -> List[slice]: + """Returns list of slices of the series. Change points must be sorted by index.""" assert all( change_points[i].index <= change_points[i + 1].index for i in range(len(change_points) - 1) @@ -82,22 +535,25 @@ class SignificanceTester(Generic[GenericStats]): ] return [interval for interval in intervals if interval.start != interval.stop] - def is_significant(self, point: ChangePointOtava[GenericStats]) -> bool: - '''Compares ChangePointOtava to level of significance max_pvalue''' + def is_significant(self, point: ChangePoint[GenericStats]) -> bool: + """Compares ChangePoint to level of significance max_pvalue""" return point.stats.pvalue <= self.max_pvalue - def change_point(self, candidate: CandidateChangePoint, series: NDArray, intervals: List[slice]) -> ChangePointOtava[GenericStats]: - '''Computes stats for a change point candidate and wraps it into ChangePointOtava class''' + def change_point( + self, candidate: CandidateChangePoint, series: NDArray, intervals: List[slice] + ) -> ChangePoint[GenericStats]: + """Computes stats for a change point candidate and wraps it into ChangePoint class""" ... class Calculator: - '''Abstract class for calculator. Calculator provides an interface to get best change point candidate''' + """Abstract class for calculator. Calculator provides an interface to get best change point candidate""" + def __init__(self, series: NDArray): self.series = series def get_next_candidate(self, intervals: List[slice]) -> Optional[CandidateChangePoint]: - '''Returns list of existing change points to find next best change point candidate.''' + """Returns list of existing change points to find next best change point candidate.""" candidates = [ self.get_candidate_change_point(interval=interval) for interval in intervals @@ -109,6 +565,6 @@ class Calculator: return candidate def get_candidate_change_point(self, interval: slice) -> CandidateChangePoint: - '''Given start and end indexes return best candidate for a change point. - Note that start and end are indexes of the first and last element, i.e. a slice [start:end+1].''' + """Given start and end indexes return best candidate for a change point. + Note that start and end are indexes of the first and last element, i.e. a slice [start:end+1].""" ... diff --git a/otava/change_point_divisive/detector.py b/otava/change_point_divisive/detector.py index e2026e9..8945372 100644 --- a/otava/change_point_divisive/detector.py +++ b/otava/change_point_divisive/detector.py @@ -21,7 +21,7 @@ import numpy as np from otava.change_point_divisive.base import ( Calculator, - ChangePointOtava, + ChangePoint, GenericStats, SignificanceTester, ) @@ -32,7 +32,7 @@ class ChangePointDetector: self.tester = significance_tester self.calculator = calculator - def get_change_points(self, series: Sequence[SupportsFloat], start: Optional[int] = None, end: Optional[int] = None) -> List[ChangePointOtava[GenericStats]]: + def get_change_points(self, series: Sequence[SupportsFloat], start: Optional[int] = None, end: Optional[int] = None) -> List[ChangePoint[GenericStats]]: '''Finds change points in `series[start : end]`.''' if not isinstance(series, np.ndarray): series = np.array(series[start : end], dtype=np.float64) diff --git a/otava/change_point_divisive/significance_test.py b/otava/change_point_divisive/significance_test.py index aa5a8f2..7e348c7 100644 --- a/otava/change_point_divisive/significance_test.py +++ b/otava/change_point_divisive/significance_test.py @@ -25,7 +25,7 @@ from otava.change_point_divisive.base import ( BaseStats, Calculator, CandidateChangePoint, - ChangePointOtava, + ChangePoint, SignificanceTester, ) @@ -50,7 +50,7 @@ class PermutationsSignificanceTester(SignificanceTester): self.seed = seed self.rng = np.random.default_rng(seed) - def change_point(self, candidate: CandidateChangePoint, series: NDArray, intervals: List[slice]) -> ChangePointOtava[PermutationStats]: + def change_point(self, candidate: CandidateChangePoint, series: NDArray, intervals: List[slice]) -> ChangePoint[PermutationStats]: '''Perform permutation test within candidate cluster''' # 1. Find permutated Qhats @@ -74,4 +74,4 @@ class PermutationsSignificanceTester(SignificanceTester): extreme_qhat_perm=extreme_qhat_perm, n_perm=self.permutations ) - return ChangePointOtava.from_candidate(candidate, stats) + return ChangePoint.from_candidate(candidate, stats) diff --git a/otava/main.py b/otava/main.py index 8748ae2..11e0a31 100644 --- a/otava/main.py +++ b/otava/main.py @@ -243,9 +243,10 @@ class Otava: def update_postgres(self, test: PostgresTestConfig, series: AnalyzedSeries): postgres = self.__get_postgres() for metric_name, change_points in series.change_points.items(): - for cp in change_points: + for cpg in change_points: + cp = cpg[metric_name] attributes = series.attributes_at(cp.index) - postgres.insert_change_point(test, metric_name, attributes, cp) + postgres.insert_change_point(test, metric_name, attributes, cpg) def update_bigquery(self, test: BigQueryTestConfig, series: AnalyzedSeries): bigquery = self.__get_bigquery() diff --git a/otava/postgres.py b/otava/postgres.py index 5a30aa1..cbf22af 100644 --- a/otava/postgres.py +++ b/otava/postgres.py @@ -21,7 +21,7 @@ from typing import Dict import pg8000 -from otava.analysis import ChangePointOtava +from otava.change_point_divisive.base import ChangePointGroup, ChangePointSerializer from otava.test_config import PostgresTestConfig @@ -88,10 +88,11 @@ class Postgres: test: PostgresTestConfig, metric_name: str, attributes: Dict, - change_point: ChangePointOtava, + change_point_group: ChangePointGroup, ): cursor = self.__get_conn().cursor() - kwargs = {**attributes, **{test.time_column: datetime.utcfromtimestamp(change_point.time)}} + change_point = ChangePointSerializer(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( update_stmt, diff --git a/otava/report.py b/otava/report.py index d9cf3c4..1244af0 100644 --- a/otava/report.py +++ b/otava/report.py @@ -21,7 +21,8 @@ from typing import List from tabulate import tabulate -from otava.series import ChangePointGroup, Series +from otava.change_point_divisive.base import ChangePoints, ChangePointSerializer +from otava.series import Series from otava.util import format_timestamp, insert_multiple, remove_common_prefix @@ -37,9 +38,9 @@ class ReportType(Enum): class Report: __series: Series - __change_points: List[ChangePointGroup] + __change_points: ChangePoints - def __init__(self, series: Series, change_points: List[ChangePointGroup]): + def __init__(self, series: Series, change_points: ChangePoints): self.__series = series self.__change_points = change_points @@ -74,19 +75,19 @@ class Report: """Returns test log with change points marked as horizontal lines""" lines = self.__format_log().split("\n") col_widths = self.__column_widths(lines) - indexes = [cp.index for cp in self.__change_points] + indexes = [list(cpg.changes.values())[0].index for cpg in self.__change_points] separators = [] columns = list( OrderedDict.fromkeys(["time", *self.__series.attributes, *self.__series.data]) ) - for cp in self.__change_points: + for cpg in self.__change_points: separator = "" info = "" for col_index, col_name in enumerate(columns): col_width = col_widths[col_index] - change = [c for c in cp.changes if c.metric == col_name] + change = [c for m, c in cpg.changes.items() if m == col_name] if change: - change = change[0] + change = ChangePointSerializer(change[0]) change_percent = change.forward_change_percent() separator += "·" * col_width + " " info += f"{change_percent:+.1f}%".rjust(col_width) + " " diff --git a/otava/series.py b/otava/series.py index 1318b8d..729bb14 100644 --- a/otava/series.py +++ b/otava/series.py @@ -18,7 +18,6 @@ import logging from dataclasses import dataclass from datetime import datetime, timezone -from itertools import groupby from typing import Any, Dict, Iterable, List, Optional from otava.analysis import ( @@ -27,7 +26,13 @@ from otava.analysis import ( compute_change_points_orig, fill_missing, ) -from otava.change_point_divisive.base import ChangePointOtava +from otava.change_point_divisive.base import ( + ChangePoint, + ChangePointGroup, + ChangePoints, + ChangePointsByMetric, + ChangePointsByTime, +) @dataclass @@ -48,7 +53,7 @@ class AnalysisOptions: "window_len": self.window_len, "max_pvalue": self.max_pvalue, "min_magnitude": self.min_magnitude, - "orig_edivisive": self.orig_edivisive + "orig_edivisive": self.orig_edivisive, } @@ -64,90 +69,7 @@ class Metric: self.unit = "" def to_json(self): - return { - "direction": self.direction, - "scale": self.scale, - "unit": self.unit - } - - -@dataclass -class ChangePointHunter(ChangePointOtava[TTestStats]): - """A change-point for a single metric""" - metric: str - time: int - - def forward_change_percent(self) -> float: - return self.stats.forward_rel_change() * 100.0 - - def backward_change_percent(self) -> float: - return self.stats.backward_rel_change() * 100.0 - - def magnitude(self): - return self.stats.change_magnitude() - - def mean_before(self): - return self.stats.mean_1 - - def mean_after(self): - return self.stats.mean_2 - - def stddev_before(self): - return self.stats.std_1 - - def stddev_after(self): - return self.stats.std_2 - - def pvalue(self): - return self.stats.pvalue - - def to_json(self, rounded=True): - if rounded: - return { - "metric": self.metric, - "index": int(self.index), - "time": self.time, - "forward_change_percent": f"{self.forward_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), - "time": self.time, - "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 -class ChangePointGroup: - """A group of change points on multiple metrics, at the same time""" - - index: int - time: float - prev_time: int - attributes: Dict[str, str] - prev_attributes: Dict[str, str] - changes: List[ChangePointHunter] - - def to_json(self, rounded=False): - return { - "time": self.time, - "attributes": self.attributes, - "changes": [cp.to_json(rounded=rounded) for cp in self.changes], - } + return {"direction": self.direction, "scale": self.scale, "unit": self.unit} class Series: @@ -184,7 +106,7 @@ class Series: def attributes_at(self, index: int) -> Dict[str, str]: result = {} - for (k, v) in self.attributes.items(): + for k, v in self.attributes.items(): result[k] = v[index] return result @@ -215,13 +137,16 @@ class AnalyzedSeries: __series: Series options: AnalysisOptions - change_points: Dict[str, List[ChangePointHunter]] - change_points_by_time: List[ChangePointGroup] + change_points: Dict[str, ChangePointGroup] + change_points_by_time: ChangePoints change_points_timestamp: Any - def __init__(self, series: Series, options: AnalysisOptions, change_points: Dict[str, ChangePointHunter] = None): + 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(tz=timezone.utc) self.change_points = None if change_points is not None: @@ -235,7 +160,8 @@ class AnalyzedSeries: @staticmethod def __compute_change_points( series: Series, options: AnalysisOptions - ) -> Dict[str, List[ChangePointHunter]]: + ) -> (ChangePointsByMetric, ChangePointsByMetric): + # To find change points, go one metric at a time result = {} weak_change_points = {} for metric in series.data.keys(): @@ -248,7 +174,15 @@ class AnalyzedSeries: values, max_pvalue=options.max_pvalue, ) - result[metric] = change_points + result[metric] = [] + for c in change_points: + c.metric = metric + cpg = ChangePointGroup( + time=series.time[c.index], + attributes=series.attributes_at(c.index), + changes={metric: c}, + ) + result[metric].append(cpg) else: change_points, weak_cps = compute_change_points( values, @@ -257,43 +191,29 @@ class AnalyzedSeries: min_magnitude=options.min_magnitude, ) for c in weak_cps: - weak_change_points[metric].append( - ChangePointHunter( - index=c.index, qhat=0.0, time=series.time[c.index], metric=metric, stats=c.stats - ) + c.metric = metric + cpg = ChangePointGroup( + time=series.time[c.index], + attributes=series.attributes_at(c.index), + changes={metric: c}, ) + weak_change_points[metric].append(cpg) for c in change_points: - result[metric].append( - ChangePointHunter( - index=c.index, qhat=0.0, time=series.time[c.index], metric=metric, stats=c.stats - ) + c.metric = metric + cpg = ChangePointGroup( + time=series.time[c.index], + attributes=series.attributes_at(c.index), + changes={metric: c}, ) - # If you got an exception and are wondering about the next row... - # weak_cps is an optimization which you can ignore - return result, weak_change_points + result[metric].append(cpg) + + return ChangePointsByMetric(result), ChangePointsByMetric(weak_change_points) @staticmethod def __group_change_points_by_time( - series: Series, change_points: Dict[str, List[ChangePointHunter]] - ) -> List[ChangePointGroup]: - changes: List[ChangePointHunter] = [] - for metric in change_points.keys(): - changes += change_points[metric] - - changes.sort(key=lambda c: c.index) - points = [] - for k, g in groupby(changes, key=lambda c: c.index): - cp = ChangePointGroup( - index=k, - time=series.time[k], - prev_time=series.time[k - 1], - attributes=series.attributes_at(k), - prev_attributes=series.attributes_at(k - 1), - changes=list(g), - ) - points.append(cp) - - return points + series: Series, change_points: ChangePoints + ) -> ChangePointsByTime: + return ChangePointsByTime(change_points) def get_stable_range(self, metric: str, index: int) -> (int, int): """ @@ -306,13 +226,13 @@ class AnalyzedSeries: It follows that there are no change points between A and B. """ begin = 0 - for cp in self.change_points[metric]: + for cp in self.change_points.get_change_points_for_metric(metric): if cp.index > index: break begin = cp.index end = len(self.time()) - for cp in reversed(self.change_points[metric]): + for cp in reversed(self.change_points.get_change_points_for_metric(metric)): if cp.index <= index: break end = cp.index @@ -337,7 +257,9 @@ class AnalyzedSeries: max_time = max(self.__series.time) for t in time: if t <= max_time: - return ValueError("time must be monotonously increasing if you use append() time={}".format(time)) + return ValueError( + "time must be monotonously increasing if you use append() time={}".format(time) + ) return None @@ -367,7 +289,7 @@ class AnalyzedSeries: for metric in self.__series.data.keys(): if metric not in new_data: - weak_change_points[metric] = self.weak_change_points[metric] + weak_change_points[metric] = self.weak_change_points.select_metrics(metric) continue change_points, weak_cps = compute_change_points( @@ -376,32 +298,40 @@ class AnalyzedSeries: max_pvalue=self.options.max_pvalue, min_magnitude=self.options.min_magnitude, new_data=len(new_data[metric]), - old_weak_cp=self.weak_change_points.get(metric, []) + old_weak_cp=self.weak_change_points.get_change_points_for_metric(metric), ) - result[metric] = [] + if metric not in result: + result[metric] = [] for c in change_points: result[metric].append( - ChangePointHunter( - index=c.index, qhat=0.0, time=self.__series.time[c.index], metric=metric, stats=c.stats + ChangePointGroup( + time=self.__series.time[c.index], + changes={metric: c}, + attributes=self.__series.attributes_at(c.index), ) ) - weak_change_points[metric] = [] + if metric not in weak_change_points: + weak_change_points[metric] = [] for c in weak_cps: weak_change_points[metric].append( - ChangePointHunter( - index=c.index, qhat=0.0, time=self.__series.time[c.index], metric=metric, stats=c.stats + ChangePointGroup( + time=self.__series.time[c.index], + changes={metric: c}, + attributes=self.__series.attributes_at(c.index), ) ) + # TODO: Remove this. It should not be a requirement that metrics have the same history. fill_missing(self.__series.data[metric]) - # If some metrics didn't participate in this round, we still keep them, but update the ones - # We did recompute - for metric in result.keys(): - self.change_points[metric] = result[metric] - for metric in weak_change_points.keys(): - self.weak_change_points[metric] = weak_change_points[metric] - self.change_points_by_time = self.__group_change_points_by_time(self.__series, self.change_points) - return result, weak_change_points + r = ChangePointsByMetric(result) + w = ChangePointsByMetric(weak_change_points) + # print("#", self.change_points.change_points) + # print("¤", r.change_points) + # print("%", w.change_points) + # r has a subset of all metrics, so can't just set change_points to r + for metric, cpglist in r.change_points.items(): + self.change_points.change_points[metric] = cpglist + return r, w def test_name(self) -> str: return self.__series.test_name @@ -435,12 +365,12 @@ class AnalyzedSeries: def to_json(self): change_points_json = {} - for metric, cps in self.change_points.items(): - change_points_json[metric] = [cp.to_json(rounded=False) for cp in cps] + for cps in self.change_points: + change_points_json = [cp.to_json(rounded=False) for cp in cps] weak_change_points_json = {} - for metric, cps in self.weak_change_points.items(): - weak_change_points_json[metric] = [cp.to_json(rounded=False) for cp in cps] + for cps in self.weak_change_points: + weak_change_points_json = [cp.to_json(rounded=False) for cp in cps] data_json = {} for metric, datapoints in self.__series.data.items(): @@ -456,7 +386,7 @@ class AnalyzedSeries: "attributes": self.__series.attributes, "data": self.__series.data, "change_points": change_points_json, - "weak_change_points": weak_change_points_json + "weak_change_points": weak_change_points_json, } @classmethod @@ -472,7 +402,7 @@ class AnalyzedSeries: analyzed_json["time"], new_metrics, analyzed_json["data"], - analyzed_json["attributes"] + analyzed_json["attributes"], ) new_options = AnalysisOptions() @@ -493,9 +423,7 @@ class AnalyzedSeries: pvalue=cp["pvalue"], ) new_list.append( - ChangePointHunter( - index=cp["index"], time=cp["time"], metric=cp["metric"], stats=stat - ) + ChangePoint(index=cp["index"], time=cp["time"], metric=cp["metric"], stats=stat) ) new_change_points[metric] = new_list @@ -511,9 +439,7 @@ class AnalyzedSeries: pvalue=cp["pvalue"], ) new_list.append( - ChangePointHunter( - index=cp["index"], time=cp["time"], metric=cp["metric"], stats=stat - ) + ChangePoint(index=cp["index"], time=cp["time"], metric=cp["metric"], stats=stat) ) new_weak_change_points[metric] = new_list @@ -522,6 +448,8 @@ class AnalyzedSeries: if "change_points_timestamp" in analyzed_json.keys(): analyzed_series.change_points_timestamp = analyzed_json["change_points_timestamp"] - analyzed_series.change_points_by_time = AnalyzedSeries.__group_change_points_by_time(analyzed_series.__series, analyzed_series.change_points) + 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/otava/slack.py b/otava/slack.py index a0d4bc3..441f083 100644 --- a/otava/slack.py +++ b/otava/slack.py @@ -23,8 +23,9 @@ from typing import Dict, List from pytz import UTC from slack_sdk import WebClient +from otava.change_point_divisive.base import ChangePointGroup, ChangePointSerializer from otava.data_selector import DataSelector -from otava.series import AnalyzedSeries, ChangePointGroup +from otava.series import AnalyzedSeries @dataclass @@ -202,8 +203,9 @@ class SlackNotification: for test_name, group in test_changes.items(): fields.append(f"*{test_name}*") summary = "" - for change in group.changes: - change_percent = change.forward_change_percent() + for metric, change in group.changes.items(): + c = ChangePointSerializer(change) + change_percent = c.forward_change_percent() change_emoji = self.__get_change_emoji(test_name, change) if isinf(change_percent): report_percent = change_percent @@ -228,7 +230,8 @@ class SlackNotification: def __get_change_emoji(self, test_name, change): metric_direction = self.test_analyzed_series[test_name].metric(change.metric).direction - regression = metric_direction * change.forward_change_percent() + c = ChangePointSerializer(change) + regression = metric_direction * c.forward_change_percent() if regression >= 0: return ":large_blue_circle:" else: diff --git a/tests/change_point_divisive_test.py b/tests/change_point_divisive_test.py index 950d982..45d4431 100644 --- a/tests/change_point_divisive_test.py +++ b/tests/change_point_divisive_test.py @@ -19,7 +19,7 @@ import numpy as np import pytest from otava.analysis import TTestSignificanceTester, TTestStats -from otava.change_point_divisive.base import ChangePointOtava +from otava.change_point_divisive.base import ChangePoint from otava.change_point_divisive.calculator import PairDistanceCalculator from otava.change_point_divisive.detector import ChangePointDetector from otava.change_point_divisive.significance_test import PermutationsSignificanceTester @@ -124,9 +124,9 @@ def test_get_intervals_requires_sorted_change_points(): # Sorted change points should work sorted_cps = [ - ChangePointOtava(index=5, qhat=1.0, stats=stats), - ChangePointOtava(index=10, qhat=1.0, stats=stats), - ChangePointOtava(index=15, qhat=1.0, stats=stats), + ChangePoint(index=5, qhat=1.0, stats=stats), + ChangePoint(index=10, qhat=1.0, stats=stats), + ChangePoint(index=15, qhat=1.0, stats=stats), ] intervals = tester.get_intervals(sorted_cps) assert len(intervals) == 4 @@ -137,9 +137,9 @@ def test_get_intervals_requires_sorted_change_points(): # Unsorted change points should raise AssertionError unsorted_cps = [ - ChangePointOtava(index=10, qhat=1.0, stats=stats), - ChangePointOtava(index=5, qhat=1.0, stats=stats), - ChangePointOtava(index=15, qhat=1.0, stats=stats), + ChangePoint(index=10, qhat=1.0, stats=stats), + ChangePoint(index=5, qhat=1.0, stats=stats), + ChangePoint(index=15, qhat=1.0, stats=stats), ] with pytest.raises(AssertionError, match="Change points must be sorted by index"): tester.get_intervals(unsorted_cps) diff --git a/tests/report_test.py b/tests/report_test.py index a220eb5..6e8b483 100644 --- a/tests/report_test.py +++ b/tests/report_test.py @@ -86,8 +86,7 @@ def test_json_report(report): 'metric': 'series2', 'pvalue': '0.000000', 'stddev_after': '0.026954', - 'stddev_before': '0.011180', - 'time': 4}], + 'stddev_before': '0.011180'}], 'time': 4}, {'attributes': {}, 'changes': [{'forward_change_percent': '-49', @@ -98,8 +97,10 @@ def test_json_report(report): 'metric': 'series1', 'pvalue': '0.000000', 'stddev_after': '0.025768', - 'stddev_before': '0.067495', - 'time': 6}], + 'stddev_before': '0.067495'}], 'time': 6}]} assert isinstance(obj, dict) + from pprint import pprint + pprint(obj) + pprint(expected) assert obj == expected diff --git a/tests/series_test.py b/tests/series_test.py index 94fbe54..5111238 100644 --- a/tests/series_test.py +++ b/tests/series_test.py @@ -20,6 +20,7 @@ from random import random import pytest +from otava.change_point_divisive.base import ChangePointSerializer from otava.series import AnalysisOptions, Metric, Series @@ -36,12 +37,66 @@ def test_change_point_detection(): attributes={}, ) - change_points = test.analyze().change_points_by_time - assert len(change_points) == 2 - assert change_points[0].index == 4 - assert change_points[0].changes[0].metric == "series2" - assert change_points[1].index == 6 - assert change_points[1].changes[0].metric == "series1" + cps = test.analyze().change_points_by_time + assert len(cps) == 2 + assert cps.change_points[0].time == 4 + assert cps.change_points[0].changes["series2"].metric == "series2" + assert cps.change_points[1].time == 6 + assert cps.change_points[1].changes["series1"].metric == "series1" + + +def test_change_point_detection_many(): + series_3 = [ + 1, + 1, + 1, + 1, + 1, + 5, + 5, + 5, + 5, + 5, + 5, + 5, + 5, + 9, + 9, + 9, + 9, + 9, + 9, + 9, + 9, + 3, + 3, + 3, + 3, + 3, + 3, + 3, + 3, + 3, + ] + time = list(range(len(series_3))) + test = Series( + "test", + branch=None, + time=time, + metrics={"series3": Metric(1, 1.0)}, + data={"series3": series_3}, + attributes={}, + ) + + options = AnalysisOptions() + options.min_magnitude = 0.0 + options.max_pvalue = 0.05 + analyzed_series = test.analyze(options) + assert len(list(analyzed_series.change_points)) == 3 + cps_by_time = analyzed_series.change_points_by_time + assert len(cps_by_time.change_points) == 3 + assert analyzed_series.change_points[0].time == 5 + assert "series3" in analyzed_series.change_points[0].changes def test_change_point_min_magnitude(): @@ -59,16 +114,16 @@ def test_change_point_min_magnitude(): options = AnalysisOptions() options.min_magnitude = 0.2 - change_points = test.analyze(options).change_points_by_time - assert len(change_points) == 1 - assert change_points[0].index == 6 - assert change_points[0].changes[0].metric == "series1" + cps = test.analyze(options).change_points_by_time + assert len(cps) == 1 + assert cps.change_points[0].time == 6 + assert "series1" in cps[0].changes - for change_point in change_points: - for change in change_point.changes: - assert ( - change.magnitude() >= options.min_magnitude - ), f"All change points must have magnitude greater than {options.min_magnitude}" + for change_point in cps: + for metric, change in change_point.changes.items(): + assert ChangePointSerializer(change).magnitude() >= options.min_magnitude, ( + f"All change points must have magnitude greater than {options.min_magnitude}" + ) # Divide by zero is only a RuntimeWarning, but for testing we want to make sure it's a failure @@ -90,7 +145,7 @@ def test_div_by_zero(): cpjson = analyzed_series.to_json() assert cpjson assert len(change_points) == 2 - assert change_points[0].index == 3 + assert change_points[0].time == 3 def test_change_point_detection_performance(): @@ -151,20 +206,22 @@ def test_incremental_otava(): ) analyzed_series = test.analyze() - analyzed_series.append(time=[len(time)], new_data={"series1": [0.5], "series2": [1.97]}, attributes={}) + analyzed_series.append( + time=[len(time)], new_data={"series1": [0.5], "series2": [1.97]}, attributes={} + ) change_points = analyzed_series.change_points - assert [c.index for c in change_points["series1"]] == [6] - assert [c.index for c in change_points["series2"]] == [4] + assert [c.index for c in change_points.get_change_points_for_metric("series1")] == [6] + assert [c.index for c in change_points.get_change_points_for_metric("series2")] == [4] analyzed_series.append(time=[len(time)], new_data={"series1": [0.51]}, attributes={}) change_points = analyzed_series.change_points - assert [c.index for c in change_points["series1"]] == [6] - assert [c.index for c in change_points["series2"]] == [4] + assert [c.index for c in change_points.get_change_points_for_metric("series1")] == [6] + assert [c.index for c in change_points.get_change_points_for_metric("series2")] == [4] analyzed_series.append(time=[len(time)], new_data={"series2": [33.33, 46.46]}, attributes={}) change_points = analyzed_series.change_points - assert [c.index for c in change_points["series1"]] == [6] - assert [c.index for c in change_points["series2"]] == [4, 12] + assert [c.index for c in change_points.get_change_points_for_metric("series1")] == [6] + assert [c.index for c in change_points.get_change_points_for_metric("series2")] == [4, 12] def test_validate(): @@ -190,13 +247,19 @@ def test_validate(): 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={}) + 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={}) + analyzed_series.append( + time=[len(time)], new_data={"series1": [0.5], "series2": [1.97]}, attributes={} + ) - err = analyzed_series._validate_append(time=[len(time)], new_data={"series1": [0.51]}, attributes={}) + err = analyzed_series._validate_append( + time=[len(time)], new_data={"series1": [0.51]}, attributes={} + ) assert err is None err = analyzed_series._validate_append(time=[5], new_data={"series1": [0.51]}, attributes={}) @@ -220,7 +283,9 @@ def test_can_append(): ) analyzed_series = test.analyze() - analyzed_series.append(time=[len(time)], new_data={"series1": [0.5], "series2": [1.97]}, attributes={}) + analyzed_series.append( + time=[len(time)], new_data={"series1": [0.5], "series2": [1.97]}, attributes={} + ) can = analyzed_series.can_append(time=[len(time)], new_data={"series1": [0.51]}, attributes={}) assert can
