"""Part A: the offline analysis pass and its result cache, checked on real footage.

Three things are being defended here, and each one is a way the feature could be
silently wrong rather than visibly broken:

* the stored payload is **what the live selector returns for that frame** — not a
  close approximation of it, and not a different frame;
* a result is either complete or unreadable, so a cancelled pass can never be drawn;
* the pass still costs what it cost when it was measured, so a cache that is correct
  but slower than re-running the chain is caught.

The inputs are real: the 33-frame box recording under ``dist/`` (its own ``frames.csv``,
``session.json`` and ``raw_0001.mp4``) and the shipped finger configuration and asset
bundle. The pass is driven with the **box** chain on it for the offset and round-trip
checks, and with the **finger** chain on the same footage for the keyframe, reset and
size checks — the finger chain over box footage is exactly what the Step 0 measurement
did, and it is the only way to get a validity mask (and therefore a static candidate
grid) out of this recording.
"""

from __future__ import annotations

import hashlib
import inspect
import json
import os
import shutil
import threading
from pathlib import Path

import numpy as np
import pytest
from orisys_app_shared.analysis import model as analysis_model
from orisys_app_shared.analysis.cache import (
    INCOMPLETE_MARKER,
    META_NAME,
    ResultCache,
    evict,
    scan_results,
)
from orisys_app_shared.analysis.model import frame_payload_bytes, result_key
from orisys_app_shared.analysis.reader import (
    AnalysisIncompleteError,
    AnalysisReader,
)
from orisys_app_shared.analysis.service import (
    AnalysisService,
    ChainSpec,
    chain_spec_for,
)
from orisys_app_shared.scene3d.visual_policy import VisualPolicy

REPO = Path(__file__).resolve().parents[2]
STUDIO = REPO / "dist" / "OrisysStudio"
RECORDING = (
    STUDIO
    / "outputs"
    / "recordings"
    / "20260930T015028Z_session0002_40cb8e7441434feab48bb646afdba011"
)
FINGER_CONFIG = STUDIO / "config" / "sensors" / "finger" / "sensor.json"
FINGER_EXPORT = STUDIO / "config" / "sensors" / "finger" / "assets" / "30fps-0703"

#: Step 0's digest of the finger validity mask, over ``np.packbits(mask)`` with
#: blake2b-16: one value across three videos and 327 frames. Reproduced here from the
#: stored keyframe, which is what makes the mask keyframe a *verified* copy rather than
#: a plausible-looking one. (The result stores a blake2b-8 digest of the same bytes; the
#: longer digest is the comparison because that is the one on record.)
STEP0_MASK_DIGEST = "d76349a3b901aea50915eb1308901c2e"

#: The frame id the reset is applied at. The recording's own ids run 20…52, so this is
#: deliberately not a small number: an off-by-one in the reset rule would land on a
#: frame that does not exist and be invisible.
REFERENCE_FRAME = 36


def _require_inputs() -> None:
    missing = [
        str(path)
        for path in (
            RECORDING / "frames.csv",
            RECORDING / "raw_0001.mp4",
            FINGER_CONFIG,
        )
        if not path.exists()
    ]
    if missing:
        pytest.skip("analysis inputs are not present: " + ", ".join(missing))
    pytest.importorskip("orisys", reason="the SDK is needed to run the analysis pass")


def _finger_chain() -> ChainSpec:
    return ChainSpec(
        sensor_type="finger", config_ref=FINGER_CONFIG, export_dir=FINGER_EXPORT
    )


def _box_service(root: Path, **kwargs) -> AnalysisService:
    from orisys_app_shared.record.playback import RecordingReader

    chain = chain_spec_for(RecordingReader(RECORDING).session())
    return AnalysisService(RECORDING, chain=chain, cache=ResultCache(root), **kwargs)


def _finger_service(root: Path, **kwargs) -> AnalysisService:
    return AnalysisService(
        RECORDING, chain=_finger_chain(), cache=ResultCache(root), **kwargs
    )


def _directory_bytes(directory: Path) -> int:
    return sum(path.stat().st_size for path in directory.rglob("*") if path.is_file())


