orisys-app/shared/tests/test_image_backend.py

"""Acceptance for the single 2D image backend.

The point of this module is that live, recomputed, replayed and previewed frames go
through *one* renderer and that the view modes come from what the frame carries. These
tests pin both halves: a plan derived from fields (not from a view preset), and
pixel-for-pixel agreement with the primitives the application used before.
"""

from __future__ import annotations

from pathlib import Path

import numpy as np
import pytest
from orisys_app_shared.acquisition import image_backend as backend
from orisys_app_shared.acquisition.arrows import (
    depth_map_to_bgr,
    draw_arrow_display,
    flow_colormap,
)


def _image(h: int = 12, w: int = 16) -> np.ndarray:
    return np.arange(h * w * 3, dtype=np.uint8).reshape(h, w, 3)


def _flow(h: int = 12, w: int = 16) -> np.ndarray:
    flow = np.zeros((h, w, 2), dtype=np.float32)
    flow[..., 0] = 3.0
    flow[..., 1] = -1.5
    return flow


# ------------------------------------------------------------------------- plans
def test_a_live_frame_offers_the_derived_modes_and_needs_no_note():
    plan = backend.plan_image(
        backend.ImageFrame(_image(), flow=_flow(), depth=np.ones((12, 16)))
    )
    assert plan.available_modes == ("arrows_pure", "arrows_stitched", "depth", "depth_arrows")
    assert plan.default_mode == "arrows_pure"
    assert plan.missing == ()
    assert plan.note_key == ""


def test_depth_arrows_draws_the_flow_on_the_depth_picture():
    depth = np.linspace(0.0, 1.0, 12 * 16, dtype=np.float32).reshape(12, 16)
    frame = backend.ImageFrame(_image(), flow=_flow(), depth=depth)
    plain = backend.render_image(frame, mode="depth", draw_style="finger")
    over = backend.render_image(frame, mode="depth_arrows", draw_style="finger")
    assert plain is not None and over is not None
    assert over.shape == plain.shape
    assert not np.array_equal(over, plain)


def test_a_recorded_frame_is_raw_and_says_so():
    """No flow and no depth: the frame *is* the picture, and the note explains it."""
    plan = backend.plan_image(backend.ImageFrame(_image(), source=backend.SOURCE_REPLAY))
    assert plan.available_modes == ("recorded",)
    assert plan.default_mode == "recorded"
    assert plan.missing == ("flow", "depth")
    assert plan.note_key == backend.RAW_NOTE_KEY
    assert plan.source == "replay"


def test_a_derived_source_keeps_every_mode_and_reports_what_is_missing():
    """A dropped frame must not silently switch the operator's view.

    Live and recomputed frames may be missing a field for one frame; the mode stays
    selectable and ``missing`` says which field is absent, so the panel can state it
    instead of the view looking broken.
    """
    flow_only = backend.plan_image(backend.ImageFrame(_image(), flow=_flow()))
    assert flow_only.available_modes == backend.DERIVED_MODES
    assert flow_only.missing == ("depth",)
    assert flow_only.default_mode == "arrows_pure"
    assert flow_only.note_key == backend.MISSING_NOTE_KEY
    assert flow_only.note_fields == ("depth",)
    assert flow_only.needs("depth") == "depth"
    assert flow_only.needs("arrows_pure") is None

    depth_only = backend.plan_image(backend.ImageFrame(_image(), depth=np.ones((12, 16))))
    assert depth_only.available_modes == backend.DERIVED_MODES
    assert depth_only.missing == ("flow",)
    assert depth_only.default_mode == "depth"
    assert depth_only.needs("arrows_pure") == "flow"

    complete = backend.plan_image(
        backend.ImageFrame(_image(), flow=_flow(), depth=np.ones((12, 16)))
    )
    assert complete.missing == ()
    assert complete.note_key == ""


def test_a_raw_source_cannot_offer_a_derived_view_at_all():
    """A recording and a camera preview have no derived field, ever."""
    for source in (backend.SOURCE_REPLAY, backend.SOURCE_PREVIEW):
        plan = backend.plan_image(backend.ImageFrame(_image(), source=source))
        assert plan.available_modes == ("recorded",)
        assert plan.note_key == backend.RAW_NOTE_KEY
        assert plan.note_fields == ()


def test_the_plan_carries_the_frame_identity_for_the_panel():
    plan = backend.plan_image(
        backend.ImageFrame(
            _image(), source=backend.SOURCE_RECOMPUTE, frame_id=42, t_rel_s=1.25, stale=True
        )
    )
    assert (plan.source, plan.frame_id, plan.t_rel_s, plan.stale) == ("recompute", 42, 1.25, True)


def test_a_mode_the_plan_does_not_offer_falls_back_to_the_default():
    """Never draw a view the source cannot have at all."""
    recorded = backend.plan_image(
        backend.ImageFrame(_image(), source=backend.SOURCE_REPLAY)
    )
    assert recorded.resolve("arrows_pure") == "recorded"
    assert recorded.resolve("nonsense") == "recorded"
    live = backend.plan_image(backend.ImageFrame(_image(), flow=_flow()))
    assert live.resolve("recorded") == "arrows_pure", "a derived source has no raw mode"
    assert live.resolve("arrows_stitched") == "arrows_stitched"


