festifun/backend/tests/test_resolve.py

"""M8 annotation -> 3D resolution tests (integration phase).

The headline acceptance (spec M8): annotating the *same* stage corner from two videos
produces an anchor within 0.2 scene units of the true corner. We build the synthetic fixture
without ffmpeg (``run_ffmpeg=False`` populates the DB + point cloud only), forward-project a
known corner into each camera to get faithful bbox centers, then resolve and check the error.

Also covers: the single-view nearest-point fallback, the near-parallel/gap rejection path,
and the ``update_event`` correction helper.
"""

from __future__ import annotations

import numpy as np
import pytest

from festival4d import config, db, resolve, synthetic
from festival4d.geometry import quat_to_mat


@pytest.fixture(scope="module")
def fixture():
    # No ffmpeg: this writes videos/poses/anchors/events into the DB and points.ply to WORK_DIR.
    synthetic.build(run_ffmpeg=False)
    return synthetic.STAGE_CORNERS


def _project(video, pose, world_pt):
    """Forward-project a world point into a video, returning (px, py, z_cam).

    Mirrors the COLMAP pinhole the fixture uses: x_cam = R@X + t; px = fx*x/z + cx (y-down).
    """
    q = [pose.qw, pose.qx, pose.qy, pose.qz]
    t = np.array([pose.tx, pose.ty, pose.tz])
    R = quat_to_mat(q)
    x_cam = R @ np.asarray(world_pt, dtype=float) + t
    z = x_cam[2]
    px = pose.fx * x_cam[0] / z + pose.cx
    py = pose.fy * x_cam[1] / z + pose.cy
    return px, py, z


def _bbox_center(px, py, W, H, half=0.02):
    """A small normalized bbox centered on the projected pixel (clamped to [0,1])."""
    cx, cy = px / W, py / H
    return [
        max(0.0, cx - half), max(0.0, cy - half),
        min(1.0, cx + half), min(1.0, cy + half),
    ]


def _views_seeing(corner):
    """Videos whose first pose projects `corner` in front of the camera and inside the frame."""
    seen = []
    for v in db.get_videos():
        poses = db.get_poses(v.id)
        if not poses:
            continue
        p0 = poses[0]
        px, py, z = _project(v, p0, corner)
        if z > 0 and 0 <= px < v.width and 0 <= py < v.height:
            seen.append((v, p0, px, py))
    return seen


def test_two_view_triangulation_recovers_corner(fixture):
    corners = fixture
    # Pick a corner visible in at least two cameras.
    chosen = None
    for label, world in corners.items():
        views = _views_seeing(world)
        if len(views) >= 2:
            chosen = (label, np.asarray(world, dtype=float), views)
            break
    assert chosen is not None, "no stage corner is visible in >=2 synthetic cameras"
    label, world, views = chosen

    event = db.add_event(t_global_s=5.0, event_type="artist_moment", source="ai",
                         description=f"located {label}")

    # Annotate the corner from the first two cameras that see it, then resolve the second one
    # (which triangulates against the first).
    (v_a, pose_a, pxa, pya), (v_b, pose_b, pxb, pyb) = views[0], views[1]
    bbox_a = _bbox_center(pxa, pya, v_a.width, v_a.height)
    bbox_b = _bbox_center(pxb, pyb, v_b.width, v_b.height)

    db.add_annotation(v_a.id, pose_a.t_video_s, *bbox_a, event_id=event.id)
    res = resolve.resolve_annotation(v_b.id, pose_b.t_video_s, bbox_b, event_id=event.id)

    assert res.method == "triangulated", f"expected triangulation, got {res.method}"
    err = float(np.linalg.norm(np.asarray(res.point) - world))
    assert err < 0.2, f"triangulated {label} off by {err:.3f} units (>0.2)"
    # Sanity: the mutual-approach gap is tight for exact projections.
    assert res.gap is not None and res.gap < 0.2


