Gerrrr commented on code in PR #161:
URL: https://github.com/apache/otava/pull/161#discussion_r3742390917
##########
otava/change_point_divisive/base.py:
##########
@@ -40,39 +155,634 @@ class BaseStats:
@dataclass
class ChangePoint(CandidateChangePoint, Generic[GenericStats]):
- '''Change point class, defined by index and signigicance test statistic.'''
+ """
+ ChangePoint class.
+
+ Defined by index and significance 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'''
- return isinstance(other, self.__class__) and self.index == other.index
+ def copy(self):
+ """
+ Copy constructor.
+
+ :return: A deep copy of self, recursively calls also stats.copy().
+ """
+ return ChangePoint(
+ index=self.index, qhat=self.qhat, stats=self.stats.copy(),
metric=self.metric
+ )
@classmethod
- def from_candidate(cls, candidate: CandidateChangePoint, stats:
GenericStats) -> 'ChangePoint[GenericStats]':
+ def from_candidate(
+ cls, candidate: CandidateChangePoint, stats: GenericStats
+ ) -> "ChangePoint[GenericStats]":
return cls(
index=candidate.index,
qhat=candidate.qhat,
stats=stats,
)
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}",
+ "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(),
+ }
+
+
+@dataclass
+class ChangePointGroup:
+ """
+ A group of change points on multiple metrics, at the same point in time.
+
+ If you think of ChangePoints as a 2D table, then ChangePointGroup are
rows, and the metrics are the columns.
+ One ChangePointGroup has only the metrics where a ChangePoint was found at
this time(stamp).
+ Note that there is no .index at this level. This is because each metric is
an independent sequence of results,
+ and each such sequence may have started at different times in the past,
and may also be missing observations
+ where others do have some. Because of this, each ChangePoint has its own
.index in
+ ChangePointGroup.changes[metric_name].index
+
+ :param time: The unix timestamp for the results that ChangePoint(s) were
found at. Decimals are milliseconds.
+ :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):
+ 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 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
+
+ def __getitem__(self, metric):
+ return self.changes[metric]
+
+ def metrics(self):
+ return self.changes.keys()
+
+ def commit(self):
+ return self.attributes.get("commit")
+
+ def datetime(self):
+ return datetime.fromtimestamp(self.time, timezone.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 only
+ holds ChangePoint objects at the time and metric (row and column...) that
one was found.
+
+ Specifically: it is NOT guaranteed that each row (each GhangePointGroup)
has each metric.
+
+ 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 used.
+ """
+
+ def __init__(self):
+ """
+ The constructor only creates an empty container. To build one from
+ existing data, use the explicit from_list() / from_dict() classmethods.
+ """
+ self._change_points = []
+
+ @classmethod
+ 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)}."
+ )
+ for cpg in cps:
+ if not isinstance(cpg, ChangePointGroup):
+ raise TypeError(
+ f"from_list() takes a list of ChangePointGroup objects.
Got {type(cpg)}."
+ )
+ obj = cls()
+ for cpg in sorted(cps, key=lambda cpg: cpg.time):
+ obj.append(cpg)
+ return obj
+
+ @classmethod
+ def from_dict(cls, cps: dict):
+ """
+ Build a ChangePointsByMetric from a dict[str, list[ChangePointGroup]].
+
+ A dict is inherently keyed by metric, so from_dict() returns a
+ ChangePointsByMetric. (This is asymmetric with from_list(), which
returns
+ whichever class you call it on.) ChangePointsByTime overrides this to
reject
+ a dict outright.
+ """
+ if not isinstance(cps, dict):
+ raise TypeError(
+ f"from_dict() takes a dict[str, list[ChangePointGroup]]. Got
{type(cps)}."
+ )
+ obj = ChangePointsByMetric()
+ 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]]."
+ )
+ 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}")
+ # store each metric's groups sorted by timestamp
+ obj._change_points[metric] = sorted(cpglist, key=lambda cpg:
cpg.time)
+ return obj
+
+ def copy(self):
+ """
+ Copy constructor.
+
+ Returns a deep copy of this object.
+ """
+ new_obj = ChangePointsByTime()
+ new_obj._change_points = [
+ cpg.copy() for cpg in sorted(self._change_points, key=lambda cpg:
cpg.time)
+ ]
+ return new_obj
+
+ def by_time(self):
+ return self
+
+ def by_metric(self):
+ return self.pivot()
+
+ 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: list[ChangePointGroup]):
+ 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 __contains__(self, metric):
+ """
+ True if this container holds at least one ChangePoint where
metric==metric
+
+ This makes more sense, and is more efficient, for ChangePointsByMetric
class, but we provide
+ the same functionality here for consistency.
+ """
+ return metric in self.metrics()
+
+ def metrics(self) -> set:
+ all_metrics = set()
+ for row in self._change_points:
+ all_metrics.update(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 subset of the columns.
+
+ Note: The internal data structure doesn't do anything to make this
+ request efficient. This will loop over all ChangePointGroup s.
+ 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)
+ metric_change_points = []
+ 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}")
+ metric_change_points.append(cp)
+ return 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)
+ return by_metric
+
+
+class ChangePointsByTime(ChangePoints):
+ """
+ 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.
+
+ The pivot() method will return a new object (a copy) holding the same
data, but ordered by metrics
+ as the primary and optimized axis. The method by_metric() can be used
for the same purpose. Note
+ that the method by_time() is a no-op and returns self, it doesn't even
do a copy.
+ """
+ @classmethod
+ def from_dict(cls, cps: dict):
+ """
+ A ChangePointsByTime is stored as a flat list of rows, not keyed by
+ metric, so a dict has no meaningful representation here.
+ """
+ raise TypeError(
+ "ChangePointsByTime doesn't accept a dict. Did you want
ChangePointsByMetric.from_dict()?"
+ )
+
+
+class ChangePointsByMetric(ChangePoints):
+ """
+ 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.
+
+ The pivot() method will return a new object (a copy) holding the same
data, but ordered by time
+ as the primary and optimized axis. The method by_time() can be used
for the same purpose. Note
+ that the method by_metric() is a no-op and returns self, it doesn't
even do a copy.
+ """
+
+ def __init__(self):
+ """
+ The constructor only creates an empty container. To build one from
+ existing data, use the explicit from_list() / from_dict() classmethods.
+ from_list() is inherited from ChangePoints; append() shuffles each
+ group's metrics into their own column.
+ """
+ self._change_points = {}
+
+ # from_dict() is inherited from ChangePoints: it already builds and
returns a
+ # ChangePointsByMetric, which is exactly what we want here.
+
+ def copy(self):
+ """
+ Copy constructor.
+
+ Returns a deep copy of this object.
+ """
+ new_obj = ChangePointsByMetric()
+ new_obj._change_points = {
+ metric: [cpg.copy() for cpg in cpglist]
+ for metric, cpglist in self._change_points.items()
+ }
+ return new_obj
+
+ def pivot(self):
+ """
+ Pivot (metric,time) to (time,metric) so that we return
ChangePointsByTime() 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():
+ if metric not in self._change_points:
+ 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}")
+ 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:
+ raise ValueError(
+ "ChangePoints.extend() can only be used such that time is
monotonically increasing"
+ )
+
+ def extend(self, cps: list[ChangePointGroup]):
+ errmsg = "ChangePoints.extend() takes as argument a list of
ChangePointGroup objects."
+ if not isinstance(cps, list):
+ raise TypeError(errmsg)
+ for cpg in cps:
+ if not isinstance(cpg, 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}")
+ 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:
+ self._change_points[metric1].append(cpg)
+ else:
+ raise ValueError(
+ "ChangePoints.extend() can only be used such that time
is monotonically increasing"
+ )
+
+ def by_time(self):
+ return self.pivot()
+
+ def by_metric(self):
+ return self
+
+ def items(self):
+ return self._change_points.items()
+
+ def keys(self):
+ return self._change_points.keys()
+
+ def __iter__(self):
+ return self.pivot().__iter__()
+
+ def __len__(self):
+ return max([len(cpg) for metric, cpg in self._change_points.items()])
Review Comment:
90eb9d1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