# --------------------------------------------------------------- one renderer
def test_the_2d_canvas_draws_from_the_policy_it_is_given():
    """The threshold is the caller's — the pane's — not a per-view default.

    ``arrow_policy`` is the policy the 3D surface samples its overlay with
    (``visual_policy.policy_for_settings`` over the display settings). A sample inside
    the hysteresis band is the case that used to differ between the two figures, so it
    is the case pinned here: with a 0.4 px threshold no arrow is drawn, with 0.3 px the
    same frame carries one.

    "No arrow" is asserted as *no colour off the flow ramp*, not as "one colour on the
    canvas": the below-threshold sample now carries its idle marker
    (:data:`~orisys_app_shared.acquisition.arrows.DRAW_IDLE_GRID_DOTS`), which is off the
    ramp by design, so the strict frame holds exactly the canvas tone and the idle tint.
    """
    from orisys_app_shared.acquisition.arrows import (
        ARROW_BG_BGR,
        FLOW_IDLE_HEX,
        hex_to_bgr,
    )
    from orisys_app_shared.scene3d.contracts import DisplaySettings
    from orisys_app_shared.scene3d.visual_policy import policy_for_settings

    h, w = 12, 16
    flow = np.zeros((h, w, 2), dtype=np.float32)
    flow[..., 0] = 0.175  # |2*flow| = 0.35 px: above flow_off_px, below flow_on_px
    frame = backend.ImageFrame(_image(h, w), flow=flow, flow_valid=True)
    base = np.zeros((h, w, 3), dtype=np.uint8)

    def tones(image: np.ndarray) -> set[tuple[int, ...]]:
        return {
            tuple(int(v) for v in tone)
            for tone in np.unique(image.reshape(-1, 3), axis=0)
        }

    canvas_tone = tuple(int(v) for v in ARROW_BG_BGR)
    tint = tuple(int(v) for v in hex_to_bgr(FLOW_IDLE_HEX))

    strict = policy_for_settings(DisplaySettings(flow_on_px=0.4, flow_off_px=0.3))
    loose = policy_for_settings(DisplaySettings(flow_on_px=0.3, flow_off_px=0.2))
    strict_image = backend.render_image(
        frame, mode="arrows_pure", draw_style="finger", arrow_policy=strict
    )
    loose_image = backend.render_image(
        frame, mode="arrows_pure", draw_style="finger", arrow_policy=loose
    )
    assert tones(strict_image) == {canvas_tone, tint}, (
        f"a 0.4 px threshold drew an arrow for a 0.35 px sample: {tones(strict_image)}"
    )
    assert tint not in tones(loose_image), (
        "the sample cleared the threshold and still carries an idle marker"
    )
    assert tones(loose_image) - {canvas_tone}, (
        f"a 0.3 px threshold drew nothing for a 0.35 px sample: {tones(loose_image)}"
    )
    # The same policy governs the primitive directly, which is what the window calls
    # through the backend.
    direct = draw_arrow_display(base, flow, draw_style="finger", policy=strict)
    assert np.array_equal(direct, strict_image)


def test_the_2d_canvas_colours_one_magnitude_the_same_in_every_frame():
    """The colour range is fixed, so two frames are comparable by colour.

    The canvas used to normalise each frame against its own maximum: the same magnitude
    was one colour in a frame with a fast sample and another in a frame without one, so
    two frames could look alike while showing different motion. ``magnitude_range``
    defaults to :data:`arrows.FLOW_MAGNITUDE_RANGE`, and the old re-normalisation is
    available only when a caller asks for it.
    """
    from orisys_app_shared.acquisition.arrows import magnitude_to_ramp_index

    # One probe magnitude in two frames whose maxima differ by a factor of four.
    weak = magnitude_to_ramp_index(np.array([0.5, 1.0]))
    strong = magnitude_to_ramp_index(np.array([0.5, 4.0]))
    assert weak[0] == strong[0], (
        "the same 0.5 px magnitude took two different ramp positions because the frame's "
        f"maximum changed: {weak[0]} against {strong[0]}"
    )
    # The range is the documented default, not one frame's own span.
    assert magnitude_to_ramp_index(np.array([0.0, 2.0])).tolist() == [0, 255]
    # The per-frame mode is still available, and is what made the colours drift.
    assert magnitude_to_ramp_index(np.array([0.5, 4.0]), per_frame=True)[0] != weak[0]
    # A caller may pin the range it wants instead: 1.0 of a 0..4.0 range is a quarter up.
    assert magnitude_to_ramp_index(np.array([1.0]), value_range=(0.0, 4.0))[0] == 63


def test_the_2d_canvas_draws_the_same_magnitude_in_the_same_colour():
    """The painted canvas obeys the fixed range through :func:`draw_arrow_display`.

    The helper's contract is one thing; this is the pixel an operator looks at. Two
    frames carry the same magnitude at the same sample and different maxima, and the
    sample's pixel is the same colour in both.
    """
    from orisys_app_shared.acquisition.arrows import ARROW_BG_BGR, flow_colormap
    from orisys_app_shared.scene3d.visual_policy import VisualPolicy

    h, w = 12, 12
    policy = VisualPolicy(flow_on_px=0.4, flow_grid_spacing=1, flow_edge_margin=0)

    def canvas(probe: float, strongest: float) -> np.ndarray:
        flow = np.zeros((h, w, 2), dtype=np.float64)
        flow[8, 8] = (probe, 0.0)
        flow[2, 8] = (strongest, 0.0)
        return draw_arrow_display(
            np.zeros((h, w, 3), dtype=np.uint8),
            flow,
            draw_style="finger",
            policy=policy,
            colormap=flow_colormap("dark"),
        )

    def sample_colour(drawn: np.ndarray, row: int, col: int) -> tuple[int, ...]:
        """The colour this sample was drawn in.

        The canvas is one flat colour and this sample is drawn at the centre of the
        patch, so any pixel that is not the canvas is the drawn sample.
        """
        margin = 2
        patch = drawn[row - margin : row + margin + 1, col - margin : col + margin + 1]
        background = np.array(ARROW_BG_BGR, dtype=np.uint8)
        drawn_pixels = patch[np.any(patch != background, axis=-1)]
        assert drawn_pixels.size, f"nothing was drawn at {(row, col)}"
        return tuple(int(v) for v in drawn_pixels[0])

    quiet = canvas(0.25, 0.5)  # probe |2*flow| = 0.5 px, frame maximum 1.0 px
    loud = canvas(0.25, 4.0)  # same probe, frame maximum 8.0 px
    assert sample_colour(quiet, 8, 8) == sample_colour(loud, 8, 8), (
        "one magnitude took two colours because another sample moved faster"
    )


