"""Minimal HTTP API so Qwythos / agents can request spatial reports. Stdlib only (no Flask). Default bind: 127.0.0.1:8765 Endpoints --------- GET /health GET /v1/spatial/schema — OpenAI-style tool descriptor POST /v1/spatial/analyze — JSON body → SpatialReport POST /v1/spatial/analyze_file — {path, mode, az, el, ...} POST /v1/spatial/demo_scene — synthetic multi-source report POST /v1/spatial/vision — Phase 3: boxes/rays → spatial report POST /v1/spatial/fuse — merge audio + vision report dicts OpenAI tools (for agents with bash): also install ``spatial-report`` on PATH. """ from __future__ import annotations import json import traceback from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any from urllib.parse import urlparse from . import __version__ from .report import ( REPORT_SCHEMA_VERSION, report_from_ambix_wav, report_from_hoa, report_from_mono_wav, report_from_scene, ) from .stream import SourceSpec from .synth import envelope_adsr, tone DEFAULT_HOST = "127.0.0.1" DEFAULT_PORT = 8765 TOOL_SCHEMA = { "type": "function", "function": { "name": "spatial_analyze", "description": ( "Analyze spatial audio (Ambix HOA or mono plane-wave) or visual " "bounding boxes on the sphere. Returns a compact spatial report " "(DOA, bands, frames) from the HOA-7 calculator — not an LLM." ), "parameters": { "type": "object", "properties": { "mode": { "type": "string", "enum": ["ambix_file", "mono_file", "demo_scene", "vision", "fuse"], "description": "Analysis mode", }, "path": { "type": "string", "description": "WAV path for ambix_file or mono_file", }, "az": { "type": "number", "description": "Azimuth degrees for mono plane-wave encode (0=front, +90=left)", }, "el": { "type": "number", "description": "Elevation degrees (0=horizon, +90=zenith)", }, "order": { "type": "integer", "description": "Max HOA order 0..7 (default 3 for speed, 7 full)", "default": 3, }, "boxes": { "type": "array", "description": "Vision boxes: [{az, el, w_deg?, h_deg?, weight?, label?}]", "items": {"type": "object"}, }, "audio_report": { "type": "object", "description": "Existing audio SpatialReport dict for fuse mode", }, "vision_report": { "type": "object", "description": "Existing vision SpatialReport dict for fuse mode", }, }, "required": ["mode"], }, }, } def _json_response(handler: BaseHTTPRequestHandler, code: int, obj: Any) -> None: body = json.dumps(obj, indent=None).encode("utf-8") handler.send_response(code) handler.send_header("Content-Type", "application/json") handler.send_header("Content-Length", str(len(body))) handler.send_header("Access-Control-Allow-Origin", "*") handler.end_headers() handler.wfile.write(body) def _read_json(handler: BaseHTTPRequestHandler) -> dict: n = int(handler.headers.get("Content-Length", "0")) raw = handler.rfile.read(n) if n else b"{}" if not raw: return {} return json.loads(raw.decode("utf-8")) def handle_analyze(body: dict) -> dict: mode = body.get("mode", "demo_scene") order = int(body.get("order", 3)) order = max(0, min(7, order)) if mode == "demo_scene": sr = int(body.get("sample_rate", 48000)) dur = float(body.get("duration", 0.4)) n = int(sr * dur) env = envelope_adsr(n, sr) sources = [ SourceSpec(0.0, 0.0, tone(440, dur, sr, amplitude=0.4) * env, "front"), SourceSpec(90.0, 10.0, tone(660, dur, sr, amplitude=0.25) * env, "left"), ] rep = report_from_scene(sources, sr, max_order=order) d = rep.to_dict() d["one_liner"] = rep.one_liner() return d if mode == "ambix_file": path = body.get("path") or body.get("file") if not path: raise ValueError("path required for ambix_file") rep = report_from_ambix_wav(path, max_order=order) d = rep.to_dict() d["one_liner"] = rep.one_liner() return d if mode == "mono_file": path = body.get("path") or body.get("file") if not path: raise ValueError("path required for mono_file") if body.get("az") is None: raise ValueError("az required for mono_file plane-wave encode") rep = report_from_mono_wav( path, float(body["az"]), float(body.get("el", 0.0)), max_order=order, ) d = rep.to_dict() d["one_liner"] = rep.one_liner() return d if mode == "vision": from .vision import report_from_boxes boxes = body.get("boxes") or [] rep = report_from_boxes(boxes, max_order=order) d = rep.to_dict() d["one_liner"] = rep.one_liner() return d if mode == "fuse": from .vision import fuse_reports ar = body.get("audio_report") or {} vr = body.get("vision_report") or {} return fuse_reports(ar, vr) if mode == "detect": from .detector import detect_to_sphere, load_boxes_json from .vision import report_from_boxes if body.get("boxes_path"): boxes = load_boxes_json(body["boxes_path"]) elif body.get("image"): boxes = detect_to_sphere( body["image"], backend=body.get("backend", "auto"), score_thresh=float(body.get("score", 0.5)), hfov_deg=float(body.get("hfov", 90)), vfov_deg=float(body.get("vfov", 60)), ) else: boxes = body.get("boxes") or [] rep = report_from_boxes(boxes, max_order=order) d = rep.to_dict() d["one_liner"] = rep.one_liner() d["boxes"] = boxes return d if mode == "live": from .live_audio import live_report rep = live_report( duration_sec=float(body.get("duration", 2.0)), sample_rate=int(body.get("sample_rate", 48000)), channels=int(body.get("channels", 1)), source=body.get("source"), az_deg=float(body.get("az", 0.0)), el_deg=float(body.get("el", 0.0)), max_order=order, keep_wav=body.get("write_wav"), ) d = rep.to_dict() d["one_liner"] = rep.one_liner() return d if mode == "panner": # UI spherical panner: az/el + W amplitude → spatial report (+ optional condition) from .conditioning import build_conditioning, panner_report rep = panner_report( float(body.get("az", body.get("az_deg", 0.0))), float(body.get("el", body.get("el_deg", 0.0))), float(body.get("w", body.get("w_amplitude", body.get("energy", 0.5)))), ) if body.get("condition") or body.get("prompt"): return build_conditioning( rep, base_prompt=str(body.get("prompt", "")), style=str(body.get("style", "natural")), ) return rep if mode == "condition": from .conditioning import build_conditioning, panner_report rep = body.get("report") if rep is None and ( "az" in body or "az_deg" in body or "w" in body or "w_amplitude" in body ): # Convenience: condition directly from panner knobs rep = panner_report( float(body.get("az", body.get("az_deg", 0.0))), float(body.get("el", body.get("el_deg", 0.0))), float(body.get("w", body.get("w_amplitude", body.get("energy", 0.5)))), ) else: rep = rep or body return build_conditioning( rep, base_prompt=str(body.get("prompt", "")), style=str(body.get("style", "natural")), ) if mode == "hoa_vector": # raw coefficients list import numpy as np coeffs = body.get("hoa") or body.get("coefficients") if coeffs is None: raise ValueError("hoa coefficients required") sr = int(body.get("sample_rate", 48000)) a = np.asarray(coeffs, dtype=np.float64) if a.ndim == 1: a = a.reshape(-1, 1) rep = report_from_hoa(a, sr, max_order=order) d = rep.to_dict() d["one_liner"] = rep.one_liner() return d raise ValueError(f"unknown mode: {mode}") class SpatialHandler(BaseHTTPRequestHandler): server_version = f"spatial-hoa/{__version__}" def log_message(self, fmt: str, *args) -> None: # quieter default sys_stderr = __import__("sys").stderr sys_stderr.write("%s - %s\n" % (self.address_string(), fmt % args)) def do_OPTIONS(self) -> None: self.send_response(204) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() def do_GET(self) -> None: path = urlparse(self.path).path if path in ("/health", "/v1/health"): _json_response( self, 200, { "status": "ok", "service": "spatial-hoa", "version": __version__, "schema": REPORT_SCHEMA_VERSION, }, ) return if path in ("/v1/spatial/schema", "/v1/tools"): _json_response( self, 200, { "tools": [TOOL_SCHEMA], "report_schema": REPORT_SCHEMA_VERSION, }, ) return _json_response(self, 404, {"error": "not found", "path": path}) def do_POST(self) -> None: path = urlparse(self.path).path try: body = _read_json(self) except Exception as e: _json_response(self, 400, {"error": f"invalid json: {e}"}) return try: if path in ( "/v1/spatial/analyze", "/v1/spatial/analyze_file", "/v1/spatial/demo_scene", "/v1/spatial/vision", "/v1/spatial/fuse", ): # map path to mode if not set if path.endswith("demo_scene") and "mode" not in body: body["mode"] = "demo_scene" elif path.endswith("vision") and "mode" not in body: body["mode"] = "vision" elif path.endswith("fuse") and "mode" not in body: body["mode"] = "fuse" elif path.endswith("analyze_file") and "mode" not in body: body["mode"] = "ambix_file" if body.get("ambix") else "mono_file" elif "mode" not in body: body["mode"] = "demo_scene" result = handle_analyze(body) _json_response(self, 200, result) return _json_response(self, 404, {"error": "not found", "path": path}) except Exception as e: _json_response( self, 400, {"error": str(e), "trace": traceback.format_exc()[-800:]}, ) def serve(host: str = DEFAULT_HOST, port: int = DEFAULT_PORT) -> None: httpd = ThreadingHTTPServer((host, port), SpatialHandler) print(f"spatial-hoa API http://{host}:{port} (hoa64 {__version__})") print(" GET /health") print(" GET /v1/spatial/schema") print(" POST /v1/spatial/analyze") httpd.serve_forever() def main(argv: list[str] | None = None) -> int: import argparse p = argparse.ArgumentParser(description="spatial-hoa HTTP API for Qwythos/agents") p.add_argument("--host", default=DEFAULT_HOST) p.add_argument("--port", type=int, default=DEFAULT_PORT) args = p.parse_args(argv) serve(args.host, args.port) return 0 if __name__ == "__main__": raise SystemExit(main())