"""A bare video's generated workspace: the two files the pass reads, and what they say.

The pass has one input shape — a directory with ``frames.csv``, ``session.json`` and
``raw_*.mp4`` — and it is a good shape: frame identity comes from the recording's own
counter, signal time from its own table, and the chain from its own metadata. A bare video
gets that shape *manufactured* for it rather than a second code path in the pass, so these
tests defend three properties of the manufacture: the rows are the container's real frames,
the documents are deterministic enough that reopening the video is a cache hit, and nothing
is written anywhere near the operator's file.

No SDK and no real footage: a video written by OpenCV here is the input, and the pass itself
is exercised in ``apps/studio/tests/test_bare_video_analysis.py`` over recorded footage.
"""

from __future__ import annotations

import csv
import json
from pathlib import Path

import pytest
from orisys_app_shared import paths as paths_module
from orisys_app_shared.analysis import bare_video
from orisys_app_shared.analysis.model import ForceSample, encode_frame, decode_frame
from orisys_app_shared.analysis.service import ChainSpec
from orisys_app_shared.record.playback import RecordingReader

FRAMES = 12
FPS = 30.0


# ------------------------------------------------------------------------ fixtures
def write_video(path: Path, *, frames: int = FRAMES, fps: float = FPS) -> Path:
    """A decodable video of ``frames`` distinct frames, written by OpenCV."""
    cv2 = pytest.importorskip("cv2", reason="OpenCV writes and reads the fixture video")
    import numpy as np

    path.parent.mkdir(parents=True, exist_ok=True)
    writer = cv2.VideoWriter(
        str(path), cv2.VideoWriter_fourcc(*"mp4v"), float(fps), (64, 48)
    )
    assert writer.isOpened(), "the fixture video could not be opened for writing"
    for index in range(frames):
        image = np.full((48, 64, 3), 30 + index * 8, dtype=np.uint8)
        image[::8, :, :] = 200
        image[:, ::8, :] = 90
        writer.write(image)
    writer.release()
    assert path.is_file() and path.stat().st_size > 0
    return path


class _Entry:
    """The catalog-entry fields the workspace generator asks about."""

    def __init__(self, *, family: str = "box", name: str = "Box", blocked: str = "",
                 sensor_json: Path | None = None, ref: str = "shipped/box",
                 has_profile: bool = True) -> None:
        self.product_family = family
        self.name = name
        self.ref = ref
        self.folder = ref.split("/")[-1]
        self.profile_id = f"orisys.{family}.r01"
        self.display_name = name
        self.bundle_path = Path(sensor_json).parent if sensor_json else Path(".")
        self.sensor_json_path = sensor_json
        #: ``None`` stands for a folder no bundle could be read from at all.
        self.profile = object() if has_profile else None
        self._blocked = blocked

    @property
    def start_blocked_reason(self) -> str | None:
        return self._blocked or None


@pytest.fixture
def video(tmp_path) -> Path:
    return write_video(tmp_path / "operator" / "clip.mp4")


@pytest.fixture
def workspace_root(tmp_path, monkeypatch) -> Path:
    """The app-owned root the workspace goes to, pointed at this test."""
    root = tmp_path / "derived" / "video-workspaces"
    monkeypatch.setattr(paths_module, "video_workspace_dir", lambda: root)
    return root


def _build(video: Path, **kwargs) -> bare_video.SyntheticWorkspace:
    return bare_video.build_workspace(video, profile=_Entry(**kwargs), setup=None)


def _entry_set(directory: Path) -> set:
    """Every entry under ``directory`` as ``(relative path, size)``."""
    found = set()
    for path in Path(directory).rglob("*"):
        relative = str(path.relative_to(directory))
        found.add((relative, -1 if path.is_dir() else path.stat().st_size))
    return found


