def append()

in otava/series.py [0:0]

• 46 lines of code
• 12 McCabe index (conditional complexity)

    def append(self, time, new_data, attributes):
        """
        Append new data points to the underlying series and recompute change points.

        The recompute is done efficiently, only the tail of the Series() is recomputed.

        Parameters are the same as for the constructor. Just the metrics are missing, it is required
        to have the same metrics or a subset in the new data,
        """
        err = self._validate_append(time, new_data, attributes)
        if err is not None:
            raise err

        for t in time:
            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]
        for k, v in attributes.items():
            self.__series.attributes[k].append(v)

        result = {}
        weak_change_points = {}

        for metric in self.__series.data.keys():
            if metric not in new_data:
                weak_change_points[metric] = self.weak_change_points[metric]
                continue

            change_points, weak_cps = compute_change_points(
                self.__series.data[metric],
                window_len=self.options.window_len,
                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, [])
            )
            result[metric] = []
            for c in change_points:
                result[metric].append(
                    ChangePoint(
                        index=c.index, time=self.__series.time[c.index], metric=metric, stats=c.stats
                    )
                )
            weak_change_points[metric] = []
            for c in weak_cps:
                weak_change_points[metric].append(
                    ChangePoint(
                        index=c.index, time=self.__series.time[c.index], metric=metric, stats=c.stats
                    )
                )
            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