def test_single_view_falls_back_to_point_cloud(fixture):
    corners = fixture
    world = np.asarray(corners["Stage FL"], dtype=float)
    views = _views_seeing(world)
    assert views, "Stage FL should be visible in at least one camera"
    v, pose, px, py = views[0]
    bbox = _bbox_center(px, py, v.width, v.height)

    # No event_id => no other view to triangulate => nearest point-cloud point (corner is in
    # the cloud) or centroid-depth ray. Either way a point comes back.
    res = resolve.resolve_annotation(v.id, pose.t_video_s, bbox, event_id=None)
    assert res.method in {"nearest_point", "ray_depth"}
    assert res.point is not None and len(res.point) == 3
    if res.method == "nearest_point":
        # The exact corner is seeded into the cloud, so the nearest hit should be very close.
        assert float(np.linalg.norm(np.asarray(res.point) - world)) < resolve.NEAREST_RADIUS


def test_update_event_sets_user_source(fixture):
    ev = db.add_event(t_global_s=9.0, event_type="candidate", source="audio_auto",
                      description="unlabeled")
    updated = db.update_event(ev.id, event_type="pyro", source="user")
    assert updated.event_type == "pyro"
    assert updated.source == "user"
    # Unspecified fields are preserved.
    assert updated.description == "unlabeled"


def test_missing_video_raises(fixture):
    with pytest.raises(KeyError):
        resolve.resolve_annotation(9999, 0.0, [0.4, 0.4, 0.6, 0.6], event_id=None)


# --- Phase 4: anchor supersede + delete (API-level, no ffmpeg) --------------------------------

def _views_and_bbox(corner):
    views = _views_seeing(corner)
    out = []
    for (v, p, px, py) in views:
        out.append((v, p, _bbox_center(px, py, v.width, v.height)))
    return out


def test_two_view_annotation_supersedes_to_single_anchor(fixture):
    """Two annotations of the same event from two views yield ONE anchor (the fallback is
    superseded in place by the triangulation), not a sibling pair (the Round-3 wart)."""
    from festival4d.api import AnnotationIn, create_annotation

    corner = np.asarray(fixture["Stage FR"], dtype=float)
    vb = _views_and_bbox(corner)
    assert len(vb) >= 2, "Stage FR should be visible in >=2 cameras"
    event = db.add_event(t_global_s=5.0, event_type="pyro", source="ai", description="p4 pyro")

    n0 = len(db.get_anchors())
    (va, pa, bba), (vb2, pb, bbb) = vb[0], vb[1]
    r1 = create_annotation(AnnotationIn(video_id=va.id, t_video_s=pa.t_video_s, bbox=bba, event_id=event.id))
    r2 = create_annotation(AnnotationIn(video_id=vb2.id, t_video_s=pb.t_video_s, bbox=bbb, event_id=event.id))

    assert r2["method"] == "triangulated" and r2["superseded"] is True
    assert r1["anchor_id"] == r2["anchor_id"]                 # same anchor, updated in place
    assert len(db.get_anchors()) - n0 == 1                    # ONE anchor, not two
    # Both annotations link to that single anchor.
    links = {a.resolved_anchor_id for a in db.get_annotations(event_id=event.id)}
    assert links == {r2["anchor_id"]}
    # Superseded position is the accurate triangulation.
    anchor = db.get_anchor(r2["anchor_id"])
    assert float(np.linalg.norm([anchor.x - corner[0], anchor.y - corner[1], anchor.z - corner[2]])) < 0.2


def test_delete_anchor_unlinks_annotations_and_404(fixture):
    from fastapi import HTTPException

    from festival4d.api import AnnotationIn, create_annotation, delete_anchor

    corner = np.asarray(fixture["Stage BL"], dtype=float)
    vb = _views_and_bbox(corner)
    assert vb, "Stage BL should be visible in >=1 camera"
    event = db.add_event(t_global_s=12.0, event_type="confetti", source="ai", description="p4 del")
    v, p, bb = vb[0]
    r = create_annotation(AnnotationIn(video_id=v.id, t_video_s=p.t_video_s, bbox=bb, event_id=event.id))
    aid = r["anchor_id"]
    assert aid is not None

    assert delete_anchor(aid) == {"deleted": aid}
    assert db.get_anchor(aid) is None
    # The annotation that resolved to it is unlinked, not orphaned with a dangling FK.
    assert all(a.resolved_anchor_id != aid for a in db.get_annotations(event_id=event.id))
    with pytest.raises(HTTPException) as exc:
        delete_anchor(aid)
    assert exc.value.status_code == 404