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