57 lines
2.6 KiB
Python
57 lines
2.6 KiB
Python
"""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,
|
|
}
|
|
|
|
|
|
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() != BOUNDS.keys():
|
|
raise ValueError("metric fields")
|
|
clean = {}
|
|
for name, limit in BOUNDS.items():
|
|
value = metrics[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")
|
|
clean[name] = value
|
|
if clean["elapsed_ms"] is None:
|
|
raise ValueError("elapsed missing")
|
|
return {"run_id": run_id, "sequence": sequence, "state": state, "metrics": clean}
|