#!/usr/bin/env python3 """Machine-check FG-2 PCM assets and their event/audio contract.""" from __future__ import annotations import argparse import array import json import math import sys import wave from pathlib import Path ROOT = Path(__file__).resolve().parents[1] DEFAULT_MANIFEST = ROOT / "runtime/assets/fg2/audio-manifest.json" EXPECTED_EVENTS = { "route_committed": {"asset_id": "FG2-SFX-COMMIT", "priority": 2, "cooldown_seconds": 0.18, "duck": True}, "plan_committed": {"asset_id": "FG2-SFX-COMMIT", "priority": 2, "cooldown_seconds": 0.18, "duck": True}, "allied_basic": {"asset_id": "FG2-SFX-IMPACT", "priority": 4, "cooldown_seconds": 0.09, "duck": False}, "wave_arrived": {"asset_id": "FG2-SFX-FORECAST", "priority": 1, "cooldown_seconds": 0.22, "duck": True}, "pulse_linked": {"asset_id": "FG2-SFX-FORECAST", "priority": 1, "cooldown_seconds": 0.22, "duck": True}, "pulse_family_forecast": {"asset_id": "FG2-SFX-FORECAST", "priority": 1, "cooldown_seconds": 0.22, "duck": True}, "linked_positions_disclosed": {"asset_id": "FG2-SFX-FORECAST", "priority": 1, "cooldown_seconds": 0.22, "duck": True}, "third_pulse_link": {"asset_id": "FG2-SFX-FORECAST", "priority": 1, "cooldown_seconds": 0.22, "duck": True}, "safe_index_forecast": {"asset_id": "FG2-SFX-FORECAST", "priority": 1, "cooldown_seconds": 0.22, "duck": True}, "egg_positions_disclosed": {"asset_id": "FG2-SFX-FORECAST", "priority": 1, "cooldown_seconds": 0.22, "duck": True}, "safe_index_reversed": {"asset_id": "FG2-SFX-FORECAST", "priority": 1, "cooldown_seconds": 0.22, "duck": True}, "plan_response": {"asset_id": "FG2-SFX-COMMIT", "priority": 2, "cooldown_seconds": 0.18, "duck": True}, "pulse_family_interrupted": {"asset_id": "FG2-SFX-COMMIT", "priority": 2, "cooldown_seconds": 0.18, "duck": True}, "linked_actions_staggered": {"asset_id": "FG2-SFX-COMMIT", "priority": 2, "cooldown_seconds": 0.18, "duck": True}, "damage": {"asset_id": "FG2-SFX-IMPACT", "guard_asset_id": "FG2-SFX-GUARD", "priority": 4, "guard_priority": 3, "cooldown_seconds": 0.09, "duck": False}, "fallen": {"asset_id": "FG2-SFX-CASUALTY", "priority": 0, "cooldown_seconds": 0.40, "duck": True}, "outcome": {"asset_id": "FG2-SFX-VICTORY", "priority": 2, "cooldown_seconds": 0.60, "duck": True}, "recovery": {"asset_id": "FG2-SFX-RECOVERY", "priority": 2, "cooldown_seconds": 0.30, "duck": True}, } REQUIRED_EVENTS = set(EXPECTED_EVENTS) def resource_path(path_value: str) -> Path: if not path_value.startswith("res://"): raise ValueError(f"invalid resource path: {path_value}") return ROOT / "runtime" / path_value[6:] def inspect(asset: dict, manifest: dict) -> tuple[dict, list[str]]: path = resource_path(str(asset["path"])) errors: list[str] = [] if not path.is_file(): return {"id": asset["id"], "path": asset["path"]}, [f"missing {asset['id']}"] with wave.open(str(path), "rb") as wav_file: channels = wav_file.getnchannels() width = wav_file.getsampwidth() rate = wav_file.getframerate() frames = wav_file.getnframes() raw = wav_file.readframes(frames) samples = array.array("h") samples.frombytes(raw) if sys.byteorder != "little": samples.byteswap() normalized = [sample / 32768.0 for sample in samples] peak = max(abs(value) for value in normalized) if normalized else 0.0 peak_dbfs = 20.0 * math.log10(max(peak, 1.0e-12)) rms = math.sqrt(sum(value * value for value in normalized) / max(1, len(normalized))) rms_dbfs = 20.0 * math.log10(max(rms, 1.0e-12)) silence_ratio = sum(abs(value) < 0.0015 for value in normalized) / max(1, len(normalized)) dc_offset = abs(sum(normalized) / max(1, len(normalized))) clipped = sum(abs(sample) >= 32760 for sample in samples) duration = frames / rate if rate else 0.0 edge_delta = abs(normalized[0] - normalized[-1]) if normalized else 0.0 if channels != int(manifest["channels"]): errors.append(f"{asset['id']} channels {channels}") if width != int(manifest["sample_width_bytes"]): errors.append(f"{asset['id']} width {width}") if rate != int(manifest["sample_rate"]): errors.append(f"{asset['id']} rate {rate}") if abs(duration - float(asset["duration_seconds"])) > float(asset["duration_tolerance"]): errors.append(f"{asset['id']} duration {duration:.6f}") if not float(asset["peak_dbfs_min"]) <= peak_dbfs <= float(asset["peak_dbfs_max"]): errors.append(f"{asset['id']} peak {peak_dbfs:.3f}") if silence_ratio > float(asset["silence_ratio_max"]): errors.append(f"{asset['id']} silence {silence_ratio:.4f}") if dc_offset > 0.02: errors.append(f"{asset['id']} dc {dc_offset:.5f}") if clipped: errors.append(f"{asset['id']} clipped {clipped}") if bool(asset["loop"]) and edge_delta > float(asset["loop_edge_delta_max"]): errors.append(f"{asset['id']} loop edge {edge_delta:.6f}") return { "id": asset["id"], "path": asset["path"], "bus": asset["bus"], "loop": asset["loop"], "channels": channels, "sample_width_bytes": width, "sample_rate": rate, "frames": frames, "duration_seconds": round(duration, 6), "peak_dbfs": round(peak_dbfs, 4), "rms_dbfs": round(rms_dbfs, 4), "silence_ratio": round(silence_ratio, 6), "dc_offset": round(dc_offset, 7), "clipped_samples": clipped, "loop_edge_delta": round(edge_delta, 7), }, errors def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST) parser.add_argument("--output", type=Path) args = parser.parse_args() manifest = json.loads(args.manifest.read_text(encoding="utf-8")) errors: list[str] = [] if manifest.get("kind") != "abg-fg2-audio-manifest": errors.append("unexpected manifest kind") event_rows = manifest.get("events", {}) events = set(event_rows.keys()) if events != REQUIRED_EVENTS: errors.append(f"event set mismatch: {sorted(events)}") asset_ids = {asset["id"] for asset in manifest.get("assets", [])} if len(asset_ids) != 9: errors.append(f"asset count {len(asset_ids)}") for event_id, event in event_rows.items(): expected = EXPECTED_EVENTS.get(event_id) if expected is not None and event != expected: errors.append(f"{event_id} contract mismatch") for key in ("asset_id", "guard_asset_id"): if key in event and event[key] not in asset_ids: errors.append(f"{event_id} missing reference {event[key]}") records = [] for asset in manifest.get("assets", []): record, asset_errors = inspect(asset, manifest) records.append(record) errors.extend(asset_errors) report = { "version": 1, "kind": "abg-fg2-audio-validation-report", "ok": not errors, "errors": errors, "assets": records, "events": sorted(events), } encoded = json.dumps(report, ensure_ascii=True, indent=2, sort_keys=True) + "\n" if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(encoded, encoding="utf-8") if errors: for error in errors: print(f"FG2_AUDIO_ERROR {error}") return 1 print("ABG_FG2_AUDIO_ASSET_OK assets=9 events=18 base=6 fg3_extension=12 wav=pcm16 mono=true rate=22050 loops=2 clipping=0 dc=true silence=true seam=true paid_calls=0") return 0 if __name__ == "__main__": raise SystemExit(main())