def test_a_failed_draw_is_not_an_idle_frame():
    """The base image comes back with a code on it, because the pixels cannot say so.

    A render failure and a frame with nothing to draw produced the same picture, so
    "the arrows are missing" was indistinguishable from "the device measured nothing".
    The code travels on the returned image; the caller surfaces it (this backend's own
    panel wiring is not part of this change).
    """
    from orisys_app_shared.acquisition.arrows import ARROW_RENDER_FAILED

    base = np.zeros((12, 12, 3), dtype=np.uint8)
    base[:, :] = (0x20, 0x30, 0x40)
    # ``select_flow_samples`` rejects a field that is not [H,W,2], which is a draw
    # failure rather than an empty measurement.
    failed = draw_arrow_display(base, np.zeros((12, 12), dtype=np.float32), draw_style="finger")
    assert np.array_equal(failed, base), "a failed draw must still hand back the image"
    assert failed.frame_diagnostic is not None
    assert failed.frame_diagnostic.code == ARROW_RENDER_FAILED
    assert failed.frame_diagnostic.severity == "warning"
    # The exception's own message is not carried: diagnostics travel to the UI and logs.
    assert "Traceback" not in failed.frame_diagnostic.message

    # A frame with no flow at all is a legitimate empty frame, not a failure.
    idle = draw_arrow_display(base, None, draw_style="finger")
    assert idle.frame_diagnostic is None
    assert np.array_equal(idle, base)

    # A drawn frame says nothing, and stays an ordinary array for every other caller.
    drawn = draw_arrow_display(
        base,
        np.full((12, 12, 2), 0.5, dtype=np.float32),
        draw_style="finger",
    )
    assert drawn.frame_diagnostic is None
    assert isinstance(drawn, np.ndarray)
    assert drawn.shape == base.shape and drawn.dtype == np.uint8


def test_the_backend_renders_what_the_primitives_rendered_before():
    """Live went through these two calls; the backend must not change a pixel."""
    image = _image()
    flow = _flow()
    mask = np.ones((12, 16), dtype=bool)
    mask[0, 0] = False

    frame = backend.ImageFrame(image, flow=flow, valid_mask=mask)
    plan = backend.plan_image(frame)
    for mode in ("arrows_pure", "arrows_stitched"):
        expected = draw_arrow_display(
            image,
            flow,
            background="captured" if mode == "arrows_stitched" else "pure_color",
            draw_style="finger",
            valid_mask=mask,
            colormap=flow_colormap("dark"),
        )
        got = backend.render_image(frame, plan, mode=mode, draw_style="finger")
        assert np.array_equal(got, expected), mode

    depth = np.linspace(0.0, 1.0, 12 * 16, dtype=np.float32).reshape(12, 16)
    depth_frame = backend.ImageFrame(image, depth=depth, valid_mask=mask)
    # The depth picture uses the *theme's* ramp — the same table the panel's colour bar
    # paints. The window used to leave this argument out, so the map was drawn with
    # OpenCV's JET while the legend under it showed the brand ramp: two colour tables
    # for one picture, and the legend claiming they were the same.
    expected_depth = depth_map_to_bgr(
        depth, flow_colormap("dark"), value_range=(0.0, 1.0), invalid_mask=mask
    )
    got_depth = backend.render_image(
        depth_frame, mode="depth", depth_range=(0.0, 1.0)
    )
    assert np.array_equal(got_depth, expected_depth)


def test_a_recorded_frame_is_passed_through_and_never_resized_to_fit():
    image = _image(9, 7)
    got = backend.render_image(
        backend.ImageFrame(image, source=backend.SOURCE_REPLAY), mode="recorded"
    )
    assert got is not None and got.shape == image.shape
    assert np.array_equal(got, image)


def test_nothing_to_draw_is_none_not_a_blank_picture():
    assert backend.render_image(backend.ImageFrame(), mode="recorded") is None
    assert backend.render_image(
        backend.ImageFrame(None, flow=_flow()), mode="arrows_pure"
    ) is None
    assert backend.render_image(backend.ImageFrame(None, depth=None), mode="depth") is None


def test_a_grey_base_frame_is_widened_once_by_the_backend():
    """A 2-D frame must reach the shader as BGR, and only one place may do it."""
    grey = np.zeros((8, 6), dtype=np.uint8)
    got = backend.render_image(backend.ImageFrame(grey), mode="recorded")
    assert got is not None and got.shape == (8, 6, 3)
    assert np.array_equal(got, np.stack([grey] * 3, axis=-1))


@pytest.mark.parametrize("mode", list(backend.ALL_MODES))
def test_every_offered_mode_renders_something_or_says_it_cannot(mode):
    """A mode the plan offers must never come back empty."""
    frame = backend.ImageFrame(_image(), flow=_flow(), depth=np.ones((12, 16)))
    plan = backend.plan_image(frame)
    if not plan.accepts(mode):
        assert plan.resolve(mode) in plan.available_modes
        return
    assert backend.render_image(frame, plan, mode=mode) is not None


def test_a_missing_field_is_reported_not_invented():
    """Depth with no depth says so; arrows with no flow show the base picture."""
    no_depth = backend.ImageFrame(_image(), flow=_flow())
    assert backend.render_image(no_depth, mode="depth") is None
    no_flow = backend.ImageFrame(_image(), depth=np.ones((12, 16)))
    base = backend.render_image(no_flow, mode="arrows_pure")
    assert base is not None
    assert np.array_equal(base, _image()), "the base picture must not be redrawn"


# ------------------------------------------------------------ the idle markers
def _flat_canvas(h: int, w: int) -> np.ndarray:
    """A base picture the markers are the only thing drawn on."""
    image = np.zeros((h, w, 3), dtype=np.uint8)
    image[:, :] = (0x16, 0x13, 0x12)
    return image


def _candidates(flow: np.ndarray, policy, mask: np.ndarray | None = None):
    from orisys_app_shared.scene3d.visual_policy import select_flow_samples

    samples = select_flow_samples(flow, mask, policy)
    return samples, np.asarray(samples.active, dtype=bool)


def _marker_is_there(image: np.ndarray, row: int, col: int, tint: np.ndarray) -> bool:
    """True when the radius-1 marker's own pixels are all at ``(row, col)``.

    The marker is the blob ``cv2.circle(radius=1, thickness=-1)`` fills — the centre and
    its four neighbours — so this asserts both that a marker was drawn *and* that it was
    centred on the candidate location rather than smeared over it.
    """
    return bool(
        np.all(image[row - 1 : row + 2, col] == tint)
        and np.all(image[row, col - 1 : col + 2] == tint)
    )