# --------------------------------------------------------------------------- fixtures
@pytest.fixture(scope="module")
def box_pass(tmp_path_factory):
    _require_inputs()
    root = tmp_path_factory.mktemp("analysis-box")
    service = _box_service(root)
    outcome = service.run()
    return service, outcome


@pytest.fixture(scope="module")
def finger_pass(tmp_path_factory):
    _require_inputs()
    root = tmp_path_factory.mktemp("analysis-finger")
    service = _finger_service(root)
    outcome = service.run()
    return service, outcome


@pytest.fixture(scope="module")
def finger_pass_again(tmp_path_factory):
    """A second pass over the same input, in its own cache root."""
    _require_inputs()
    root = tmp_path_factory.mktemp("analysis-finger-again")
    service = _finger_service(root)
    return service, service.run()


@pytest.fixture(scope="module")
def finger_reset_pass(tmp_path_factory):
    _require_inputs()
    root = tmp_path_factory.mktemp("analysis-finger-reset")
    service = _finger_service(root, reference_frame=REFERENCE_FRAME)
    return service, service.run()


# --------------------------------------------------------------- frame identity
def test_the_box_chain_consumes_frames_before_the_first_one_a_caller_can_read(box_pass):
    """The offset is measured, recorded, and applied — not assumed.

    Constructing a box ``Sensor`` reads seven frames through its stitch probe, so the
    first frame the pass can read is the recording's *eighth* video frame. Nothing about
    the file says so, which is why the offset comes from the capture's own decode
    position and why the resulting frame identity is checked against ``frames.csv``.
    """
    _, outcome = box_pass
    assert outcome.complete, outcome.reason
    assert outcome.provenance.decode_order_offsets == {"1": 7}
    assert outcome.provenance.decode_order_offset == 7
    assert outcome.provenance.decode_order_offset_source == "measured"
    assert outcome.provenance.counts["skipped"] == 7

    frames = _box_service(Path(outcome.directory).parent).reader().info().frames
    rows = sorted(
        (frame for frame in frames if frame.in_video and frame.segment == 1),
        key=lambda frame: frame.segment_frame,
    )
    assert len(rows) == 33
    assert outcome.provenance.first_output_frame_id == rows[7].frame_id == 27
    assert outcome.provenance.last_output_frame_id == rows[-1].frame_id == 52
    assert outcome.frames_written == 26
    assert outcome.frames_expected == 26


def test_the_box_result_round_trips_against_a_fresh_run_of_the_chain(box_pass):
    """Frame for frame, the store equals what the chain computes now."""
    service, outcome = box_pass
    assert outcome.complete, outcome.reason
    reader = AnalysisReader(outcome.directory)
    try:
        checked = 0
        for live in service.recompute():
            stored = reader.frame(live.row.frame_id)
            assert stored.t_rel_s == live.row.t_rel_s
            assert live.payload_matches(stored), f"frame {live.row.frame_id} differs"
            checked += 1
        assert checked == outcome.frames_written == len(reader) == 26
    finally:
        reader.close()


# ------------------------------------------------------------------- keyframes
def test_the_finger_result_stores_the_candidate_grid_and_the_mask_once(finger_pass):
    """330 candidates on every frame, from one grid keyframe and one mask keyframe.

    This is the measurement the format rests on: the candidate *location* set does not
    move, so it is stored once and the per-frame record carries only values.
    """
    _, outcome = finger_pass
    assert outcome.complete, outcome.reason
    reader = AnalysisReader(outcome.directory)
    try:
        assert reader.keyframe_counts == {"grid": 1, "mask": 1}
        payloads = list(reader.frames())
        assert len(payloads) == 33
        assert {payload.candidates for payload in payloads} == {330}
        assert len({payload.grid_id for payload in payloads}) == 1

        mask = reader.mask()
        assert mask is not None and mask.shape == (300, 280)
        packed = np.packbits(mask, axis=None).tobytes()
        assert hashlib.blake2b(packed, digest_size=16).hexdigest() == STEP0_MASK_DIGEST

        assert outcome.provenance.counts["candidates_min"] == 330
        assert outcome.provenance.counts["candidates_max"] == 330
        assert outcome.provenance.counts["active_zero_frames"] == 5
        assert outcome.provenance.counts["active_total"] == 152
    finally:
        reader.close()