# ------------------------------------------------------------------- the frame index
def test_the_frame_index_is_one_row_per_container_frame(video, workspace_root):
    """``frames.csv`` is the video's own frames: ids from 1, signal time from the rate.

    A bare video has one clock — the container's — so ``t_rel_s`` is that clock divided
    out, and every row is in the video because nothing was captured and nothing was lost.
    """
    workspace = _build(video)

    assert [row.frame_id for row in workspace.frames] == list(range(1, FRAMES + 1))
    assert [row.segment for row in workspace.frames] == [1] * FRAMES
    assert [row.segment_frame for row in workspace.frames] == list(range(1, FRAMES + 1))
    assert all(row.in_video for row in workspace.frames)
    assert not any(row.dropped for row in workspace.frames)
    assert workspace.frames[0].t_rel_s == 0.0
    assert workspace.frames[-1].t_rel_s == pytest.approx((FRAMES - 1) / FPS, abs=1e-9)

    # The file is read by the pass's own reader, so it has to say the same thing the
    # in-memory rows do — that equality is the whole point of generating it.
    reader = RecordingReader(workspace.directory)
    try:
        loaded = reader.info().frames
    finally:
        reader.close()
    assert [(row.frame_id, row.segment, row.segment_frame, row.in_video) for row in loaded] == [
        (row.frame_id, row.segment, row.segment_frame, row.in_video) for row in workspace.frames
    ]
    assert [row.t_rel_s for row in loaded] == pytest.approx(
        [row.t_rel_s for row in workspace.frames], abs=1e-9
    )


def test_the_session_document_says_what_the_pass_needs(video, workspace_root, tmp_path):
    """``device``/``stats``/``files`` are what the chain and the segment plan are read from."""
    config = tmp_path / "sensors" / "box" / "sensor.json"
    config.parent.mkdir(parents=True)
    config.write_text('{"camera": {"fps": 30}}', encoding="utf-8")
    workspace = _build(video, sensor_json=config)

    session = workspace.session
    assert session["device"]["sensor_type"] == "box"
    assert session["device"]["config_ref"] == str(config)
    assert session["device"]["config_sha256"]
    assert session["stats"]["frames"] == FRAMES
    assert session["stats"]["video_frames"] == FRAMES
    assert session["stats"]["segments"] == 1
    assert session["files"]["video"] == ["raw_0001.mp4"]
    assert (workspace.directory / "raw_0001.mp4").is_file()

    # ...and the pass's own chain reader agrees, from the file rather than from memory.
    reader = RecordingReader(workspace.directory)
    try:
        assert reader.session()["device"]["sensor_type"] == "box"
    finally:
        reader.close()


def test_the_workspace_is_marked_synthetic(video, workspace_root):
    """A generated workspace must be distinguishable from a recording, forever.

    ``read_synthetic`` is the question the app asks a directory before it decides how to
    treat it, and it answers ``None`` for a real recording — the difference between
    "somebody measured this" and "this was generated here".
    """
    workspace = _build(video)
    block = bare_video.read_synthetic(workspace.directory)

    assert block is not None
    assert block["generator"] == bare_video.GENERATOR
    assert block["source_video"] == str(Path(video).resolve())
    assert block["frame_index"].startswith("generated:")
    assert bare_video.is_synthetic(workspace.directory)
    assert bare_video.source_video(workspace.directory) == Path(video).resolve()

    # A real recording's session.json has no such block, and a folder without one is not
    # a workspace at all.
    plain = workspace.directory.parent / "plain"
    plain.mkdir()
    (plain / "session.json").write_text(json.dumps({"session_id": 1}), encoding="utf-8")
    assert bare_video.read_synthetic(plain) is None
    assert bare_video.read_synthetic(workspace.directory / "nothing-here") is None