def test_every_sample_below_the_threshold_is_marked_once_and_only_once():
    """One marker per candidate: an arrow where motion was measured, a dot where it was not.

    ``select_flow_samples`` hands the painter every candidate location with an ``active``
    flag, and the painter's choice is the whole of the display rule: an active sample is an
    arrow, an inactive one is a dot, and no sample carries both. The dot count is pinned
    pixel-exactly (5 px per marker), which is what "one marker per sample" means on the
    canvas rather than in the selector.
    """
    from orisys_app_shared.acquisition.arrows import FLOW_IDLE_HEX, hex_to_bgr
    from orisys_app_shared.scene3d.visual_policy import VisualPolicy

    h, w = 20, 24
    policy = VisualPolicy(
        flow_on_px=0.5, flow_off_px=0.4, flow_grid_spacing=4, flow_edge_margin=0
    )
    flow = np.zeros((h, w, 2), dtype=np.float64)
    flow[10, 10] = (0.5, 0.0)  # |2*flow| = 1.0 px: over the threshold
    flow[6, 6] = (0.1, 0.0)  # 0.2 px: under it
    canvas = _flat_canvas(h, w)
    tint = np.array(hex_to_bgr(FLOW_IDLE_HEX), dtype=np.uint8)

    drawn = draw_arrow_display(canvas, flow, background="captured", draw_style="finger", policy=policy)
    samples, active = _candidates(flow, policy)
    assert samples.rows.size == 30 and int(active.sum()) == 1, (
        f"the fixture is not one arrow among dots: {samples.rows.size} candidates, "
        f"{int(active.sum())} active"
    )

    idle_rows = samples.rows[~active]
    idle_cols = samples.cols[~active]
    active_rows = samples.rows[active]
    active_cols = samples.cols[active]
    for row, col in zip(idle_rows.tolist(), idle_cols.tolist()):
        assert _marker_is_there(drawn, row, col, tint), (
            f"the below-threshold sample at {(row, col)} has no idle marker"
        )
    for row, col in zip(active_rows.tolist(), active_cols.tolist()):
        assert not _marker_is_there(drawn, row, col, tint), (
            f"the sample at {(row, col)} cleared the threshold and still got a dot"
        )
        assert not np.all(drawn[row, col] == tint), "the arrow's own origin is a dot"
        assert not np.all(drawn[row, col] == canvas[row, col]), "no arrow was drawn"
    assert int(np.all(drawn == tint, axis=-1).sum()) == 5 * int((~active).sum()), (
        "the markers do not cover exactly one dot per below-threshold sample"
    )
    # The dots are off the flow ramp: a colour on the canvas is a measurement, and a
    # marker that repeated one would read as a small displacement.
    assert not np.any(np.all(tint == flow_colormap("dark"), axis=-1))


def test_a_dot_is_only_where_the_device_vouches_for_the_sample():
    """Inside the sensing region, and nowhere else.

    Two ways a candidate location fails the selector and must get no marker: the mask is
    false at the sample, or the arrow's own tip would land outside it. Both are the
    selector's rules (``valid &= mask[rr, cc]`` and ``valid &= mask[end_y, end_x]``), and
    the dots must obey them rather than the grid they are drawn on.
    """
    from orisys_app_shared.acquisition.arrows import FLOW_IDLE_HEX, hex_to_bgr
    from orisys_app_shared.scene3d.visual_policy import VisualPolicy

    h, w = 20, 24
    policy = VisualPolicy(
        flow_on_px=0.5, flow_off_px=0.4, flow_grid_spacing=4, flow_edge_margin=0
    )
    mask = np.ones((h, w), dtype=bool)
    mask[:3, :] = False  # a band the device does not see at all
    mask[10, 15:] = False  # where the (10, 14) sample's tip would land
    flow = np.zeros((h, w, 2), dtype=np.float64)
    flow[10, 14] = (1.0, 0.0)  # 2.0 px of displacement: tip 1.6 px to the right
    canvas = _flat_canvas(h, w)
    tint = np.array(hex_to_bgr(FLOW_IDLE_HEX), dtype=np.uint8)

    drawn = draw_arrow_display(
        canvas, flow, background="captured", draw_style="finger", policy=policy, valid_mask=mask
    )
    samples, active = _candidates(flow, policy, mask)
    marked = np.all(drawn == tint, axis=-1)
    assert not marked[:3, :].any(), (
        f"a dot was drawn outside the sensing region: {int(marked[:3, :].sum())} px in the band"
    )
    # The tip rule: (10, 14) is a grid location and its sample is inside the mask, but its
    # arrow would end outside it — so it is not a candidate at all, and carries no dot.
    assert (10, 14) not in set(zip(samples.rows.tolist(), samples.cols.tolist()))
    assert not np.any(drawn[8:13, 12:17] != canvas[8:13, 12:17]), (
        "something was drawn at a location whose arrow tip leaves the sensing region"
    )
    for row, col in zip(samples.rows[~active].tolist(), samples.cols[~active].tolist()):
        assert mask[row, col], f"a dot was drawn at {(row, col)}, which the mask rejects"
    assert int(np.all(drawn == tint, axis=-1).sum()) == 5 * int((~active).sum())


def test_an_empty_frame_and_a_failed_draw_carry_no_idle_dots():
    """No flow is a legitimate empty frame; a failed draw is not an idle one.

    Neither may put a grid on the canvas: the dots are a statement about samples the
    device measured and found quiet, and there are no samples in either case.
    """
    from orisys_app_shared.acquisition.arrows import (
        ARROW_RENDER_FAILED,
        FLOW_IDLE_HEX,
        hex_to_bgr,
    )

    base = _flat_canvas(12, 12)
    tint = np.array(hex_to_bgr(FLOW_IDLE_HEX), dtype=np.uint8)

    empty = draw_arrow_display(base, None, draw_style="finger")
    assert empty.frame_diagnostic is None
    assert np.array_equal(empty, base)
    assert not np.any(np.all(empty == tint, axis=-1))

    failed = draw_arrow_display(base, np.zeros((12, 12), dtype=np.float32), draw_style="finger")
    assert failed.frame_diagnostic is not None
    assert failed.frame_diagnostic.code == ARROW_RENDER_FAILED
    assert np.array_equal(failed, base), "a failed draw must still hand back the base image"
    assert not np.any(np.all(failed == tint, axis=-1)), "a failed draw left markers behind"


