File size: 2,377 Bytes
570b87b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 | #!/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()
|