impl/python.py

from typing import List, Sequence

from .monitor_series_window_types import MetricSample


def series_window(samples: Sequence[MetricSample], from_: int, to: int) -> List[MetricSample]:
    """The samples with from <= at < to. Half-open, so consecutive windows
    (the last hour, the hour before) never count a sample twice."""
    for v in (from_, to):
        if not isinstance(v, int) or isinstance(v, bool):
            raise ValueError("from and to must be whole seconds")
    if from_ > to:
        raise ValueError("from must not be after to: %d > %d" % (from_, to))
    out: List[MetricSample] = []
    for i, s in enumerate(samples):
        # Checked over the whole series, not just the window: a misordered
        # store gives wrong answers everywhere, so it should fail everywhere.
        if i > 0 and s.at <= samples[i - 1].at:
            raise ValueError("samples must be in strictly ascending time order: %d follows %d" % (s.at, samples[i - 1].at))
        if from_ <= s.at < to:
            out.append(MetricSample(at=s.at, value=s.value))
    return out