def test_every_stored_frame_reconstructs_what_the_live_selector_returns(finger_pass):
    """The keyframe path is exact: the grid is the one the selector chose, per frame."""
    service, outcome = finger_pass
    assert outcome.complete, outcome.reason
    reader = AnalysisReader(outcome.directory)
    try:
        policy = service.policy
        assert policy.flow_grid_spacing == 10 and policy.max_arrows == 400
        for live in service.recompute():
            stored = reader.frame(live.row.frame_id)
            assert live.payload_matches(stored), f"frame {live.row.frame_id} differs"
            rebuilt = reader.samples(live.row.frame_id)
            assert np.array_equal(rebuilt.rows, np.asarray(live.samples.rows, np.int32))
            assert np.array_equal(rebuilt.cols, np.asarray(live.samples.cols, np.int32))
            assert np.array_equal(
                np.asarray(rebuilt.magnitude, np.float32),
                np.asarray(live.samples.magnitude, np.float32),
            )
            assert np.array_equal(rebuilt.active, np.asarray(live.samples.active, bool))
    finally:
        reader.close()


def test_a_threshold_change_is_served_without_rebuilding_the_result(finger_pass):
    """Display settings are not in the key because the record can answer a new threshold.

    Activity is stored, but magnitude is stored too — so a viewer with a different
    ``flow_on_px`` gets a correct answer from the same file.
    """
    _, outcome = finger_pass
    reader = AnalysisReader(outcome.directory)
    try:
        frame_id = int(reader.frame_ids[0])
        recorded = reader.samples(frame_id)
        impossible = reader.samples(frame_id, policy=VisualPolicy(flow_on_px=1e9))
        assert impossible.active_count == 0
        permissive = reader.samples(
            frame_id, policy=VisualPolicy(flow_on_px=0.0, flow_off_px=0.0)
        )
        assert permissive.active_count >= recorded.active_count
        assert np.array_equal(permissive.magnitude, recorded.magnitude)
        assert np.array_equal(
            permissive.active, np.asarray(permissive.magnitude >= 0.0, bool)
        )
    finally:
        reader.close()


# ------------------------------------------------------------------ determinism
def test_two_passes_over_the_same_input_store_byte_identical_payloads(
    finger_pass, finger_pass_again
):
    """Re-running the pass produces the same bytes, so a cache hit cannot drift.

    The determinism contract is the **frame records** and the **keyframe digests**. The
    metadata is deliberately outside it: ``elapsed_s``, ``open_seconds``, ``started_at``
    and ``finished_at`` describe the run, not the measurement, and two runs of the same
    pass cannot share them.
    """
    _, first = finger_pass
    _, second = finger_pass_again
    assert first.complete and second.complete
    reader_a = AnalysisReader(first.directory)
    reader_b = AnalysisReader(second.directory)
    try:
        ids = reader_a.frame_ids
        assert np.array_equal(ids, reader_b.frame_ids)
        assert reader_a.frame_record_bytes == reader_b.frame_record_bytes
        for frame_id in ids:
            assert (
                reader_a.frame(int(frame_id)).canonical_bytes()
                == reader_b.frame(int(frame_id)).canonical_bytes()
            ), f"frame {frame_id} is not reproducible"
        assert {key: value.digest for key, value in reader_a.keyframes.items()} == {
            key: value.digest for key, value in reader_b.keyframes.items()
        }
        assert reader_a.mask_digest == reader_b.mask_digest
    finally:
        reader_a.close()
        reader_b.close()


