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

Present replay 3D with the cursor and open Analyze on a remembered video

Browse Source 
Replay left the 3D pane on the rest mesh or a sparse overlay, and Analyze skipped the setup dialog unless the file was re-picked. Live Run no longer starts a chain on a video path. Verified with Studio tests for flush, overlay mask, Analyze open, live-video refusal, and locales.
This commit is contained in:
ykcheungab
2026-10-05 16:52:27 +08:00
parent 3433eb0fec
commit c0a10e6adb
26 changed files with 2226 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shared/tests/test_bare_video.py
+505
View File
@@ -0,0 +1,505 @@
"""A bare video's generated workspace: the two files the pass reads, and what they say.
The pass has one input shape — a directory with ``frames.csv``, ``session.json`` and
``raw_*.mp4`` — and it is a good shape: frame identity comes from the recording's own
counter, signal time from its own table, and the chain from its own metadata. A bare video
gets that shape *manufactured* for it rather than a second code path in the pass, so these
tests defend three properties of the manufacture: the rows are the container's real frames,
the documents are deterministic enough that reopening the video is a cache hit, and nothing
is written anywhere near the operator's file.
No SDK and no real footage: a video written by OpenCV here is the input, and the pass itself
is exercised in ``apps/studio/tests/test_bare_video_analysis.py`` over recorded footage.
"""
from __future__ import annotations
import csv
import json
from pathlib import Path
import pytest
from orisys_app_shared import paths as paths_module
from orisys_app_shared.analysis import bare_video
from orisys_app_shared.analysis.model import ForceSample, encode_frame, decode_frame
from orisys_app_shared.analysis.service import ChainSpec
from orisys_app_shared.record.playback import RecordingReader
FRAMES = 12
FPS = 30.0
# ------------------------------------------------------------------------ fixtures
def write_video(path: Path, *, frames: int = FRAMES, fps: float = FPS) -> Path:
"""A decodable video of ``frames`` distinct frames, written by OpenCV."""
cv2 = pytest.importorskip("cv2", reason="OpenCV writes and reads the fixture video")
import numpy as np
path.parent.mkdir(parents=True, exist_ok=True)
writer = cv2.VideoWriter(
str(path), cv2.VideoWriter_fourcc(*"mp4v"), float(fps), (64, 48)
)
assert writer.isOpened(), "the fixture video could not be opened for writing"
for index in range(frames):
image = np.full((48, 64, 3), 30 + index * 8, dtype=np.uint8)
image[::8, :, :] = 200
image[:, ::8, :] = 90
writer.write(image)
writer.release()
assert path.is_file() and path.stat().st_size > 0
return path
class _Entry:
"""The catalog-entry fields the workspace generator asks about."""
def __init__(self, *, family: str = "box", name: str = "Box", blocked: str = "",
sensor_json: Path | None = None, ref: str = "shipped/box",
has_profile: bool = True) -> None:
self.product_family = family
self.name = name
self.ref = ref
self.folder = ref.split("/")[-1]
self.profile_id = f"orisys.{family}.r01"
self.display_name = name
self.bundle_path = Path(sensor_json).parent if sensor_json else Path(".")
self.sensor_json_path = sensor_json
#: ``None`` stands for a folder no bundle could be read from at all.
self.profile = object() if has_profile else None
self._blocked = blocked
@property
def start_blocked_reason(self) -> str | None:
return self._blocked or None
@pytest.fixture
def video(tmp_path) -> Path:
return write_video(tmp_path / "operator" / "clip.mp4")
@pytest.fixture
def workspace_root(tmp_path, monkeypatch) -> Path:
"""The app-owned root the workspace goes to, pointed at this test."""
root = tmp_path / "derived" / "video-workspaces"
monkeypatch.setattr(paths_module, "video_workspace_dir", lambda: root)
return root
def _build(video: Path, **kwargs) -> bare_video.SyntheticWorkspace:
return bare_video.build_workspace(video, profile=_Entry(**kwargs), setup=None)
def _entry_set(directory: Path) -> set:
"""Every entry under ``directory`` as ``(relative path, size)``."""
found = set()
for path in Path(directory).rglob("*"):
relative = str(path.relative_to(directory))
found.add((relative, -1 if path.is_dir() else path.stat().st_size))
return found
# ------------------------------------------------------------------- the frame index
def test_the_frame_index_is_one_row_per_container_frame(video, workspace_root):
"""``frames.csv`` is the video's own frames: ids from 1, signal time from the rate.
A bare video has one clock — the container's — so ``t_rel_s`` is that clock divided
out, and every row is in the video because nothing was captured and nothing was lost.
"""
workspace = _build(video)
assert [row.frame_id for row in workspace.frames] == list(range(1, FRAMES + 1))
assert [row.segment for row in workspace.frames] == [1] * FRAMES
assert [row.segment_frame for row in workspace.frames] == list(range(1, FRAMES + 1))
assert all(row.in_video for row in workspace.frames)
assert not any(row.dropped for row in workspace.frames)
assert workspace.frames[0].t_rel_s == 0.0
assert workspace.frames[-1].t_rel_s == pytest.approx((FRAMES - 1) / FPS, abs=1e-9)
# The file is read by the pass's own reader, so it has to say the same thing the
# in-memory rows do — that equality is the whole point of generating it.
reader = RecordingReader(workspace.directory)
try:
loaded = reader.info().frames
finally:
reader.close()
assert [(row.frame_id, row.segment, row.segment_frame, row.in_video) for row in loaded] == [
(row.frame_id, row.segment, row.segment_frame, row.in_video) for row in workspace.frames
]
assert [row.t_rel_s for row in loaded] == pytest.approx(
[row.t_rel_s for row in workspace.frames], abs=1e-9
)
def test_the_session_document_says_what_the_pass_needs(video, workspace_root, tmp_path):
"""``device``/``stats``/``files`` are what the chain and the segment plan are read from."""
config = tmp_path / "sensors" / "box" / "sensor.json"
config.parent.mkdir(parents=True)
config.write_text('{"camera": {"fps": 30}}', encoding="utf-8")
workspace = _build(video, sensor_json=config)
session = workspace.session
assert session["device"]["sensor_type"] == "box"
assert session["device"]["config_ref"] == str(config)
assert session["device"]["config_sha256"]
assert session["stats"]["frames"] == FRAMES
assert session["stats"]["video_frames"] == FRAMES
assert session["stats"]["segments"] == 1
assert session["files"]["video"] == ["raw_0001.mp4"]
assert (workspace.directory / "raw_0001.mp4").is_file()
# ...and the pass's own chain reader agrees, from the file rather than from memory.
reader = RecordingReader(workspace.directory)
try:
assert reader.session()["device"]["sensor_type"] == "box"
finally:
reader.close()
def test_the_workspace_is_marked_synthetic(video, workspace_root):
"""A generated workspace must be distinguishable from a recording, forever.
``read_synthetic`` is the question the app asks a directory before it decides how to
treat it, and it answers ``None`` for a real recording — the difference between
"somebody measured this" and "this was generated here".
"""
workspace = _build(video)
block = bare_video.read_synthetic(workspace.directory)
assert block is not None
assert block["generator"] == bare_video.GENERATOR
assert block["source_video"] == str(Path(video).resolve())
assert block["frame_index"].startswith("generated:")
assert bare_video.is_synthetic(workspace.directory)
assert bare_video.source_video(workspace.directory) == Path(video).resolve()
# A real recording's session.json has no such block, and a folder without one is not
# a workspace at all.
plain = workspace.directory.parent / "plain"
plain.mkdir()
(plain / "session.json").write_text(json.dumps({"session_id": 1}), encoding="utf-8")
assert bare_video.read_synthetic(plain) is None
assert bare_video.read_synthetic(workspace.directory / "nothing-here") is None
def test_reopening_the_same_video_is_the_same_analysis_key(video, workspace_root, tmp_path):
"""Determinism is not tidiness: the key is what makes a second open a cache read.
``recording_identity`` hashes the session's own facts, so anything derived from the
clock would give the same video a new result directory on every open — and the pass
would run again every time.
"""
config = tmp_path / "sensors" / "box" / "sensor.json"
config.parent.mkdir(parents=True)
config.write_text("{}", encoding="utf-8")
chain = ChainSpec(sensor_type="box", config_ref=config)
first = _build(video, sensor_json=config)
again = _build(video, sensor_json=config)
assert again.directory == first.directory
assert again.session["session_id"] == first.session["session_id"]
from orisys_app_shared.analysis.service import AnalysisService
keys = [
AnalysisService(item.directory, chain=chain).dependency_report()["key"].digest
for item in (first, again)
]
assert keys[0] == keys[1], "reopening the video produced a different result key"
# A different video is a different workspace and a different key.
other = write_video(video.parent / "other.mp4", frames=FRAMES + 3)
third = _build(other, sensor_json=config)
assert third.directory != first.directory
assert (
AnalysisService(third.directory, chain=chain).dependency_report()["key"].digest
!= keys[0]
)
def test_the_calibration_is_part_of_the_key_by_its_content(video, workspace_root, tmp_path):
"""Coefficients edited in place must not keep serving a result computed from the old ones."""
config = tmp_path / "sensors" / "box" / "sensor.json"
config.parent.mkdir(parents=True)
config.write_text("{}", encoding="utf-8")
calibration = tmp_path / "force_calibration.json"
calibration.write_text('{"gain": 1}', encoding="utf-8")
first = bare_video.build_workspace(
video,
profile=_Entry(sensor_json=config),
setup=bare_video.BareVideoSetup(profile_ref="shipped/box", calibration=calibration),
)
calibration.write_text('{"gain": 2}', encoding="utf-8")
second = bare_video.build_workspace(
video,
profile=_Entry(sensor_json=config),
setup=bare_video.BareVideoSetup(profile_ref="shipped/box", calibration=calibration),
)
assert first.session["calibration_id"] != second.session["calibration_id"]
# The name still travels with the digest, so a reader knows which file it was.
assert second.session["calibration_id"].startswith("force_calibration.json@")
raw = bare_video.build_workspace(
video,
profile=_Entry(sensor_json=config),
setup=bare_video.BareVideoSetup(profile_ref="shipped/box", calibration=None),
)
assert raw.session["calibration_id"] == ""
def test_a_chosen_calibration_that_is_not_there_is_refused(video, workspace_root, tmp_path):
with pytest.raises(bare_video.BareVideoError):
bare_video.build_workspace(
video,
profile=_Entry(),
setup=bare_video.BareVideoSetup(
profile_ref="shipped/box", calibration=tmp_path / "absent.json"
),
)
# ------------------------------------------------------------------- what is refused
def test_a_split_recording_is_refused_rather_than_half_analysed(tmp_path):
"""More than one file is one measurement, not one input: refuse it, by name.
Analysing the first segment of a split recording and calling the result that
recording's measurement is a measurement error, and it is silent — the only visible
symptom is a short timeline.
"""
segments = tmp_path / "split"
segments.mkdir()
write_video(segments / "raw_0001.mp4")
write_video(segments / "raw_0002.mp4", frames=4)
with pytest.raises(bare_video.BareVideoError) as failure:
bare_video.resolve_video(segments / "raw_0001.mp4")
assert "raw_0001.mp4" in str(failure.value) and "raw_0002.mp4" in str(failure.value)
assert "split recording" in str(failure.value)
# A folder is read as "the video in here", and a folder of two is the same refusal.
with pytest.raises(bare_video.BareVideoError) as folder_failure:
bare_video.resolve_video(segments)
assert "holds 2 video files" in str(folder_failure.value)
def test_a_lone_video_in_a_folder_is_that_video(tmp_path):
"""One file is unambiguous, however it is pointed at."""
folder = tmp_path / "single"
folder.mkdir()
only = write_video(folder / "clip.mp4")
assert bare_video.resolve_video(folder) == only
assert bare_video.resolve_video(only) == only
def test_something_that_is_not_a_video_is_refused_with_its_reason(tmp_path):
text = tmp_path / "notes.txt"
text.write_text("notes", encoding="utf-8")
with pytest.raises(bare_video.BareVideoError) as failure:
bare_video.resolve_video(text)
assert "not a video" in str(failure.value)
with pytest.raises(bare_video.BareVideoError):
bare_video.resolve_video(tmp_path / "absent.mp4")
empty = tmp_path / "empty"
empty.mkdir()
with pytest.raises(bare_video.BareVideoError):
bare_video.resolve_video(empty)
def test_a_container_without_a_timeline_is_refused(tmp_path):
"""No frame count or rate means every ``t_rel_s`` would be a guess."""
broken = tmp_path / "broken.mp4"
broken.write_bytes(b"not a container")
with pytest.raises(bare_video.BareVideoError):
bare_video.probe(broken)
def test_a_profile_with_no_chain_is_listed_with_its_own_reason(tmp_path):
"""Nothing is hidden: every entry is offered, and the refused ones say why.
Two different refusals, from two owners — the SDK's validation and this app's chain
table — and both have to reach the operator as a sentence rather than as an entry that
is simply missing from the list.
"""
entries = [
_Entry(family="box", ref="shipped/box"),
_Entry(family="finger", ref="shipped/finger"),
_Entry(family="finger-mems", ref="shipped/finger-mems", name="MEMS"),
_Entry(
family="box",
ref="user/box",
name="Box (user)",
blocked="the SDK rejected this profile (failed)",
),
]
assert [entry.ref for entry in bare_video.profiles_for_video(entries)] == [
"shipped/box",
"shipped/finger",
"shipped/finger-mems",
"user/box",
]
assert bare_video.setup_blocked_reason(entries[0]) is None
assert bare_video.setup_blocked_reason(entries[1]) is None
assert "finger-mems" in str(bare_video.setup_blocked_reason(entries[2]))
assert bare_video.setup_blocked_reason(entries[3]) == (
"the SDK rejected this profile (failed)"
), "the SDK's own verdict must win over this app's chain limit"
# A folder no bundle could be read from has no profile behind it, so there is nothing
# to process with — and nothing to refuse *by* either.
assert bare_video.profiles_for_video([_Entry(has_profile=False)]) == []
with pytest.raises(bare_video.BareVideoError) as failure:
bare_video.build_workspace(
write_video(tmp_path / "clip.mp4"), profile=_Entry(family="uv_unroll")
)
assert "no chain" in str(failure.value)
# ------------------------------------------------------------------------- the files
def test_nothing_is_written_beside_the_operator_s_video(video, workspace_root):
"""The frame index is the app's derived data, and the video's folder is the operator's."""
before = _entry_set(video.parent)
workspace = _build(video)
assert _entry_set(video.parent) == before, "the workspace wrote into the video's folder"
assert workspace.directory.parent == workspace_root
assert sorted(path.name for path in workspace.directory.iterdir()) == [
"frames.csv",
"raw_0001.mp4",
"session.json",
]
def test_the_video_inside_the_workspace_is_the_operator_s_own_file(video, workspace_root):
"""A link when the filesystem allows one, and the copy is the same bytes when it does not."""
workspace = _build(video)
inside = workspace.directory / "raw_0001.mp4"
assert inside.read_bytes() == video.read_bytes()
mode = workspace.session["synthetic"]["video_in_workspace"]
assert mode in ("link", "copy", "existing")
# Rebuilding finds the same file rather than stacking a second copy.
again = _build(video)
assert again.session["synthetic"]["video_in_workspace"] == "existing"
def test_the_frame_index_is_written_as_the_columns_the_reader_wants(video, workspace_root):
workspace = _build(video)
with (workspace.directory / "frames.csv").open(encoding="utf-8", newline="") as handle:
rows = list(csv.DictReader(handle))
assert list(rows[0]) == list(bare_video.FRAME_COLUMNS)
assert len(rows) == FRAMES
assert rows[0]["frame_id"] == "1"
assert rows[0]["in_video"] == "1"
assert rows[0]["dropped"] == "0"
assert rows[0]["acquired_at"] == ""
# ------------------------------------------------------------------ the force channel
def test_a_frame_carries_its_force_sample_when_the_pass_collected_one():
"""Six values and two verdicts, appended last so an older reader still reads the frame."""
sample = ForceSample(
normal=1.5,
shear_x=-0.25,
shear_y=0.0,
normal_raw=1200.0,
shear_x_raw=-30.0,
shear_y_raw=0.0,
valid=True,
calibrated=True,
)
record = encode_frame(
frame_id=7,
t_rel_s=0.25,
grid_id=0,
dx=[1.0, 2.0],
dy=[0.0, 0.0],
magnitude=[1.0, 2.0],
active=[True, False],
force=sample,
)
decoded = decode_frame(record)
assert decoded.frame_id == 7
assert decoded.force == sample
assert decoded.force.newtons() == (1.5, -0.25, 0.0)
assert decoded.force.raw() == (1200.0, -30.0, 0.0)
assert decoded.force.calibrated is True
# A frame the pass did not collect force for carries none, and says so by absence.
without = decode_frame(
encode_frame(
frame_id=8,
t_rel_s=0.5,
grid_id=0,
dx=[1.0],
dy=[0.0],
magnitude=[1.0],
active=[True],
)
)
assert without.force is None
# A readout the chain could not make is stored as "no measurement", never as zero.
empty = ForceSample.from_reading(
{
"normal_force": float("nan"),
"shear_x": float("nan"),
"shear_y": float("nan"),
"valid": False,
"force_calibrated": False,
}
)
assert not empty.any_finite and not empty.valid
restored = decode_frame(
encode_frame(
frame_id=9,
t_rel_s=0.75,
grid_id=0,
dx=[1.0],
dy=[0.0],
magnitude=[1.0],
active=[True],
force=empty,
)
).force
assert restored is not None and not restored.any_finite and restored.valid is False
def test_depth_does_not_swallow_the_force_sample_that_follows_it():
"""Force is appended after depth. The depth codec must stop at its own payload."""
import numpy as np
sample = ForceSample(
normal=0.5,
shear_x=0.25,
shear_y=-0.125,
normal_raw=80.0,
shear_x_raw=10.0,
shear_y_raw=-4.0,
valid=True,
calibrated=False,
)
sparse = np.zeros((16, 16), dtype=np.float32)
sparse[2, 3] = 0.03
noise = np.linspace(0.0, 1.0, 16 * 16, dtype=np.float32).reshape(16, 16)
for depth in (sparse, noise, np.zeros((8, 8), dtype=np.float32)):
decoded = decode_frame(
encode_frame(
frame_id=3,
t_rel_s=0.1,
grid_id=0,
dx=[1.0],
dy=[0.0],
magnitude=[1.0],
active=[True],
depth=depth,
force=sample,
)
)
assert decoded.force == sample
assert decoded.depth is not None
restored = decoded.depth_map()
assert restored is not None and restored.shape == depth.shape