examples/demo_phase3.py

#!/usr/bin/env python3
"""Phase 3 + Qwythos wiring demo: vision boxes, fuse with audio, HTTP API."""

from __future__ import annotations

import json
import sys
import urllib.request
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from hoa64.report import report_from_scene
from hoa64.stream import SourceSpec
from hoa64.synth import envelope_adsr, tone
from hoa64.vision import fuse_reports, report_from_boxes


def main() -> None:
    sr = 48000
    dur = 0.4
    n = int(sr * dur)
    env = envelope_adsr(n, sr)
    audio_rep = report_from_scene(
        [
            SourceSpec(20.0, 0.0, tone(500, dur, sr, amplitude=0.5) * env, "beep"),
        ],
        sr,
        max_order=3,
    )
    vision_rep = report_from_boxes(
        [
            {"az": 25.0, "el": 2.0, "w_deg": 12, "h_deg": 12, "weight": 1.0, "label": "speaker"},
            {"az": -100.0, "el": 0.0, "w_deg": 8, "h_deg": 8, "weight": 0.3, "label": "clutter"},
        ],
        max_order=3,
    )
    fused = fuse_reports(audio_rep.to_dict(), {**vision_rep.to_dict(), "one_liner": vision_rep.one_liner()})

    print("AUDIO ", audio_rep.one_liner())
    print("VISION", vision_rep.one_liner())
    print("FUSE  ", fused["one_liner"])
    print(json.dumps({"agreement": fused["agreement"], "sep_deg": fused["angular_separation_deg"]}, indent=2))

    # Optional live API check
    try:
        req = urllib.request.Request(
            "http://127.0.0.1:8765/health",
            method="GET",
        )
        with urllib.request.urlopen(req, timeout=1) as r:
            health = json.loads(r.read().decode())
        print("API   ", health)
        req = urllib.request.Request(
            "http://127.0.0.1:8765/v1/spatial/analyze",
            data=json.dumps({"mode": "vision", "order": 3, "boxes": [{"az": 0, "el": 0, "kind": "point"}]}).encode(),
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        with urllib.request.urlopen(req, timeout=5) as r:
            api = json.loads(r.read().decode())
        print("API vision one_liner:", api.get("one_liner"))
    except Exception as e:
        print(f"API not running yet ({e}); start: systemctl --user start spatial-hoa")

    print("\nPhase 3 vision + fuse ready. Qwythos: curl :8765 or spatial-report CLI.")


if __name__ == "__main__":
    main()