def test_reopening_the_same_video_is_the_same_analysis_key(video, workspace_root, tmp_path):
    """Determinism is not tidiness: the key is what makes a second open a cache read.

    ``recording_identity`` hashes the session's own facts, so anything derived from the
    clock would give the same video a new result directory on every open — and the pass
    would run again every time.
    """
    config = tmp_path / "sensors" / "box" / "sensor.json"
    config.parent.mkdir(parents=True)
    config.write_text("{}", encoding="utf-8")
    chain = ChainSpec(sensor_type="box", config_ref=config)

    first = _build(video, sensor_json=config)
    again = _build(video, sensor_json=config)
    assert again.directory == first.directory
    assert again.session["session_id"] == first.session["session_id"]

    from orisys_app_shared.analysis.service import AnalysisService

    keys = [
        AnalysisService(item.directory, chain=chain).dependency_report()["key"].digest
        for item in (first, again)
    ]
    assert keys[0] == keys[1], "reopening the video produced a different result key"

    # A different video is a different workspace and a different key.
    other = write_video(video.parent / "other.mp4", frames=FRAMES + 3)
    third = _build(other, sensor_json=config)
    assert third.directory != first.directory
    assert (
        AnalysisService(third.directory, chain=chain).dependency_report()["key"].digest
        != keys[0]
    )


def test_the_calibration_is_part_of_the_key_by_its_content(video, workspace_root, tmp_path):
    """Coefficients edited in place must not keep serving a result computed from the old ones."""
    config = tmp_path / "sensors" / "box" / "sensor.json"
    config.parent.mkdir(parents=True)
    config.write_text("{}", encoding="utf-8")
    calibration = tmp_path / "force_calibration.json"
    calibration.write_text('{"gain": 1}', encoding="utf-8")

    first = bare_video.build_workspace(
        video,
        profile=_Entry(sensor_json=config),
        setup=bare_video.BareVideoSetup(profile_ref="shipped/box", calibration=calibration),
    )
    calibration.write_text('{"gain": 2}', encoding="utf-8")
    second = bare_video.build_workspace(
        video,
        profile=_Entry(sensor_json=config),
        setup=bare_video.BareVideoSetup(profile_ref="shipped/box", calibration=calibration),
    )
    assert first.session["calibration_id"] != second.session["calibration_id"]
    # The name still travels with the digest, so a reader knows which file it was.
    assert second.session["calibration_id"].startswith("force_calibration.json@")

    raw = bare_video.build_workspace(
        video,
        profile=_Entry(sensor_json=config),
        setup=bare_video.BareVideoSetup(profile_ref="shipped/box", calibration=None),
    )
    assert raw.session["calibration_id"] == ""


def test_a_chosen_calibration_that_is_not_there_is_refused(video, workspace_root, tmp_path):
    with pytest.raises(bare_video.BareVideoError):
        bare_video.build_workspace(
            video,
            profile=_Entry(),
            setup=bare_video.BareVideoSetup(
                profile_ref="shipped/box", calibration=tmp_path / "absent.json"
            ),
        )


# ------------------------------------------------------------------- what is refused
def test_a_split_recording_is_refused_rather_than_half_analysed(tmp_path):
    """More than one file is one measurement, not one input: refuse it, by name.

    Analysing the first segment of a split recording and calling the result that
    recording's measurement is a measurement error, and it is silent — the only visible
    symptom is a short timeline.
    """
    segments = tmp_path / "split"
    segments.mkdir()
    write_video(segments / "raw_0001.mp4")
    write_video(segments / "raw_0002.mp4", frames=4)

    with pytest.raises(bare_video.BareVideoError) as failure:
        bare_video.resolve_video(segments / "raw_0001.mp4")
    assert "raw_0001.mp4" in str(failure.value) and "raw_0002.mp4" in str(failure.value)
    assert "split recording" in str(failure.value)

    # A folder is read as "the video in here", and a folder of two is the same refusal.
    with pytest.raises(bare_video.BareVideoError) as folder_failure:
        bare_video.resolve_video(segments)
    assert "holds 2 video files" in str(folder_failure.value)