def test_the_idle_dot_switch_hides_the_dots_and_leaves_the_arrows_alone():
    """Hiding them changes the dots and nothing else.

    The switch is the painter's, so it cannot reach the arrows: every pixel that is not an
    idle marker is byte-identical between the two draws, the result keeps its shape, dtype
    and ``frame_diagnostic``, and the switch works against the module default as well as
    with it.
    """
    from orisys_app_shared.acquisition import arrows as arrow_mod
    from orisys_app_shared.acquisition.arrows import (
        DRAW_IDLE_GRID_DOTS,
        FLOW_IDLE_HEX,
        hex_to_bgr,
    )
    from orisys_app_shared.scene3d.visual_policy import VisualPolicy

    h, w = 20, 24
    policy = VisualPolicy(
        flow_on_px=0.5, flow_off_px=0.4, flow_grid_spacing=4, flow_edge_margin=0
    )
    flow = np.zeros((h, w, 2), dtype=np.float64)
    flow[10, 10] = (0.5, 0.0)
    flow[6, 6] = (0.1, 0.0)
    canvas = _flat_canvas(h, w)
    tint = np.array(hex_to_bgr(FLOW_IDLE_HEX), dtype=np.uint8)

    assert DRAW_IDLE_GRID_DOTS is True, "the decided display behaviour is that dots are shown"
    drawn = draw_arrow_display(canvas, flow, background="captured", draw_style="finger", policy=policy)
    hidden = draw_arrow_display(
        canvas, flow, background="captured", draw_style="finger", policy=policy, idle_dots=False
    )
    marked = np.all(drawn == tint, axis=-1)
    assert marked.sum() > 0, "the default draw has no idle markers to hide"
    assert not np.any(np.all(hidden == tint, axis=-1)), "the switch did not hide the dots"
    assert np.array_equal(drawn[~marked], hidden[~marked]), "hiding the dots moved the arrows"
    assert (hidden.shape, hidden.dtype) == (drawn.shape, drawn.dtype)
    assert hidden.frame_diagnostic is None and drawn.frame_diagnostic is None

    # A caller may also switch them on when the build default is off, and the default
    # itself is read at call time rather than frozen into the parameter.
    original = arrow_mod.DRAW_IDLE_GRID_DOTS
    arrow_mod.DRAW_IDLE_GRID_DOTS = False
    try:
        off_by_default = draw_arrow_display(
            canvas, flow, background="captured", draw_style="finger", policy=policy
        )
        assert np.array_equal(off_by_default, hidden)
        forced_on = draw_arrow_display(
            canvas, flow, background="captured", draw_style="finger", policy=policy, idle_dots=True
        )
        assert np.array_equal(forced_on, drawn)
    finally:
        arrow_mod.DRAW_IDLE_GRID_DOTS = original

    # The arrows themselves still come from the shared selection, which the switch does
    # not touch: same locations, same magnitudes, same verdicts.
    samples, active = _candidates(flow, policy)
    assert int(active.sum()) == 1 and samples.rows.size == 30


def test_the_idle_marker_is_the_marker_a_box_sample_gets():
    """One marker shape and one tint across sensor styles.

    Box and finger both use the shared per-sample painter. With the same policy they
    must put the same plus-shaped idle blob on the same quiet sample.
    """
    from orisys_app_shared.acquisition.arrows import FLOW_IDLE_HEX, hex_to_bgr
    from orisys_app_shared.scene3d.visual_policy import VisualPolicy

    # One candidate location on a 20 px grid with a 15 px edge: a 60x60 frame has
    # exactly the (30, 30) sample inside the margin.
    size = 60
    flow = np.zeros((size, size, 2), dtype=np.float64)
    flow[30, 30] = (0.1, 0.0)  # 0.2 px: under the threshold
    canvas = _flat_canvas(size, size)
    policy = VisualPolicy(flow_on_px=0.4, flow_grid_spacing=20, flow_edge_margin=15)

    box = draw_arrow_display(
        canvas, flow, background="captured", draw_style="box", policy=policy
    )
    finger = draw_arrow_display(
        canvas, flow, background="captured", draw_style="finger", policy=policy
    )
    tint = np.array(hex_to_bgr(FLOW_IDLE_HEX), dtype=np.uint8)
    box_marked = np.all(box == tint, axis=-1)
    assert box_marked.sum() == 5, f"the box path drew {int(box_marked.sum())} marker px"
    assert np.array_equal(box, finger), (
        "the finger's idle marker is not the marker the box path draws: "
        f"{(box != finger).any(axis=-1).sum()} px differ"
    )
    assert _marker_is_there(finger, 30, 30, tint)


def test_raising_the_threshold_turns_arrows_into_dots_without_moving_a_marker():
    """The sweep: the marker set is the policy's grid, the threshold only splits it.

    Every sample is a candidate location whatever the threshold is, so raising it moves
    samples from the arrow side to the dot side one at a time and lowers it moves them
    back — the total never changes, and neither does any location. This is the property
    that makes the two markers comparable at a glance: the same grid is on the canvas in
    both cases, and only the colour of the verdict differs.
    """
    from orisys_app_shared.acquisition.arrows import FLOW_IDLE_HEX, hex_to_bgr
    from orisys_app_shared.scene3d.visual_policy import VisualPolicy

    h, w = 40, 40
    flow = np.zeros((h, w, 2), dtype=np.float64)
    # One moving sample per step, on the 4 px grid (2 + 4k) the policy samples at, with
    # displacements of 0.1, 0.2, 0.4, 0.7, 1.0 and 1.5 px. Every threshold below sits
    # between two of them, so each step of the sweep moves exactly one sample across.
    for index, value in enumerate((0.05, 0.1, 0.2, 0.35, 0.5, 0.75)):
        row = col = 6 + 4 * index
        flow[row, col] = (value, 0.0)
    canvas = _flat_canvas(h, w)
    tint = np.array(hex_to_bgr(FLOW_IDLE_HEX), dtype=np.uint8)

    counts: list[tuple[float, int, int, int]] = []
    for on_px in (0.2, 0.4, 0.6, 0.9, 1.2):
        policy = VisualPolicy(
            flow_on_px=on_px, flow_off_px=max(0.0, on_px - 0.1), flow_grid_spacing=4,
            flow_edge_margin=0,
        ).normalised()
        samples, active = _candidates(flow, policy)
        drawn = draw_arrow_display(
            canvas, flow, background="captured", draw_style="finger", policy=policy
        )
        dots = int((~active).sum())
        counts.append((on_px, int(samples.rows.size), int(active.sum()), dots))
        assert int(np.all(drawn == tint, axis=-1).sum()) == 5 * dots, (
            f"the canvas drew other than one dot per idle sample at {on_px} px"
        )
        for row, col in zip(samples.rows[active].tolist(), samples.cols[active].tolist()):
            assert not np.all(drawn[row, col] == tint), (
                f"an arrow at {(row, col)} also carried a dot at threshold {on_px}"
            )

    markers = {row[1] for row in counts}
    assert len(markers) == 1, f"the marker set changed with the threshold: {counts}"
    arrows = [row[2] for row in counts]
    assert arrows == sorted(arrows, reverse=True), f"raising the threshold added arrows: {counts}"
    assert len(set(arrows)) == len(arrows), f"a step of the sweep moved nothing: {counts}"
    for _on_px, markers_at, arrows_at, dots_at in counts:
        assert markers_at == arrows_at + dots_at, "a sample carried no marker at all"


