orisys-tech/orisys-app
3
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Record the live box arrow grid so replay matches live paint

Browse Source 
Box analysis used the finger sample grid, so replay packed arrows into the contact zone. The pass now stores draw_arrows cells; stale box caches miss and re-run. Verified with shared and Studio tests, including a live-vs-replay canvas check.
This commit is contained in:
ykcheungab
2026-10-05 15:01:48 +08:00
parent 28e955a817
commit 3433eb0fec
10 changed files with 1614 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shared/tests/test_image_backend.py
+141
View File
@@ -919,3 +919,144 @@ def test_resetting_the_latch_clears_the_session_boundary():
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():
"""Where a frame came from decides the painter, not which sensor's style is selected.
The SDK painter walks the dense sampling grid, so on a field that is finite only at a
stored result's candidate cells it finds no sample at all and paints nothing: measured
on a real box recording, **0 px** against the shared painter's **1,486 px** on the same
frame. So the routing is by ``ImageFrame.source`` — the frame came from a result — and
the box *style* still owns the magnification and the idle-dot tint.
"""
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)
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(),
shared_painter=True,
)
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"
# The same array handed over as a *live* field keeps the SDK path — which is what the
# sensor family's draw style has always meant, and which paints nothing here.
sdk = backend.render_image(
live_frame, mode=backend.ARROW_PURE, draw_style="box", arrow_policy=_policy()
)
assert _painted(sdk) == 0, "the SDK painter drew a recovered field"
assert not np.array_equal(np.asarray(drawn), np.asarray(sdk))
# 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 live painter's grid, so filling the gaps and calling it
back draws the same picture as 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)
length = np.sqrt(dx * dx + dy * dy)
flow_mag = np.asarray(samples.magnitude, np.float64) / DISPLACEMENT_PX_PER_FLOW
safe = np.maximum(length, 1e-12)
moving = length > 0.0
sparse = np.full(dense.shape, np.nan, dtype=np.float64)
sparse[samples.rows, samples.cols, 0] = np.where(moving, dx / safe * flow_mag, 0.0)
sparse[samples.rows, samples.cols, 1] = np.where(moving, dy / safe * flow_mag, 0.0)
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,
)
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
shared/tests/test_replay_analysis.py
+10 -9
View File
@@ -13,9 +13,8 @@ The inputs are real: the 33-frame box recording under ``dist/`` (its own ``frame
``session.json`` and ``raw_0001.mp4``) and the shipped finger configuration and asset
bundle. The pass is driven with the **box** chain on it for the offset and round-trip
checks, and with the **finger** chain on the same footage for the keyframe, reset and
size checks — the finger chain over box footage is exactly what the Step 0 measurement
did, and it is the only way to get a validity mask (and therefore a static candidate
grid) out of this recording.
size checks. The finger chain is what produces a validity mask; the box chain's
painter grid is static without one.
"""
from __future__ import annotations
@@ -530,14 +529,16 @@ def test_the_stored_size_is_the_measured_encoding(box_pass, finger_pass):
_, box = box_pass
box_reader = AnalysisReader(box.directory)
try:
# The box chain has no validity mask, so there is no digest and every frame
# carries its own grid keyframe: 4,882 B of record per frame at 400 candidates,
# plus whatever depth it carries.
# The box chain has no validity mask. Its sample grid is the live painter's
# (spacing 20 over the 365 stitch, every cell kept): 18 x 18 = 324 candidates,
# one keyframe, and the per-frame record is values plus whatever depth it carries.
frames = list(box_reader.frames())
assert {item.candidates for item in frames} == {324}
assert box_reader.frame_record_bytes == sum(
32 + 400 * 12 + 50 + item.depth_stored_bytes
for item in box_reader.frames()
frame_payload_bytes(324, with_validity=False) + item.depth_stored_bytes
for item in frames
)
assert box_reader.keyframe_counts == {"grid": 26}
assert box_reader.keyframe_counts == {"grid": 1}
assert box_reader.mask() is None
finally:
box_reader.close()