FabledSteward/steward/core/formatting.py

"""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"


# Every state a page shows, and the tone it is drawn in: ok (green), warn
# (amber), crit (red), info (blue) or dim. The status_badge and status_dot
# macros read this, so one state looks the same on every page.
STATES: dict[str, tuple[str, str]] = {
    "up": ("ok", "Up"),
    "ok": ("ok", "OK"),
    "resolved": ("ok", "Resolved"),
    "down": ("crit", "Down"),
    "firing": ("crit", "Firing"),
    "crit": ("crit", "Critical"),
    "warn": ("warn", "Warning"),
    "pending": ("warn", "Pending"),
    "stale": ("warn", "Stale"),
    "no_data": ("warn", "No data"),
    "acknowledged": ("info", "Acknowledged"),
    "inactive": ("dim", "Inactive"),
    "unknown": ("dim", "Unknown"),
}


def status_info(state: Any) -> tuple[str, str]:
    """(tone, label) for a state name or enum. A state the map doesn't know
    is drawn dim under its own name rather than breaking the page."""
    key = str(getattr(state, "value", state) or "unknown").lower()
    return STATES.get(key, ("dim", key.replace("_", " ").capitalize()))


def required(value: Any, what: str) -> Any:
    """For a macro argument that must be given: raise if it is empty."""
    if value is None or not str(value).strip():
        raise ValueError(f"{what} is required")
    return value