ykcheungab
c0a10e6adb
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.
506 lines
20 KiB
Python
506 lines
20 KiB
Python
"""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
|