From aa526c62a9e531777ac7ddb628c263279339be41 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Sun, 4 Oct 2026 11:09:19 +0800 Subject: [PATCH] Mark every below-threshold finger sample with an idle dot The selector now returns candidate locations with an activity flag, so the quiet half of the finger's sampling grid can be drawn at last: one radius-1 dot per candidate, in the idle tint and off the flow ramp, on no sample that also carries an arrow. The dots are `DRAW_IDLE_GRID_DOTS` (on) and `draw_arrow_display(idle_dots=...)`; box and mini keep their own SDK path. --- .../tests/test_finger_2d_orientation.py | 14 +- .../orisys_app_shared/acquisition/arrows.py | 93 +++- shared/tests/test_image_backend.py | 433 +++++++++++++++++- shared/tests/test_visual_policy.py | 40 ++ 4 files changed, 548 insertions(+), 32 deletions(-) diff --git a/apps/studio/tests/test_finger_2d_orientation.py b/apps/studio/tests/test_finger_2d_orientation.py index 60c09fe..0dc7f1b 100644 --- a/apps/studio/tests/test_finger_2d_orientation.py +++ b/apps/studio/tests/test_finger_2d_orientation.py @@ -233,11 +233,17 @@ def _patch_flow(mask: np.ndarray, dx: float, dy: float): def _drawn_displacement(image: np.ndarray, origin: tuple[int, int]) -> tuple[float, float]: """The drawn arrow's displacement, in display pixels, from its own origin. - The origin is where the canvas would have put the sample; every non-background - pixel belongs to the one arrow drawn, and their centroid sits along it. The sign - is the whole assertion, but the magnitude is returned so a failure is readable. + The origin is where the canvas would have put the sample. The canvas is black apart + from what it drew, so the arrow's pixels are the non-background ones **minus the idle + markers**: every candidate below the threshold is marked with an idle dot, and + counting those would measure the marker grid instead of the arrow. The sign is the + whole assertion, but the magnitude is returned so a failure is readable. """ - pixels = np.argwhere(np.any(image != 0, axis=2)) + from orisys_app_shared.acquisition.arrows import FLOW_IDLE_HEX, hex_to_bgr + + idle_tint = np.asarray(hex_to_bgr(FLOW_IDLE_HEX), dtype=np.uint8) + drawn = np.any(image != 0, axis=2) & ~np.all(image == idle_tint, axis=2) + pixels = np.argwhere(drawn) assert pixels.size, "the canvas drew no arrow at all" centre = pixels.mean(axis=0) return float(centre[1] - origin[1]), float(centre[0] - origin[0]) diff --git a/shared/src/orisys_app_shared/acquisition/arrows.py b/shared/src/orisys_app_shared/acquisition/arrows.py index d1f3031..30994a4 100644 --- a/shared/src/orisys_app_shared/acquisition/arrows.py +++ b/shared/src/orisys_app_shared/acquisition/arrows.py @@ -35,9 +35,13 @@ NO_DEPTH_BGR = (0x8A, 0x8A, 0x8A) #: value and passes it as ``magnitude_range`` to :func:`draw_arrow_display`. FLOW_MAGNITUDE_RANGE: tuple[float, float] = (0.0, 2.0) -#: Whether an inactive sample is drawn as a quiet dot. Off: the live 2D canvas draws -#: arrows and nothing else, so splitting activity out of the selector changed no pixel. -DRAW_IDLE_GRID_DOTS = False +#: Whether an inactive sample is drawn as a quiet dot. **On**, which is the decided +#: display behaviour: a sample below the display threshold is a statement about the +#: sensing region ("sampled here, nothing to show"), and it is the one statement the +#: finger canvas could not make before — the selector had dropped the location, so an +#: idle grid could not be drawn at all. Off hides the dots again and changes nothing +#: else; :func:`draw_arrow_display` takes the same choice per call as ``idle_dots``. +DRAW_IDLE_GRID_DOTS = True #: Machine-readable code for "this frame could not be drawn": the caller gets the base #: image back and must surface the code instead of presenting it as an idle frame. The @@ -284,6 +288,33 @@ def _policy_from_style( ).normalised() +#: The pixels ``cv2.circle(..., radius=1, thickness=-1)`` fills, as ``(dy, dx)`` offsets +#: from the centre it is given. One marker shape for every sensor type: the SDK's +#: ``orisys.util.draw_arrows`` paints a below-threshold sample with exactly that call, so +#: a dot on the finger canvas has to be the same plus-shaped 5-pixel blob a box dot is. +#: Painting the offsets in one vectorised write, instead of one ``cv2.circle`` per dot, +#: is what keeps a full finger grid cheap: on a 276-candidate island the pass costs 47 us +#: against 223 us for the call loop, and the whole draw gains only ~35 us. +_IDLE_DOT_OFFSETS = ((-1, 0), (0, -1), (0, 0), (0, 1), (1, 0)) + + +def _paint_idle_dots( + image: np.ndarray, rows: np.ndarray, cols: np.ndarray, color: tuple[int, int, int] +) -> None: + """Paint one radius-1 marker at every sample, clipped at the image edge. + + The footprint is :data:`_IDLE_DOT_OFFSETS`, which is what ``cv2.circle`` fills for + ``radius=1, thickness=-1``; a dot that would fall outside the image contributes only + the pixels that are inside it, exactly as the clipped ``cv2.circle`` call does. + """ + h, w = image.shape[:2] + for dy, dx in _IDLE_DOT_OFFSETS: + rr = rows + dy + cc = cols + dx + keep = (rr >= 0) & (rr < h) & (cc >= 0) & (cc < w) + image[rr[keep], cc[keep]] = color + + def _paint_shared_flow_arrows( canvas: np.ndarray, flow: np.ndarray, @@ -294,23 +325,38 @@ def _paint_shared_flow_arrows( below_threshold_color: tuple[int, int, int], magnitude_range: tuple[float, float] | None = None, per_frame_colour: bool = False, + idle_dots: bool = True, ) -> np.ndarray: """Draw arrows from :func:`select_flow_samples` so 2D matches 3D sample points. The selector returns *candidate locations* and an ``active`` flag per sample; the - drawing choice is this function's own. Arrows are drawn for active samples, so what - reaches the canvas is what reached it before the contract split. + drawing choice is this function's own, and it is **one marker per sample**: an + active sample is drawn as an arrow and nothing else, an inactive one as a quiet dot + and nothing else, so no sample carries both. - An inactive sample is drawn as a quiet dot in ``below_threshold_color`` only when - :data:`DRAW_IDLE_GRID_DOTS` asks for it, and this canvas does not: a live finger frame - at 60 Hz would otherwise gain a grid of dots under the arrows. Flipping that constant - is what the split made possible — the dots used to depend on a below-threshold branch - that the selector's own filtering had already made unreachable. + The dot rules, all of them consequences of the selector's own contract: + + * **density** — one dot per candidate location, i.e. the policy's sampling grid + (``flow_grid_spacing``, 10 px on the finger's shared policy) rather than one per UV + pixel. A realistic finger island holds 28345 pixels and 276 candidates, so the grid + is what makes an idle frame readable instead of noise; + * **region** — the selector has already clipped the candidates to ``valid_mask`` at + the sample *and* at the arrow tip and to ``flow_edge_margin``, so a dot cannot land + where the device vouches for nothing or where the tip would fall outside; + * **colour** — ``below_threshold_color`` (:data:`FLOW_IDLE_HEX`), deliberately off the + flow ramp: colour is a measurement, and a dot states the absence of one; + * **surface** — a 1 px dot per 10 px cell leaves the island's own fill visible, so the + canvas is not darkened or recoloured to make the dots readable. + + ``idle_dots`` is the switch, defaulting to :data:`DRAW_IDLE_GRID_DOTS`. Arrows are + painted after the dots, so a dot from a neighbouring grid point can never take a pixel + off an arrow's tip. ``magnitude_range`` is the **fixed** colour range in uv pixels of displacement (:data:`FLOW_MAGNITUDE_RANGE`); ``per_frame_colour`` opts in to the old re-normalisation of every frame against its own maximum, which makes one magnitude - a different colour from one frame to the next. + a different colour from one frame to the next. Neither touches the dots: they are off + the ramp on purpose. """ from orisys_app_shared.scene3d.visual_policy import select_flow_samples @@ -319,14 +365,14 @@ def _paint_shared_flow_arrows( if samples.rows.size == 0: return image active = np.asarray(samples.active, dtype=bool) - if DRAW_IDLE_GRID_DOTS: - for i in np.flatnonzero(~active): - cv2.circle( + if idle_dots: + inactive = np.flatnonzero(~active) + if inactive.size: + _paint_idle_dots( image, - (int(samples.cols[i]), int(samples.rows[i])), - radius=1, - color=below_threshold_color, - thickness=-1, + samples.rows[inactive], + samples.cols[inactive], + tuple(int(v) for v in below_threshold_color), ) if not np.any(active): return image @@ -379,6 +425,7 @@ def draw_arrow_display( policy=None, magnitude_range: tuple[float, float] | None = None, per_frame_colour: bool = False, + idle_dots: bool | None = None, ) -> np.ndarray: """Draw flow arrows on a solid canvas or over the processed frame. @@ -404,6 +451,12 @@ def draw_arrow_display( back in to normalising every frame against its own maximum, which makes one magnitude a different colour from one frame to the next. + ``idle_dots`` shows or hides the below-threshold markers of the **shared finger + canvas**; ``None`` means :data:`DRAW_IDLE_GRID_DOTS`, which is on. Box/Mini draw + through the SDK's own ``orisys.util.draw_arrows``, which paints its own below-threshold + dots (same tint, same radius-1 marker) and has no such switch: this one governs the + canvas that draws here, and Box keeps the appearance it has. + Returns the drawn image, which carries ``frame_diagnostic``: ``None`` when the frame was drawn, and otherwise a :class:`~orisys_app_shared.scene3d.contracts.Diagnostic` whose ``code`` is :data:`ARROW_RENDER_FAILED`. **The failure must be surfaced**: the @@ -472,6 +525,10 @@ def draw_arrow_display( below_threshold_color=tuple(draw_kw["below_threshold_color"]), magnitude_range=magnitude_range, per_frame_colour=per_frame_colour, + # Read here rather than as a parameter default: the module constant is + # the display decision, and a caller that flips it (a build, a test) + # must be obeyed without every call site having to pass it. + idle_dots=DRAW_IDLE_GRID_DOTS if idle_dots is None else bool(idle_dots), ) ) diff --git a/shared/tests/test_image_backend.py b/shared/tests/test_image_backend.py index 0d4d2fd..ffbfe56 100644 --- a/shared/tests/test_image_backend.py +++ b/shared/tests/test_image_backend.py @@ -8,6 +8,8 @@ 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 @@ -116,9 +118,19 @@ def test_the_2d_canvas_draws_from_the_policy_it_is_given(): ``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 nothing is drawn, with 0.3 px the - same frame carries arrows. + 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 @@ -128,6 +140,15 @@ def test_the_2d_canvas_draws_from_the_policy_it_is_given(): 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( @@ -136,16 +157,19 @@ def test_the_2d_canvas_draws_from_the_policy_it_is_given(): loose_image = backend.render_image( frame, mode="arrows_pure", draw_style="finger", arrow_policy=loose ) - assert len(np.unique(strict_image.reshape(-1, 3), axis=0)) == 1, ( - "a 0.4 px threshold drew something for a 0.35 px sample" + 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 len(np.unique(loose_image.reshape(-1, 3), axis=0)) > 1, ( - "a 0.3 px threshold drew nothing for a 0.35 px sample" + 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 len(np.unique(direct.reshape(-1, 3), axis=0)) == 1 + assert np.array_equal(direct, strict_image) def test_the_2d_canvas_colours_one_magnitude_the_same_in_every_frame(): @@ -339,6 +363,374 @@ def test_a_missing_field_is_reported_not_invented(): 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 types. + + Box and Mini draw through the SDK's ``orisys.util.draw_arrows``, which paints a + below-threshold sample with ``cv2.circle(radius=1, thickness=-1)`` in the same + ``below_threshold_color`` this module passes it. The finger canvas has to put the same + blob on the same kind of location, or a quiet sample would look like two different + things depending on the sensor; this compares the two paths' pixels for one + below-threshold sample. + """ + pytest.importorskip("orisys", reason="the SDK path needs the SDK") + 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, shared by both paths: the box draw's own grid is 20 px with + # a 15 px edge margin, so a 60x60 frame has exactly the (30, 30) sample. + size = 60 + flow = np.zeros((size, size, 2), dtype=np.float64) + flow[30, 30] = (0.1, 0.0) # 0.2 px: under both paths' thresholds + 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") + 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 overlay_on.origins.shape[0] == int(active.sum()), ( + "the 3D overlay is not drawing the active samples the 2D canvas draws" + ) + 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" @@ -403,6 +795,10 @@ def test_a_wobbling_domain_mask_does_not_repaint_the_boundary(): 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() @@ -425,14 +821,31 @@ def test_a_wobbling_domain_mask_does_not_repaint_the_boundary(): frame, mode="depth", depth_range=(0.0, 1.0), domain_mask=used ) ) - assert all(np.array_equal(arrows[0], other) for other in arrows), ( - "the arrow canvas repainted a boundary that had not moved" - ) 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.""" diff --git a/shared/tests/test_visual_policy.py b/shared/tests/test_visual_policy.py index bdc703b..7d24b32 100644 --- a/shared/tests/test_visual_policy.py +++ b/shared/tests/test_visual_policy.py @@ -97,6 +97,46 @@ def test_the_location_cap_is_a_location_budget() -> None: ) +def test_the_threshold_decides_the_marker_and_never_the_location() -> None: + """The two markers are one grid read two ways: only the verdict moves with a threshold. + + The 2D canvas draws an arrow for an ``active`` sample and a quiet dot for an inactive + one, both at the candidate location this selector returns. So the candidate set — the + marker set, once each sample carries exactly one marker — is the policy's grid and + nothing else, while ``flow_on_px`` decides how many of them are arrows. A threshold + that moved a location would move the idle grid with it, which is what makes the dots + comparable from frame to frame. + """ + h, w = 30, 30 + flow = np.zeros((h, w, 2), dtype=np.float64) + # 6 px displacements of 0.1 .. 1.5 px, each on the 5 px grid (2 + 5k) the policy + # samples at, so each threshold step moves exactly one sample across. + for index, value in enumerate((0.05, 0.1, 0.2, 0.35, 0.5, 0.75)): + flow[2 + 5 * index, 2 + 5 * index] = (value, 0.0) + + locations: list[tuple[int, int]] | None = None + arrows: list[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=5, + flow_edge_margin=0, + ).normalised() + samples = select_flow_samples(flow, None, policy) + here = list(zip(samples.rows.tolist(), samples.cols.tolist())) + if locations is None: + locations = here + assert here == locations, ( + f"the candidate locations moved with the threshold at {on_px} px" + ) + # One marker per sample, and which marker is the threshold's whole decision. + assert samples.active.size == len(here) + assert samples.active_count + int((~samples.active).sum()) == len(here) + arrows.append(samples.active_count) + assert arrows == [5, 4, 3, 2, 1], f"the sweep did not move one sample per step: {arrows}" + + def test_paint_domain_background_colours() -> None: mask = np.zeros((4, 4), dtype=bool) mask[1:3, 1:3] = True