def test_hiding_the_idle_dots_leaves_the_3d_figure_where_it_was():
    """The switch is the 2D painter's, and the 3D overlay is not built by the painter.

    Both figures read one :func:`select_flow_samples` result and draw the samples it marks
    active (``composer.SceneComposer`` through ``active_only()``), so the canvas's dots
    cannot move a 3D arrow. Composed around both draws on the shipped finger island, the
    overlay is identical — which is the check that fails if the marker choice is ever moved
    into the shared selection, where it would reach the 3D figure too.
    """
    from orisys_app_shared.scene3d.composer import SceneComposer
    from orisys_app_shared.scene3d.contracts import (
        DisplaySettings,
        FieldDescriptor,
        FrameIdentity,
        SurfaceFieldFrame,
        freeze,
    )
    from orisys_app_shared.scene3d.package import read_package, scene_from_contents
    from orisys_app_shared.scene3d.visual_policy import policy_for_settings

    package = (
        Path(__file__).resolve().parents[2]
        / "apps/studio/config/sensors/finger/scene/finger-display-1.0.0.oryscene"
    )
    if not package.is_file():
        pytest.skip(f"the shipped finger package is missing: {package}")
    scene = scene_from_contents(read_package(package), None)
    mask = np.asarray(scene.surface_mapping.valid, dtype=bool)
    h, w = mask.shape
    yy, xx = np.mgrid[0:h, 0:w].astype(np.float64)
    cy, cx = h / 2.0, w / 2.0
    radius = np.hypot(yy - cy, xx - cx)
    speed = 1.1 * np.clip(1.0 - radius / float(radius.max()), 0.0, 1.0)
    flow = np.stack(
        [-speed * (yy - cy) / max(cy, 1.0), speed * (xx - cx) / max(cx, 1.0)], axis=-1
    )
    settings = DisplaySettings(
        flow_arrows_enabled=True, flow_on_px=0.4, flow_off_px=0.3, flow_grid_spacing=10
    )
    policy = policy_for_settings(settings)
    frame = SurfaceFieldFrame(
        identity=FrameIdentity("live", 1, 1, 0.1, generation=1),
        geometry_id=scene.geometry_id,
        field_domain_id=scene.field_domain_id,
        scalar=freeze(np.zeros((h, w), dtype=np.float32), dtype=np.float32),
        scalar_descriptor=FieldDescriptor("depth_response", "1", "approximate", False),
        valid=freeze(mask.copy(), dtype=np.bool_),
        flow_uv=freeze(flow.astype(np.float32), dtype=np.float32),
        flow_valid_mask=freeze(mask.copy(), dtype=np.bool_),
    )
    composer = SceneComposer()
    canvas = _flat_canvas(h, w)
    args = dict(background="captured", draw_style="finger", policy=policy, valid_mask=mask)
    dots_on = draw_arrow_display(canvas, flow, **args)
    overlay_on = composer.compose(scene, frame, settings).flow_arrows
    dots_off = draw_arrow_display(canvas, flow, idle_dots=False, **args)
    overlay_off = composer.compose(scene, frame, settings).flow_arrows

    samples, active = _candidates(flow, policy, mask)
    assert 0 < int(active.sum()) < samples.rows.size, (
        "the fixture must hold both markers: "
        f"{int(active.sum())} active of {samples.rows.size} candidates"
    )
    assert overlay_on is not None and overlay_off is not None
    assert int(np.count_nonzero(overlay_on.lengths > 0.0)) == int(active.sum()), (
        "the 3D overlay is not drawing the active samples the 2D canvas draws"
    )
    assert int(np.count_nonzero(overlay_on.lengths < 0.0)) > 0, (
        "inactive samples are not drawn as grey points"
    )
    assert np.array_equal(overlay_on.origins, overlay_off.origins)
    assert np.array_equal(overlay_on.directions, overlay_off.directions)
    assert not np.array_equal(dots_on, dots_off), "the switch changed nothing to hide"


# ------------------------------------------------------- the latched boundary
#: A domain the same size and the same kind of shape as the shipped finger sensor's
#: 300x280 UV island: one large blob with a long smooth rim, so "a one-pixel wobble"
#: and "the island moved" have the overlap the real package has (measured on the
#: shipped island: a 1 px wobble scores IoU 0.9633, a 6 px move 0.9102).
DOMAIN_HW = (300, 280)
ISLAND_RADIUS_PX = 95


def _island(h: int = DOMAIN_HW[0], w: int = DOMAIN_HW[1]) -> np.ndarray:
    yy, xx = np.mgrid[0:h, 0:w]
    return ((yy - h / 2.0) ** 2 + (xx - w / 2.0) ** 2) <= float(ISLAND_RADIUS_PX) ** 2


def _blank_image(h: int = DOMAIN_HW[0], w: int = DOMAIN_HW[1]) -> np.ndarray:
    """A base picture the two views do not read: only depth and the mask are drawn."""
    return np.zeros((h, w, 3), dtype=np.uint8)


def _shift(mask: np.ndarray, dy: int, dx: int) -> np.ndarray:
    out = np.zeros_like(mask)
    h, w = mask.shape
    ys = slice(max(0, dy), h + min(0, dy))
    xs = slice(max(0, dx), w + min(0, dx))
    ys2 = slice(max(0, -dy), h + min(0, -dy))
    xs2 = slice(max(0, -dx), w + min(0, -dx))
    out[ys, xs] = mask[ys2, xs2]
    return out


