bvandeusenandClaude Opus 5.5
bf08169176
#5093 "Shared format filters and status_class driven by Settings → Thresholds". - steward/core/formatting.py: human_bytes (binary units), human_rate, human_duration (two largest units), ago (a <time> titled with the absolute UTC time) and status_class(value, kind, thresholds) -> ok/warn/crit/unknown, which raises on a kind the thresholds don't name. - app.py registers them as the bytes, rate, duration and ago filters and the status_class global; threshold_style is now graded by status_class (and caps its weight at 500). - Replaces the copies: docker's _human_bytes/_human_uptime, core alerts' _duration, the host-agent fmt_bps/fmt_bytes macros, the Traefik and UniFi byte-rate arithmetic, the swarm-failures ago macro and two uptime spans. - Tests: a table per helper, and every template compiled against the app's filters, so an unknown filter fails CI rather than a page. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
"""steward/core/formatting.py
|
|
|
|
How Steward writes a size, a rate, a span of time and a moment, and how it
|
|
grades a reading against Settings → Thresholds — once, for every page and
|
|
plugin (#5093). Registered as Jinja filters (|bytes, |rate, |duration, |ago)
|
|
and the status_class global in app.py; Python callers import them from here.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from markupsafe import Markup, escape
|
|
|
|
NONE = "—"
|
|
_UNITS = ("B", "KiB", "MiB", "GiB", "TiB", "PiB")
|
|
_SPANS = (("d", 86400), ("h", 3600), ("m", 60), ("s", 1))
|
|
|
|
|
|
def human_bytes(n: float | None) -> str:
|
|
"""Binary units: '0 B', '512 B', '1.5 KiB', '1.4 GiB'."""
|
|
if n is None:
|
|
return NONE
|
|
size = float(n)
|
|
for unit in _UNITS:
|
|
if abs(size) < 1024 or unit == _UNITS[-1]:
|
|
return f"{int(size)} B" if unit == "B" else f"{size:.1f} {unit}"
|
|
size /= 1024
|
|
raise AssertionError("unreachable")
|
|
|
|
|
|
def human_rate(bytes_per_s: float | None) -> str:
|
|
"""Bytes per second in binary units: '12.3 MiB/s'."""
|
|
return NONE if bytes_per_s is None else f"{human_bytes(bytes_per_s)}/s"
|
|
|
|
|
|
def human_duration(seconds: float | None) -> str:
|
|
"""The two largest units of a span: '45s', '8m', '1m 30s', '2h 5m', '3d 4h'.
|
|
The second unit is dropped when it is zero. Negative spans read as None."""
|
|
if seconds is None or seconds < 0:
|
|
return NONE
|
|
left = int(seconds)
|
|
if left == 0:
|
|
return "0s"
|
|
parts = []
|
|
for i, (unit, size) in enumerate(_SPANS):
|
|
count, left = divmod(left, size)
|
|
if count:
|
|
parts.append(f"{count}{unit}")
|
|
nxt = _SPANS[i + 1] if i + 1 < len(_SPANS) else None
|
|
if nxt and left // nxt[1]:
|
|
parts.append(f"{left // nxt[1]}{nxt[0]}")
|
|
break
|
|
return " ".join(parts)
|
|
|
|
|
|
def _utc(when: datetime) -> datetime:
|
|
return when.replace(tzinfo=timezone.utc) if when.tzinfo is None else when.astimezone(timezone.utc)
|
|
|
|
|
|
def seconds_since(when: datetime | None, now: datetime | None = None) -> float | None:
|
|
"""Seconds from `when` to now; a naive datetime is taken as UTC."""
|
|
if when is None:
|
|
return None
|
|
return ((now or datetime.now(timezone.utc)) - _utc(when)).total_seconds()
|
|
|
|
|
|
def ago(when: datetime | None, now: datetime | None = None) -> Markup:
|
|
"""'3m ago' (or 'in 3m' for a future moment) in a <time> element whose
|
|
title is the absolute UTC time, so a hover gives the exact moment."""
|
|
if when is None:
|
|
return Markup(NONE)
|
|
when = _utc(when)
|
|
secs = seconds_since(when, now)
|
|
if abs(secs) < 1:
|
|
text = "just now"
|
|
elif secs > 0:
|
|
text = f"{human_duration(secs)} ago"
|
|
else:
|
|
text = f"in {human_duration(-secs)}"
|
|
return Markup('<time datetime="{}" title="{}">{}</time>').format(
|
|
when.strftime("%Y-%m-%dT%H:%M:%SZ"), when.strftime("%Y-%m-%d %H:%M:%S UTC"), escape(text))
|
|
|
|
|
|
def status_class(value: Any, kind: str, thresholds: dict) -> str:
|
|
"""Grade a reading against Settings → Thresholds (to_thresholds_cfg):
|
|
'ok', 'warn' or 'crit', or 'unknown' when there is no reading.
|
|
|
|
A kind the thresholds don't name is a mistake in the caller, not a
|
|
reading to colour as healthy, so it raises."""
|
|
if value is None:
|
|
return "unknown"
|
|
th = thresholds[kind]
|
|
warn, crit = th["warn"], th["crit"]
|
|
if th.get("dir", "high") == "low":
|
|
return "crit" if value < crit else "warn" if value < warn else "ok"
|
|
return "crit" if value >= crit else "warn" if value >= warn else "ok"
|