atlas-iac/services/bstein-dev-home/av-observer/report-service/validation.py

"""Strict aggregate-only receipt schema: unknown fields and text are refused."""
import math


STATES = {"running", "completed", "hidden", "closed", "device-change", "stopped", "failed"}
BOUNDS = {
    "elapsed_ms": 180_000,
    "video_width": 8192, "video_height": 8192, "video_fps": 240,
    "audio_rate": 384_000, "audio_channels": 32, "audio_context_rate": 384_000,
    "video_callbacks": 100_000, "video_unobserved": 100_000, "video_resets": 100_000,
    "video_interval_p95_ms": 180_000, "video_gap_max_ms": 180_000,
    "audio_input_frames": 100_000_000, "audio_processed_frames": 100_000_000,
    "audio_zero_frames": 100_000_000, "audio_rms": 32, "audio_peak": 32,
    "paired_audio_signals": 256, "paired_video_signals": 256,
    "paired_matches": 256, "paired_ambiguous": 256, "paired_overflow": 100_000,
    "paired_estimate_audio_late_ms": 10_000, "paired_estimate_audio_early_ms": 10_000,
    "marker_frames": 100_000, "marker_bad": 100_000, "marker_ambiguous": 100_000,
    "marker_missing_fraction": 1, "marker_duplicate_fraction": 1,
    "pilot_windows": 18_000, "pilot_gap_max_ms": 180_000,
    "pilot_missing_fraction": 1, "pilot_trailing_gap_ms": 180_000,
    "pilot_discontinuities": 100_000_000,
    "pilot_recent_windows": 1000, "pilot_recent_observed_ms": 100_000,
    "pilot_recent_age_ms": 180_000, "pilot_recent_absent_fraction": 1,
    "pilot_recent_absence_max_ms": 100_000, "pilot_recent_trailing_ms": 100_000,
    "pilot_recent_interruptions": 100_000_000, "pilot_recent_coverage_complete": 1,
}
INTEGER_METRICS = frozenset((
    "video_width", "video_height", "audio_rate", "audio_channels", "audio_context_rate",
    "video_callbacks", "video_unobserved", "video_resets", "audio_input_frames",
    "audio_processed_frames", "audio_zero_frames", "paired_audio_signals",
    "paired_video_signals", "paired_matches", "paired_ambiguous", "paired_overflow",
    "marker_frames", "marker_bad", "marker_ambiguous", "pilot_windows", "pilot_discontinuities",
    "pilot_recent_windows", "pilot_recent_interruptions", "pilot_recent_coverage_complete",
))
RECENT_METRICS = frozenset(key for key in BOUNDS if key.startswith("pilot_recent_"))
LEGACY_METRICS = BOUNDS.keys() - RECENT_METRICS


def no_duplicate_fields(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate field")
        result[key] = value
    return result


def validate_report(data):
    """Return a rebuilt, bounded numeric receipt, never caller-owned text."""
    if type(data) is not dict or set(data) != {"run_id", "sequence", "state", "metrics"}:
        raise ValueError("report shape")
    run_id, sequence, state = data["run_id"], data["sequence"], data["state"]
    if type(run_id) is not str or len(run_id) != 32 or any(c not in "0123456789abcdef" for c in run_id):
        raise ValueError("run id")
    if type(sequence) is not int or not 1 <= sequence <= 64:
        raise ValueError("sequence")
    if type(state) is not str or state not in STATES:
        raise ValueError("state")
    metrics = data["metrics"]
    if type(metrics) is not dict or metrics.keys() not in (BOUNDS.keys(), LEGACY_METRICS):
        raise ValueError("metric fields")
    clean = {}
    for name, limit in BOUNDS.items():
        # A cached older page or retained receipt has no recent window. Only
        # its exact previous schema is accepted; missing evidence is not zero.
        value = metrics.get(name)
        if value is not None and (type(value) not in (int, float) or not 0 <= value <= limit or not math.isfinite(value)):
            raise ValueError("metric value")
        if value is not None and name in INTEGER_METRICS and type(value) is not int:
            raise ValueError("integer metric required")
        clean[name] = value
    if clean["elapsed_ms"] is None:
        raise ValueError("elapsed missing")
    return {"run_id": run_id, "sequence": sequence, "state": state, "metrics": clean}