ykcheungab
1102 lines
48 KiB
Python
1102 lines
48 KiB
Python
"""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_depth_display_threshold_uses_base_colour_without_changing_source():
|
|
depth = np.array([[0.1, 0.4, 0.8]], dtype=np.float32)
|
|
original = depth.copy()
|
|
frame = backend.ImageFrame(_image(1, 3), depth=depth)
|
|
plain = backend.render_image(frame, mode="depth", depth_range=(0.0, 1.0))
|
|
tuned = backend.render_image(
|
|
frame, mode="depth", depth_range=(0.0, 1.0), depth_threshold=0.5
|
|
)
|
|
assert plain is not None and tuned is not None
|
|
assert np.array_equal(tuned[0, 0], tuned[0, 1])
|
|
assert np.array_equal(tuned[0, 2], plain[0, 2])
|
|
assert np.array_equal(depth, original)
|
|
|
|
|
|
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
|
|
|
|
|
|
@pytest.mark.parametrize("draw_style", ["box", "mini"])
|
|
def test_box_family_keeps_sparse_idle_dots_and_distinct_motion_colours(draw_style):
|
|
from orisys_app_shared.acquisition.arrows import FLOW_IDLE_HEX, hex_to_bgr, sdk_box_policy
|
|
|
|
policy = sdk_box_policy()
|
|
base = np.zeros((100, 120, 3), np.uint8)
|
|
flow = np.zeros((100, 120, 2), np.float32)
|
|
flow[50, 50, 0] = 1.25 # 2.5 px displacement, just over the box floor.
|
|
flow[50, 90, 0] = 3.0 # 6 px displacement on the same frame.
|
|
drawn = draw_arrow_display(
|
|
base, flow, draw_style=draw_style, policy=policy,
|
|
colormap=flow_colormap("dark"),
|
|
)
|
|
tint = np.array(hex_to_bgr(FLOW_IDLE_HEX), np.uint8)
|
|
assert _marker_is_there(drawn, 30, 30, tint)
|
|
assert not _marker_is_there(drawn, 50, 50, tint)
|
|
assert not np.array_equal(drawn[50, 50], drawn[50, 90])
|