def test_a_lone_video_in_a_folder_is_that_video(tmp_path):
    """One file is unambiguous, however it is pointed at."""
    folder = tmp_path / "single"
    folder.mkdir()
    only = write_video(folder / "clip.mp4")
    assert bare_video.resolve_video(folder) == only
    assert bare_video.resolve_video(only) == only


def test_something_that_is_not_a_video_is_refused_with_its_reason(tmp_path):
    text = tmp_path / "notes.txt"
    text.write_text("notes", encoding="utf-8")
    with pytest.raises(bare_video.BareVideoError) as failure:
        bare_video.resolve_video(text)
    assert "not a video" in str(failure.value)

    with pytest.raises(bare_video.BareVideoError):
        bare_video.resolve_video(tmp_path / "absent.mp4")

    empty = tmp_path / "empty"
    empty.mkdir()
    with pytest.raises(bare_video.BareVideoError):
        bare_video.resolve_video(empty)


def test_a_container_without_a_timeline_is_refused(tmp_path):
    """No frame count or rate means every ``t_rel_s`` would be a guess."""
    broken = tmp_path / "broken.mp4"
    broken.write_bytes(b"not a container")
    with pytest.raises(bare_video.BareVideoError):
        bare_video.probe(broken)


def test_a_profile_with_no_chain_is_listed_with_its_own_reason(tmp_path):
    """Nothing is hidden: every entry is offered, and the refused ones say why.

    Two different refusals, from two owners — the SDK's validation and this app's chain
    table — and both have to reach the operator as a sentence rather than as an entry that
    is simply missing from the list.
    """
    entries = [
        _Entry(family="box", ref="shipped/box"),
        _Entry(family="finger", ref="shipped/finger"),
        _Entry(family="finger-mems", ref="shipped/finger-mems", name="MEMS"),
        _Entry(
            family="box",
            ref="user/box",
            name="Box (user)",
            blocked="the SDK rejected this profile (failed)",
        ),
    ]
    assert [entry.ref for entry in bare_video.profiles_for_video(entries)] == [
        "shipped/box",
        "shipped/finger",
        "shipped/finger-mems",
        "user/box",
    ]
    assert bare_video.setup_blocked_reason(entries[0]) is None
    assert bare_video.setup_blocked_reason(entries[1]) is None
    assert "finger-mems" in str(bare_video.setup_blocked_reason(entries[2]))
    assert bare_video.setup_blocked_reason(entries[3]) == (
        "the SDK rejected this profile (failed)"
    ), "the SDK's own verdict must win over this app's chain limit"

    # A folder no bundle could be read from has no profile behind it, so there is nothing
    # to process with — and nothing to refuse *by* either.
    assert bare_video.profiles_for_video([_Entry(has_profile=False)]) == []
    with pytest.raises(bare_video.BareVideoError) as failure:
        bare_video.build_workspace(
            write_video(tmp_path / "clip.mp4"), profile=_Entry(family="uv_unroll")
        )
    assert "no chain" in str(failure.value)


# ------------------------------------------------------------------------- the files
def test_nothing_is_written_beside_the_operator_s_video(video, workspace_root):
    """The frame index is the app's derived data, and the video's folder is the operator's."""
    before = _entry_set(video.parent)
    workspace = _build(video)

    assert _entry_set(video.parent) == before, "the workspace wrote into the video's folder"
    assert workspace.directory.parent == workspace_root
    assert sorted(path.name for path in workspace.directory.iterdir()) == [
        "frames.csv",
        "raw_0001.mp4",
        "session.json",
    ]


def test_the_video_inside_the_workspace_is_the_operator_s_own_file(video, workspace_root):
    """A link when the filesystem allows one, and the copy is the same bytes when it does not."""
    workspace = _build(video)
    inside = workspace.directory / "raw_0001.mp4"

    assert inside.read_bytes() == video.read_bytes()
    mode = workspace.session["synthetic"]["video_in_workspace"]
    assert mode in ("link", "copy", "existing")

    # Rebuilding finds the same file rather than stacking a second copy.
    again = _build(video)
    assert again.session["synthetic"]["video_in_workspace"] == "existing"


