wake-alarm/wake_alarm/_audit_checks.py

"""Expected-vs-observed for one morning: pure, evidence in, checks out.

Each check is ``ok`` (True), a mismatch (False) or not applicable (None).
Escapes and mismatches become the report's ``## Flags`` -- escapes first,
because every use of the hatch must be seen.
"""

from __future__ import annotations

from datetime import timedelta
from typing import TYPE_CHECKING

from wake_alarm._audit_evidence import (
    Check,
    Evidence,
    bad,
    na,
    ok,
    pc_on_day,
    phone_on_day,
)
from wake_alarm._audit_stick import check_lockdown, check_penalty
from wake_alarm._constants import RTC_WAKE_LEAD
from wake_alarm._enforcement_payload import is_honoured
from wake_alarm._enforcement_records import parse_instant

if TYPE_CHECKING:
    from datetime import date, datetime

    from wake_alarm._audit_sources import PcEvent

_FINAL_STOPS = frozenset({"escape", "session_started", "cap", "free_day", "sick_day"})
_RING_KINDS = frozenset({"ring_screen_opened", "ring_started_published"})
# alarm_fired.data.reason: only the daily ring is the morning's ring -- a
# free_day / sick_day firing rings nothing, and a rering presupposes an
# earlier ring.
_NOT_THE_MORNING_RING = frozenset({"free_day", "sick_day", "rering"})


def alarm_from_events(pc: list[PcEvent], day: date) -> datetime | None:
    """The ``A`` the fallback tick used on ``day``, if it ran."""
    for event in pc:
        alarm = event.fields.get("alarm")
        if event.event == "fallback_tick" and isinstance(alarm, str):
            at = parse_instant(alarm)
            if at is not None and at.date() == day:
                return at
    return None


def check_alarm(ev: Evidence) -> Check:
    """The alarm for the day is known."""
    if ev.alarm_at is None:
        return bad("alarm known", "no alarm time known for this day")
    return ok("alarm known", f"{ev.alarm_at:%H:%M} ({ev.alarm_source})")


def check_rtc(ev: Evidence) -> Check:
    """On a stick day the RTC was armed the night before."""
    if not ev.stick_day or ev.free_day:
        return na("RTC armed the night before", "not a stick day")
    if ev.sick_day:
        return na("RTC armed the night before", "sick day")
    armed = [
        e
        for e in ev.pc
        if e.event == "rtc_armed" and e.fields.get("target_date") == ev.day.isoformat()
    ]
    if not armed:
        return bad("RTC armed the night before", "no rtc_armed for this day")
    return ok(
        "RTC armed the night before",
        f"{armed[-1].fields.get('wake_at')} (logged {armed[-1].at:%H:%M})",
    )


def check_up_by_alarm(ev: Evidence) -> Check:
    """On a stick day the PC was up by ``A``."""
    if not ev.stick_day or ev.free_day or ev.alarm_at is None:
        return na("PC up by A", "not a stick day or no alarm")
    if ev.sick_day:
        return na("PC up by A", "sick day")
    alarm = ev.alarm_at
    lead = alarm - RTC_WAKE_LEAD - timedelta(minutes=5)
    ups = [u for u in ev.ups if u.at.date() == ev.day and u.at <= alarm]
    if ups:
        return ok("PC up by A", f"{ups[-1].kind} at {ups[-1].at:%H:%M:%S}")
    alive = [at for at in ev.pc_seen if lead <= at <= alarm]
    if alive:
        return ok("PC up by A", f"already up (logged at {alive[0]:%H:%M})")
    return bad("PC up by A", f"no boot, resume or PC event by {alarm:%H:%M}")


def check_ring(ev: Evidence) -> Check:
    """Something rang: the phone's ring screen, or the PC fallback."""
    if ev.free_day:
        return na("something rang", "free day")
    if ev.sick_day:
        return na("something rang", "sick day")
    fallback = pc_on_day(ev, "fallback_ring_start")
    phone = ev.ring_started or any(
        e.kind in _RING_KINDS
        or (
            e.kind == "alarm_fired"
            and e.data.get("reason") not in _NOT_THE_MORNING_RING
        )
        for e in phone_on_day(ev)
    )
    if not (phone or fallback):
        return bad("something rang", "no ring_started and no fallback ring")
    parts = (["phone"] if phone else []) + (["PC fallback"] if fallback else [])
    return ok("something rang", " + ".join(parts))


def check_fallback_stop(ev: Evidence) -> Check:
    """Every fallback ring stopped, and for a contract reason."""
    starts = pc_on_day(ev, "fallback_ring_start")
    if not starts:
        return na("fallback stopped correctly", "fallback did not ring")
    stops = pc_on_day(ev, "fallback_ring_stop")
    reasons = [str(e.fields.get("stop_reason")) for e in stops]
    if len(stops) < len(starts) or (reasons and reasons[-1] not in _FINAL_STOPS):
        return bad(
            "fallback stopped correctly",
            f"{len(starts)} start(s), stops: {', '.join(reasons) or 'none'}",
        )
    return ok("fallback stopped correctly", ", ".join(reasons))


CHECKS = (
    check_alarm,
    check_rtc,
    check_up_by_alarm,
    check_ring,
    check_fallback_stop,
    check_penalty,
    check_lockdown,
)


def run_checks(ev: Evidence) -> list[Check]:
    """Every check, in report order."""
    return [check(ev) for check in CHECKS]


def flags(ev: Evidence, checks: list[Check]) -> list[str]:
    """Escapes first, then every mismatch."""
    lines = []
    for escape in ev.escapes:
        honoured = "" if is_honoured(escape) else " (IGNORED by the PC: "
        if honoured:
            honoured += "after the last day or reason too short)"
        lines.append(f"ESCAPE {escape.at or escape.date}: {escape.reason!r}{honoured}")
    lines.extend(f"MISMATCH {c.name}: {c.detail}" for c in checks if c.ok is False)
    return lines