ykcheungab
d4b66c65a1
Drop orisys.util.draw_arrows; box live and replay both use _paint_shared_flow_arrows with sdk_box_policy when no policy is passed. Recovered box flow uses the same inverse as finger. Verified with shared/tests/test_image_backend.py arrow/idle/recovered cases.
782 lines
31 KiB
Python
782 lines
31 KiB
Python
"""T1: a bare video, analysed end to end, through the window's own action.
|
|
|
|
The workflow this defends is the minimum one an operator needs today: point Studio at a
|
|
video that has no recording beside it, answer one prompt, and get a **measurement** — arrows,
|
|
a depth map and a force trace that can be scrubbed in both directions, because they are read
|
|
from a stored result rather than from a re-run of the chain.
|
|
|
|
What makes it a *bare* video is the input: the clip is copied into a folder of its own with
|
|
nothing else in it (no ``forces.csv``, no ``frames.csv``, no ``session.json``). What makes it
|
|
analysable is the workspace the app generates for it (``analysis.bare_video``), written under
|
|
the app's own derived-data root — which this module points at a temporary directory, so the
|
|
test can assert *where* it went instead of trusting it.
|
|
|
|
Where a failure would be invisible:
|
|
|
|
* the prompt's answers decide what is measured, so the profile, the calibration and the
|
|
reference frame are read back from the dialog's own widgets;
|
|
* a recomputed force trace and a recorded one are different claims, so the charts' source
|
|
kind and the provenance's ``sources``/``synthetic`` are asserted, not just the pixels;
|
|
* a box result is recorded on the box sample grid, so the canvas is compared with
|
|
``draw_arrow_display`` on the recovered field.
|
|
|
|
The fixture clip is the project's recorded box footage under ``dist/``, copied into the test's
|
|
own directory. If it is absent the test **fails** (``_require_clip``) rather than skipping: a
|
|
silently skipped end-to-end test is how a broken workflow ships.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from orisys_app_shared import paths as paths_module
|
|
from orisys_app_shared.acquisition.arrows import (
|
|
depth_map_to_bgr,
|
|
draw_arrow_display,
|
|
flow_colormap,
|
|
hex_to_bgr,
|
|
)
|
|
from orisys_app_shared.acquisition.image_backend import (
|
|
ARROW_PURE,
|
|
ARROW_STITCHED,
|
|
DEPTH,
|
|
RECORDED,
|
|
RESULT_FAILED,
|
|
SOURCE_REPLAY,
|
|
ImageFrame,
|
|
render_image,
|
|
)
|
|
from orisys_app_shared.analysis import bare_video
|
|
from orisys_app_shared.analysis.cache import scan_results
|
|
from orisys_app_shared.i18n import t
|
|
from orisys_app_shared.scene3d.contracts import DisplaySettings
|
|
from orisys_app_shared.sensors import (
|
|
CatalogEntry,
|
|
SensorCatalog,
|
|
UserRootDecision,
|
|
scan_sensor_catalog,
|
|
)
|
|
from orisys_app_shared.theme import color
|
|
from PySide6.QtWidgets import QApplication, QDialog, QDialogButtonBox
|
|
|
|
from orisys_studio.ui.video_analysis_dialog import VideoAnalysisDialog
|
|
|
|
APP = Path(__file__).resolve().parents[1]
|
|
REPO = APP.parents[1]
|
|
EVIDENCE = (
|
|
REPO
|
|
/ "dist"
|
|
/ "OrisysStudio"
|
|
/ "outputs"
|
|
/ "recordings"
|
|
/ "20260930T015028Z_session0002_40cb8e7441434feab48bb646afdba011"
|
|
)
|
|
SOURCE_CLIP = EVIDENCE / "raw_0001.mp4"
|
|
|
|
|
|
def _require_clip() -> Path:
|
|
"""The one real input: recorded box footage, copied rather than faked.
|
|
|
|
A synthetic clip cannot stand in for it. The box chain's stitch probe finds its lenslet
|
|
grid in a real image; on generated pixels it does not converge — measured, a synthetic
|
|
896² clip sent the chain into minutes of work where the real 33-frame clip takes two
|
|
seconds. So the input is the project's own recording, and its absence is a **failure**:
|
|
skipping would leave "a bare video can be analysed" untested, which is the state this
|
|
task exists to change.
|
|
"""
|
|
assert SOURCE_CLIP.is_file(), (
|
|
"the bare-video test needs recorded box footage at "
|
|
f"{SOURCE_CLIP}; copy it there or run the build that produces dist/"
|
|
)
|
|
pytest.importorskip("orisys", reason="the analysis pass drives the SDK chain")
|
|
return SOURCE_CLIP
|
|
|
|
|
|
# ------------------------------------------------------------------------ fixtures
|
|
@pytest.fixture(scope="module")
|
|
def cache_root(tmp_path_factory):
|
|
"""The analysis cache for this module, in place of the user's own."""
|
|
root = tmp_path_factory.mktemp("bare-video-cache") / "analysis-cache"
|
|
root.mkdir(parents=True)
|
|
patch = pytest.MonkeyPatch()
|
|
patch.setattr(paths_module, "analysis_cache_dir", lambda: root)
|
|
yield root
|
|
patch.undo()
|
|
|
|
|
|
@pytest.fixture
|
|
def own_cache(tmp_path, monkeypatch) -> Path:
|
|
"""A cache root this test alone owns, so its first open really does run a pass."""
|
|
root = tmp_path / "own-cache"
|
|
root.mkdir()
|
|
monkeypatch.setattr(paths_module, "analysis_cache_dir", lambda: root)
|
|
return root
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def workspace_root(tmp_path_factory):
|
|
"""The app-owned root the generated workspace goes to, in place of the user's own."""
|
|
root = tmp_path_factory.mktemp("bare-video-workspaces") / "video-workspaces"
|
|
patch = pytest.MonkeyPatch()
|
|
patch.setattr(paths_module, "video_workspace_dir", lambda: root)
|
|
yield root
|
|
patch.undo()
|
|
|
|
|
|
@pytest.fixture
|
|
def bare_clip(tmp_path) -> Path:
|
|
"""One video, alone in a folder: a bare video and nothing else."""
|
|
folder = tmp_path / "operator-footage"
|
|
folder.mkdir()
|
|
clip = folder / "session-clip.mp4"
|
|
shutil.copy2(_require_clip(), clip)
|
|
return clip
|
|
|
|
|
|
@pytest.fixture
|
|
def window(make_window, qt_app, cache_root, workspace_root):
|
|
"""The application, in the replay view, as the operator meets it."""
|
|
view = make_window(sensor_type="box", source_mode="camera")
|
|
view.resize(1440, 880)
|
|
view.show()
|
|
QApplication.processEvents()
|
|
view.brand_bar.btn_playback.click()
|
|
QApplication.processEvents()
|
|
assert view.layout_ctl.in_replay_view(), "the window did not switch to the replay view"
|
|
return view
|
|
|
|
|
|
# ------------------------------------------------------------------------- helpers
|
|
def answer_prompt(
|
|
monkeypatch,
|
|
*,
|
|
profile: str = "shipped/box",
|
|
calibration: str = "default",
|
|
reference: int | None = None,
|
|
accept: bool = True,
|
|
seen: list | None = None,
|
|
):
|
|
"""Answer the setup prompt through the real dialog's own widgets.
|
|
|
|
The dialog class is the real one and the window builds it; only ``exec`` is replaced, so
|
|
what a test answers is exactly what an operator could have answered, and the defaults can
|
|
be read back from the widgets the operator sees.
|
|
"""
|
|
|
|
def _answer(dialog_self):
|
|
if seen is not None:
|
|
seen.append(dialog_self)
|
|
assert isinstance(dialog_self, VideoAnalysisDialog)
|
|
assert dialog_self.select_profile(profile), (
|
|
f"the prompt does not offer {profile}: "
|
|
f"{[entry.ref for entry in dialog_self.entries]}"
|
|
)
|
|
if reference is not None:
|
|
dialog_self.reference_spin.setValue(int(reference))
|
|
if calibration == "none":
|
|
index = dialog_self.calibration_combo.findData("none")
|
|
assert index >= 0
|
|
dialog_self.calibration_combo.setCurrentIndex(index)
|
|
elif calibration != "default":
|
|
dialog_self.select_calibration(Path(calibration))
|
|
return QDialog.DialogCode.Accepted if accept else QDialog.DialogCode.Rejected
|
|
|
|
monkeypatch.setattr(VideoAnalysisDialog, "exec", _answer)
|
|
|
|
|
|
def panel_bgr(window) -> np.ndarray:
|
|
"""The 2D panel's picture, in BGR, from the end of the real display path."""
|
|
image = window.arrow_panel._source_image
|
|
assert image is not None and not image.isNull(), "the 2D panel shows no picture"
|
|
width, height = image.width(), image.height()
|
|
rows = np.frombuffer(bytes(image.constBits()), dtype=np.uint8)
|
|
rows = rows.reshape(height, image.bytesPerLine())
|
|
return rows[:, : width * 3].reshape(height, width, 3)[:, :, ::-1].copy()
|
|
|
|
|
|
def sink_frames(window, monkeypatch) -> list:
|
|
captured: list = []
|
|
original = window._show_image_frame
|
|
|
|
def spy(frame):
|
|
captured.append(frame)
|
|
return original(frame)
|
|
|
|
monkeypatch.setattr(window, "_show_image_frame", spy)
|
|
return captured
|
|
|
|
|
|
def set_mode(window, mode: str) -> None:
|
|
index = window.arrow_panel.view_combo.findData(mode)
|
|
assert index >= 0, f"the mode {mode} is not offered: {window.arrow_panel.view_modes()}"
|
|
window.arrow_panel.view_combo.setCurrentIndex(index)
|
|
QApplication.processEvents()
|
|
assert window.arrow_panel.view_mode() == mode
|
|
|
|
|
|
def canvas_bgr(window) -> tuple[int, int, int]:
|
|
return hex_to_bgr(color(window._theme, "arrow_canvas", "#0B1830"))
|
|
|
|
|
|
def painted(image, canvas) -> int:
|
|
array = np.asarray(image)
|
|
return int(np.count_nonzero(np.any(array != np.asarray(canvas, np.uint8), axis=-1)))
|
|
|
|
|
|
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
|
|
|
|
|
|
def open_bare_video(window, clip: Path, monkeypatch, **answer) -> None:
|
|
"""The operator's sequence up to "the pass is running": open, and answer the prompt."""
|
|
answer_prompt(monkeypatch, **answer)
|
|
assert window.replay_ctl.load_video(clip) is True, "the video did not open"
|
|
QApplication.processEvents()
|
|
|
|
|
|
def settle(window, *, timeout_s: float = 300.0) -> None:
|
|
"""Let a pass finish if one is running, and check that a result came out of it."""
|
|
if window.replay_ctl.analysis_running():
|
|
assert window.replay_ctl.wait_for_analysis(timeout_s), "the pass did not finish"
|
|
QApplication.processEvents()
|
|
assert window.replay_ctl.analysis_error() == "", (
|
|
f"the pass left no usable result: {window.replay_ctl.analysis_error()}"
|
|
)
|
|
assert window.replay_ctl.available_result() is not None, "no stored result was adopted"
|
|
|
|
|
|
# --------------------------------------------------------------------- the workflow
|
|
def test_a_bare_video_is_analysed_and_scrubbed_in_both_directions(
|
|
bare_clip, window, monkeypatch, own_cache, workspace_root
|
|
):
|
|
"""Open a bare video → answer the prompt → the pass runs → scrub it backwards."""
|
|
before = entry_set(bare_clip.parent)
|
|
captured = sink_frames(window, monkeypatch)
|
|
seen: list = []
|
|
open_bare_video(window, bare_clip, monkeypatch, seen=seen)
|
|
|
|
assert window.replay_ctl.analysis_running(), (
|
|
"no pass was started for a bare video with no stored result"
|
|
)
|
|
assert window.analysis_progress.state() == "running"
|
|
settle(window)
|
|
|
|
# ---- the prompt's fields and their defaults -----------------------------
|
|
dialog = seen[0]
|
|
assert [entry.ref for entry in dialog.entries] == [
|
|
entry.ref for entry in bare_video.profiles_for_video(scan_sensor_catalog().entries)
|
|
], "the prompt offered a different set of profiles than the catalog can analyse with"
|
|
assert "flow" in dialog.capability_label.text(), (
|
|
"the row does not say what the profile can do"
|
|
)
|
|
assert dialog.calibration_combo.currentData() == "default"
|
|
assert dialog.calibration_combo.currentText() == t(
|
|
"video_analysis.calibration_default", name="force_calibration.json"
|
|
), "the calibration did not default to the profile's declared one"
|
|
assert dialog.reference_spin.value() == 1, (
|
|
"the reference frame did not default to the first"
|
|
)
|
|
chosen = dialog.choice()
|
|
assert chosen is not None and chosen.reference_frame is None, (
|
|
"the first frame must stay the pass's own default, not a reset it cannot apply"
|
|
)
|
|
assert chosen.calibration is not None and chosen.calibration.name == "force_calibration.json"
|
|
|
|
# ---- the workspace, and where it went -----------------------------------
|
|
workspace = Path(window.replay_ctl.recording_directory() or "")
|
|
assert workspace.is_dir(), "the pass was not given a generated workspace"
|
|
assert workspace.parent == workspace_root, f"the workspace went to {workspace.parent}"
|
|
assert workspace.parent != bare_clip.parent
|
|
assert sorted(path.name for path in workspace.iterdir()) == [
|
|
"frames.csv",
|
|
"raw_0001.mp4",
|
|
"session.json",
|
|
]
|
|
assert entry_set(bare_clip.parent) == before, (
|
|
"opening and analysing a bare video wrote beside the operator's footage"
|
|
)
|
|
|
|
session = window.replay_ctl.available_result()
|
|
provenance = session.reader.provenance
|
|
assert provenance.synthetic is True, (
|
|
"the result does not say its input was a generated workspace"
|
|
)
|
|
assert provenance.synthetic_source == str(bare_clip), (
|
|
"the provenance does not name the video the numbers were computed from"
|
|
)
|
|
assert provenance.sources["flow"] == "recomputed"
|
|
assert provenance.sources["forces"] == "recomputed", (
|
|
"a recomputed force trace was recorded as a recorded measurement"
|
|
)
|
|
|
|
# ---- the charts: the recomputed force trace -----------------------------
|
|
assert window.force_plots.curve_kind() == "analysis", (
|
|
"the charts are not reading the result the pass wrote"
|
|
)
|
|
assert window.force_plots.is_recording_source()
|
|
assert window.force_plots.mode_label.text() == t("chart.mode_recomputed"), (
|
|
"the charts do not say that this trace is a recomputation"
|
|
)
|
|
table = session.reader.forces()
|
|
assert table is not None and len(table) > 1, "the result stores no force at all"
|
|
assert np.isfinite(table.newtons).any(), "the stored force trace holds no number"
|
|
unit = window.force_plots.unit()
|
|
xs, ys = window.force_plots.curves[0].getData()
|
|
assert xs is not None and len(xs) == len(table), "the FN curve does not hold the trace"
|
|
assert np.allclose(
|
|
np.asarray(ys, float), table.value_for_unit(unit)[:, 0], equal_nan=True
|
|
)
|
|
|
|
# ---- the 2D canvas: arrows and depth, from the cache --------------------
|
|
canvas = canvas_bgr(window)
|
|
ids = [int(value) for value in session.frame_ids]
|
|
last_id, middle_id = ids[-1], ids[len(ids) // 2]
|
|
assert session.frame(last_id).active_count > 0
|
|
|
|
set_mode(window, ARROW_PURE)
|
|
window.replay.set_position(session.time_of(middle_id))
|
|
QApplication.processEvents()
|
|
assert captured[-1].frame_id == middle_id
|
|
assert captured[-1].source == SOURCE_REPLAY
|
|
assert captured[-1].result_available is True
|
|
|
|
# ...then scrub *backwards*, from the end of the timeline to that same frame: it must
|
|
# resolve to the stored frame for the position, drawn from the cache, with no pass.
|
|
window.replay.set_position(window.replay.duration_s() + 1.0)
|
|
QApplication.processEvents()
|
|
assert captured[-1].frame_id == last_id, (
|
|
"the cursor at the end of the timeline did not resolve to the last stored frame"
|
|
)
|
|
window.replay.set_position(session.time_of(middle_id))
|
|
QApplication.processEvents()
|
|
assert captured[-1].frame_id == middle_id, (
|
|
"scrubbing backwards did not resolve to the stored frame at that position"
|
|
)
|
|
assert not window.replay_ctl.analysis_running(), "scrubbing ran an analysis pass"
|
|
middle_shot = panel_bgr(window)
|
|
|
|
frame = session.frame(middle_id)
|
|
policy = session.paint_policy()
|
|
expected = render_image(
|
|
ImageFrame(
|
|
frame.base_bgr,
|
|
flow=frame.flow,
|
|
depth=frame.depth,
|
|
valid_mask=frame.valid_mask,
|
|
source=SOURCE_REPLAY,
|
|
frame_id=int(frame.frame_id),
|
|
flow_valid=True,
|
|
depth_valid=bool(frame.has_depth),
|
|
result_available=True,
|
|
arrow_policy=policy,
|
|
),
|
|
mode=ARROW_PURE,
|
|
theme=window._theme,
|
|
draw_style=window._draw_style(),
|
|
canvas_bgr=canvas,
|
|
domain_mask=window._domain_mask.mask,
|
|
arrow_policy=policy,
|
|
)
|
|
assert expected is not None
|
|
assert painted(expected, canvas) > 0, "the cached arrows painted nothing"
|
|
assert np.array_equal(middle_shot, np.asarray(expected)), (
|
|
"the panel's arrow canvas is not the stored frame's own drawing"
|
|
)
|
|
raw = window.replay.video_frame_bgr()
|
|
assert raw is not None and not np.array_equal(middle_shot, np.asarray(raw)), (
|
|
"the arrow view is the video's raw picture"
|
|
)
|
|
|
|
# ---- overlay: arrows on the captured frame, not on the black placeholder -
|
|
set_mode(window, ARROW_STITCHED)
|
|
QApplication.processEvents()
|
|
overlay = panel_bgr(window)
|
|
assert painted(overlay, canvas) > 0, "the overlay painted no arrows"
|
|
assert float(np.mean(overlay)) > 8.0, (
|
|
"the overlay is still a near-black canvas; the captured frame was not under the arrows"
|
|
)
|
|
assert not np.array_equal(overlay, middle_shot), (
|
|
"the overlay view is the pure-colour canvas"
|
|
)
|
|
|
|
# ---- the depth view, from the same stored frame -------------------------
|
|
set_mode(window, DEPTH)
|
|
QApplication.processEvents()
|
|
expected_depth = depth_map_to_bgr(
|
|
frame.depth,
|
|
flow_colormap(window._theme),
|
|
value_range=window._depth_range(),
|
|
invalid_mask=window._domain_mask.mask,
|
|
)
|
|
assert expected_depth is not None
|
|
assert np.array_equal(panel_bgr(window), expected_depth), (
|
|
"the depth view is not depth_map_to_bgr of the stored frame"
|
|
)
|
|
|
|
# ---- the recording's own picture is still one step away -----------------
|
|
set_mode(window, RECORDED)
|
|
assert np.array_equal(panel_bgr(window), np.asarray(raw))
|
|
assert window.arrow_panel.note_label.text() == "", (
|
|
"the panel says there is nothing to draw while it is drawing a result"
|
|
)
|
|
|
|
|
|
def test_the_box_arrow_canvas_is_painted_by_the_live_painter(
|
|
bare_clip, window, monkeypatch
|
|
):
|
|
"""A box pass stores the box sample grid; the panel redraws with that policy.
|
|
|
|
The recovered field is NaN off the recorded cells. The shared selector only samples
|
|
finite cells, so the panel matches ``draw_arrow_display`` on that sparse field.
|
|
"""
|
|
open_bare_video(window, bare_clip, monkeypatch)
|
|
settle(window)
|
|
session = window.replay_ctl.available_result()
|
|
canvas = canvas_bgr(window)
|
|
policy = session.paint_policy()
|
|
assert int(policy.flow_grid_spacing) == 20
|
|
assert abs(float(policy.flow_arrow_scale) - 0.5) < 1e-9
|
|
common = {
|
|
"background": "pure_color",
|
|
"canvas_bgr": canvas,
|
|
"draw_style": window._draw_style(),
|
|
"colormap": flow_colormap(window._theme),
|
|
"policy": policy,
|
|
}
|
|
|
|
set_mode(window, ARROW_PURE)
|
|
first = next(frame for frame in session.frames() if frame.active_count > 0)
|
|
window.replay.set_position(session.time_of(first.frame_id))
|
|
QApplication.processEvents()
|
|
shown = panel_bgr(window)
|
|
assert painted(shown, canvas) > 0, "the operator's arrow canvas is blank"
|
|
expected = draw_arrow_display(
|
|
first.base_bgr,
|
|
first.flow,
|
|
valid_mask=first.valid_mask,
|
|
**common,
|
|
)
|
|
assert np.array_equal(shown, np.asarray(expected)), (
|
|
"the panel's canvas is not the shared painter's output for this frame"
|
|
)
|
|
|
|
|
|
def test_declining_the_prompt_leaves_the_video_playing_with_nothing_promised(
|
|
bare_clip, window, monkeypatch
|
|
):
|
|
"""Cancelling is a state, not a crash: the picture plays and the panel says no pass ran."""
|
|
before = entry_set(bare_clip.parent)
|
|
captured = sink_frames(window, monkeypatch)
|
|
open_bare_video(window, bare_clip, monkeypatch, accept=False)
|
|
|
|
assert window.replay.has_video(), "declining the prompt closed the video"
|
|
assert window.replay_ctl.available_result() is None
|
|
assert window.replay_ctl.recording_directory() is None, (
|
|
"a dismissed prompt still claimed a workspace as this timeline's input"
|
|
)
|
|
assert not window.replay_ctl.analysis_running(), "a pass ran after the prompt was dismissed"
|
|
state, reason = window.replay_ctl.result_state()
|
|
assert state == RESULT_FAILED
|
|
assert reason == t("playback.analysis_setup_declined")
|
|
assert captured and captured[-1].result_state == RESULT_FAILED
|
|
assert entry_set(bare_clip.parent) == before
|
|
|
|
|
|
def test_a_split_recording_is_refused_with_a_clear_message(
|
|
bare_clip, window, monkeypatch
|
|
):
|
|
"""A split recording is a measurement error: refuse it rather than half-analyse it.
|
|
|
|
The files are named the way a split recording is (``take_0001.mp4`` beside
|
|
``take_0002.mp4``), and the chosen one is opened as a video first — the refusal comes
|
|
from the setup step, which is where "what exactly is this input" is asked.
|
|
"""
|
|
folder = bare_clip.parent
|
|
bare_clip.unlink()
|
|
chosen = folder / "take_0001.mp4"
|
|
shutil.copy2(_require_clip(), chosen)
|
|
shutil.copy2(_require_clip(), folder / "take_0002.mp4")
|
|
captured = sink_frames(window, monkeypatch)
|
|
|
|
open_bare_video(window, chosen, monkeypatch)
|
|
|
|
assert not window.replay_ctl.analysis_running(), "a split recording was analysed anyway"
|
|
_state, reason = window.replay_ctl.result_state()
|
|
assert "split recording" in reason, f"the refusal does not say why: {reason!r}"
|
|
assert "take_0001.mp4" in reason and "take_0002.mp4" in reason
|
|
assert window.replay.has_video(), "the refusal closed the video"
|
|
assert captured, "the refusal left the panel with nothing"
|
|
|
|
|
|
def test_reopening_the_video_reuses_the_result_and_runs_no_second_pass(
|
|
bare_clip, window, monkeypatch, cache_root
|
|
):
|
|
"""The cache is the point: a second open is a file read, not a second pass.
|
|
|
|
``recording_identity`` is derived from the video and the chosen profile, so a workspace
|
|
generated again is the *same* input — the key does not move and no pass runs.
|
|
"""
|
|
open_bare_video(window, bare_clip, monkeypatch)
|
|
settle(window)
|
|
directories = {entry.directory for entry in scan_results(cache_root)}
|
|
assert directories, "the first open stored no result"
|
|
workspace = window.replay_ctl.recording_directory()
|
|
|
|
window.replay.close()
|
|
QApplication.processEvents()
|
|
assert window.replay_ctl.recording_directory() is None
|
|
|
|
started: list = []
|
|
original_start = window.analysis_progress.start
|
|
|
|
def spy_start(*args, **kwargs):
|
|
started.append(True)
|
|
return original_start(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(window.analysis_progress, "start", spy_start)
|
|
open_bare_video(window, bare_clip, monkeypatch)
|
|
settle(window)
|
|
|
|
assert not started, "a second pass ran for a video that already had a result"
|
|
assert window.analysis_progress.state() == "reused"
|
|
assert {entry.directory for entry in scan_results(cache_root)} == directories
|
|
assert window.replay_ctl.recording_directory() == workspace, (
|
|
"the second open generated a different workspace"
|
|
)
|
|
assert window.force_plots.curve_kind() == "analysis"
|
|
|
|
|
|
def test_a_setting_that_moves_the_grid_redraws_and_writes_nothing(
|
|
bare_clip, window, monkeypatch, cache_root
|
|
):
|
|
"""A display setting is a read; a grid setting cannot be served and must not blank it."""
|
|
open_bare_video(window, bare_clip, monkeypatch)
|
|
settle(window)
|
|
session = window.replay_ctl.available_result()
|
|
set_mode(window, ARROW_PURE)
|
|
before = {entry.directory for entry in scan_results(cache_root)}
|
|
captured = sink_frames(window, monkeypatch)
|
|
|
|
window.replay_ctl.on_display_settings(DisplaySettings(flow_on_px=0.9))
|
|
QApplication.processEvents()
|
|
assert captured, "the threshold change drew nothing"
|
|
assert session.threshold() == pytest.approx(0.9)
|
|
assert {entry.directory for entry in scan_results(cache_root)} == before, (
|
|
"a display change wrote an analysis result"
|
|
)
|
|
|
|
window.replay_ctl.on_display_settings(DisplaySettings(max_arrows=50))
|
|
QApplication.processEvents()
|
|
policy = captured[-1].arrow_policy
|
|
assert policy is not None, (
|
|
"the frame does not say which policy its recovered field can be sampled with"
|
|
)
|
|
assert policy.max_arrows == session.paint_policy().max_arrows, (
|
|
"the display's grid setting reached the recovered field's sampling policy"
|
|
)
|
|
assert painted(panel_bgr(window), canvas_bgr(window)) > 0, (
|
|
"a refused grid setting blanked the canvas"
|
|
)
|
|
assert {entry.directory for entry in scan_results(cache_root)} == before
|
|
|
|
|
|
def test_the_workspace_carries_the_profile_the_operator_chose(
|
|
bare_clip, window, monkeypatch
|
|
):
|
|
"""The prompt's answer is what the pass is driven with — read back from the workspace."""
|
|
open_bare_video(window, bare_clip, monkeypatch, calibration="none")
|
|
settle(window)
|
|
workspace = Path(window.replay_ctl.recording_directory())
|
|
document = json.loads((workspace / "session.json").read_text(encoding="utf-8"))
|
|
|
|
assert document["device"]["sensor_type"] == "box"
|
|
config_ref = Path(document["device"]["config_ref"])
|
|
assert config_ref.name == "sensor.json" and "sensors" in config_ref.parts
|
|
assert document["synthetic"]["profile_ref"] == "shipped/box"
|
|
assert document["synthetic"]["calibration"] == ""
|
|
assert document["calibration_id"] == "", (
|
|
"choosing raw force must not leave a calibration in the result's key"
|
|
)
|
|
assert document["stats"]["segments"] == 1
|
|
assert (workspace / "frames.csv").is_file()
|
|
# The chain was told, too: no calibration means the raw triplet, not a substitute curve.
|
|
assert window.replay_ctl.available_result() is not None
|
|
|
|
|
|
# ------------------------------------------------------------------- the prompt itself
|
|
def _catalog(entry: CatalogEntry) -> SensorCatalog:
|
|
return SensorCatalog(
|
|
roots=(),
|
|
entries=(entry,),
|
|
user_root_decision=UserRootDecision(
|
|
path=Path("."), kind="explicit", writable=True, reason="test"
|
|
),
|
|
)
|
|
|
|
|
|
def _blocked_entry() -> CatalogEntry:
|
|
"""The shipped box bundle as the catalog reports one the SDK's validation rejected.
|
|
|
|
Built from the *real* entry so the profile behind it is real too: the verdict is what is
|
|
replaced, and the reason is the sentence the SDK would have produced.
|
|
"""
|
|
import dataclasses
|
|
|
|
real = next(
|
|
entry for entry in scan_sensor_catalog().entries if entry.folder == "box"
|
|
)
|
|
return dataclasses.replace(
|
|
real,
|
|
valid=False,
|
|
state="failed",
|
|
reason="assets.computational-missing: the required unroll map is missing",
|
|
)
|
|
|
|
|
|
def test_a_profile_the_sdk_rejected_cannot_be_chosen_and_says_why(
|
|
bare_clip, window, qt_app, tmp_path
|
|
):
|
|
"""The catalog's ``start_blocked_reason`` reaches the prompt instead of no caller."""
|
|
facts = bare_video.probe(bare_video.resolve_video(bare_clip))
|
|
dialog = VideoAnalysisDialog(
|
|
video=bare_clip, facts=facts, catalog=_catalog(_blocked_entry()), window=window
|
|
)
|
|
try:
|
|
entry = dialog.entries[0]
|
|
blocked = bare_video.setup_blocked_reason(entry)
|
|
assert blocked is not None and "unroll map" in blocked
|
|
item = dialog.profile_combo.model().item(0)
|
|
assert not item.isEnabled(), "a profile the SDK rejected is offered as startable"
|
|
assert not dialog.buttons.button(QDialogButtonBox.StandardButton.Ok).isEnabled(), (
|
|
"OK is offered for a profile that cannot be started"
|
|
)
|
|
assert dialog.note_label.text() == t("video_analysis.blocked", reason=blocked), (
|
|
"the prompt does not carry the SDK's own reason"
|
|
)
|
|
assert dialog.choice() is None, "the refused profile produced a setup anyway"
|
|
finally:
|
|
dialog.deleteLater()
|
|
|
|
|
|
def test_the_first_frame_is_the_prompt_s_default_reference(
|
|
bare_clip, window, qt_app
|
|
):
|
|
"""A reference frame that is not asked for must not become a reset the pass cannot make."""
|
|
facts = bare_video.probe(bare_video.resolve_video(bare_clip))
|
|
dialog = VideoAnalysisDialog(
|
|
video=bare_clip, facts=facts, catalog=scan_sensor_catalog(), window=window
|
|
)
|
|
try:
|
|
assert dialog.reference_spin.minimum() == 1
|
|
assert dialog.reference_spin.maximum() == facts.frame_count
|
|
assert dialog.reference_spin.value() == 1
|
|
assert dialog.choice().reference_frame is None
|
|
dialog.reference_spin.setValue(4)
|
|
assert dialog.choice().reference_frame == 4
|
|
assert dialog.reference_spin.value() == 4
|
|
finally:
|
|
dialog.deleteLater()
|
|
|
|
|
|
def test_the_prompt_restores_a_prior_reference_and_no_calibration(
|
|
bare_clip, window, qt_app
|
|
):
|
|
"""Reopening Analysis setup must show the answers from the last pass."""
|
|
facts = bare_video.probe(bare_video.resolve_video(bare_clip))
|
|
dialog = VideoAnalysisDialog(
|
|
video=bare_clip,
|
|
facts=facts,
|
|
catalog=scan_sensor_catalog(),
|
|
window=window,
|
|
initial_profile="shipped/box",
|
|
prefer_no_calibration=True,
|
|
initial_reference_frame=4,
|
|
)
|
|
try:
|
|
assert dialog.reference_spin.value() == 4
|
|
assert dialog.choice().reference_frame == 4
|
|
assert dialog.calibration_combo.currentData() == "none"
|
|
assert dialog.choice().calibration is None
|
|
finally:
|
|
dialog.deleteLater()
|
|
|
|
|
|
def test_analysis_setup_reopens_the_prompt_after_the_first_pass(
|
|
bare_clip, window, monkeypatch
|
|
):
|
|
"""Analysis ▸ Analysis setup… asks again; a new reference frame starts a new pass."""
|
|
open_bare_video(window, bare_clip, monkeypatch, reference=None)
|
|
settle(window)
|
|
first = window.replay_ctl.recording_directory()
|
|
assert first is not None
|
|
assert bare_video.is_synthetic(first)
|
|
|
|
# The box stitch probe consumes the first decoded frames; a reference that names
|
|
# one of them is refused. Frame 10 is past that offset on this clip.
|
|
seen: list = []
|
|
answer_prompt(monkeypatch, profile="shipped/box", reference=10, seen=seen)
|
|
window.replay_ctl.on_analysis_setup()
|
|
settle(window)
|
|
|
|
assert seen, "Analysis setup did not open the dialog"
|
|
second = window.replay_ctl.recording_directory()
|
|
assert second is not None
|
|
block = bare_video.read_synthetic(second)
|
|
assert block is not None
|
|
assert int(block["reference_frame"]) == 10
|
|
assert window.replay_ctl.available_result() is not None
|
|
|
|
|
|
def test_a_finger_calibration_file_must_name_the_same_device(qt_app, tmp_path):
|
|
"""A browsed Finger file is accepted only when the device id matches its SN."""
|
|
import cv2
|
|
|
|
clip = tmp_path / "clip.mp4"
|
|
writer = cv2.VideoWriter(str(clip), cv2.VideoWriter_fourcc(*"mp4v"), 10.0, (32, 24))
|
|
for _ in range(4):
|
|
writer.write(np.zeros((24, 32, 3), dtype=np.uint8))
|
|
writer.release()
|
|
facts = bare_video.probe(bare_video.resolve_video(clip))
|
|
catalog = scan_sensor_catalog()
|
|
finger = next(
|
|
(
|
|
entry
|
|
for entry in catalog.entries
|
|
if str(entry.product_family) == "finger"
|
|
and bare_video.setup_blocked_reason(entry) is None
|
|
),
|
|
None,
|
|
)
|
|
assert finger is not None, "the catalog has no startable Finger profile"
|
|
path = tmp_path / "DD02-14.json"
|
|
path.write_text(
|
|
json.dumps({"meta": {"sensor_type": "finger", "sn": "DD02-14"}}),
|
|
encoding="utf-8",
|
|
)
|
|
dialog = VideoAnalysisDialog(video=clip, facts=facts, catalog=catalog)
|
|
try:
|
|
assert dialog.select_profile(finger.ref)
|
|
dialog.select_calibration(path)
|
|
assert not dialog.device_edit.isHidden()
|
|
assert dialog.device_edit.text() == "DD02-14"
|
|
chosen = dialog.choice()
|
|
assert chosen is not None
|
|
assert chosen.calibration == path
|
|
assert chosen.device_id == "DD02-14"
|
|
dialog.device_edit.setText("OTHER")
|
|
assert dialog.choice() is None
|
|
assert dialog.note_label.text() == t(
|
|
"video_analysis.device_mismatch", sn="DD02-14"
|
|
)
|
|
finally:
|
|
dialog.deleteLater()
|