tests/test_integration_extras.py

"""Detector, live pipeline (file), conditioning tests."""

from __future__ import annotations

import json
import sys
import tempfile
from pathlib import Path

import numpy as np

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

from hoa64.analysis import angular_error_deg
from hoa64.audio_io import write_wav
from hoa64.conditioning import build_conditioning, comfy_txt2img_payload, spatial_prompt_fragment
from hoa64.detector import (
    Detection,
    detections_to_sphere_boxes,
    load_yolo_labels,
    write_demo_image_with_box,
)
from hoa64.live_audio import audio_to_hoa_stream, live_report_from_file, stereo_to_order1_hoa
from hoa64.synth import tone
from hoa64.vision import fuse_reports, report_from_boxes


def test_demo_image_detect_sphere():
    with tempfile.TemporaryDirectory() as td:
        img = Path(td) / "demo.png"
        det = write_demo_image_with_box(img)
        boxes = detections_to_sphere_boxes([det])
        assert boxes[0]["label"] == "demo_object"
        # center of box ~ (0.45, 0.5) → slight left az positive? cx=0.45 → az=(0.5-0.45)*90=+4.5
        assert abs(boxes[0]["az"] - 4.5) < 1.0
        rep = report_from_boxes(boxes, max_order=3)
        assert rep.energy > 0


def test_yolo_labels():
    with tempfile.TemporaryDirectory() as td:
        p = Path(td) / "labels.txt"
        # class 0 center front-ish
        p.write_text("0 0.5 0.5 0.2 0.2 0.9\n")
        dets = load_yolo_labels(p)
        assert len(dets) == 1
        boxes = detections_to_sphere_boxes(dets)
        assert abs(boxes[0]["az"]) < 1.0  # center → az≈0


def test_live_pipeline_from_synth_wav():
    with tempfile.TemporaryDirectory() as td:
        wav = Path(td) / "m.wav"
        sr = 16000
        sig = tone(800, 0.3, sr, amplitude=0.4)
        write_wav(wav, sig, sr)
        rep = live_report_from_file(wav, az_deg=30.0, el_deg=0.0, max_order=1)
        err = angular_error_deg(30.0, 0.0, rep.doa_az_deg, rep.doa_el_deg)
        assert err < 8.0, f"err={err}"


def test_stereo_pseudo_hoa():
    sr = 8000
    L = tone(400, 0.2, sr, amplitude=0.5)
    R = tone(400, 0.2, sr, amplitude=0.1)
    hoa = stereo_to_order1_hoa(L, R, width_az_deg=40)
    assert hoa.shape[0] == 64
    # left-dominant → positive Y energy
    assert np.mean(hoa[1] ** 2) > 0


def test_conditioning_from_fuse():
    audio = {"doa_az_deg": 15.0, "doa_el_deg": 0.0, "energy": 1.0, "kind": "spatial_field"}
    vision = {
        "doa_az_deg": 18.0,
        "doa_el_deg": 2.0,
        "energy": 0.5,
        "kind": "spatial_vision",
        "peak_az_deg": 18.0,
        "peak_el_deg": 2.0,
    }
    fused = fuse_reports(audio, vision)
    cond = build_conditioning(fused, base_prompt="cinematic room", style="natural")
    assert "cinematic room" in cond["positive_prompt"]
    assert cond["schema"] == "spatial-hoa.conditioning.v1"
    assert "control_vector" in cond
    frag = spatial_prompt_fragment(fused, style="tags")
    assert "spatial-az" in frag
    # offline payload with explicit checkpoint (no Comfy required)
    wf = comfy_txt2img_payload(
        cond, checkpoint="sd_xl_base_1.0.safetensors", auto_checkpoint=False
    )
    assert "3" in wf and wf["6"]["inputs"]["text"]
    assert wf["4"]["inputs"]["ckpt_name"] == "sd_xl_base_1.0.safetensors"
    assert wf["5"]["inputs"]["width"] == 1024  # XL default size


if __name__ == "__main__":
    test_demo_image_detect_sphere()
    print("OK test_demo_image_detect_sphere")
    test_yolo_labels()
    print("OK test_yolo_labels")
    test_live_pipeline_from_synth_wav()
    print("OK test_live_pipeline_from_synth_wav")
    test_stereo_pseudo_hoa()
    print("OK test_stereo_pseudo_hoa")
    test_conditioning_from_fuse()
    print("OK test_conditioning_from_fuse")