# ------------------------------------------------------------------ cancellation
def test_an_interrupted_pass_is_incomplete_and_is_not_served(tmp_path):
    """A cancelled pass leaves a directory that refuses to open by default."""
    _require_inputs()
    root = tmp_path / "cache"
    cancelled = threading.Event()
    seen: list[int] = []

    def on_progress(progress):
        seen.append(progress.frame_id)
        if len(seen) >= 5:
            cancelled.set()

    service = _finger_service(root, on_progress=on_progress, cancelled=cancelled)
    outcome = service.run()

    assert outcome.complete is False
    assert outcome.reason == "cancelled"
    assert 0 < outcome.frames_written < 33
    assert (outcome.directory / INCOMPLETE_MARKER).is_file()
    meta = json.loads((outcome.directory / META_NAME).read_text(encoding="utf-8"))
    assert meta["complete"] is False
    assert meta["stopped_reason"] == "cancelled"
    assert meta["provenance"]["complete"] is False
    assert [item.complete for item in scan_results(root)] == [False]

    with pytest.raises(AnalysisIncompleteError):
        AnalysisReader(outcome.directory)

    # The partial data is still inspectable on purpose, and says what it is.
    partial = AnalysisReader(outcome.directory, allow_incomplete=True)
    try:
        assert partial.complete is False
        assert len(partial) == outcome.frames_written
        assert "incomplete" in partial.stale_reasons()
    finally:
        partial.close()


def test_a_declared_segment_that_is_missing_makes_the_result_incomplete(tmp_path):
    """A segment the session declares and the directory lacks is a failure, not a short read."""
    _require_inputs()
    recording = tmp_path / "recording"
    recording.mkdir()
    shutil.copy2(RECORDING / "frames.csv", recording / "frames.csv")
    shutil.copy2(RECORDING / "forces.csv", recording / "forces.csv")
    shutil.copy2(RECORDING / "events.csv", recording / "events.csv")
    session = json.loads((RECORDING / "session.json").read_text(encoding="utf-8"))
    session["stats"]["segments"] = 2
    (recording / "session.json").write_text(json.dumps(session), encoding="utf-8")
    try:
        os.link(RECORDING / "raw_0001.mp4", recording / "raw_0001.mp4")
    except OSError:
        shutil.copy2(RECORDING / "raw_0001.mp4", recording / "raw_0001.mp4")

    service = AnalysisService(
        recording, chain=_finger_chain(), cache=ResultCache(tmp_path / "cache")
    )
    outcome = service.run()
    assert outcome.complete is False
    assert outcome.reason == "segments_missing"
    assert outcome.frames_written == 0
    assert outcome.provenance.segments == [1]
    assert outcome.provenance.segments_declared == 2
    assert outcome.provenance.segments_missing == [2]
    with pytest.raises(AnalysisIncompleteError):
        AnalysisReader(outcome.directory)


def test_a_reference_frame_outside_the_recording_is_refused(tmp_path):
    """The reset is applied at a named frame or the pass does not run.

    Frame ids come from the acquisition session and need not start at zero or one: this
    recording's run 20…52. A requested reference of 16 therefore names nothing, and
    ignoring that would store a result whose ``reference_frame_id`` is not the frame
    that was asked for.
    """
    _require_inputs()
    service = _finger_service(tmp_path / "cache", reference_frame=16)
    with pytest.raises(ValueError, match="reference frame 16"):
        service.run()
    assert not (tmp_path / "cache").exists(), (
        "a refused pass must not leave a directory"
    )


# ------------------------------------------------------------------ reference reset
def test_resetting_the_reference_records_the_frame_and_clears_the_tracker(
    finger_pass, finger_reset_pass
):
    """The reset is exact at the named frame and touches nothing before it.

    ``Sensor.reset`` clears the optical-flow tracker, so the frame the reset is applied
    at measures against an empty reference: its displacement is exactly zero — the
    Step 0 signature. The frames before it are byte-identical to a pass with no reset,
    which is what makes "reference frame" a statement about a position rather than about
    a slightly different run.
    """
    _, plain = finger_pass
    _, reset = finger_reset_pass
    assert reset.complete, reset.reason
    assert reset.provenance.reference_is_explicit is True
    assert reset.provenance.reference_frame_id == REFERENCE_FRAME
    assert reset.provenance.first_output_frame_id == 20
    assert reset.provenance.counts["reset_count"] == 1
    assert reset.provenance.counts["reset_seconds"] >= 0.0
    # With no explicit reset the reference is the pass's own clean start.
    assert plain.provenance.reference_is_explicit is False
    assert plain.provenance.reference_frame_id == plain.provenance.first_output_frame_id

    reader = AnalysisReader(plain.directory)
    reset_reader = AnalysisReader(reset.directory)
    try:
        before = int(REFERENCE_FRAME) - 1
        assert reset_reader.frame(REFERENCE_FRAME).magnitude.max() == 0.0
        assert reader.frame(REFERENCE_FRAME).magnitude.max() > 0.0
        assert reset_reader.frame(before).magnitude.max() > 0.0

        for frame_id in reader.frame_ids:
            if int(frame_id) < REFERENCE_FRAME:
                assert (
                    reader.frame(int(frame_id)).canonical_bytes()
                    == reset_reader.frame(int(frame_id)).canonical_bytes()
                ), f"frame {frame_id} before the reference was disturbed by the reset"
        assert (
            reader.frame(int(REFERENCE_FRAME)).canonical_bytes()
            != reset_reader.frame(int(REFERENCE_FRAME)).canonical_bytes()
        )
    finally:
        reader.close()
        reset_reader.close()


