lesavka/scripts/manual/validate_local_hevc_bundle_audit.py

#!/usr/bin/env python3
"""Validate the local synthetic HEVC+audio bundle audit artifact.

The local audit is our passwordless proof that the client can generate the
same coded flash/tone train we will later send through the WAN and RCT path.
This validator keeps the acceptance rules in one reusable place so later
hardware failures can be compared against a known-good client-origin manifest.
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

EXPECTED_EVENTS = 16
EXPECTED_AUDIO_PACKETS_PER_EVENT = 2
EXPECTED_VIDEO_PERIOD_US = 1_000_000
MAX_AUDIO_VIDEO_SKEW_US = 120_000


def fail(message: str) -> None:
    """Exit with a compact validation error for shell scripts and operators."""

    raise SystemExit(f"local HEVC bundle audit failed: {message}")


def require(condition: bool, message: str) -> None:
    """Keep validation checks readable while preserving precise error context."""

    if not condition:
        fail(message)


def load_manifest(path: Path) -> dict:
    """Read the manifest JSON generated by the local Rust bundle preflight."""

    try:
        return json.loads(path.read_text())
    except FileNotFoundError:
        fail(f"missing audit manifest: {path}")
    except json.JSONDecodeError as exc:
        fail(f"invalid audit JSON: {exc}")


def validate_manifest(data: dict) -> dict:
    """Validate manifest-level and per-event timing invariants.

    Input: the `lesavka.local-hevc-bundle-audit.v1` JSON object. Output: the
    summary object for caller reporting. The checks intentionally match the
    analyzer evidence floor and sync-probe event train: sixteen ordered coded
    HEVC frames, two nearby audio packets per frame, and no per-bundle audio
    skew outside the coded pulse width.
    """

    require(
        data.get("schema") == "lesavka.local-hevc-bundle-audit.v1",
        f"unexpected schema {data.get('schema')!r}",
    )
    summary = data.get("summary") or {}
    events = data.get("events") or []

    expected_summary = {
        "video_codec": "hevc",
        "metadata_mode": "1920x1080@30",
        "bundles": EXPECTED_EVENTS,
        "coded_video_events": EXPECTED_EVENTS,
        "annex_b_video_events": EXPECTED_EVENTS,
        "audio_packets": EXPECTED_EVENTS * EXPECTED_AUDIO_PACKETS_PER_EVENT,
        "bundles_with_audio_before_video": EXPECTED_EVENTS,
        "bundles_with_audio_after_video": EXPECTED_EVENTS,
        "monotonic_bundle_sequences": True,
    }
    for key, expected in expected_summary.items():
        require(
            summary.get(key) == expected,
            f"summary {key} expected {expected!r}, got {summary.get(key)!r}",
        )

    require(len(events) == EXPECTED_EVENTS, f"expected {EXPECTED_EVENTS} events, got {len(events)}")
    previous_seq = 0
    previous_video_pts = None
    for index, event in enumerate(events, start=1):
        seq = int(event.get("bundle_seq", -1))
        code = int(event.get("event_code", -1))
        video_pts = int(event.get("video_capture_pts_us", -1))
        send_pts = int(event.get("video_send_pts_us", -1))
        capture_start = int(event.get("capture_start_us", -1))
        capture_end = int(event.get("capture_end_us", -1))
        audio_pts = [int(value) for value in event.get("audio_capture_pts_us") or []]
        max_skew = int(event.get("max_audio_video_skew_us", -1))

        require(seq == index, f"event {index} has bundle_seq={seq}")
        require(code == index, f"event {index} has event_code={code}")
        require(seq > previous_seq, f"event {index} sequence is not monotonic")
        previous_seq = seq
        require(event.get("has_annex_b_start_code") is True, f"event {index} lacks Annex-B")
        require(
            int(event.get("audio_packets", -1)) == EXPECTED_AUDIO_PACKETS_PER_EVENT,
            f"event {index} has wrong audio packet count",
        )
        require(len(audio_pts) == EXPECTED_AUDIO_PACKETS_PER_EVENT, f"event {index} missing audio PTS")
        require(capture_start <= video_pts <= capture_end, f"event {index} video outside bounds")
        require(all(capture_start <= pts <= capture_end for pts in audio_pts), f"event {index} audio outside bounds")
        require(any(pts < video_pts for pts in audio_pts), f"event {index} lacks pre-video audio")
        require(any(pts > video_pts for pts in audio_pts), f"event {index} lacks post-video audio")
        require(send_pts >= video_pts, f"event {index} send PTS precedes capture PTS")
        require(max_skew <= MAX_AUDIO_VIDEO_SKEW_US, f"event {index} audio/video skew {max_skew}us")

        if previous_video_pts is not None:
            period = video_pts - previous_video_pts
            require(
                period == EXPECTED_VIDEO_PERIOD_US,
                f"event {index} period {period}us != {EXPECTED_VIDEO_PERIOD_US}us",
            )
        previous_video_pts = video_pts

    return summary


def main(argv: list[str]) -> int:
    """Validate one manifest path and print a concise operator summary."""

    if len(argv) != 2:
        print(f"usage: {argv[0]} AUDIT_JSON", file=sys.stderr)
        return 2

    path = Path(argv[1])
    summary = validate_manifest(load_manifest(path))
    print("local HEVC bundle audit validation: pass")
    print(f"- schema: lesavka.local-hevc-bundle-audit.v1")
    print(f"- mode: {summary['metadata_mode']} codec={summary['video_codec']}")
    print(f"- bundles: {summary['bundles']}")
    print(f"- coded video events: {summary['coded_video_events']}/{EXPECTED_EVENTS}")
    print(f"- event codes: 1..{EXPECTED_EVENTS}")
    print(f"- Annex-B video events: {summary['annex_b_video_events']}/{EXPECTED_EVENTS}")
    print(f"- audio packets: {summary['audio_packets']}")
    print(
        "- bundles with audio before/after video: "
        f"{summary['bundles_with_audio_before_video']}/{summary['bundles_with_audio_after_video']}"
    )
    print(f"- monotonic bundle sequences: {summary['monotonic_bundle_sequences']}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))