Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.
from typing import List, Optional, Sequence
from .math_round_div import round_div ← from math.round-div ^1.0.0 · built alongside by fune
from .monitor_series_window import MetricSample, series_window ← from monitor.series-window ^1.0.0 · built alongside by fune
from .monitor_rollup_types import Bucket
_MAX_BUCKETS = 10000
def rollup(samples: Sequence[MetricSample], from_: int, to: int, bucket_seconds: int) -> List[Bucket]:
"""One row per bucket from `from` to `to`, empty buckets included so a chart
shows gaps as gaps. Buckets are aligned to `from`; the last is clipped to `to`."""
if not isinstance(bucket_seconds, int) or isinstance(bucket_seconds, bool):
raise ValueError("bucketSeconds must be a whole number of seconds")
if bucket_seconds < 1:
raise ValueError("bucketSeconds must be at least 1, received %d" % bucket_seconds)
inside = series_window(samples, from_, to)
n = -((from_ - to) // bucket_seconds)
if n > _MAX_BUCKETS:
raise ValueError("too many buckets: %d, at most %d" % (n, _MAX_BUCKETS))
out: List[Bucket] = []
j = 0
for i in range(n):
start = from_ + i * bucket_seconds
end = min(start + bucket_seconds, to)
count = 0
total = 0
lo: Optional[int] = None
hi: Optional[int] = None
last: Optional[int] = None
while j < len(inside) and inside[j].at < end:
v = inside[j].value
count += 1
total += v
lo = v if lo is None or v < lo else lo
hi = v if hi is None or v > hi else hi
last = v
j += 1
avg = None if count == 0 else round_div(total, count, "half-up")
out.append(Bucket(start=start, end=end, count=count, min=lo, max=hi, avg=avg, sum=total, last=last))
return out