orisys-app/apps/autocal/tests/test_frames_index.py

"""A force row belongs to a frame by identity, never by its line number."""

from __future__ import annotations

import pytest

from evidence_helpers import (
    BASE_NS,
    force_row,
    frame_row,
    write_capture,
)
from orisys_autocal.frames_index import (
    DUPLICATE_REASON,
    joined_samples,
    read_frame_index,
    window_samples,
)


def _capture(tmp_path, frames, forces, **kwargs):
    return write_capture(tmp_path / "capture", frames, forces, video_segments=0, **kwargs)


def test_samples_are_paired_by_frame_id_not_by_row_order(tmp_path):
    frames = [
        frame_row(1, BASE_NS + 10),
        frame_row(2, BASE_NS + 20),
        frame_row(3, BASE_NS + 30),
    ]
    # The force rows are reversed on disk, as an out-of-order write would leave them.
    forces = [force_row(3, fn=30.0), force_row(2, fn=20.0), force_row(1, fn=10.0)]
    _capture(tmp_path, frames, forces)

    samples, report = joined_samples(tmp_path / "capture")

    assert [sample.frame_id for sample in samples] == [1, 2, 3]
    assert [sample.t_host_mono_ns for sample in samples] == [
        BASE_NS + 10, BASE_NS + 20, BASE_NS + 30,
    ]
    assert [sample.value("FN") for sample in samples] == [10.0, 20.0, 30.0]
    assert report.unjoined_force_rows == 0

    window = window_samples(samples, BASE_NS + 15, BASE_NS + 30)
    assert [sample.frame_id for sample in window] == [2]


def test_a_repeated_frame_id_excludes_the_samples_it_would_pair(tmp_path):
    """A camera that restarted reuses frame numbers; neither copy is the measurement."""
    frames = [
        frame_row(1, BASE_NS + 10),
        frame_row(2, BASE_NS + 20),
        frame_row(1, BASE_NS + 1_000),
        frame_row(2, BASE_NS + 1_010),
    ]
    forces = [
        force_row(1, fn=10.0),
        force_row(2, fn=20.0),
        force_row(1, fn=11.0),
        force_row(2, fn=21.0),
    ]
    _capture(tmp_path, frames, forces)

    samples, report = joined_samples(tmp_path / "capture")

    assert samples == []
    assert report.duplicate_frame_ids == 2
    assert DUPLICATE_REASON in report.reasons


def test_frames_outside_the_video_are_kept_only_when_asked_for(tmp_path):
    frames = [
        frame_row(1, BASE_NS + 10, in_video=1),
        frame_row(2, BASE_NS + 20, in_video=0),
        frame_row(3, BASE_NS + 30, in_video=1),
    ]
    forces = [force_row(1, fn=1.0), force_row(2, fn=2.0), force_row(3, fn=3.0)]
    _capture(tmp_path, frames, forces)

    with_video, report = joined_samples(tmp_path / "capture", require_video=True)
    everything, _ = joined_samples(tmp_path / "capture")

    assert [sample.frame_id for sample in with_video] == [1, 3]
    assert [sample.frame_id for sample in everything] == [1, 2, 3]
    assert report.frames_not_in_video == 1


def test_a_dropped_frame_row_has_no_time_to_pair_with(tmp_path):
    frames = [
        frame_row(1, BASE_NS + 10),
        frame_row(2, 0, dropped=1),
        frame_row(3, BASE_NS + 30),
    ]
    forces = [force_row(1, fn=1.0), force_row(2, fn=2.0), force_row(3, fn=3.0)]
    _capture(tmp_path, frames, forces)

    samples, report = joined_samples(tmp_path / "capture")

    assert [sample.frame_id for sample in samples] == [1, 3]
    assert report.unjoined_force_rows == 1
    assert report.frames_dropped == 1


def test_a_recording_without_an_in_video_column_still_joins(tmp_path):
    """Older recordings have no membership column; unknown is not "outside"."""
    fields = [
        name for name in (
            "frame_id", "source_frame_id", "t_rel_s", "t_host_mono_ns", "t_host_utc_ns",
            "acquired_at", "time_basis", "media_time_ms", "segment", "segment_frame",
            "dropped", "capture_dropped",
        )
    ]
    frames = [
        {"frame_id": 1, "t_host_mono_ns": BASE_NS + 10, "dropped": 0, "t_rel_s": "0.0"},
        {"frame_id": 2, "t_host_mono_ns": BASE_NS + 20, "dropped": 0, "t_rel_s": "0.0"},
    ]
    _capture(tmp_path, frames, [force_row(1, fn=1.0), force_row(2, fn=2.0)], fields=fields)

    samples, report = joined_samples(tmp_path / "capture", require_video=True)

    assert [sample.frame_id for sample in samples] == [1, 2]
    assert report.frames_not_in_video == 0


def test_a_missing_axis_is_empty_not_zero(tmp_path):
    frames = [frame_row(1, BASE_NS + 10), frame_row(2, BASE_NS + 20)]
    forces = [
        force_row(1, fn=10.0, fx=0.5, fy=-0.25),
        force_row(2, fn=None, fx=None, fy=None, valid=0),
    ]
    _capture(tmp_path, frames, forces)

    samples, _ = joined_samples(tmp_path / "capture")

    assert samples[0].values == (10.0, 0.5, -0.25)
    assert samples[1].value("FN") is None
    assert samples[1].valid is False
    assert samples[1].usable() is False
    assert samples[0].usable() is True


def test_the_frame_index_keeps_repeated_ids_and_unknown_video_membership(tmp_path):
    frames = [
        frame_row(1, BASE_NS + 10, in_video=1),
        frame_row(1, BASE_NS + 20, in_video=0),
    ]
    _capture(tmp_path, frames, [force_row(1, fn=1.0)])

    index = read_frame_index(tmp_path / "capture")

    assert [row.frame_id for row in index] == [1, 1]
    assert [row.in_video for row in index] == [True, False]


def test_a_missing_capture_directory_reads_as_empty(tmp_path):
    samples, report = joined_samples(tmp_path / "absent")

    assert samples == []
    assert report.frames_total == 0 and report.force_rows == 0


@pytest.mark.parametrize("frame_id", [0, 120])
def test_frame_ids_are_used_verbatim(tmp_path, frame_id):
    """Zero and large ids are still identities; nothing is renumbered."""
    frames = [frame_row(frame_id, BASE_NS + 10)]
    _capture(tmp_path, frames, [force_row(frame_id, fn=2.0)])

    samples, _ = joined_samples(tmp_path / "capture")

    assert samples[0].frame_id == frame_id