import math from dataclasses import fields from app.sample import Sample RowAgg = dict[str, float | int | None] def window(snap: list[Sample], max_points: int) -> list[tuple[float, dict[str, RowAgg]]]: """Window-average a sample list down to at most `max_points` points. The samples are split into consecutive chunks of ceil(n / max_points) and each numeric Sample field is reduced to {avg, min, max} per chunk; whole-number fields (byte counts) stay ints, fractional fields are rounded to 0.1. Each point is stamped with the timestamp of the last sample in its chunk. Args: snap: samples oldest first (HistoryStore.snapshot). max_points: maximum number of points to emit. Returns: (timestamp, field aggregations) pairs, oldest first. """ n = len(snap) w = max(1, math.ceil(n / max_points)) out: list[tuple[float, dict[str, RowAgg]]] = [] for start in range(0, n, w): chunk = snap[start : start + w] vals: dict[str, list[int | float]] = {} for sample in chunk: for f in fields(sample): if f.name == "ts": continue v = getattr(sample, f.name) if isinstance(v, (int, float)) and not isinstance(v, bool): vals.setdefault(f.name, []).append(v) row: dict[str, RowAgg] = {} for k, lst in vals.items(): ints = all(isinstance(v, int) for v in lst) avg = sum(lst) / len(lst) row[k] = { "avg": round(avg) if ints else round(avg, 1), "min": min(lst) if ints else round(min(lst), 1), "max": max(lst) if ints else round(max(lst), 1), } out.append((chunk[-1].ts, row)) return out