# ------------------------------------------------------------------ the key
def test_the_key_follows_the_inputs_that_change_the_result_and_no_others(finger_pass):
    """Five components, and display settings are deliberately not one of them."""
    base = dict(
        recording="rec-a",
        sensor_config="config-a",
        calibration="cal-a",
        reference_frame=36,
    )
    digest = result_key(**base).digest
    assert result_key(**base).digest == digest
    assert result_key(**{**base, "sensor_config": "config-b"}).digest != digest
    assert result_key(**{**base, "algorithm": "other-chain"}).digest != digest
    assert result_key(**{**base, "calibration": "cal-b"}).digest != digest
    assert result_key(**{**base, "reference_frame": None}).digest != digest
    assert result_key(**{**base, "recording": "rec-b"}).digest != digest

    # Nothing in the signature can carry a display setting, so a change to one cannot
    # move the key even by accident.
    parameters = set(inspect.signature(result_key).parameters)
    assert not {name for name in parameters if "policy" in name or "display" in name}
    assert "policy" not in parameters

    # And the concrete consequence: two services that differ only in the policy the
    # samples are selected with address the same stored result.
    service = _finger_service(finger_pass[1].directory.parent)
    other = _finger_service(finger_pass[1].directory.parent)
    other.policy = VisualPolicy(flow_on_px=1.5, max_arrows=1)
    assert (
        service.dependency_report()["key"].digest
        == other.dependency_report()["key"].digest
        == finger_pass[1].key.digest
    )
    recorded = finger_pass[1].provenance.policy
    assert recorded["flow_on_px"] == 0.4 and recorded["max_arrows"] == 400


# ------------------------------------------------------------------ size
def test_the_stored_size_is_the_measured_encoding(box_pass, finger_pass):
    """State the encoding, then hold it: 4,034 B per frame at the finger grid.

    Implicit candidate grid (a keyframe) + float32 ``dx``/``dy``/``magnitude`` +
    bit-packed ``active`` + a 24-byte header + the 8-byte validity digest:
    ``24 + 330*12 + 42 + 8 = 4,034`` B. The alternative measured encodings are float16
    ``dxy``+magnitude (2,022 B) and dropping magnitude (4,002 B with the grid, no
    header); float32 is chosen because the round trip has to be *exact*, and the
    magnitude is what serves a changed threshold.
    """
    assert frame_payload_bytes(330, with_validity=True) == 4034
    assert frame_payload_bytes(330, with_validity=False) == 4026
    assert frame_payload_bytes(400, with_validity=False) == 4874

    _, finger = finger_pass
    reader = AnalysisReader(finger.directory)
    try:
        assert reader.bytes_per_frame == 4034.0
        assert reader.frame_record_bytes == 33 * 4034
        assert reader.bytes_per_frame <= 4096, "the stated per-frame bound"
        total_per_frame = _directory_bytes(finger.directory) / len(reader)
        assert total_per_frame <= 4600, "keyframes and metadata included"
    finally:
        reader.close()

    _, box = box_pass
    box_reader = AnalysisReader(box.directory)
    try:
        # The box chain has no validity mask, so there is no digest and every frame
        # carries its own grid keyframe: 4,874 B of record per frame at 400 candidates.
        assert box_reader.frame_record_bytes == 26 * 4874
        assert box_reader.keyframe_counts == {"grid": 26}
        assert box_reader.mask() is None
    finally:
        box_reader.close()