def test_overlap_of_two_identical_masks_is_one():
    mask = _island()
    assert backend.mask_overlap(mask, mask) == 1.0
    assert backend.mask_overlap(mask, ~mask) == 0.0
    # Two frames that agree there is no island at all are not a change.
    empty = np.zeros((4, 4), dtype=bool)
    assert backend.mask_overlap(empty, empty) == 1.0


def test_the_gate_separates_a_rebuilt_mask_from_a_moved_one():
    """The threshold itself, on the two cases it exists to tell apart."""
    island = _island()
    for dy, dx in ((1, 0), (0, 1), (1, 1), (-1, 1), (2, 0)):
        assert backend.mask_overlap(island, _shift(island, dy, dx)) >= (
            backend.DOMAIN_MASK_MIN_IOU
        ), f"a {dy},{dx} rebuild would be taken for a moved boundary"
    assert backend.mask_overlap(island, _shift(island, 6, 0)) < backend.DOMAIN_MASK_MIN_IOU


def test_a_stable_domain_mask_is_latched_once_for_the_session():
    """The device's island is geometry: rebuilt per frame, decided once."""
    latch = backend.DomainMaskLatch()
    mask = _island()
    seen = {latch.update(mask).tobytes() for _ in range(20)}
    assert len(seen) == 1, "a stable mask was rebuilt into a different array"
    assert np.array_equal(latch.mask, mask)
    assert (latch.adoptions, latch.held) == (1, 19)


def test_a_wobbling_domain_mask_does_not_repaint_the_boundary():
    """A one-pixel rim wobble is the mask being rebuilt, not the island moving.

    The measured case: on the shipped finger island this is a 0.73% area flip that
    repaints 69% of the boundary band and 35% of the rendered edge pixels.

    The canvas also carries the idle markers of this frame's own quiet samples, and those
    *are* a per-frame reading: the rim wobble moves the markers at the rim. What the latch
    is for is the boundary paint, and that is what is compared here.
    """
    latch = backend.DomainMaskLatch()
    image = _blank_image()
    # A zero field, so nothing but the boundary can change the picture.
    flow = np.zeros((*DOMAIN_HW, 2), dtype=np.float32)
    depth = np.ones(DOMAIN_HW, dtype=np.float32)
    wobbles = [(0, 0), (1, 0), (0, -1), (1, 1), (-1, 0), (0, 1), (-1, -1)]
    arrows, maps = [], []
    for dy, dx in wobbles:
        mask = _shift(_island(), dy, dx)
        frame = backend.ImageFrame(image, flow=flow, depth=depth, valid_mask=mask)
        used = latch.update(mask)
        arrows.append(
            backend.render_image(
                frame, mode="arrows_pure", draw_style="finger", domain_mask=used
            )
        )
        maps.append(
            backend.render_image(
                frame, mode="depth", depth_range=(0.0, 1.0), domain_mask=used
            )
        )
    assert all(np.array_equal(maps[0], other) for other in maps), (
        "the depth view's no-data boundary repainted a boundary that had not moved"
    )
    assert latch.adoptions == 1, "a rim wobble was taken for a moved boundary"

    # The arrow canvas draws one more thing from the frame's *own* mask: the idle markers
    # of samples the device vouched for and found quiet. They are this frame's samples —
    # not the latched boundary's — so a rim wobble moves the markers at the rim by design
    # (measured: 276 marker pixels on the un-wobbled island, 282 on the (1, 1) shift). The
    # *boundary paint* is what the latch exists for, and it must be byte-identical: every
    # pixel that differs between two frames has to be one of those two frames' own
    # markers. A repainted island or exterior tone is not a marker pixel.
    from orisys_app_shared.acquisition.arrows import FLOW_IDLE_HEX, hex_to_bgr

    tint = np.array(hex_to_bgr(FLOW_IDLE_HEX), dtype=np.uint8)
    for other in arrows:
        changed = np.any(arrows[0] != other, axis=-1)
        explained = np.all(arrows[0][changed] == tint, axis=-1) | np.all(
            other[changed] == tint, axis=-1
        )
        assert bool(np.all(explained)), (
            "the arrow canvas repainted a boundary that had not moved: "
            f"{int((~explained).sum())} px are neither frame's idle marker"
        )


def test_a_real_move_of_the_domain_is_followed():
    """The latch must not freeze a boundary that actually moved."""
    latch = backend.DomainMaskLatch()
    island = _island()
    latch.update(island)
    moved = _shift(island, 6, 0)
    used = latch.update(moved)
    assert np.array_equal(used, moved), "a moved island kept the old boundary"
    assert latch.adoptions == 2

    frame = backend.ImageFrame(
        _blank_image(), flow=np.zeros((*DOMAIN_HW, 2), np.float32)
    )
    before = backend.render_image(
        frame, mode="arrows_pure", draw_style="finger", domain_mask=island
    )
    after = backend.render_image(
        frame, mode="arrows_pure", draw_style="finger", domain_mask=used
    )
    assert not np.array_equal(before, after)


def test_a_frame_without_a_mask_cannot_move_the_boundary():
    latch = backend.DomainMaskLatch()
    assert latch.update(None) is None, "a session with no mask invented one"
    island = _island()
    latch.update(island)
    assert np.array_equal(latch.update(None), island), (
        "a frame that states no mask dropped the session's boundary"
    )
    assert (latch.adoptions, latch.held) == (1, 0)


def test_the_latched_boundary_never_gates_the_arrows():
    """Latching the boundary must not put an arrow on an unvouched-for sample."""
    image = _blank_image()
    flow = np.zeros((*DOMAIN_HW, 2), dtype=np.float32)
    flow[::10, ::10, 0] = 3.0
    # The frame claims nothing; the session's boundary is the whole domain.
    frame = backend.ImageFrame(image, flow=flow, valid_mask=np.zeros(DOMAIN_HW, bool))
    boundary = np.ones(DOMAIN_HW, dtype=bool)
    got = backend.render_image(
        frame, mode="arrows_pure", draw_style="finger", domain_mask=boundary
    )
    expected = draw_arrow_display(
        image,
        flow,
        background="pure_color",
        draw_style="finger",
        valid_mask=frame.valid_mask,
        domain_mask=boundary,
        colormap=flow_colormap("dark"),
    )
    assert np.array_equal(got, expected)
    # A whole-domain boundary paints one tone; an empty frame mask means no arrows,
    # so the picture stays that one tone rather than gaining arrow pixels.
    tones = {tuple(int(v) for v in tone) for tone in np.unique(got.reshape(-1, 3), axis=0)}
    assert tones == {(0x16, 0x13, 0x12)}, (
        f"arrows were drawn from a mask the frame does not claim: {tones}"
    )


