festifun/backend/tests/test_synthetic.py

"""Fixture tests for synthetic.py (foundation-owned; synthetic.py is a frozen contract).

These lock the synthetic fixture's contract: known audio offsets are recoverable, bang
times stand out for M7 candidate detection, the PLY round-trips, generated poses face the
stage under the frozen conversion, and ``build`` populates the DB + files as ground truth.
"""

from __future__ import annotations

import json
import shutil

import numpy as np
import pytest
from scipy.signal import correlate

from festival4d import config, db, synthetic
from festival4d.geometry import colmap_to_threejs


def _estimate_offset_ms(clip_ref, clip_v, sr):
    """Recover clip_v's offset vs the reference via cross-correlation (see derivation)."""
    corr = correlate(clip_v, clip_ref, mode="full")
    lag = int(np.argmax(corr)) - (len(clip_ref) - 1)
    return -lag / sr * 1000.0


def test_audio_slices_recover_known_offsets():
    sr = config.SYNTH_AUDIO_SR
    master, _ = synthetic.synth_master_audio(sr)
    ref = synthetic.slice_for_offset(master, sr, 0.0, config.SYNTH_DURATION_S)
    for offset_ms in config.SYNTH_OFFSETS_MS:
        clip = synthetic.slice_for_offset(master, sr, offset_ms, config.SYNTH_DURATION_S)
        recovered = _estimate_offset_ms(ref, clip, sr)
        assert abs(recovered - offset_ms) < 1.0, (offset_ms, recovered)


def test_bang_times_dominate_energy():
    """Loud bangs (M7 candidates) must clearly exceed the regular-beat energy floor."""
    sr = config.SYNTH_AUDIO_SR
    master, gt = synthetic.synth_master_audio(sr)

    def window_rms(center_s, half=0.05):
        i0 = int((center_s - synthetic.G_START - half) * sr)
        i1 = int((center_s - synthetic.G_START + half) * sr)
        return float(np.sqrt(np.mean(master[i0:i1] ** 2)))

    bang_rms = [window_rms(t) for t in gt["bang_times_global_s"]]
    # a spread of non-bang beat times for the baseline
    beat_rms = [window_rms(t) for t in (1.0, 4.5, 7.0, 11.5, 14.0)]
    assert min(bang_rms) > 3.0 * np.median(beat_rms)


def test_ply_roundtrip(tmp_path):
    pts, cols = synthetic.generate_point_cloud()
    path = tmp_path / "points.ply"
    synthetic.write_ply(path, pts, cols)
    pts2, cols2 = synthetic.read_ply(path)
    assert pts2.shape == pts.shape and cols2.shape == cols.shape
    np.testing.assert_allclose(pts2, pts, atol=1e-6)
    np.testing.assert_array_equal(cols2, cols)


def test_generated_poses_face_the_stage():
    """Each synthetic pose, run through the frozen conversion, must look at the stage."""
    poses = synthetic.camera_track(1, config.SYNTH_DURATION_S, config.SYNTH_FPS,
                                   config.SYNTH_VIDEO_W, config.SYNTH_VIDEO_H)
    target = np.array(synthetic.STAGE_TARGET)
    for p in poses:
        q = [p["qw"], p["qx"], p["qy"], p["qz"]]
        t = [p["tx"], p["ty"], p["tz"]]
        position, R = colmap_to_threejs(q, t)
        look = R @ np.array([0.0, 0.0, -1.0])          # Three.js camera looks down -z
        expected = target - position
        expected /= np.linalg.norm(expected)
        np.testing.assert_allclose(look, expected, atol=1e-9)


def test_build_no_ffmpeg_populates_db_and_files(tmp_path):
    summary = synthetic.build(base_dir=tmp_path, duration_s=5.0, run_ffmpeg=False)

    videos = db.get_videos()
    assert len(videos) == 3
    by_name = {v.filename: v for v in videos}
    assert by_name["cam0.mp4"].offset_ms == 0.0
    assert by_name["cam1.mp4"].offset_ms == 1370.0
    assert by_name["cam2.mp4"].offset_ms == -842.0
    for v in videos:
        assert len(db.get_poses(v.id)) > 0
        assert db.get_poses(v.id)[0].registered is True

    assert db.has_poses()
    assert len(db.get_anchors()) == 4
    assert len(db.get_events()) == len(synthetic.SEED_EVENTS)

    ply = tmp_path / "work" / "points.ply"
    assert ply.exists()
    pts, _ = synthetic.read_ply(ply)
    assert len(pts) == summary["point_count"] > 1000

    gt = json.loads((tmp_path / "work" / "ground_truth.json").read_text())
    assert [vv["offset_ms"] for vv in gt["videos"]] == [0.0, 1370.0, -842.0]
    assert gt["reference_filename"] == "cam0.mp4"


@pytest.mark.skipif(shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None,
                    reason="ffmpeg/ffprobe required for the full fixture build")
def test_build_full_ffmpeg_renders_seekable_videos(tmp_path):
    synthetic.build(base_dir=tmp_path, duration_s=4.0, run_ffmpeg=True)
    raw = tmp_path / "raw"
    for name in ("cam0.mp4", "cam1.mp4", "cam2.mp4"):
        path = raw / name
        assert path.exists() and path.stat().st_size > 10_000
    videos = db.get_videos()
    assert len(videos) == 3
    for v in videos:
        assert 3.5 < v.duration_s < 4.6         # ~4 s clips
        assert v.width == 640 and v.height == 360