def test_the_frame_index_is_written_as_the_columns_the_reader_wants(video, workspace_root):
    workspace = _build(video)
    with (workspace.directory / "frames.csv").open(encoding="utf-8", newline="") as handle:
        rows = list(csv.DictReader(handle))
    assert list(rows[0]) == list(bare_video.FRAME_COLUMNS)
    assert len(rows) == FRAMES
    assert rows[0]["frame_id"] == "1"
    assert rows[0]["in_video"] == "1"
    assert rows[0]["dropped"] == "0"
    assert rows[0]["acquired_at"] == ""


# ------------------------------------------------------------------ the force channel
def test_a_frame_carries_its_force_sample_when_the_pass_collected_one():
    """Six values and two verdicts, appended last so an older reader still reads the frame."""
    sample = ForceSample(
        normal=1.5,
        shear_x=-0.25,
        shear_y=0.0,
        normal_raw=1200.0,
        shear_x_raw=-30.0,
        shear_y_raw=0.0,
        valid=True,
        calibrated=True,
    )
    record = encode_frame(
        frame_id=7,
        t_rel_s=0.25,
        grid_id=0,
        dx=[1.0, 2.0],
        dy=[0.0, 0.0],
        magnitude=[1.0, 2.0],
        active=[True, False],
        force=sample,
    )
    decoded = decode_frame(record)
    assert decoded.frame_id == 7
    assert decoded.force == sample
    assert decoded.force.newtons() == (1.5, -0.25, 0.0)
    assert decoded.force.raw() == (1200.0, -30.0, 0.0)
    assert decoded.force.calibrated is True

    # A frame the pass did not collect force for carries none, and says so by absence.
    without = decode_frame(
        encode_frame(
            frame_id=8,
            t_rel_s=0.5,
            grid_id=0,
            dx=[1.0],
            dy=[0.0],
            magnitude=[1.0],
            active=[True],
        )
    )
    assert without.force is None

    # A readout the chain could not make is stored as "no measurement", never as zero.
    empty = ForceSample.from_reading(
        {
            "normal_force": float("nan"),
            "shear_x": float("nan"),
            "shear_y": float("nan"),
            "valid": False,
            "force_calibrated": False,
        }
    )
    assert not empty.any_finite and not empty.valid
    restored = decode_frame(
        encode_frame(
            frame_id=9,
            t_rel_s=0.75,
            grid_id=0,
            dx=[1.0],
            dy=[0.0],
            magnitude=[1.0],
            active=[True],
            force=empty,
        )
    ).force
    assert restored is not None and not restored.any_finite and restored.valid is False


def test_depth_does_not_swallow_the_force_sample_that_follows_it():
    """Force is appended after depth. The depth codec must stop at its own payload."""
    import numpy as np

    sample = ForceSample(
        normal=0.5,
        shear_x=0.25,
        shear_y=-0.125,
        normal_raw=80.0,
        shear_x_raw=10.0,
        shear_y_raw=-4.0,
        valid=True,
        calibrated=False,
    )
    sparse = np.zeros((16, 16), dtype=np.float32)
    sparse[2, 3] = 0.03
    noise = np.linspace(0.0, 1.0, 16 * 16, dtype=np.float32).reshape(16, 16)
    for depth in (sparse, noise, np.zeros((8, 8), dtype=np.float32)):
        decoded = decode_frame(
            encode_frame(
                frame_id=3,
                t_rel_s=0.1,
                grid_id=0,
                dx=[1.0],
                dy=[0.0],
                magnitude=[1.0],
                active=[True],
                depth=depth,
                force=sample,
            )
        )
        assert decoded.force == sample
        assert decoded.depth is not None
        restored = decoded.depth_map()
        assert restored is not None and restored.shape == depth.shape