def test_resetting_the_latch_clears_the_session_boundary():
    latch = backend.DomainMaskLatch()
    latch.update(_island())
    latch.reset()
    assert latch.mask is None
    assert (latch.adoptions, latch.held) == (0, 0)
    other = _shift(_island(), 5, 5)
    assert np.array_equal(latch.update(other), other), (
        "a new session kept the previous session's island"
    )


# --------------------------------------------- a field recovered from a stored result
def _policy():
    from orisys_app_shared.scene3d.visual_policy import VisualPolicy

    return VisualPolicy(flow_grid_spacing=10, max_arrows=400).normalised()


def _recovered_flow(height: int = 60, width: int = 80):
    """A flow field as an analysis result hands one over: finite only at its own cells.

    This is what the replay's reconstruction produces — the pass recorded candidate
    *locations* and their displacements, and the field is rebuilt with those values at
    those cells and ``NaN`` everywhere else. A dense field is the opposite case, and the
    difference between them is what the painter has to be chosen by.
    """
    from orisys_app_shared.scene3d.visual_policy import select_flow_samples

    dense = np.zeros((height, width, 2), dtype=np.float64)
    dense[..., 0] = 1.5
    dense[..., 1] = -0.75
    samples = select_flow_samples(dense, None, _policy())
    sparse = np.full((height, width, 2), np.nan, dtype=np.float64)
    sparse[samples.rows, samples.cols, 0] = samples.dx
    sparse[samples.rows, samples.cols, 1] = samples.dy
    return sparse, samples


def _painted(image, canvas=(0x16, 0x13, 0x12)) -> int:
    """Pixels the painter changed from the canvas colour."""
    array = np.asarray(image)
    return int(
        np.count_nonzero(np.any(array != np.asarray(canvas, np.uint8), axis=-1))
    )


def test_a_recovered_field_is_drawn_by_the_shared_painter_whatever_the_style_is():
    """Live and analysis sources use the same painter on a sparse recovered field.

    The field is finite only at the stored candidate cells. The shared selector walks
    those cells, so both sources paint the same picture under the same policy.
    """
    from orisys_app_shared.scene3d.visual_policy import select_flow_samples

    sparse, samples = _recovered_flow()
    base = np.zeros((60, 80, 3), dtype=np.uint8)
    base[:, :] = (0x16, 0x13, 0x12)
    policy = _policy()

    result_frame = backend.ImageFrame(
        base, flow=sparse, source=backend.SOURCE_ANALYSIS, flow_valid=True
    )
    live_frame = backend.ImageFrame(
        base, flow=sparse, source=backend.SOURCE_LIVE, flow_valid=True
    )
    drawn = backend.render_image(
        result_frame, mode=backend.ARROW_PURE, draw_style="box", arrow_policy=policy
    )
    expected = draw_arrow_display(
        base,
        sparse,
        background="pure_color",
        draw_style="box",
        # The backend hands the theme's ramp to the painter, so the reference call has to
        # as well; the default is ``cv2.COLORMAP_JET``, which is a different picture.
        colormap=flow_colormap("dark"),
        policy=policy,
    )
    assert np.array_equal(np.asarray(drawn), np.asarray(expected)), (
        "a result's canvas is not the shared painter's output"
    )
    assert _painted(drawn) > 0, "the recovered field painted nothing"

    live = backend.render_image(
        live_frame, mode=backend.ARROW_PURE, draw_style="box", arrow_policy=policy
    )
    assert np.array_equal(np.asarray(drawn), np.asarray(live)), (
        "live and analysis disagree on the same sparse field"
    )

    # One marker per recorded cell at most: the recovered field carries no other value, so
    # nothing outside those cells can become a sample.
    assert select_flow_samples(sparse, None, policy).rows.size == samples.rows.size


def test_a_box_result_on_the_sdk_grid_matches_the_live_painter():
    """A box pass stores the box grid; the idempotent inverse redraws the dense field."""
    from orisys_app_shared.acquisition.arrows import sdk_box_policy
    from orisys_app_shared.scene3d.visual_policy import (
        DISPLACEMENT_PX_PER_FLOW,
        select_flow_samples,
    )

    policy = sdk_box_policy()
    dense = np.zeros((80, 100, 2), dtype=np.float64)
    dense[20:60, 20:80, 0] = 1.5
    dense[30:40, 30:50, 0] = 8.0
    samples = select_flow_samples(dense, None, policy)
    assert samples.rows.size > 0
    dx = np.asarray(samples.dx, np.float64)
    dy = np.asarray(samples.dy, np.float64)
    scale = float(policy.flow_arrow_scale) or 1.0
    divisor = (
        DISPLACEMENT_PX_PER_FLOW * float(policy.flow_length_margin) / scale
    )
    sparse = np.full(dense.shape, np.nan, dtype=np.float64)
    sparse[samples.rows, samples.cols, 0] = dx / divisor
    sparse[samples.rows, samples.cols, 1] = dy / divisor
    base = np.zeros(dense.shape[:2] + (3,), np.uint8)
    base[:] = (0x16, 0x13, 0x12)
    cmap = flow_colormap("dark")
    live = draw_arrow_display(
        base,
        dense,
        draw_style="box",
        background="pure_color",
        canvas_bgr=(0x16, 0x13, 0x12),
        colormap=cmap,
        policy=policy,
    )
    frame = backend.ImageFrame(
        base,
        flow=sparse,
        source=backend.SOURCE_ANALYSIS,
        flow_valid=True,
        result_available=True,
    )
    drawn = backend.render_image(
        frame,
        mode=backend.ARROW_PURE,
        draw_style="box",
        arrow_policy=policy,
        canvas_bgr=(0x16, 0x13, 0x12),
    )
    assert np.array_equal(np.asarray(drawn), np.asarray(live))
    assert _painted(drawn) > 0