File size: 3,109 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 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | #!/usr/bin/env python3
"""End-to-end pack for manual testing: audio scene, vision detect, fuse, condition.
Writes artifacts under /tmp/spatial_hoa_e2e/ and prints commands for Qwythos/Comfy.
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from hoa64.audio_io import write_wav
from hoa64.conditioning import build_conditioning, comfy_txt2img_payload, save_conditioning
from hoa64.detector import detections_to_sphere_boxes, write_demo_image_with_box
from hoa64.report import report_from_scene
from hoa64.stream import SourceSpec, encode_scene
from hoa64.synth import envelope_adsr, tone
from hoa64.vision import fuse_reports, report_from_boxes
def main() -> None:
out = Path("/tmp/spatial_hoa_e2e")
out.mkdir(parents=True, exist_ok=True)
sr = 48000
dur = 0.5
n = int(sr * dur)
env = envelope_adsr(n, sr)
sources = [
SourceSpec(15.0, 0.0, tone(520, dur, sr, amplitude=0.45) * env, "beep"),
SourceSpec(-80.0, 5.0, tone(780, dur, sr, amplitude=0.2) * env, "side"),
]
audio_rep = report_from_scene(sources, sr, max_order=3)
audio_rep.save(out / "audio_report.json")
hoa = encode_scene(sources, max_order=3)
write_wav(out / "scene_ambix4.wav", hoa[:4], sr)
img = out / "demo_frame.png"
det = write_demo_image_with_box(img)
boxes = detections_to_sphere_boxes([det])
(out / "boxes.json").write_text(json.dumps(boxes, indent=2) + "\n")
vision_rep = report_from_boxes(boxes, max_order=3)
vision_rep.save(out / "vision_report.json")
fused = fuse_reports(
{**audio_rep.to_dict(), "one_liner": audio_rep.one_liner()},
{**vision_rep.to_dict(), "one_liner": vision_rep.one_liner()},
)
(out / "fuse_report.json").write_text(json.dumps(fused, indent=2) + "\n")
cond = build_conditioning(
fused,
base_prompt="cinematic interior, soft window light, photoreal",
style="natural",
)
save_conditioning(cond, out / "conditioning.json")
wf = comfy_txt2img_payload(cond, width=512, height=512, steps=20)
(out / "comfy_workflow.json").write_text(json.dumps(wf, indent=2) + "\n")
print("=== E2E artifacts ===")
for p in sorted(out.iterdir()):
print(f" {p}")
print("\nAUDIO ", audio_rep.one_liner())
print("VISION", vision_rep.one_liner())
print("FUSE ", fused["one_liner"])
print("PROMPT", cond["positive_prompt"])
print(
"""
Test commands:
curl -s http://127.0.0.1:8765/health
spatial-report analyze /tmp/spatial_hoa_e2e/scene_ambix4.wav --ambix -o /tmp/a.json
spatial-report detect --demo-image /tmp/spatial_hoa_e2e/frame2.png -o /tmp/v.json
spatial-report condition /tmp/spatial_hoa_e2e/fuse_report.json -o /tmp/c.json --prompt 'moody hall'
spatial-report live --duration 1.5 -o /tmp/live.json --write-wav /tmp/live.wav
# optional Comfy (if running):
spatial-report condition /tmp/spatial_hoa_e2e/fuse_report.json --comfy --write-workflow /tmp/wf.json
"""
)
if __name__ == "__main__":
main()
|