from typing import Optional, Sequence from .monitor_check_status import Check from .monitor_uptime_types import UptimeReport def _whole(value: object) -> bool: return isinstance(value, int) and not isinstance(value, bool) def uptime(checks: Sequence[Check], from_: int, to: int, max_gap_seconds: int) -> UptimeReport: """Seconds up, degraded, down and unknown over [from, to). Each check's status holds from its `at` until the next check, but for at most max_gap_seconds: a monitor that stopped reporting has not proved the target was up.""" if not _whole(from_) or not _whole(to): raise ValueError("from and to must be whole seconds") if from_ > to: raise ValueError("from must not be after to: %d > %d" % (from_, to)) if not _whole(max_gap_seconds) or max_gap_seconds < 1: raise ValueError("maxGapSeconds must be a whole number of at least 1, received %s" % (max_gap_seconds,)) up = degraded = down = 0 count = down_count = 0 for i, c in enumerate(checks): if c.status not in ("up", "degraded", "down"): raise ValueError("unknown check status: %s" % (c.status,)) if i > 0 and c.at <= checks[i - 1].at: raise ValueError("checks must be in strictly ascending time order: %d follows %d" % (c.at, checks[i - 1].at)) if from_ <= c.at < to: count += 1 if c.status == "down": down_count += 1 # The span this check vouches for, clipped to the window. end = c.at + max_gap_seconds if i + 1 < len(checks) and checks[i + 1].at < end: end = checks[i + 1].at s = max(c.at, from_) e = min(end, to) if e > s: if c.status == "up": up += e - s elif c.status == "degraded": degraded += e - s else: down += e - s known = up + degraded + down # Floored: never 100.00% while a down second exists. bp: Optional[int] = None if known == 0 else (up + degraded) * 10000 // known return UptimeReport( up_seconds=up, degraded_seconds=degraded, down_seconds=down, unknown_seconds=to - from_ - known, uptime_basis_points=bp, checks=count, down_checks=down_count, )