# ------------------------------------------------------------------ reader
def test_a_stored_result_is_served_without_the_recording_or_a_replay(
    finger_pass, tmp_path
):
    """A seek is a file read: the result answers with the recording and SDK out of reach."""
    _, outcome = finger_pass
    copy = tmp_path / "copied-result"
    shutil.copytree(outcome.directory, copy)

    original = AnalysisReader(outcome.directory)
    copied = AnalysisReader(copy)
    try:
        assert np.array_equal(original.frame_ids, copied.frame_ids)
        for frame_id in original.frame_ids:
            assert (
                original.frame(int(frame_id)).canonical_bytes()
                == copied.frame(int(frame_id)).canonical_bytes()
            )
        # Random access by frame id and by timestamp both land on the right frame.
        times = copied.times
        middle = len(times) // 2
        assert copied.frame(int(copied.frame_ids[middle])).t_rel_s == times[middle]
        assert copied.frame_at(float(times[middle])).frame_id == int(
            copied.frame_ids[middle]
        )
        assert copied.frame_at(float(times[middle]) + 1e-6).frame_id == int(
            copied.frame_ids[middle]
        )
        assert copied.frame_at(-1.0).frame_id == int(copied.frame_ids[0])
        assert copied.frame_at(float(times[-1]) + 100.0).frame_id == int(
            copied.frame_ids[-1]
        )
        window = list(
            copied.frames(start=int(copied.frame_ids[1]), stop=int(copied.frame_ids[2]))
        )
        assert [payload.frame_id for payload in window] == [
            int(copied.frame_ids[1]),
            int(copied.frame_ids[2]),
        ]
        assert copied.has(int(copied.frame_ids[0]))
        assert not copied.has(10_000)
    finally:
        original.close()
        copied.close()


def test_eviction_removes_whole_results_against_the_cap(tmp_path):
    """The cap is enforced by removing complete results, oldest first, never a mistake."""
    root = tmp_path / "cache"
    root.mkdir()
    made: list[Path] = []
    for index in range(3):
        directory = root / f"result-{index}"
        directory.mkdir()
        (directory / META_NAME).write_text(
            json.dumps({"complete": True, "key": f"result-{index}", "format": "x"}),
            encoding="utf-8",
        )
        (directory / "blob.bin").write_bytes(b"\0" * 4096)
        os.utime(directory / META_NAME, (1_600_000_000 + index, 1_600_000_000 + index))
        made.append(directory)

    def size_of(path: Path) -> int:
        return sum(item.stat().st_size for item in path.rglob("*") if item.is_file())

    kept = made[2]
    # A cap that one removal satisfies: the oldest complete result goes, and the
    # protected one is never a candidate whatever the arithmetic says.
    cap = sum(size_of(item) for item in made) - size_of(made[0])
    removed = evict(root, max_bytes=cap, keep=[kept])
    assert removed == [made[0]]
    assert made[0].exists() is False
    assert kept.is_dir() and made[1].is_dir()
    assert [item.directory.name for item in scan_results(root)] == [
        "result-1",
        "result-2",
    ]

    # A dry run reports without deleting, and still never proposes the protected result.
    assert evict(root, max_bytes=1, dry_run=True, keep=[kept]) == [made[1]]
    assert made[1].is_dir()
    assert kept.is_dir()


# ------------------------------------------------------------------ incompleteness gate
def test_a_result_without_metadata_is_never_served(tmp_path):
    """The only thing that opens is a directory whose metadata says it finished."""
    directory = tmp_path / "empty-result"
    directory.mkdir()
    from orisys_app_shared.analysis.reader import AnalysisNotFoundError

    with pytest.raises(AnalysisNotFoundError):
        AnalysisReader(directory)

    (directory / META_NAME).write_text(
        json.dumps({"format": analysis_model.ANALYSIS_FORMAT, "format_version": 1}),
        encoding="utf-8",
    )
    with pytest.raises(AnalysisIncompleteError):
        AnalysisReader(directory)