tests/test_timing.py

"""Synthetic, data-free tests for the clip/trajectory/video time-base mapping.

Covers two things found to be buggy on real data (see fpgm.types.ClipTiming's
docstring): the mp4-fps trap (video_frames_per_step) and the annotation-stride
trap (annotation_stride) -- and the fpgm.data.pointworld.verify_annotation_stride
helper that measures the latter rather than assuming it.
"""

from __future__ import annotations

import numpy as np
import pytest

from fpgm.data.pointworld import verify_annotation_stride
from fpgm.types import ClipTiming, DataError


class TestClipTimingAnnotationStride:
    def test_dt_is_stride_over_trajectory_fps(self):
        timing = ClipTiming(
            clip_start_frame=50, clip_end_frame=61, trajectory_fps=15.0, annotation_stride=2
        )
        assert timing.dt == pytest.approx(2.0 / 15.0)

    def test_dt_defaults_to_one_over_fps_at_stride_one(self):
        timing = ClipTiming(
            clip_start_frame=0, clip_end_frame=11, trajectory_fps=15.0, annotation_stride=1
        )
        assert timing.dt == pytest.approx(1.0 / 15.0)

    def test_clip_frame_to_trajectory_frame_matches_measured_bug(self):
        # The exact case from the bug report: clip "50:61", frame t=0 -> row 100.
        timing = ClipTiming(
            clip_start_frame=50, clip_end_frame=61, trajectory_fps=15.0, annotation_stride=2
        )
        t = np.arange(11)
        rows = timing.clip_frame_to_trajectory_frame(t)
        assert np.array_equal(rows, 2 * (50 + t))
        assert rows[0] == 100
        assert rows[-1] == 120

    def test_clip_frame_to_video_frame_composes_both_traps(self):
        # video_frames_per_step != 1 (the mp4-fps trap) must stack multiplicatively
        # with annotation_stride (the annotation trap), not replace it.
        timing = ClipTiming(
            clip_start_frame=10,
            clip_end_frame=21,
            trajectory_fps=15.0,
            video_frames_per_step=0.5,
            annotation_stride=2,
        )
        assert timing.clip_frame_to_video_frame(0) == pytest.approx(10 * 2 * 0.5)
        assert timing.clip_frame_to_video_frame(3) == pytest.approx(13 * 2 * 0.5)

    def test_clip_frame_to_seconds_spacing_is_dt(self):
        timing = ClipTiming(
            clip_start_frame=0, clip_end_frame=11, trajectory_fps=15.0, annotation_stride=2
        )
        seconds = timing.clip_frame_to_seconds(np.arange(11))
        diffs = np.diff(seconds)
        assert np.allclose(diffs, timing.dt)
        assert np.allclose(diffs, 2.0 / 15.0)

    def test_video_frame_to_clip_frame_roundtrip(self):
        timing = ClipTiming(
            clip_start_frame=25,
            clip_end_frame=36,
            trajectory_fps=15.0,
            video_frames_per_step=1.0,
            annotation_stride=2,
        )
        t = np.arange(11)
        video_frames = timing.clip_frame_to_video_frame(t)
        back = timing.video_frame_to_clip_frame(video_frames)
        assert np.allclose(back, t)

    def test_from_counts_defaults_and_accepts_explicit_stride(self):
        timing = ClipTiming.from_counts(
            clip_start=0,
            clip_end=11,
            trajectory_length=128,
            video_frame_count=127,
            annotation_stride=2,
        )
        assert timing.annotation_stride == 2
        assert timing.video_frames_per_step == pytest.approx(127 / 128)


class TestVerifyAnnotationStride:
    """Uses tiny synthetic joint arrays -- no real dataset needed."""

    def _trajectory(self, n_rows: int = 30, n_joints: int = 7) -> np.ndarray:
        rng = np.random.default_rng(0)
        return rng.normal(size=(n_rows, n_joints)).astype(np.float32)

    def test_detects_stride_two(self):
        trajectory = self._trajectory(n_rows=30)
        clip_start = 5
        n = 8
        rows = clip_start * 2 + 2 * np.arange(n)
        clip_joint_positions = trajectory[rows]

        detected = verify_annotation_stride(clip_joint_positions, trajectory, clip_start)
        assert detected == 2

    def test_detects_stride_one(self):
        trajectory = self._trajectory(n_rows=30)
        clip_start = 3
        n = 6
        rows = clip_start * 1 + 1 * np.arange(n)
        clip_joint_positions = trajectory[rows]

        detected = verify_annotation_stride(clip_joint_positions, trajectory, clip_start)
        assert detected == 1

    def test_raises_when_nothing_matches(self):
        trajectory = self._trajectory(n_rows=30)
        clip_start = 2
        n = 6
        # Independent random data: no candidate stride should line up with this.
        rng = np.random.default_rng(99)
        bogus_clip = rng.normal(size=(n, 7)).astype(np.float32)

        with pytest.raises(DataError):
            verify_annotation_stride(bogus_clip, trajectory, clip_start)

    def test_raises_when_every_candidate_runs_past_trajectory_end(self):
        trajectory = self._trajectory(n_rows=5)
        clip_start = 10  # already past the end of a 5-row trajectory at any stride
        clip_joint_positions = self._trajectory(n_rows=4)

        with pytest.raises(DataError):
            verify_annotation_stride(clip_joint_positions, trajectory, clip_start)

    def test_raises_on_empty_clip(self):
        trajectory = self._trajectory(n_rows=10)
        with pytest.raises(DataError):
            verify_annotation_stride(np.zeros((0, 7)), trajectory, 0)

    def test_prefers_exact_match_over_atol_noise(self):
        # A candidate with tiny floating point noise (float32 round-trip) should
        # still be picked over one that doesn't match at all.
        trajectory = self._trajectory(n_rows=30)
        clip_start = 1
        n = 5
        rows = clip_start * 2 + 2 * np.arange(n)
        clip_joint_positions = (trajectory[rows].astype(np.float64) + 1e-7).astype(np.float32)

        detected = verify_annotation_stride(clip_joint_positions, trajectory, clip_start)
        assert detected == 2