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

Store the depth channel in an analysis result and serve it to the 2D depth view

Browse Source 
A replay could draw arrows but not depth, and the recording view said there was
nothing to draw even when a result existed. Depth is now quantised per frame
(14 B empty, 486 B at the worst frame seen), read from the cache by the same
draw path live uses, and the raw-recording note appears only when no result
exists. Format version 2; see apps/studio/docs/replay-analysis.md.
This commit is contained in:
ykcheungab
2026-10-04 14:11:22 +08:00
parent cf49b0836e
commit 9e20fd0277
17 changed files with 1412 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shared/tests/test_analysis_depth.py
+454
View File
@@ -0,0 +1,454 @@
"""The depth channel in an analysis result: what it costs, and that it comes back.
What is being defended:
1. **The round trip is within the stated error, frame for frame, on real footage.** The
chain is walked once more, independently of the pass, and every stored depth map is
compared with the map the chain produced for the same frame id. The accepted error is
half one quantisation step of that frame's own range, ``(max - min) / 510`` — below
half a step of the 256-entry colour ramp the depth view paints with, so a
reconstruction cannot be coloured differently from the original.
2. **The size is stated and held.** 14 B per frame whose depth map is exactly all zeros
(the frame header's statistics plus a 6-byte payload header) and the header plus a
zlib'd uint8 plane for a frame that has one. The alternative encodings were measured
and rejected; the numbers are in the module docstring of ``analysis.model``.
3. **"Empty" and "absent" are different answers.** A frame whose depth map is all zeros
still carries the channel and states its shape; a frame whose chain exposed no depth
channel at all carries none. A view that draws the same blank picture for both is the
one that misleads, so the reader distinguishes them and the statistics travel with
the record.
4. **A corrupt depth payload is refused, not decoded into a plausible array.**
The fixtures use real footage — the 33-frame box recording under ``dist/`` and the
264-frame eight-times-slow clip under ``.scratch/sdk-stream-pipeline``, both driven
through the **finger** chain. Where the SDK or the footage is absent the tests skip
rather than fake a result: a synthetic array would defend neither the statistics
observed on this footage nor the size they produce.
"""
from __future__ import annotations
import json
import os
import tempfile
from pathlib import Path
import numpy as np
import pytest
from orisys_app_shared.analysis.cache import ResultCache, scan_results
from orisys_app_shared.analysis.model import (
CODEC_EMPTY,
CODEC_RAW,
CODEC_ZLIB,
DEPTH_HEADER,
decode_depth,
encode_depth,
encode_frame,
frame_payload_bytes,
)
from orisys_app_shared.analysis.reader import AnalysisReader
from orisys_app_shared.analysis.service import (
AnalysisService,
ChainSpec,
_OpenChain,
)
REPO = Path(__file__).resolve().parents[2]
STUDIO = REPO / "dist" / "OrisysStudio"
RECORDING = (
STUDIO
/ "outputs"
/ "recordings"
/ "20260930T015028Z_session0002_40cb8e7441434feab48bb646afdba011"
)
FINGER_CONFIG = STUDIO / "config" / "sensors" / "finger" / "sensor.json"
FINGER_EXPORT = STUDIO / "config" / "sensors" / "finger" / "assets" / "30fps-0703"
#: The 264-frame clip Step 0 measured the pass's throughput on. It lives in another
#: task's scratch, so it is optional: without it the clip half of the round trip is
#: reported as skipped and the box recording still runs.
CLIP = REPO / ".scratch" / "sdk-stream-pipeline" / "long_8x.mp4"
CLIP_FRAMES = 264
CLIP_FPS = 30.0
#: Measured on the two recordings, at the finger chain's own 300 x 280 depth grid.
#: Empty frames and the worst non-empty frame are the two numbers the encoding is
#: chosen on, so they are asserted rather than described.
DEPTH_GRID = (300, 280)
EMPTY_PAYLOAD_BYTES = DEPTH_HEADER.size
EMPTY_RECORD_BYTES = frame_payload_bytes(330, with_validity=True, with_depth=True)
FIXED_RECORD_BYTES = EMPTY_RECORD_BYTES - DEPTH_HEADER.size
#: The worst non-empty frame either recording produced: 336 of 84,000 pixels on the clip,
#: whose quantised plane zlib's to 486 B. The bound below is what the format promises;
#: the measured number is what the recordings delivered, and both are asserted.
WORST_NONEMPTY_PIXELS = 336
WORST_NONEMPTY_PAYLOAD_BYTES = 486
def _require_inputs() -> None:
missing = [
str(path)
for path in (RECORDING / "frames.csv", RECORDING / "raw_0001.mp4", FINGER_CONFIG)
if not path.exists()
]
if missing:
pytest.skip("analysis inputs are not present: " + ", ".join(missing))
pytest.importorskip("orisys", reason="the SDK is needed to run the analysis pass")
def _finger_chain() -> ChainSpec:
return ChainSpec(
sensor_type="finger", config_ref=FINGER_CONFIG, export_dir=FINGER_EXPORT
)
def _result_directory(root: Path) -> Path:
directories = [entry.directory for entry in scan_results(root) if entry.complete]
assert directories, f"no complete result under {root}"
return directories[0]
def _clip_recording(tmp_path: Path) -> Path:
"""The clip as a one-segment recording: a link to the video plus a frames index.
A bare container has no ``frames.csv``, and frame identity comes from the
recording's own counter (Part A, section 6). Writing the index here is what makes
the clip addressable by the same code path as a real recording instead of by an
index standing in for an id.
"""
directory = tmp_path / "clip-recording"
directory.mkdir(parents=True, exist_ok=True)
video = directory / "raw_0001.mp4"
if not video.exists():
try:
os.link(CLIP, video)
except OSError:
video.write_bytes(CLIP.read_bytes())
index = directory / "frames.csv"
rows = ["frame_id,t_rel_s,segment,segment_frame,in_video,dropped"]
rows.extend(
f"{number},{number / CLIP_FPS:.6f},1,{number},1,0"
for number in range(1, CLIP_FRAMES + 1)
)
index.write_text("\n".join(rows) + "\n", encoding="utf-8")
return directory
def _chain_depth(recording: Path, video: str = "raw_0001.mp4") -> list[np.ndarray]:
"""Walk the chain once more and return the depth map it produced for every frame.
Deliberately a *separate* walk from the pass: comparing a stored result with the
object the pass held would only prove the writer copied what it was given.
"""
from orisys_app_shared.acquisition.sdk_worker import process_chain
produced: list[np.ndarray] = []
with tempfile.TemporaryDirectory() as working:
with _OpenChain(_finger_chain(), recording / video, Path(working)) as opened:
while True:
ok, frame = opened.read()
if not ok or frame is None:
break
if opened.feed(frame) is None:
raise AssertionError("the chain could not unroll a frame")
depth, error = process_chain(opened.sensor)
assert not error, error
assert depth is not None, "the finger chain exposed no depth channel"
array = np.asarray(depth, dtype=np.float32)
while array.ndim > 2 and array.shape[0] == 1:
array = array[0]
produced.append(np.ascontiguousarray(array))
return produced
def _tolerance(frame) -> float:
"""Half one quantisation step of this frame's own stored range."""
return (float(frame.depth_max) - float(frame.depth_min)) / 510.0
def _compare(stored, produced: list[np.ndarray]) -> dict[str, int]:
"""Compare every stored depth map with the chain's, frame for frame.
The order is the pass's own: the finger chain consumes no frame before the first
readable one, so decoded frame *k* is stored frame *k*.
"""
frames = list(stored)
assert len(frames) == len(produced), (
f"{len(frames)} stored frames against {len(produced)} decoded"
)
stats = {"frames": len(frames), "empty": 0, "nonempty": 0, "worst_pixels": 0}
for frame, original in zip(frames, produced):
assert frame.has_depth, f"frame {frame.frame_id} lost its depth channel"
assert frame.depth.shape == DEPTH_GRID, frame.depth.shape
assert original.shape == frame.depth.shape
if frame.depth_nonzero:
stats["nonempty"] += 1
stats["worst_pixels"] = max(stats["worst_pixels"], frame.depth_nonzero)
else:
stats["empty"] += 1
assert not bool(np.any(original)), (
f"frame {frame.frame_id} is stored as an empty depth map but the chain "
"produced a non-zero one"
)
# The statistic the record states is the statistic of the map the chain made.
assert frame.depth_nonzero == int(np.count_nonzero(original))
assert float(frame.depth_max) == float(original.max())
assert float(frame.depth_min) == float(original.min())
error = float(np.max(np.abs(np.asarray(frame.depth) - original)))
assert error <= _tolerance(frame), (
f"frame {frame.frame_id} is {error} away from the chain's map, over the "
f"{_tolerance(frame)} the encoding accepts"
)
# The reconstruction carries the same zero/non-zero verdict everywhere: every
# observed non-zero value is far above the half step, so the count survives too.
assert int(np.count_nonzero(frame.depth)) == frame.depth_nonzero
return stats
# ------------------------------------------------------------------ the round trip
@pytest.fixture(scope="module")
def box_depth(tmp_path_factory):
"""One real pass over the evidence recording, kept for the round trip and the size."""
_require_inputs()
root = tmp_path_factory.mktemp("analysis-depth-box")
outcome = AnalysisService(
RECORDING, chain=_finger_chain(), cache=ResultCache(root)
).run()
assert outcome.complete, outcome.reason
return root
def test_the_box_recording_round_trips_its_depth_within_the_stated_error(box_depth):
produced = _chain_depth(RECORDING)
reader = AnalysisReader(_result_directory(box_depth))
try:
stats = _compare(reader.frames(), produced)
# The observed split, not a hope: 32 of the 33 frames carry a map that is
# exactly all zeros, and the one that does not holds 76 pixels of 84,000.
assert stats == {
"frames": 33,
"empty": 32,
"nonempty": 1,
"worst_pixels": 76,
}
finally:
reader.close()
def test_the_clip_round_trips_its_depth_within_the_stated_error(tmp_path):
if not CLIP.is_file():
pytest.skip(f"the clip is not present: {CLIP}")
_require_inputs()
recording = _clip_recording(tmp_path)
root = tmp_path / "cache"
outcome = AnalysisService(
recording, chain=_finger_chain(), cache=ResultCache(root)
).run()
assert outcome.complete, outcome.reason
produced = _chain_depth(recording)
reader = AnalysisReader(_result_directory(root))
try:
stats = _compare(reader.frames(), produced)
assert stats == {
"frames": CLIP_FRAMES,
"empty": 241,
"nonempty": 23,
"worst_pixels": WORST_NONEMPTY_PIXELS,
}
finally:
reader.close()
# ------------------------------------------------------------------------- the size
def test_the_depth_cost_is_the_measured_encoding(box_depth):
"""14 B per empty frame, the plane's own zlib'd bytes for a frame that has one.
The record without depth is 4,034 B at the finger grid; the depth channel adds the
eight bytes of per-frame statistics the header now carries (``depth_min``,
``depth_max``) and the six-byte payload header, and then whatever the plane costs.
Ten minutes at 30 fps is 18,000 frames, so the empty case is 72,864,000 B = 69.5 MiB
against 69.3 MiB before: the number is small because the *common* case is a map that
is all zeros and a payload that is a header.
"""
assert DEPTH_HEADER.size == 6
assert FIXED_RECORD_BYTES == 4042
assert EMPTY_RECORD_BYTES == 4048
reader = AnalysisReader(_result_directory(box_depth))
try:
frames = list(reader.frames())
emptied = [item for item in frames if item.depth_nonzero == 0]
carrying = [item for item in frames if item.depth_nonzero]
assert len(emptied) + len(carrying) == len(frames)
assert {item.depth_stored_bytes for item in emptied} == {EMPTY_PAYLOAD_BYTES}
for item in carrying:
assert item.depth_stored_bytes == DEPTH_HEADER.size + len(
_payload_after_header(item.depth, item.depth_min, item.depth_max)
)
assert reader.frame_record_bytes == sum(
FIXED_RECORD_BYTES + item.depth_stored_bytes for item in frames
)
assert reader.bytes_per_frame <= 4096.0, "the stated per-frame bound"
# Worst case, every frame the worst non-empty frame observed on the clip.
worst = FIXED_RECORD_BYTES + DEPTH_HEADER.size + WORST_NONEMPTY_PAYLOAD_BYTES
assert worst == 4534
assert worst * 18_000 == 81_612_000, "81.6 MB per ten minutes, upper bound"
# The rate the footage actually produced, over the 264-frame clip: 241 empty
# frames and 23 at the worst non-empty size.
typical = (241 * EMPTY_RECORD_BYTES + 23 * worst) / 264
assert typical == pytest.approx(4090.34, abs=0.01)
assert typical * 18_000 == pytest.approx(73_626_136, abs=1.0)
assert typical * 18_000 < 73_700_000, "70.2 MB per ten minutes at the observed rate"
assert typical < 4096.0
finally:
reader.close()
def _payload_after_header(depth: np.ndarray, low: float, high: float) -> bytes:
"""What ``encode_depth`` would store for a map, payload only. Shared with the codec."""
return encode_depth(depth)[0][DEPTH_HEADER.size :]
# ------------------------------------------------------------------- the codec itself
def test_an_empty_map_costs_a_header_and_states_its_shape():
depth = np.zeros(DEPTH_GRID, dtype=np.float32)
payload, nonzero, low, high = encode_depth(depth)
assert (nonzero, low, high) == (0, 0.0, 0.0)
assert payload == DEPTH_HEADER.pack(CODEC_EMPTY, 8, *DEPTH_GRID)
assert len(payload) == EMPTY_PAYLOAD_BYTES
decoded = decode_depth(payload)
assert decoded.is_empty and decoded.nonzero == 0
assert (decoded.height, decoded.width) == DEPTH_GRID
from orisys_app_shared.analysis.model import depth_values
restored = depth_values(decoded, low=low, high=high)
assert restored.shape == DEPTH_GRID
assert not bool(np.any(restored)), "an empty map comes back as zeros, not as NaNs"
def test_a_single_plane_is_quantised_over_its_own_range_and_stays_within_half_a_step():
rng = np.random.default_rng(20261004)
depth = np.zeros(DEPTH_GRID, dtype=np.float32)
rows = rng.integers(0, DEPTH_GRID[0], 500)
cols = rng.integers(0, DEPTH_GRID[1], 500)
depth[rows, cols] = rng.random(500).astype(np.float32) * 0.05
payload, nonzero, low, high = encode_depth(depth)
assert nonzero == int(np.count_nonzero(depth))
assert (low, high) == (float(depth.min()), float(depth.max()))
assert len(payload) < 4096, "a 500-pixel plane is not 84 kB of payload"
decoded = decode_depth(payload, low=low, high=high)
assert decoded.codec in (CODEC_ZLIB, CODEC_RAW)
from orisys_app_shared.analysis.model import depth_values
restored = depth_values(decoded, low=low, high=high)
step = (high - low) / 255.0
assert float(np.max(np.abs(restored - depth))) <= step / 2.0 + 1e-12
# The count survives for every value the quantiser can actually represent: a value
# above half a step lands on a code of at least one. A value *below* half a step can
# round to zero, and that is the accepted error rather than a lost measurement — it
# is smaller than the error the encoding already allows.
assert int(np.count_nonzero(restored[depth > step / 2.0])) == int(
np.count_nonzero(depth > step / 2.0)
)
assert not bool(np.any(restored[depth == 0.0]))
# The display paints this range onto a 256-entry ramp, so half a quantisation step
# is half a ramp step: the reconstruction cannot change a pixel's colour.
assert step / 2.0 <= (high - low) / 255.0
def test_a_plane_that_does_not_compress_is_stored_raw_rather_than_larger():
rng = np.random.default_rng(7)
depth = rng.random(DEPTH_GRID).astype(np.float32)
payload, nonzero, low, high = encode_depth(depth)
decoded = decode_depth(payload, low=low, high=high)
assert decoded.codec == CODEC_RAW
assert decoded.stored_bytes == depth.size
assert len(payload) == DEPTH_HEADER.size + depth.size
def test_depth_statistics_are_per_frame_because_the_range_is():
"""Two frames with different ranges each keep all 256 levels."""
flat = np.zeros(DEPTH_GRID, dtype=np.float32)
flat[10, 10] = 0.001
steep = np.zeros(DEPTH_GRID, dtype=np.float32)
steep[10, 10] = 0.0
steep[20, 20] = 1.0
for frame in (flat, steep):
payload, _, low, high = encode_depth(frame)
decoded = decode_depth(payload, low=low, high=high)
from orisys_app_shared.analysis.model import depth_values
restored = depth_values(decoded, low=low, high=high)
assert float(np.max(np.abs(restored - frame))) <= (high - low) / 510.0 + 1e-12
def test_a_corrupt_depth_payload_is_refused_rather_than_decoded():
good, _, low, high = encode_depth(
np.full(DEPTH_GRID, 0.25, dtype=np.float32)
)
# A plane that decodes to the wrong number of bytes for the shape it declares.
with pytest.raises(ValueError):
decode_depth(good[:-1], low=low, high=high)
# A shape no sensor has, refused before it becomes an allocation.
with pytest.raises(ValueError):
decode_depth(DEPTH_HEADER.pack(CODEC_EMPTY, 8, 0xFFFF, 0xFFFF))
# A bit depth this build does not store.
with pytest.raises(ValueError):
decode_depth(DEPTH_HEADER.pack(CODEC_EMPTY, 16, *DEPTH_GRID))
# And a header that is not even there.
with pytest.raises(ValueError):
decode_depth(b"")
def test_a_frame_without_a_depth_channel_is_not_a_frame_with_an_empty_one():
"""The two facts have to stay apart: only one of them can be drawn as a measurement."""
blank = encode_frame(
frame_id=1,
t_rel_s=0.0,
grid_id=0,
dx=np.zeros(2, np.float32),
dy=np.zeros(2, np.float32),
magnitude=np.zeros(2, np.float32),
active=np.zeros(2, bool),
depth=np.zeros(DEPTH_GRID, np.float32),
)
absent = encode_frame(
frame_id=2,
t_rel_s=0.0,
grid_id=0,
dx=np.zeros(2, np.float32),
dy=np.zeros(2, np.float32),
magnitude=np.zeros(2, np.float32),
active=np.zeros(2, bool),
)
assert len(blank) - len(absent) == EMPTY_PAYLOAD_BYTES
from orisys_app_shared.analysis.model import decode_frame
empty_frame = decode_frame(blank)
absent_frame = decode_frame(absent)
assert empty_frame.depth is not None and empty_frame.depth.is_empty
assert empty_frame.depth_map() is not None
assert absent_frame.depth is None and absent_frame.depth_map() is None
def test_the_provenance_counts_what_was_empty(box_depth):
"""The operator's "there is no depth here" has to be answerable from the result."""
reader = AnalysisReader(_result_directory(box_depth))
try:
counts = reader.counts
assert counts["depth_frames"] == 33
assert counts["depth_nonzero_frames"] == 1
assert counts["depth_nonzero_max"] == 76
assert counts["depth_pixels"] == DEPTH_GRID[0] * DEPTH_GRID[1]
assert counts["frame_record_bytes"] == reader.frame_record_bytes
provenance = json.loads(
(_result_directory(box_depth) / "meta.json").read_text(encoding="utf-8")
)["provenance"]
assert provenance["sources"]["depth"] == "recomputed"
finally:
reader.close()
shared/tests/test_replay_analysis.py
+47 -15
View File
@@ -38,7 +38,11 @@ from orisys_app_shared.analysis.cache import (
evict,
scan_results,
)
from orisys_app_shared.analysis.model import frame_payload_bytes, result_key
from orisys_app_shared.analysis.model import (
FRAME_HEADER,
frame_payload_bytes,
result_key,
)
from orisys_app_shared.analysis.reader import (
AnalysisIncompleteError,
AnalysisReader,
@@ -477,24 +481,43 @@ def test_the_key_follows_the_inputs_that_change_the_result_and_no_others(finger_
# ------------------------------------------------------------------ size
def test_the_stored_size_is_the_measured_encoding(box_pass, finger_pass):
"""State the encoding, then hold it: 4,034 B per frame at the finger grid.
"""State the encoding, then hold it: 4,048 B per frame at the finger grid.
Implicit candidate grid (a keyframe) + float32 ``dx``/``dy``/``magnitude`` +
bit-packed ``active`` + a 24-byte header + the 8-byte validity digest:
``24 + 330*12 + 42 + 8 = 4,034`` B. The alternative measured encodings are float16
``dxy``+magnitude (2,022 B) and dropping magnitude (4,002 B with the grid, no
header); float32 is chosen because the round trip has to be *exact*, and the
magnitude is what serves a changed threshold.
bit-packed ``active`` + a 32-byte header + the 8-byte validity digest + the 6-byte
depth header: ``32 + 330*12 + 42 + 8 + 6 = 4,048`` B, of which the depth part is
``14`` B over the 4,034 B this record cost before the depth channel was stored. A
frame whose depth map is *not* all zeros adds its own zlib'd uint8 plane on top.
The alternative measured encodings are float16 ``dxy``+magnitude (2,022 B) and
dropping magnitude (4,002 B with the grid, no header); float32 is chosen because
the round trip has to be *exact*, and the magnitude is what serves a changed
threshold.
"""
assert frame_payload_bytes(330, with_validity=True) == 4034
assert frame_payload_bytes(330, with_validity=False) == 4026
assert frame_payload_bytes(400, with_validity=False) == 4874
fixed = FRAME_HEADER.size + 330 * 12 + 42 + 8
assert fixed == 4042
assert frame_payload_bytes(330, with_validity=True) == 4042
assert frame_payload_bytes(330, with_validity=True, with_depth=True) == 4048
assert frame_payload_bytes(330, with_validity=False, with_depth=True) == 4040
assert frame_payload_bytes(400, with_validity=False) == 4882
assert frame_payload_bytes(400, with_validity=False, with_depth=True) == 4888
_, finger = finger_pass
reader = AnalysisReader(finger.directory)
try:
assert reader.bytes_per_frame == 4034.0
assert reader.frame_record_bytes == 33 * 4034
# Every frame states the depth channel; the ones whose map is all zeros pay the
# 6-byte header and nothing else.
frames = list(reader.frames())
emptied = [item for item in frames if item.has_depth and not item.depth_nonzero]
carrying = [item for item in frames if item.depth_nonzero]
assert len(emptied) + len(carrying) == len(frames)
assert reader.frame_record_bytes == sum(
fixed + item.depth_stored_bytes for item in frames
)
assert reader.frame_record_bytes == (
len(emptied) * 4048
+ len(carrying) * fixed
+ sum(item.depth_stored_bytes for item in carrying)
)
assert reader.bytes_per_frame <= 4096, "the stated per-frame bound"
total_per_frame = _directory_bytes(finger.directory) / len(reader)
assert total_per_frame <= 4600, "keyframes and metadata included"
@@ -505,8 +528,12 @@ def test_the_stored_size_is_the_measured_encoding(box_pass, finger_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,874 B of record per frame at 400 candidates.
assert box_reader.frame_record_bytes == 26 * 4874
# carries its own grid keyframe: 4,882 B of record per frame at 400 candidates,
# plus whatever depth it carries.
assert box_reader.frame_record_bytes == sum(
32 + 400 * 12 + 50 + item.depth_stored_bytes
for item in box_reader.frames()
)
assert box_reader.keyframe_counts == {"grid": 26}
assert box_reader.mask() is None
finally:
@@ -608,7 +635,12 @@ def test_a_result_without_metadata_is_never_served(tmp_path):
AnalysisReader(directory)
(directory / META_NAME).write_text(
json.dumps({"format": analysis_model.ANALYSIS_FORMAT, "format_version": 1}),
json.dumps(
{
"format": analysis_model.ANALYSIS_FORMAT,
"format_version": analysis_model.ANALYSIS_FORMAT_VERSION,
}
),
encoding="utf-8",
)
with pytest.raises(AnalysisIncompleteError):