重构 Autocal 采集页:单一队列与控制器驱动的主按钮

选点、对准、加载与录像分散在两个面板,状态靠隐藏相位判断。现改由不依赖 Qt 的采集控制器驱动,并加入录像选项、汇总平铺副本、保持到时自动结束、会话结束回到空闲,以及无设备时的模拟采集。已运行 Autocal 非 Qt 测试并通过范围内门禁;界面测试待 Windows 复核,未做实机验收。
This commit is contained in:
ykcheungab
2026-10-18 16:04:32 +08:00
parent 85d0b396e3
commit 06e06126ca
44 changed files with 7769 additions and 2218 deletions
+34 -9
View File
@@ -290,6 +290,7 @@ def _write_multipoint_trial(
raw_per_n: float,
samples_per_hold: int,
frame_start: int,
capture_status: str | None = None,
) -> list[dict]:
write_csv(
path / "reference.csv",
@@ -302,14 +303,17 @@ def _write_multipoint_trial(
["t_host_mono_ns", "t_host_utc_ns", "kind", "detail"],
[],
)
write_json_file(path / "session.json", {
manifest = {
"trial_id": f"{site_id}-repeat-{repeat_index:03d}",
"site_id": site_id,
"status": "complete",
"uv_px": {"u": 128, "v": 64},
"images": {"captured": with_capture,
"reason": "" if with_capture else "camera not attached yet"},
})
}
if capture_status is not None:
manifest["capture_status"] = capture_status
write_json_file(path / "session.json", manifest)
if not with_capture:
return []
frames: list[dict] = []
@@ -324,8 +328,13 @@ def _write_multipoint_trial(
)
frames.extend(hold_frames)
forces.extend(hold_forces)
write_capture(path / "capture", frames, forces, fields=None if video else FRAMES_FIELDS,
video_segments=1)
if not video:
# A data-only recording: the rows exist, no frame was written to a video.
frames = [
{**row, "segment": 0, "segment_frame": 0, "in_video": 0} for row in frames
]
write_capture(path / "capture", frames, forces, fields=FRAMES_FIELDS,
video_segments=1 if video else 0)
return []
@@ -343,8 +352,17 @@ def multipoint_session(
simulate: bool = False,
trial_status: str = "complete",
extra_holds: list[Hold] | None = None,
recording_block: dict | None = None,
capture_status: str | None = None,
) -> Path:
"""A complete multi-point session directory through the review contract."""
"""A complete multi-point session directory through the review contract.
``video=False`` writes a data-only recording: frame rows exist with
``in_video=0`` and no MP4 is created. ``recording_block`` writes the root
manifest's ``recording`` block; ``None`` keeps the fixture shaped like a
session recorded before that block existed. ``capture_status`` puts the
lifecycle into the trial records, as the campaign does.
"""
session_id = f"finger-SN1_{root.name}_test0001"
trials: list[dict] = []
completion: list[dict] = []
@@ -379,21 +397,25 @@ def multipoint_session(
raw_per_n=raw_per_n,
samples_per_hold=samples_per_hold,
frame_start=1,
capture_status=capture_status,
)
trials.append({
trial_record = {
"trial_id": f"{site_id}-repeat-{repeat:03d}",
"site_id": site_id,
"repeat_index": repeat,
"directory": directory,
"status": trial_status,
"reason": "" if trial_status == "complete" else "stopped by the operator",
})
}
if capture_status is not None:
trial_record["capture_status"] = capture_status
trials.append(trial_record)
if trial_status == "complete":
complete += 1
completion.append({"site_id": site_id, "repeats_planned": repeats,
"repeats_complete": complete,
"complete": complete >= repeats})
write_json_file(root / "session.json", {
manifest = {
"format_version": "autocal-multipoint-1.0",
"session_id": session_id,
"sensor_model": "finger",
@@ -411,7 +433,10 @@ def multipoint_session(
"[t_start_host_mono_ns, t_end_host_mono_ns). Rows are not paired."},
"platform_motion": False,
"automatic_force_control": False,
})
}
if recording_block is not None:
manifest["recording"] = recording_block
write_json_file(root / "session.json", manifest)
return root
+10
View File
@@ -34,6 +34,14 @@ FORBIDDEN_BINDINGS = {"PyQt5", "PyQt6", "PySide2"}
#: The one binding allowed, and where it may be named.
ALLOWED_BINDING = "PySide6"
#: ``ui/`` modules that deliberately name no binding. ``reveal.py`` hands a path
#: to the OS file manager; ``site_export.py`` writes the queue's CSV point sheet.
#: Neither builds a widget, so neither has a binding to declare. The rule below
#: exists so a UI module cannot *use* a binding it does not name; a module that
#: uses none cannot break it, and importing PySide6 only to satisfy the rule would
#: be the false declaration the rule is meant to prevent.
_BINDING_FREE_UI = {"reveal.py", "site_export.py"}
#: Scoped enum holders we care about, with the value names actually used.
#:
#: Hand-maintained, and therefore incomplete by construction: the PySide6 migration
@@ -147,6 +155,8 @@ def test_every_ui_module_imports_the_allowed_binding():
for path in _python_files(APP_SRC / "orisys_autocal" / "ui"):
if path.name == "__init__.py":
continue # a re-export barrel, not a widget module
if path.name in _BINDING_FREE_UI:
continue
if ALLOWED_BINDING not in _imported_packages(path):
offenders.append(path.name)
assert not offenders, f"these UI modules do not import {ALLOWED_BINDING}: {offenders}"
File diff suppressed because it is too large Load Diff
+736
View File
@@ -0,0 +1,736 @@
"""The capture page end to end: one queue, one primary button, one message line.
These are interface tests in the literal sense: they click the widgets an operator
clicks, in the order the controller offers them, and check the files the session
wrote. The platform move is stubbed at the window's move planner (the real move is
hardware-adjacent and already covered elsewhere); everything else runs as shipped.
"""
from __future__ import annotations
import csv
import json
import re
import time
from pathlib import Path
from types import SimpleNamespace
import numpy as np
import pytest
from orisys_app_shared.i18n import t
from orisys_autocal.sensor.uv_flat import FlatSurface
POINT = (4.872, 5.444, 16.176)
NORMAL = (0.595, 0.673, 0.439)
#: The only direct Campaign calls the frozen contract keeps in ``ui/``: the
#: recorder's lifecycle is not a workflow decision and not rendered data.
_ALLOWED_CAMPAIGN_MEMBERS = frozenset({"attach_capture", "flush_close"})
def _campaign_accesses(source: str) -> list[str]:
"""Every ``campaign.<member>`` access in one module's source, in order."""
return [match.group(1) for match in re.finditer(r"campaign\.(\w+)", source)]
def _campaign_offenders(directory) -> list[str]:
"""UI modules reaching past the controller for campaign state or actions."""
offenders = []
for path in sorted(directory.rglob("*.py")):
for member in _campaign_accesses(path.read_text(encoding="utf-8")):
if member not in _ALLOWED_CAMPAIGN_MEMBERS:
offenders.append(f"{path.name}: campaign.{member}")
return offenders
def _flat() -> FlatSurface:
"""A stand-in rasterised surface: every pixel is the same measured point."""
size = 8
positions = np.tile(np.asarray(POINT, dtype=np.float64), (size, size, 1))
normals = np.tile(np.asarray(NORMAL, dtype=np.float64), (size, size, 1))
return FlatSurface(
size_hw=(size, size),
image_rgb=np.zeros((size, size, 3), dtype=np.uint8),
positions_mm=positions,
normals=normals,
face_index=np.zeros((size, size), dtype=np.int32),
uv_min=(0.0, 0.0),
uv_max=(1.0, 1.0),
)
def _pump(app, seconds: float = 0.3) -> None:
deadline = time.monotonic() + seconds
while time.monotonic() < deadline:
app.processEvents()
time.sleep(0.005)
def _page(
window,
qt_app,
tmp_path,
*,
max_force: str = "20",
align: bool = True,
flat: bool = True,
):
"""A capture page whose readiness passes in simulation mode.
``align=True`` (the default) makes the page behave as it does with a platform
attached: the point must be aligned before a session may start, so the tests
that are about the align step keep covering it. ``align=False`` is the page as
it is with nothing connected, and ``flat=False`` leaves out the calibration
bundle too.
"""
window.resize(1440, 900)
window.show()
window.workspace.set_current_key("sensor")
qt_app.processEvents()
page = window.calibration_panel
page.settings.simulate.setChecked(True)
page.settings.sensor_sn.setText("SN1")
page.settings.max_force.setText(max_force)
page.settings.output_path.setText(str(tmp_path))
if align:
# The page treats a simulation as needing no move; these tests are about
# the step that does, so they ask for the platform-attached behaviour.
page._simulating = lambda: False # noqa: SLF001 - the seam under test
if flat:
page.set_flat(_flat(), "bundle")
# A hold of a few hundredths of a second keeps the test fast; the session
# contract is the same.
page.hold.setRange(0.05, 30)
page.hold.setValue(0.05)
page.levels.setText("5")
page.repeats.setValue(1)
return page
def _started_page(window, qt_app, tmp_path, *, sites=("P1", "P2")):
"""Two points added through the page, first aligned, session started."""
_stub_moves(window, succeeds=True)
page = _page(window, qt_app, tmp_path)
for index, site_id in enumerate(sites):
_add_point(page, site_id, index, index, qt_app)
_click_primary(page, qt_app) # align the first point
assert page.controller.primary_action().key == "start_session"
_click_primary(page, qt_app) # start the session
assert page.controller.poll()["phase"] == "load"
return page
def _stub_moves(window, *, succeeds: bool = True) -> list[dict]:
"""Replace the window's move planner: record the plan, report the outcome."""
plans: list[dict] = []
def fake_start(plan, title, *, report=None):
plans.append({"title": title, "roles": [step.role for step in plan.steps]})
if succeeds:
window.calibration_panel.alignment_succeeded()
else:
window.calibration_panel.alignment_failed("refused by the test")
window._start_move = fake_start # noqa: SLF001 - the seam under test is the page
return plans
def _click_primary(page, qt_app, timeout: float = 5.0) -> str:
"""Press the primary button once, after filling whatever it needs first."""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
qt_app.processEvents()
action = page.controller.primary_action()
if action.enabled and not (
action.key == "confirm_hold" and page.controller.poll()["remaining_s"] > 0
):
# The page redraws on its own timer; read the button the operator would
# click, not just the controller's answer, or a fast step races the redraw.
page._refresh() # noqa: SLF001 - the widget state is what is under test
if not page.primary_button.isEnabled():
_pump(qt_app, 0.05)
continue
if action.key == "confirm_target":
target = page.controller.model().progress.get("target_n") or 0.0
page.reference.setText(f"{float(target):g}")
page.primary_button.click()
_pump(qt_app, 0.1)
return action.key
_pump(qt_app, 0.05)
raise AssertionError(
f"the primary action never became available: "
f"{page.controller.primary_action()} / {page.message.text()}"
)
def _add_point(page, site_id: str, u: int, v: int, qt_app) -> None:
page.u_spin.setValue(u)
page.v_spin.setValue(v)
page.site_id.setText(site_id)
page.add_point_button.click()
_pump(qt_app, 0.05)
def test_two_point_capture_walkthrough_in_simulation(make_window, qt_app, tmp_path):
"""Align, start, load every step, hold, liftoff, align, and finish - by clicking."""
window = make_window(move_confirm=False)
plans = _stub_moves(window, succeeds=True)
page = _page(window, qt_app, tmp_path)
_add_point(page, "P1", 0, 0, qt_app)
_add_point(page, "P2", 1, 1, qt_app)
assert [row.site_id for row in page.controller.model().sites] == ["P1", "P2"]
# The first point must be aligned before the session may start.
assert page.controller.primary_action().key == "align_point"
assert t("cap.action.align") == page.primary_button.text()
_click_primary(page, qt_app)
# The press axis is held because its relation to a contact height is not
# calibrated on the test bench; the four movable roles still travel together.
assert plans and plans[-1]["roles"] == ["yaw", "pitch", "x", "y"]
assert page.controller.primary_action().key == "start_session"
assert t("cap.action.start") == page.primary_button.text()
_click_primary(page, qt_app) # start the session
state = page.controller.poll()
assert state["phase"] == "load"
# Starting a capture does not switch the workspace page.
assert window.workspace.current_key() == "sensor"
# The recording options are locked once the session exists.
assert not page.settings.video_check.isEnabled()
assert not page.settings.locked_label.isHidden()
assert page.settings.locked_label.text() == t("cap.recording.locked")
seen: list[str] = []
for _ in range(60):
action = page.controller.primary_action()
seen.append(action.key)
if action.key == "finish_session":
_click_primary(page, qt_app)
break
_click_primary(page, qt_app)
else: # pragma: no cover - a stuck walkthrough is the failure to report
raise AssertionError(f"the session never finished: {seen[-6:]}")
# The hold ends by itself at its planned time, so the button may or may not be
# caught showing "confirm hold" between two ticks; the order of the steps is
# what matters.
assert seen[0] == "confirm_target"
assert seen.count("confirm_target") >= 3
# Two points, so the second one needs its own alignment between them.
assert "align_point" in seen
assert "confirm_liftoff" in seen
directory = page._session_directory() # noqa: SLF001 - the recorded folder under test
assert directory is not None
manifest = json.loads((directory / "session.json").read_text(encoding="utf-8"))
assert manifest["status"] == "complete"
assert [site["site_id"] for site in manifest["sites"]] == ["P1", "P2"]
assert [trial["status"] for trial in manifest["trials"]] == ["complete", "complete"]
assert (directory / "stable_means.csv").is_file()
# A simulated run writes no video; that is the recorded state, not a failure.
assert page.controller.poll()["capture_state"] == "not_recorded"
# The end of the session publishes the candidate summary (P1's flat copy).
deadline = time.monotonic() + 20.0
while not (directory / "summary-latest.json").is_file() and time.monotonic() < deadline:
_pump(qt_app, 0.05)
assert (directory / "summary-latest.json").is_file()
window.close()
def test_a_simulation_runs_with_nothing_connected(make_window, qt_app, tmp_path):
"""No bundle, no platform, no camera: type two points, start, finish, go again."""
window = make_window(move_confirm=False)
plans = _stub_moves(window, succeeds=True)
page = _page(window, qt_app, tmp_path, align=False, flat=False)
_add_point(page, "P1", 0, 0, qt_app)
_add_point(page, "P2", 3, 4, qt_app)
assert [row.site_id for row in page.controller.model().sites] == ["P1", "P2"]
# Every ready check passes without a bundle, a bus or an input frame.
assert all(check.ok for check in page.controller.readiness()), [
check for check in page.controller.readiness() if not check.ok
]
# No align step stands between the queue and the session.
assert page.controller.primary_action().key == "start_session"
assert t("cap.action.start") == page.primary_button.text()
_click_primary(page, qt_app) # start the session
assert page.controller.poll()["phase"] == "load"
seen: list[str] = []
for _ in range(60):
action = page.controller.primary_action()
seen.append(action.key)
_click_primary(page, qt_app)
if action.key == "finish_session":
break
else: # pragma: no cover - a stuck walkthrough is the failure to report
raise AssertionError(f"the session never finished: {seen[-6:]}")
assert "align_point" not in seen
assert "confirm_liftoff" in seen
assert plans == [] # nothing asked the window to move an axis
directory = page._session_directory() # noqa: SLF001
manifest = json.loads((directory / "session.json").read_text(encoding="utf-8"))
assert manifest["simulate"] is True
assert manifest["status"] == "complete"
assert {site["pose"]["normal_aligned_to"] for site in manifest["sites"]} == {
"simulated"
}
# Closing out leaves the page ready to run the same queue again.
assert page.controller.poll()["phase"] == "idle"
assert page.controller.primary_action().key == "start_session"
window.close()
def test_a_refused_alignment_keeps_start_disabled(make_window, qt_app, tmp_path):
"""A move the operator did not confirm must never unlock "start capture"."""
window = make_window(move_confirm=False)
_stub_moves(window, succeeds=False)
page = _page(window, qt_app, tmp_path)
_add_point(page, "P1", 0, 0, qt_app)
_click_primary(page, qt_app) # align: the window refuses the move
assert page.controller.primary_action().key == "align_point"
assert page.controller.primary_action().key != "start_session"
assert t("cap.action.start") != page.primary_button.text()
assert t("cap.message.align_failed", detail="refused by the test") in page.message.text()
# A retry whose move is confirmed is what unlocks the session.
_stub_moves(window, succeeds=True)
_click_primary(page, qt_app)
assert page.controller.primary_action().key == "start_session"
window.close()
def test_the_capture_page_never_reaches_past_the_controller():
"""P4 exit condition: ``ui/`` renders controller output, not campaign state.
The guard is a member-access scan rather than a fixed string list: any new
``campaign.<member>`` in a UI module is a finding unless it is one of the two
recorder-lifecycle methods the frozen contract keeps (interfaces 1.11). That
covers ``phase``, ``sites``, ``_*``, every queue edit and every flow action —
including ones added later.
"""
from pathlib import Path
ui = Path(__file__).resolve().parents[1] / "src" / "orisys_autocal" / "ui"
assert _campaign_offenders(ui) == []
def test_the_campaign_guard_catches_a_regression():
"""The guard above must fail on a forbidden access, not merely pass by luck."""
assert _campaign_accesses("x = self.campaign.sites\n") == ["sites"]
assert _campaign_accesses("if self.campaign.phase == 'idle':") == ["phase"]
assert _campaign_accesses("self.campaign._skipped") == ["_skipped"]
assert _campaign_accesses("self.campaign.add_solved_site(a)") == ["add_solved_site"]
assert _campaign_accesses("self.campaign.confirm_reference(1)") == ["confirm_reference"]
assert _campaign_accesses("self.campaign.attach_capture(w)") == ["attach_capture"]
for member in (
"phase",
"sites",
"_skipped",
"add_solved_site",
"update_site",
"remove_site",
"move_site",
"store_solved_pose",
"set_force_rotation",
"confirm_reference",
"complete_hold",
"stop",
"remeasure_step",
"skip_current_point",
):
assert member not in _ALLOWED_CAMPAIGN_MEMBERS, member
def test_unavailable_codecs_are_disabled_with_their_reason(make_window):
"""A codec this machine cannot open is refused here, never silently swapped."""
window = make_window()
settings = window.calibration_panel.settings
settings.set_availability({"mp4v": True, "avc1": False, "H264": False})
model = settings.codec.model()
for index in range(settings.codec.count()):
item = model.item(index)
codec = settings.codec.itemData(index)
assert item.isEnabled() is (codec == "mp4v")
assert settings.codec.currentData() == "mp4v"
for codec in ("avc1", "H264"):
assert t("cap.recording.unavailable", codec=codec) in settings.codec_reason.text()
window.close()
def test_the_size_estimate_warns_below_twice_the_planned_volume(make_window, monkeypatch):
from orisys_autocal.ui import capture_settings_dialog as module
window = make_window()
settings = window.calibration_panel.settings
settings.set_availability({"mp4v": True, "avc1": False, "H264": False})
settings.video_check.setChecked(True)
monkeypatch.setattr(
module.shutil,
"disk_usage",
lambda path: SimpleNamespace(total=0, used=0, free=1_000_000_000),
)
# One hour of mp4v at the measured bench bitrate is far more than 1 GB.
settings.set_plan(3600.0)
assert t("cap.recording.disk_short") in settings.estimate.text()
assert "12.15 GB" in settings.estimate.text()
monkeypatch.setattr(
module.shutil,
"disk_usage",
lambda path: SimpleNamespace(total=0, used=0, free=10**14),
)
settings.set_plan(3601.0)
assert t("cap.recording.disk_short") not in settings.estimate.text()
# Recording no video makes the size question disappear, not a zero.
settings.video_check.setChecked(False)
assert settings.estimate.text().startswith(t("cap.recording.off"))
window.close()
def test_a_session_recorded_without_video_says_so_in_the_review(qt_app, tmp_path):
from evidence_helpers import multipoint_session
from orisys_autocal.session_review import review_session
from orisys_autocal.ui.record_overview_dialog import RecordOverviewDialog
root = multipoint_session(
tmp_path / "quiet",
sites=("P1",),
repeats=1,
levels=(5.0,),
recording_block={"video": False, "codec": "mp4v", "lossless_frames": False},
)
review = review_session(root)
assert review["recording"]["video"] is False
dialog = RecordOverviewDialog(root, review=review)
try:
assert t("overview.video_not_recorded") in dialog.identity.text()
finally:
dialog.close()
# A session that kept its video says nothing about a missing one.
with_video = multipoint_session(
tmp_path / "loud", sites=("P1",), repeats=1, levels=(5.0,)
)
dialog = RecordOverviewDialog(with_video, review=review_session(with_video))
try:
assert t("overview.video_not_recorded") not in dialog.identity.text()
finally:
dialog.close()
def test_the_page_refuses_start_until_the_ready_checks_pass(make_window, qt_app):
"""Every ✗ check names itself and blocks the start with a stated reason."""
window = make_window(move_confirm=False)
page = window.calibration_panel
window.workspace.set_current_key("sensor")
qt_app.processEvents()
action = page.controller.primary_action()
reason = t(action.blocked_reason)
assert action.enabled is False
assert page.primary_reason.text() == reason
checks = {check.key: check for check in page.controller.readiness()}
assert checks["max_force"].ok is False
assert checks["sensor_sn"].ok is False
for key in ("calibration_bundle", "input_frames", "bus_ready"):
assert checks[key].ok is False
assert page.ready_labels[key].text().startswith("✗")
# Filling one fact turns only that chip over.
page.settings.sensor_sn.setText("SN1")
checks = {check.key: check for check in page.controller.readiness()}
assert checks["sensor_sn"].ok is True
assert page.controller.primary_action().enabled is False
window.close()
def test_a_point_needs_a_maximum_force_and_the_controller_is_told(
make_window, qt_app, tmp_path
):
"""The queue edit is the controller's, including its maximum-force refusal."""
window = make_window(move_confirm=False)
page = _page(window, qt_app, tmp_path, max_force="")
page._refresh() # noqa: SLF001 - the page's own render is the unit under test
page.add_point_button.click()
_pump(qt_app, 0.1)
assert page.controller.model().sites == []
assert page.message.text().endswith(t("cap.block.no_max_force"))
# The dialog value reaches the controller through set_max_force, not a rebuild.
page.settings.max_force.setText("20")
page._refresh() # noqa: SLF001
checks = {check.key: check for check in page.controller.readiness()}
assert checks["max_force"].ok is True
_add_point(page, "P1", 0, 0, qt_app)
assert [row.site_id for row in page.controller.model().sites] == ["P1"]
window.close()
def test_a_skipped_point_is_rendered_from_the_controller_row(
make_window, qt_app, tmp_path
):
"""The skip mark is the controller's; the page only draws ``SiteRow.skipped``."""
window = make_window(move_confirm=False)
page = _started_page(window, qt_app, tmp_path)
assert page.controller.model().sites[0].skipped is False
page._refresh() # noqa: SLF001 - render the freshly enabled button
page.skip_button.click()
_pump(qt_app, 0.2)
assert page.controller.model().sites[0].skipped is True
assert page.queue_table.item(0, 4).text() == t("cap.queue.state_skipped")
assert page.message.text().endswith(t("cap.message.skipped", site="P1"))
window.close()
def test_the_stop_button_leaves_the_reason_to_the_controller(
make_window, qt_app, tmp_path
):
"""The stop button passes no reason; the controller records the default one."""
window = make_window(move_confirm=False)
page = _started_page(window, qt_app, tmp_path, sites=("P1",))
directory = page._session_directory() # noqa: SLF001
page._refresh() # noqa: SLF001 - render the freshly enabled button
page.stop_button.click()
_pump(qt_app, 0.2)
manifest = json.loads((directory / "session.json").read_text(encoding="utf-8"))
assert manifest["reason"] == "operator stopped"
assert page.controller.primary_action().key == "finish_session"
window.close()
def test_closing_the_window_records_the_application_reason(
make_window, qt_app, tmp_path
):
"""Window shutdown names its own reason, so the data can tell the two apart."""
window = make_window(move_confirm=False)
page = _started_page(window, qt_app, tmp_path, sites=("P1",))
directory = page._session_directory() # noqa: SLF001
window.close()
manifest = json.loads((directory / "session.json").read_text(encoding="utf-8"))
assert manifest["reason"] == "application closed"
def test_updating_a_point_retakes_it_and_keeps_its_queue_position(
make_window, qt_app, tmp_path
):
"""A retaken point goes through the controller and stays where the operator put it."""
window = make_window(move_confirm=False)
page = _page(window, qt_app, tmp_path)
for index, site_id in enumerate(("P1", "P2", "P3")):
_add_point(page, site_id, index, index, qt_app)
page.queue_table.selectRow(1) # the operator's row selection
qt_app.processEvents()
assert page._selected_site == "P2" # noqa: SLF001
page.u_spin.setValue(5)
page.v_spin.setValue(5)
page.site_id.setText("P2b")
page.update_point_button.click()
_pump(qt_app, 0.2)
rows = page.controller.model().sites
assert [row.site_id for row in rows] == ["P1", "P2b", "P3"]
assert rows[1].site_id == "P2b"
assert rows[1].uv_px is not None
assert page._selected_site == "P2b" # noqa: SLF001
window.close()
# --------------------------------------------------- P5: geometry, sheet, reach
def _drive_to_liftoff(page) -> None:
"""Finish the first point's repeat directly, leaving the session at liftoff."""
campaign = page.controller.campaign
while campaign.phase == "load":
target = campaign.progress()["target_n"]
campaign.confirm_reference(target)
campaign.complete_hold(
now_ns=campaign._hold_started_ns + int(campaign._hold_s * 1e9)
)
assert campaign.phase == "liftoff"
def test_changing_the_mounting_resolves_every_queued_pose(make_window, qt_app, tmp_path):
"""stackChanged re-solves the queue and writes the fresh poses back while idle."""
window = make_window(move_confirm=False)
page = _page(window, qt_app, tmp_path)
_add_point(page, "P1", 0, 0, qt_app)
_add_point(page, "P2", 1, 1, qt_app)
before = {row.site_id: row.pose["z"] for row in page.controller.model().sites}
spin = window.finger_panel.sensor_z_spin
spin.setValue(spin.value() + 4.0) # the operator changes the mounting height
_pump(qt_app, 0.2)
rows = page.controller.model().sites
assert all(row.reachable for row in rows)
for row in rows:
assert row.pose["z"] != pytest.approx(before[row.site_id]), row.site_id
fresh = page._solve_uv(*row.uv_px)[2] # noqa: SLF001 - the solver's answer
assert row.pose["z"] == pytest.approx(fresh["z"])
window.close()
def test_an_unreachable_point_is_marked_and_cannot_be_aligned(
make_window, qt_app, tmp_path
):
"""A travel limit that makes a point unreachable disables its align, visibly."""
window = make_window(move_confirm=False)
emitted = _stub_moves(window, succeeds=True)
page = _page(window, qt_app, tmp_path)
_add_point(page, "P1", 0, 0, qt_app)
assert page.controller.model().sites[0].reachable is True
emitted.clear()
window.finger_panel.stack.yaw_limit_deg = 0.0 # the stage can no longer turn
window.finger_panel.stackChanged.emit()
_pump(qt_app, 0.2)
row = page.controller.model().sites[0]
assert row.reachable is False
assert page.queue_table.item(0, 3).text() == t("cap.queue.unreachable")
action = page.controller.primary_action()
assert (action.key, action.enabled) == ("align_point", False)
assert action.blocked_reason == "cap.block.no_pose"
page._align() # noqa: SLF001 - a direct call must be refused too
assert not emitted
assert t("cap.block.pose_unreachable") in page.message.text()
window.close()
def test_a_frozen_queue_is_not_rewritten_when_the_mounting_changes(
make_window, qt_app, tmp_path
):
"""Between trials the queue is frozen: reachability is shown, poses are not stored."""
window = make_window(move_confirm=False)
page = _started_page(window, qt_app, tmp_path)
_drive_to_liftoff(page)
page._refresh() # noqa: SLF001
recorded = page.controller.model().sites[1].pose["z"]
spin = window.finger_panel.sensor_z_spin
spin.setValue(spin.value() + 4.0)
_pump(qt_app, 0.2)
rows = page.controller.model().sites
assert rows[1].pose["z"] == pytest.approx(recorded) # nothing written back
assert page._live_reachable # noqa: SLF001 - the display-only answer exists
fresh = page._solve_uv(*rows[1].uv_px)[2] # noqa: SLF001
assert page._live_reachable[rows[1].site_id] == bool( # noqa: SLF001
fresh["ok"] and fresh["in_range"]
)
window.close()
def test_the_point_sheet_matches_the_documented_fields(make_window, qt_app, tmp_path):
"""P5 exit: the operation table's fields are the data-format document's, in order."""
from orisys_autocal.sensor.uv_flat import mapping_sha256, uv_origin
from orisys_autocal.ui.site_export import OPERATION_TABLE_FIELDS, write_operation_table
document = (
Path(__file__).resolve().parents[1]
/ "docs"
/ "calibration-data-format.zh.md"
).read_text(encoding="utf-8")
# The document spells out the first two blocks; the rest are described as a group.
explicit = (
"order", "site_id", "u_px", "v_px",
"x_sensor_mm", "y_sensor_mm", "z_sensor_mm",
"x_platform_mm", "y_platform_mm", "z_platform_mm",
"yaw_deg", "pitch_deg",
"origin_raster_x_px", "origin_raster_y_px",
)
assert OPERATION_TABLE_FIELDS[: len(explicit)] == explicit
for field in explicit:
assert field in document, f"{field} is not named in the format document"
window = make_window(move_confirm=False)
page = _page(window, qt_app, tmp_path)
_add_point(page, "P1", 0, 0, qt_app)
_add_point(page, "P2", 1, 1, qt_app)
target = tmp_path / "points.csv"
write_operation_table(target, page.controller.model().sites, page._flat)
with target.open(encoding="utf-8-sig", newline="") as handle:
header = next(csv.reader(handle))
written = list(csv.DictReader(target.open(encoding="utf-8-sig", newline="")))
assert tuple(header) == OPERATION_TABLE_FIELDS
assert [row["site_id"] for row in written] == ["P1", "P2"]
origin = uv_origin(page._flat)
assert int(written[0]["origin_raster_x_px"]) == origin[0]
assert int(written[0]["origin_raster_y_px"]) == origin[1]
assert int(written[0]["raster_width_px"]) == page._flat.width
assert int(written[0]["raster_height_px"]) == page._flat.height
assert written[0]["u_positive"] == "right"
assert written[0]["v_positive"] == "up"
assert written[0]["mapping_sha256"] == mapping_sha256(page._flat)
for record, row in zip(written, page.controller.model().sites):
assert float(record["x_sensor_mm"]) == pytest.approx(row.position["x"])
assert float(record["x_platform_mm"]) == pytest.approx(row.pose["x"])
assert float(record["yaw_deg"]) == pytest.approx(row.pose["yaw"])
# A recorded coordinate that no longer matches the map refuses the sheet.
page.controller.campaign.sites[0].x += 0.5
with pytest.raises(ValueError):
write_operation_table(tmp_path / "invalid.csv", page.controller.model().sites, page._flat)
assert not (tmp_path / "invalid.csv").exists()
window.close()
def test_the_export_button_writes_the_sheet_and_reports_it(
make_window, qt_app, tmp_path, monkeypatch
):
from orisys_autocal.ui import calibration_panel as module
window = make_window(move_confirm=False)
page = _page(window, qt_app, tmp_path)
_add_point(page, "P1", 0, 0, qt_app)
target = tmp_path / "sheet.csv"
monkeypatch.setattr(
module.QFileDialog,
"getSaveFileName",
lambda *args, **kwargs: (str(target), ""),
)
page.export_sheet_button.click()
_pump(qt_app, 0.1)
assert target.is_file()
assert page.message.text() == t("cap.queue.export_saved", path=str(target))
# The refusal path names the point and writes nothing.
page.controller.campaign.sites[0].x += 0.5
page.export_sheet_button.click()
_pump(qt_app, 0.1)
assert t("cap.queue.export_edited_point", order=1) in page.message.text()
window.close()
def test_a_failed_summary_copy_is_reported_not_called_published(make_window):
"""The copy under summary/ is a convenience; a failure must not read as success."""
window = make_window()
page = window.calibration_panel
detail = "the readable copy under summary/ was not written: no space left"
page._summary_warnings = [detail]
page._summary_finished("/some/published/summary", None)
assert page.message.text() == t("summary.copy_failed").format(detail=detail)
window.close()
def test_a_stored_unavailable_codec_does_not_become_the_selection(make_window):
"""Loading stored options must not re-select a codec the probe disabled."""
window = make_window()
settings = window.calibration_panel.settings
settings.prefs["recording"] = {
"video": True,
"codec": "avc1",
"lossless_frames": False,
}
# The dialog probes first and loads the stored options second; that order is
# what used to put the disabled avc1 back into the combo.
settings.set_availability({"mp4v": True, "avc1": False, "H264": False})
settings._load_preferences()
assert settings.codec.currentData() == "mp4v"
assert settings.recording_options().codec == "mp4v"
window.close()
+64 -136
View File
@@ -1,7 +1,9 @@
"""The finger debugger's panel: picking, refusing, and what a row contains.
"""The geometry-and-mounting screen: the single-point trial.
Interaction tests use a small dome. Geometry and native-resolution checks keep
the real export and skip individually when that optional export is unavailable.
The point queue and its operation-sheet export live on the capture page and are
covered by ``test_capture_page.py``.
"""
from __future__ import annotations
@@ -398,9 +400,9 @@ def test_the_play_button_needs_a_picked_point(panel):
assert panel.sim_button.isEnabled()
def test_clearing_the_points_disables_the_simulation_again(panel):
def test_clearing_the_point_disables_the_simulation_again(panel):
panel.handle_pick(*_first_surface_pixel(panel.view.flat))
panel.clear_points()
panel.clear_point()
assert not panel.preview.can_simulate
assert not panel.sim_button.isEnabled()
@@ -539,155 +541,81 @@ def test_the_simulation_cannot_start_without_a_pose(bundle, preview):
assert preview._sim_transform() is None # noqa: SLF001
# ------------------------------------------------------------- the point table
# ------------------------------------------------------------- the single pick
def _pick_three(panel) -> list[dict]:
"""Fill the table with three real picks, spread over the surface."""
flat = panel.view.flat
picked: list[dict] = []
step = max(1, flat.width // 7)
for y in range(step, flat.height, step):
for x in range(step, flat.width, step):
if flat.sample(x, y) is None:
def _another_surface_pixel(flat, first) -> tuple[int, int]:
"""A second surface pixel, so a replacement pick is visibly different."""
for y in range(0, flat.normals.shape[0], 5):
for x in range(0, flat.normals.shape[1], 5):
if (x, y) == first or flat.sample(x, y) is None:
continue
row = panel.handle_pick(x, y)
if row is not None:
picked.append(row)
if len(picked) == 3:
break
if len(picked) == 3:
break
assert len(picked) == 3, f"only picked {len(picked)} points"
return picked
return x, y
raise AssertionError("the flat raster has no second surface pixel")
def test_the_table_has_a_select_column_and_an_editable_body(panel):
from PySide6.QtCore import Qt
from PySide6.QtWidgets import QCheckBox
def test_the_window_names_the_screen_geometry_and_mounting(make_window):
"""P5: the debug screen is now the geometry and mounting page, in both places."""
from orisys_app_shared.i18n import t
_pick_three(panel)
assert panel.table.columnCount() == len(panel.COLUMNS) + 1
for row in range(panel.table.rowCount()):
holder = panel.table.cellWidget(row, panel.SELECT_COLUMN)
assert holder is not None and holder.findChild(QCheckBox) is not None
# The six input columns are editable; the solved ones are not.
for offset in range(len(panel.COLUMNS)):
item = panel.table.item(0, offset + 1)
editable = bool(item.flags() & Qt.ItemFlag.ItemIsEditable)
assert editable == (offset < panel.INPUT_COLUMNS), panel.COLUMNS[offset]
window = make_window()
try:
title = t("finger.title")
assert title == t("workspace.tab_finger")
assert title == "几何与安装", title # the default locale is zh_CN
assert window.finger_panel.title() == title
index = window.workspace.page_keys().index("finger")
assert window.workspace.tabText(index) == title
finally:
window.close()
def test_editing_a_coordinate_resolves_the_row_again(panel):
"""Typing a new sensor coordinate must re-solve, not leave a stale pose."""
_pick_three(panel)
before = panel.points()[0]
def test_the_screen_no_longer_keeps_a_point_table(panel):
"""P5: the queue owns points; this screen is the single-point trial.
item = panel.table.item(0, 1) # the x column
item.setText(f"{before['x'] + 1.0:.3f}")
after = panel.points()[0]
assert after["x"] == pytest.approx(before["x"] + 1.0, abs=1e-3)
# The solved columns moved with it, which is the point of re-resolving.
assert after["X"] != pytest.approx(before["X"], abs=1e-9)
assert after["residual"] < 1.0
The list, its editor and its export moved to the capture page's queue, so the
old widgets must not exist here any more.
"""
assert not hasattr(panel, "table")
assert not hasattr(panel, "export_sites")
assert not hasattr(panel, "clear_points")
assert not hasattr(panel, "acquisition_points")
assert not hasattr(panel, "export_sites_button")
def test_a_non_numeric_edit_is_refused_and_the_old_value_kept(panel):
_pick_three(panel)
before = panel.points()[0]["x"]
panel.table.item(0, 1).setText("not a number")
assert panel.points()[0]["x"] == pytest.approx(before)
assert panel.status_label.text()
def test_ticking_a_row_selects_it_and_clears_again(panel):
from PySide6.QtWidgets import QCheckBox
_pick_three(panel)
first = panel.table.cellWidget(0, panel.SELECT_COLUMN).findChild(QCheckBox)
third = panel.table.cellWidget(2, panel.SELECT_COLUMN).findChild(QCheckBox)
first.setChecked(True)
third.setChecked(True)
assert panel.selected_rows() == [0, 2]
first.setChecked(False)
assert panel.selected_rows() == [2]
def test_removing_selected_drops_exactly_those_rows(panel):
from PySide6.QtWidgets import QCheckBox
_pick_three(panel)
panel.table.cellWidget(1, panel.SELECT_COLUMN).findChild(QCheckBox).setChecked(True)
removed = panel.remove_selected()
assert removed == 1
assert panel.point_count() == 2
assert not panel.selected_rows()
def test_removing_with_nothing_ticked_drops_the_current_row(panel):
"""The obvious fallback: no tick means 'the row I am looking at'."""
_pick_three(panel)
panel.table.setCurrentCell(1, 1)
assert panel.remove_selected() == 1
assert panel.point_count() == 2
def test_a_row_can_be_moved_up_and_down(panel):
_pick_three(panel)
before = [row["x"] for row in panel.points()]
panel.table.setCurrentCell(0, 1)
assert panel.move_selected(1)
assert [row["x"] for row in panel.points()] == [before[1], before[0], before[2]]
assert panel.move_selected(-1)
assert [row["x"] for row in panel.points()] == before
def test_a_row_cannot_be_moved_off_the_end(panel):
_pick_three(panel)
before = [row["x"] for row in panel.points()]
panel.table.setCurrentCell(0, 1)
assert not panel.move_selected(-1)
assert [row["x"] for row in panel.points()] == before
def test_selection_marks_the_point_in_both_views(panel):
"""The red dot: the whole reason for selecting a row."""
_pick_three(panel)
panel.table.setCurrentCell(2, 1)
record = panel.points()[2]
assert panel.view.selected_pixel is not None
# And it is the right pixel: the one showing this row's 3D point.
pixel = panel.view.selected_pixel
point, _normal, _face = panel.view.flat.sample(*pixel)
assert point == pytest.approx([record["x"], record["y"], record["z"]], abs=0.05)
assert panel.preview._highlight_mm is not None # noqa: SLF001
def test_a_pick_marks_its_own_row_as_selected(panel):
def test_a_pick_keeps_exactly_one_point(panel):
flat = panel.view.flat
x, y = _first_surface_pixel(flat)
first = _first_surface_pixel(flat)
second = _another_surface_pixel(flat, first)
first_row = panel.handle_pick(*first)
assert first_row is not None
assert panel.point_count() == 1
assert panel.points()[0]["u_px"] == first_row["u_px"]
second_row = panel.handle_pick(*second)
assert second_row is not None
# A second click replaces the first point rather than adding a row.
assert panel.point_count() == 1
stored = panel.points()[0]
assert (stored["u_px"], stored["v_px"]) == (second_row["u_px"], second_row["v_px"])
assert stored["u_px"] != first_row["u_px"]
def test_a_pick_marks_the_point_in_both_views(panel):
x, y = _first_surface_pixel(panel.view.flat)
panel.handle_pick(x, y)
assert panel.view.selected_pixel is not None
assert panel.preview._highlight_mm is not None # noqa: SLF001
def test_clearing_the_points_removes_the_markers(panel):
_pick_three(panel)
panel.clear_points()
def test_clearing_the_point_removes_the_markers(panel):
panel.handle_pick(*_first_surface_pixel(panel.view.flat))
panel.clear_point()
assert panel.point_count() == 0
assert panel.points() == []
assert panel.view.selected_pixel is None
assert panel.preview._highlight_mm is None # noqa: SLF001
@@ -1042,13 +970,13 @@ def test_picking_with_no_bundle_says_so(qt_app):
widget.deleteLater()
def test_clearing_points_empties_the_table(panel):
def test_clearing_the_point_empties_the_screen(panel):
flat = panel.view.flat
x, y = _first_surface_pixel(flat)
panel.handle_pick(x, y)
assert panel.point_count() == 1
panel.clear_points()
panel.clear_point()
assert panel.point_count() == 0
assert panel.points() == []
@@ -1078,10 +1006,10 @@ def test_the_sensor_z_offset_reaches_the_solve(panel):
assert 0.0 < thick.contact_z_mm - thin.contact_z_mm < 4.0
assert thick.target_xy_mm == pytest.approx(thin.target_xy_mm)
# Picking through the panel uses the same fields, so the two must agree.
panel.clear_points()
panel.clear_point()
panel.set_sensor_setup((0.0, 0.0, 0.0), 0.0)
thin_row = panel.handle_pick(x, y)
panel.clear_points()
panel.clear_point()
panel.set_sensor_setup((0.0, 0.0, 4.0), 0.0)
thick_row = panel.handle_pick(x, y)
assert thin_row is not None and thick_row is not None
+153 -104
View File
@@ -243,103 +243,112 @@ def test_input_end_aborts_active_hold(trial_setup, clock):
trial.finish_session()
def test_operator_export_reproduces_uv_sites_and_rejects_edited_xyz(qt_app, flat, tmp_path):
from orisys_autocal.sensor.export_bundle import SensorBundle
from orisys_autocal.ui.sensor_panel import FingerPanel
def test_a_trial_recorded_without_video_finishes_complete(trial_setup, clock):
"""Data-only recording: the force rows are the evidence and the verdict says so."""
from orisys_autocal.recording_options import RecordingOptions
panel = FingerPanel()
bundle = SensorBundle(
path=tmp_path, vertices_mm=np.array([[2, 1, 3], [4, 1, 3], [2, 4, 3]]),
faces=np.array([[0, 1, 2]]), uv_data=np.array([[0, 0], [1, 0], [0, 1]]),
loop_to_vertex=np.array([0, 1, 2]),
)
panel.set_bundle(bundle, flat)
panel.handle_pick(2, 4)
panel.handle_pick(4, 1)
panel.table.setCurrentCell(1, 1)
panel.move_selected(-1)
destination = panel.export_sites(tmp_path / "sites.csv")
from orisys_autocal.record_overview import read_plan, plan_points
from orisys_autocal.sensor.uv_flat import mapping_sha256
definition = {"origin_raster_xy_px": uv_origin(flat), "raster_size_hw": [flat.height, flat.width],
"mapping_sha256": mapping_sha256(flat), "u_positive": "right", "v_positive": "up"}
imported = plan_points(read_plan(destination, definition), definition)
assert [point["uv_xy_px"] for point in imported] == [[1, 2], [-1, -1]]
with destination.open(encoding="utf-8-sig", newline="") as handle:
rows = list(csv.DictReader(handle))
assert [(int(r["u_px"]), int(r["v_px"])) for r in rows] == [(1, 2), (-1, -1)]
for row in rows:
site = site_at(flat, int(row["u_px"]), int(row["v_px"]))
assert site.position_sensor_mm == tuple(float(row[f"{axis}_sensor_mm"]) for axis in "xyz")
assert row["mapping_sha256"] == site.mapping_sha256
panel.table.item(0, 1).setText("999")
with pytest.raises(ValueError):
panel.export_sites(tmp_path / "invalid.csv")
assert not (tmp_path / "invalid.csv").exists()
panel.deleteLater()
def test_gui_hold_controls_and_session_lifecycle(make_window, trial_setup, clock, flat, tmp_path, qt_app):
from orisys_app_shared.i18n import t
window = make_window()
panel = window.calibration_panel
worker = Worker()
panel.source_panel.worker = worker
panel.trial.worker = worker
panel.source_panel.mode = SimpleNamespace(fps=30)
panel.source_panel.config_ref.setText(trial_setup["config_ref"])
panel.set_flat(flat, trial_setup["bundle_path"])
panel._picked(2, 1)
panel.output_path.setText(str(trial_setup["root"]))
panel.sensor_sn.setText("F-001")
assert panel.sensor_model.currentText() == "finger"
panel.levels.setText("5, 10")
panel.start_button.click()
assert panel.trial.status == "recording"
manifest = json.loads((panel.trial.session.directory / "session.json").read_text(encoding="utf-8"))
assert manifest["uv_outline"]["surface"]
assert manifest["uv_outline"]["mapping_sha256"] == manifest["uv_definition"]["mapping_sha256"]
trial = ManualTrial(worker)
directory = trial.start(**{**trial_setup, "recording": RecordingOptions(video=False)})
for index, (_branch, target) in enumerate(trial.steps):
trial.confirm(target)
# Two frames inside the second half of the hold: the summary's own rule.
# Each frame carries its own id: a repeated id has no single acquisition
# time and the review would exclude it.
for delay, frame_id in ((15, index * 2 + 2), (1, index * 2 + 3)):
clock.advance(delay)
worker.packet = replace(
worker.packet, capture_mono_ns=clock.now,
frame_id=frame_id, source_frame_id=frame_id,
)
worker.emit()
clock.advance(4)
trial.poll()
assert trial.step_index == index + 1
trial.finish()
assert trial.recorder.stop(wait=True).ok
assert trial.poll() == "complete"
assert not list(directory.glob("raw_*.mp4"))
root = json.loads(
(directory.parent.parent / "session.json").read_text(encoding="utf-8")
)
assert root["recording"] == {
"video": False, "codec": "mp4v", "lossless_frames": False}
from orisys_autocal.summary import build_summary
summary = build_summary(directory.parent.parent)
assert all(row["eligible"] for row in summary["windows"]), [
(row["excluded_reason"], row["capture_status"], row["samples_valid"])
for row in summary["windows"]
]
assert all(row["capture_status"] == "complete" for row in summary["windows"])
trial.finish_session()
def test_recording_settings_are_locked_by_the_first_trial(trial_setup, clock):
"""One session, one recording choice: the manifest must not change mid-session."""
from orisys_autocal.recording_options import RecordingOptions
worker = Worker()
trial = ManualTrial(worker)
directory = trial.start(
**{**trial_setup, "recording": RecordingOptions(video=False)}
)
root = json.loads(
(directory.parent.parent / "session.json").read_text(encoding="utf-8")
)
assert root["recording"] == {
"video": False, "codec": "mp4v", "lossless_frames": False}
worker.emit()
assert panel.views.currentWidget() is panel.source_panel
assert not panel.sensor_sn.isEnabled()
assert not panel.close_session_button.isEnabled()
for index, force in enumerate((0, 5, 10, 5, 0)):
panel.reference.setText(str(force))
panel.confirm_button.click()
assert panel.trial.holding
assert not panel.confirm_button.isEnabled()
assert not panel.finish_button.isEnabled()
assert panel.trial.step_index == index
assert t("calibration.holding").split("{")[0] in panel.target_label.text()
if index == 0:
panel.resize(1120, 740)
panel.show()
qt_app.processEvents()
assert panel.grab().save(str(tmp_path / "operator-hold.png"))
clock.advance(20)
panel._poll()
assert panel.trial.step_index == index + 1
assert panel.finish_button.isEnabled()
panel.finish_button.click()
panel.trial.recorder.stop(wait=True)
panel._poll()
assert panel.close_session_button.isEnabled()
assert not panel.sensor_sn.isEnabled()
panel.close_session_button.click()
assert panel.trial.session is None
assert panel.sensor_sn.isEnabled()
trial.finish(aborted=True, reason="test ended")
trial.recorder.stop(wait=True)
trial.poll()
with pytest.raises(ValueError, match="locked"):
trial.start(**{**trial_setup, "recording": RecordingOptions(video=True)})
# The locked settings are what a later trial of this session uses.
second = trial.start(
**{**trial_setup, "recording": RecordingOptions(video=False)}
)
assert trial.recording.video is False
assert second.parent.parent == directory.parent.parent
trial.recorder.stop(wait=True)
trial.poll()
trial.finish_session()
def test_the_capture_page_no_longer_drives_a_manual_trial(make_window):
"""D1: a new capture is a one-point queue in the ``sites/`` format.
The ManualTrial flow stays only so old ``trials/`` sessions keep reading; the
capture page owns a Campaign plus the workflow controller. The two-page
single/multi-point split it used to switch between is gone.
"""
window = make_window()
page = window.calibration_panel
assert not hasattr(page, "trial")
assert not hasattr(page, "multipoint")
assert page.campaign is not None
assert page.controller is not None
window.close()
def test_calibration_page_keeps_precise_uv_site_when_switching_to_live_view(
make_window,
def test_the_capture_page_keeps_the_picked_uv_point_while_the_live_view_is_visible(
make_window, qt_app
):
from orisys_app_shared.i18n import t
"""Both images are on the page at once; picking on the map keeps the point."""
from orisys_autocal.sensor.uv_flat import FlatSurface
window = make_window()
window.resize(1440, 900)
window.show()
qt_app.processEvents()
window.workspace.set_current_key("sensor")
qt_app.processEvents()
mask = np.full((5, 7), -1, dtype=np.int32)
mask[1:4, 2:5] = 2
positions = np.zeros((5, 7, 3), dtype=np.float64)
@@ -353,21 +362,61 @@ def test_calibration_page_keeps_precise_uv_site_when_switching_to_live_view(
uv_min=(0, 0),
uv_max=(1, 1),
)
panel = window.calibration_panel
for key, button in (
("start", panel.start_button),
("confirm", panel.confirm_button),
("finish", panel.finish_button),
("abort", panel.abort_button),
):
assert button.text() == t(f"calibration.{key}")
assert button.toolTip() == t(f"calibration.{key}_tip")
assert panel.target_label.text() == t("calibration.before_start")
panel.set_flat(flat, "bundle")
panel._picked(2, 1)
assert (panel.u_spin.value(), panel.v_spin.value()) == (-1, 2)
assert panel._site.raster_xy_px == (2, 1)
assert "2.000" in panel.position_value.text()
panel.views.setCurrentWidget(panel.source_panel)
assert panel._site.raster_xy_px == (2, 1)
page = window.calibration_panel
page.set_flat(flat, "bundle")
page._picked(2, 1)
assert (page.u_spin.value(), page.v_spin.value()) == (-1, 2)
assert page._site.raster_xy_px == (2, 1)
assert "2.000" in page.position_value.text()
# The UV map and the live 2-D image are visible in the same row: selecting one
# used to hide the other behind a tab.
assert page.flat_view.isVisible() and page.live_view.isVisible()
window.close()
def test_the_session_row_opens_the_folder_the_summary_and_the_path(
make_window, tmp_path, monkeypatch
):
"""The operator can reach the recording without copying a path by hand."""
from PySide6.QtWidgets import QApplication
from orisys_app_shared.i18n import t
from evidence_helpers import multipoint_session
from orisys_autocal.summary import build_summary
from orisys_autocal.summary_export import publish_summary
from orisys_autocal.ui import calibration_panel as panel_module
window = make_window()
panel = window.calibration_panel
opened: list = []
monkeypatch.setattr(
panel_module, "reveal", lambda path, select=None: opened.append((path, select)) or True
)
# Nothing has been recorded yet: each action is disabled and says why.
panel._refresh()
assert not panel.open_session_button.isEnabled()
assert not panel.open_summary_button.isEnabled()
assert not panel.copy_path_button.isEnabled()
assert panel.open_session_button.toolTip() == t("cap.block.no_directory")
assert panel.open_summary_button.toolTip() == t("cap.block.no_summary")
root = multipoint_session(tmp_path / "session", sites=("P1",), repeats=1, levels=(5.0,))
publish_summary(root, build_summary(root))
# The page finds the session the campaign is recording into.
panel.campaign.directory = root
panel._refresh()
assert panel._session_directory() == root
assert panel.open_session_button.isEnabled()
assert panel.open_summary_button.isEnabled()
assert panel.copy_path_button.isEnabled()
assert panel.open_session_button.toolTip() == t("cap.action.open_session")
panel.open_session_button.click()
assert opened[-1][0] == root
panel.open_summary_button.click()
assert opened[-1][0] == root / "summary"
panel.copy_path_button.click()
assert QApplication.clipboard().text() == str(root)
window.close()
+108 -83
View File
@@ -1,13 +1,16 @@
"""Aligning the normal solves the point again instead of trusting the stored pose.
"""Aligning a point re-solves it against the geometry in force now.
The stored pose is a record of what the solver said when the point was taken.
Mounting offsets, the mounting angle, the tool position and the travel limits are
all editable afterwards, so a stale pose can name a target this machine no longer
reaches. These tests drive the panel's own align path and watch what it emits.
reaches. These tests drive the capture page's own align path and watch what it
emits; the window that would plan the move is not involved.
"""
from __future__ import annotations
import json
import numpy as np
import pytest
@@ -41,13 +44,14 @@ def _prefix(key: str) -> str:
return t(key).split("{")[0]
def _add_site(panel, stack, *, site_id="P1", point=POINT, normal=NORMAL, uv=None):
def _add_site(page, stack, *, site_id="P1", point=POINT, normal=NORMAL, uv=None):
solved = stack.solve(
np.asarray(point, dtype=np.float64),
np.asarray(normal, dtype=np.float64),
)
assert solved.ok
panel.campaign.add_solved_site(
page.controller.set_max_force(20.0)
page.controller.add_point(
site_id,
float(point[0]),
float(point[1]),
@@ -68,107 +72,110 @@ def _add_site(panel, stack, *, site_id="P1", point=POINT, normal=NORMAL, uv=None
[5.0],
1.0,
1,
max_force_n=20.0,
uv_px=uv,
)
return solved
def _watch(panel, window) -> list[dict]:
"""Record what the panel emits, without letting the window plan a move."""
panel.alignRequested.disconnect()
def _watch(page) -> list[dict]:
"""Record what the page emits, without letting the window plan a move."""
page.alignRequested.disconnect()
emitted: list[dict] = []
panel.alignRequested.connect(emitted.append)
page.alignRequested.connect(emitted.append)
return emitted
def test_aligning_solves_the_point_against_the_mounting_in_force_now(make_window):
window = make_window()
panel = window.calibration_panel.multipoint
stored = _add_site(panel, window.finger_stack)
panel._selected = "P1" # the panel is the unit under test - the table's selection, set without a click
emitted = _watch(panel, window)
page = window.calibration_panel
page._flat = _flat() # the page re-solves a UV point against the current map
stored = _add_site(page, window.finger_stack, uv=(0.0, 0.0))
emitted = _watch(page)
panel._align() # the panel is the unit under test
assert emitted, panel.message.text()
page._align() # the page is the unit under test
assert emitted, page.message.text()
assert emitted[-1]["z"] == pytest.approx(stored.contact_z_mm)
# The window reports the move cancelled, so the point stays unaligned and the
# page keeps offering an align for a retry.
page.alignment_failed("cancelled")
# The operator raises the sensor on its plate: the same point now needs a
# different pose, and the panel has to ask the solver again.
# different pose, and the page has to ask the solver again.
window.finger_panel.sensor_z_spin.setValue(12.0)
panel._align() # the panel is the unit under test
assert len(emitted) == 2, panel.message.text()
fresh = panel.campaign.sites[0].pose
page._align() # the page is the unit under test
assert len(emitted) == 2, page.message.text()
fresh = page.controller.campaign.sites[0].pose
assert fresh["z"] == pytest.approx(emitted[-1]["z"])
assert fresh["z"] != pytest.approx(stored.contact_z_mm)
window.close()
def test_a_point_the_map_has_moved_asks_to_be_taken_again(make_window):
window = make_window()
panel = window.calibration_panel.multipoint
window.calibration_panel._flat = _flat() # the panel is the unit under test - the loaded surface
_add_site(panel, window.finger_stack, uv=(0.0, 0.0))
page = window.calibration_panel
page._flat = _flat()
_add_site(page, window.finger_stack, uv=(0.0, 0.0))
# The record says the point is half a millimetre from where the map puts it.
panel.campaign.sites[0].x += 0.5
panel._selected = "P1" # the panel is the unit under test
emitted = _watch(panel, window)
page.controller.campaign.sites[0].x += 0.5
emitted = _watch(page)
panel._align() # the panel is the unit under test
page._align()
assert not emitted
assert _prefix("multipoint.point_moved") in panel.message.text()
assert _prefix("multipoint.point_moved") in page.message.text()
window.close()
def test_a_point_whose_normal_has_moved_asks_to_be_taken_again(make_window):
window = make_window()
panel = window.calibration_panel.multipoint
window.calibration_panel._flat = _flat() # the panel is the unit under test
_add_site(panel, window.finger_stack, uv=(0.0, 0.0))
page = window.calibration_panel
page._flat = _flat()
_add_site(page, window.finger_stack, uv=(0.0, 0.0))
tilted = np.asarray(NORMAL, dtype=np.float64)
tilted = tilted / np.linalg.norm(tilted)
panel.campaign.sites[0].normal = (float(tilted[0]), float(tilted[1]), float(-tilted[2]))
panel._selected = "P1" # the panel is the unit under test
emitted = _watch(panel, window)
page.controller.campaign.sites[0].normal = (float(tilted[0]), float(tilted[1]), float(-tilted[2]))
emitted = _watch(page)
panel._align() # the panel is the unit under test
page._align()
assert not emitted
assert _prefix("multipoint.normal_moved") in panel.message.text()
assert _prefix("multipoint.normal_moved") in page.message.text()
window.close()
def test_a_point_without_a_usable_source_is_refused(make_window):
window = make_window()
panel = window.calibration_panel.multipoint
_add_site(panel, window.finger_stack)
panel.campaign.sites[0].position_frame = "fixture"
panel._selected = "P1" # the panel is the unit under test
emitted = _watch(panel, window)
page = window.calibration_panel
_add_site(page, window.finger_stack)
page.controller.campaign.sites[0].position_frame = "fixture"
emitted = _watch(page)
panel._align() # the panel is the unit under test
page._align()
assert not emitted
assert _prefix("multipoint.point_unmapped") in panel.message.text()
assert _prefix("multipoint.point_unmapped") in page.message.text()
window.close()
def test_a_pose_the_travel_no_longer_reaches_is_never_commanded(make_window):
window = make_window()
panel = window.calibration_panel.multipoint
_add_site(panel, window.finger_stack)
panel._selected = "P1" # the panel is the unit under test
emitted = _watch(panel, window)
page = window.calibration_panel
page._flat = _flat()
_add_site(page, window.finger_stack, uv=(0.0, 0.0))
emitted = _watch(page)
# A yaw axis with no travel cannot reach any of the measured points, and the
# stored pose still says it could.
window.finger_panel.stack.yaw_limit_deg = 0.0
panel._align() # the panel is the unit under test
page._align()
assert not emitted
assert _prefix("multipoint.pose_refused") in panel.message.text()
assert _prefix("multipoint.pose_refused") in page.message.text()
window.close()
def test_a_point_cannot_be_aligned_in_the_middle_of_a_run(make_window, tmp_path):
window = make_window()
panel = window.calibration_panel.multipoint
_add_site(panel, window.finger_stack)
panel._selected = "P1" # the panel is the unit under test
emitted = _watch(panel, window)
panel.campaign.start(
page = window.calibration_panel
_add_site(page, window.finger_stack)
emitted = _watch(page)
page.campaign.start(
tmp_path,
sensor_model="finger",
sensor_sn="SN1",
@@ -176,42 +183,60 @@ def test_a_point_cannot_be_aligned_in_the_middle_of_a_run(make_window, tmp_path)
max_force_n=20.0,
simulate=True,
)
page._refresh()
panel._align() # the panel is the unit under test
page._align()
assert not emitted
assert t("multipoint.align_busy") in panel.message.text()
assert t("multipoint.align_busy") in page.message.text()
window.close()
def test_next_site_can_be_aligned_between_trials(make_window, tmp_path):
import json
def test_next_site_is_aligned_from_its_recorded_pose_between_trials(make_window, tmp_path):
"""Between trials the queue is frozen, so the recorded pose is what moves.
Re-solving and storing is an idle-only queue edit (interfaces 1.10/1.11), so the
page does not pretend to retarget the next point mid-session: the point's stored
pose is used unchanged, and re-solving after a mounting change inside a running
session is P5's job. The guard that the point itself did not move still runs.
"""
window = make_window()
panel = window.calibration_panel.multipoint
_add_site(panel, window.finger_stack, site_id="P1")
_add_site(panel, window.finger_stack, site_id="P2")
emitted = _watch(panel, window)
panel.campaign.start(tmp_path, sensor_model="finger", sensor_sn="SN1",
coordinate_frame="sensor", max_force_n=20.0, simulate=True)
while panel.campaign.phase == "load":
target = panel.campaign.progress()["target_n"]
panel.campaign.confirm_reference(target)
panel.campaign.complete_hold(
now_ns=panel.campaign._hold_started_ns + int(panel.campaign._hold_s * 1e9))
assert panel.campaign.phase == "liftoff"
panel._selected = "P2"
panel._refresh()
assert panel.align_button.isEnabled()
previous = json.loads((panel.campaign.directory / "sites/P1/repeat-001/session.json").read_text())
page = window.calibration_panel
page._flat = _flat()
_add_site(page, window.finger_stack, site_id="P1", uv=(0.0, 0.0))
_add_site(page, window.finger_stack, site_id="P2", uv=(1.0, 1.0))
emitted = _watch(page)
campaign = page.controller.campaign
campaign.start(
tmp_path, sensor_model="finger", sensor_sn="SN1",
coordinate_frame="sensor", max_force_n=20.0, simulate=True,
)
while campaign.phase == "load":
target = campaign.progress()["target_n"]
campaign.confirm_reference(target)
campaign.complete_hold(
now_ns=campaign._hold_started_ns + int(campaign._hold_s * 1e9))
assert campaign.phase == "liftoff"
page._refresh()
assert page.controller.primary_action().key == "align_point"
previous = json.loads(
(campaign.directory / "sites/P1/repeat-001/session.json").read_text()
)
recorded = page.controller.model().sites[1].pose["z"]
window.finger_panel.sensor_z_spin.setValue(12.0)
panel._align()
page._align()
assert len(emitted) == 1
manifest = json.loads((panel.campaign.directory / "session.json").read_text())
assert manifest["sites"][1]["pose"]["z"] == emitted[0]["z"]
panel.campaign.confirm_liftoff()
repeat = json.loads((panel.campaign.directory / "sites/P2/repeat-001/session.json").read_text())
assert emitted[0]["z"] == pytest.approx(recorded)
# The move is reported finished by the window; until then no start is offered.
page.alignment_succeeded()
assert page.controller.primary_action().key == "confirm_liftoff"
page.primary_button.click() # confirm liftoff -> the next point's first repeat
assert campaign.phase == "load"
manifest = json.loads((campaign.directory / "session.json").read_text())
assert manifest["sites"][1]["pose"]["z"] == pytest.approx(recorded)
repeat = json.loads((campaign.directory / "sites/P2/repeat-001/session.json").read_text())
assert repeat["pose"] == manifest["sites"][1]["pose"]
assert json.loads((panel.campaign.directory / "sites/P1/repeat-001/session.json").read_text()) == previous
panel._selected = "P1"
panel._align()
assert json.loads((campaign.directory / "sites/P1/repeat-001/session.json").read_text()) == previous
page._align()
assert len(emitted) == 1
panel.campaign.stop("test finished")
campaign.stop("test finished")
window.close()
+116 -47
View File
@@ -15,6 +15,7 @@ import pytest
from evidence_helpers import FakeWorker
from orisys_autocal.multipoint import Campaign
from orisys_autocal.recording_options import RecordingOptions
class _Clock:
@@ -46,10 +47,18 @@ def _campaign(*, repeats: int = 2) -> Campaign:
return campaign
def _start(campaign: Campaign, root, *, simulate: bool = False) -> "object":
def _start(
campaign: Campaign,
root,
*,
simulate: bool = False,
recording: RecordingOptions | None = None,
codec_availability: dict | None = None,
) -> "object":
return campaign.start(
root, sensor_model="finger", sensor_sn="SN1",
coordinate_frame="fixture", max_force_n=20, simulate=simulate,
recording=recording, codec_availability=codec_availability,
)
@@ -323,74 +332,134 @@ def test_a_repeat_directory_that_already_exists_is_never_reused(tmp_path):
campaign._advance_after_repeat()
def test_the_confirm_button_waits_until_the_capture_is_recording(make_window, tmp_path):
"""A hold confirmed before the writer exists would be evidence with no video."""
def test_a_live_start_attaches_the_recorder_before_offering_a_confirmation(
make_window, tmp_path
):
"""A hold confirmed before the writer exists would be evidence with no video.
The page owns the attach, so by the time it reports a started session the
repeat is already recording and the controller offers the confirmation.
"""
from types import SimpleNamespace
from orisys_app_shared.i18n import t
window = make_window()
panel = window.calibration_panel
multipoint = panel.multipoint
worker = FakeWorker()
panel.source_panel.worker = worker
panel.source_panel.mode = SimpleNamespace(fps=10)
multipoint.campaign.add_site("P1", 0, 0, 1, [5.0], 1, 1, max_force_n=20)
panel.output_path.setText(str(tmp_path))
panel.sensor_sn.setText("SN1")
from orisys_autocal.motion import BusStatus
# A started campaign whose capture was never attached: not ready.
multipoint.campaign.start(
tmp_path, sensor_model="finger", sensor_sn="SN1",
coordinate_frame="sensor", max_force_n=20, simulate=False,
window = make_window(move_confirm=False)
window.poller.service._spec = object() # noqa: SLF001 - connected, without opening anything
window.poller.service._bus = BusStatus( # noqa: SLF001
bus_type="ecat", init_status=1, total_axes=5
)
panel._sync()
assert multipoint.campaign.phase == "load"
assert not panel.confirm_button.isEnabled()
assert t("multipoint.capture_wait") in panel.target_label.text()
panel.reference.setText("0")
panel._confirm()
assert multipoint.campaign.phase == "load"
page = window.calibration_panel
window.workspace.set_current_key("sensor")
worker = FakeWorker()
page.source_panel.worker = worker
page.source_panel.mode = SimpleNamespace(fps=10)
page._flat = object() # a loaded bundle is all the readiness check needs here
page.settings.output_path.setText(str(tmp_path))
page.settings.sensor_sn.setText("SN1")
page.settings.max_force.setText("20")
page.campaign.add_site("P1", 0, 0, 1, [5.0], 1, 1, max_force_n=20)
# The window confirmed the move, so the session may start.
page.alignment_succeeded()
page._refresh()
multipoint.campaign.attach_capture(worker, fps=10.0)
panel._sync()
assert panel.confirm_button.isEnabled()
assert multipoint.campaign.capture_ready
action = page.controller.primary_action()
assert action.key == "start_session", (action, page.message.text())
page.primary_button.click()
campaign = page.campaign
assert campaign.phase == "load"
assert campaign.capture_state == "recording"
assert page.controller.primary_action().key == "confirm_target"
assert t("multipoint.capture_recording") in page.capture_label.text()
window.close()
def test_a_restarted_source_is_reported_on_the_panel(make_window, tmp_path):
def test_a_restarted_source_is_reported_on_the_capture_page(make_window, tmp_path):
from types import SimpleNamespace
from orisys_app_shared.i18n import t
window = make_window()
panel = window.calibration_panel
multipoint = panel.multipoint
worker = FakeWorker()
panel.source_panel.worker = worker
panel.source_panel.mode = SimpleNamespace(fps=10)
multipoint.campaign.add_site("P1", 0, 0, 1, [5.0], 1, 1, max_force_n=20)
multipoint.campaign.start(
tmp_path, sensor_model="finger", sensor_sn="SN1",
coordinate_frame="sensor", max_force_n=20, simulate=False,
from orisys_autocal.motion import BusStatus
window = make_window(move_confirm=False)
window.poller.service._spec = object() # noqa: SLF001
window.poller.service._bus = BusStatus( # noqa: SLF001
bus_type="ecat", init_status=1, total_axes=5
)
multipoint.campaign.attach_capture(worker, fps=10.0)
multipoint.campaign.poll()
page = window.calibration_panel
window.workspace.set_current_key("sensor")
worker = FakeWorker()
page.source_panel.worker = worker
page.source_panel.mode = SimpleNamespace(fps=10)
page._flat = object()
page.settings.output_path.setText(str(tmp_path))
page.settings.sensor_sn.setText("SN1")
page.settings.max_force.setText("20")
page.campaign.add_site("P1", 0, 0, 1, [5.0], 1, 1, max_force_n=20)
page.alignment_succeeded()
page._refresh()
page.primary_button.click()
campaign = page.campaign
assert campaign.capture_state == "recording"
worker.restart()
worker.emit()
multipoint._poll()
_drain(multipoint.campaign)
multipoint._refresh()
page._poll()
_drain(campaign)
page._refresh()
assert multipoint.campaign.phase == "aborted"
assert t("multipoint.capture_incomplete") in multipoint.capture_status.text()
assert t("multipoint.capture_reason_restarted") in multipoint.capture_status.text()
assert campaign.phase == "aborted"
assert t("multipoint.capture_incomplete") in page.capture_label.text()
assert t("multipoint.capture_reason_restarted") in page.capture_label.text()
manifest = json.loads(
(multipoint.campaign.directory / "sites" / "P1" / "repeat-001" / "session.json")
(campaign.directory / "sites" / "P1" / "repeat-001" / "session.json")
.read_text(encoding="utf-8")
)
assert manifest["capture_status"] == "incomplete"
assert manifest["capture_reason"] == "source_restarted"
window.close()
def test_a_repeat_recorded_without_video_finishes_complete(tmp_path, clock):
"""Choosing data only must not make every repeat incomplete."""
from orisys_autocal.session_review import review_session
from orisys_autocal.summary import build_summary
campaign = _campaign(repeats=1)
root = _start(campaign, tmp_path, recording=RecordingOptions(video=False))
worker = FakeWorker(clock=lambda: clock.now)
campaign.attach_capture(worker, fps=10.0)
_record_repeat(campaign, worker, clock=clock)
assert campaign.phase == "done"
session = _session(root)
assert session["recording"] == {
"video": False, "codec": "mp4v", "lossless_frames": False}
trial = session["trials"][0]
assert trial["status"] == "complete"
assert trial["capture_status"] == "complete"
capture = root / trial["directory"] / "capture"
assert not list(capture.glob("raw_*.mp4"))
review = review_session(root)
assert review["recording"]["video"] is False
summary = build_summary(root)
assert all(row["eligible"] for row in summary["windows"])
assert all(row["FN_mean"] is not None for row in summary["windows"])
def test_an_unavailable_codec_is_refused_before_the_session_starts(tmp_path):
"""No silent fallback to mp4v: an encoder that failed the probe stops the start."""
campaign = _campaign(repeats=1)
availability = {"mp4v": True, "avc1": False, "H264": False}
with pytest.raises(ValueError, match="avc1"):
_start(campaign, tmp_path, recording=RecordingOptions(codec="avc1"),
codec_availability=availability)
assert campaign.directory is None
assert campaign.phase == "idle"
assert list(tmp_path.iterdir()) == []
@@ -75,3 +75,28 @@ def test_a_window_that_closes_still_lands_in_the_scratch_directory(qt_app):
after = INSTALLED.read_bytes() if INSTALLED.is_file() else None
assert after == before, "closing a test window rewrote the installed preferences"
def test_a_stored_recording_block_without_the_newer_keys_gets_each_default():
"""The upgrade path for the recording preference, through the same net.
A stored ``recording`` block replaces the whole default block, exactly like
``safety_limits``: a codec or the lossless flag added later would otherwise
read as absent and the recorder would be configured from a hole.
"""
import json
from orisys_app_shared.paths import preference_path
from orisys_autocal.preferences import load_preferences
from orisys_autocal.recording_options import RecordingOptions, default_options
path = preference_path()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps({"recording": {"video": False}}), encoding="utf-8")
options = default_options(load_preferences())
assert options == RecordingOptions(video=False)
assert options.codec == "mp4v"
assert options.lossless_frames is False
@@ -0,0 +1,100 @@
"""Recording options: the settings a session is recorded with, and what can encode.
The defaults are a decision, not a preference: the measured mp4v bitrate is the
baseline until H.264 equivalence is shown (plan D2), so a missing block must mean
"video expected, mp4v" rather than whatever the machine happens to encode.
"""
from __future__ import annotations
import pytest
from orisys_autocal.recording_options import (
CODEC_CANDIDATES,
DEFAULT_CODEC,
DEFAULT_LOSSLESS_FRAMES,
DEFAULT_VIDEO,
RecordingOptions,
default_options,
make_record_config,
probe_codecs,
require_codec_available,
)
def test_the_defaults_are_the_confirmed_recording_decision():
options = RecordingOptions()
assert options.video is DEFAULT_VIDEO is True
assert options.codec == DEFAULT_CODEC == "mp4v"
assert options.lossless_frames is DEFAULT_LOSSLESS_FRAMES is False
assert options.as_block() == {
"video": True, "codec": "mp4v", "lossless_frames": False}
def test_the_block_round_trips_through_a_session_manifest():
options = RecordingOptions(video=False, codec="avc1", lossless_frames=True)
assert RecordingOptions.from_block(options.as_block()) == options
def test_a_missing_or_partial_block_means_video_was_expected():
"""Old sessions have no block; their video is required, so they keep behaving."""
assert RecordingOptions.from_block(None) == RecordingOptions()
assert RecordingOptions.from_block({}) == RecordingOptions()
assert RecordingOptions.from_block("not a block") == RecordingOptions()
# One stored key is not a reason to discard the others.
assert RecordingOptions.from_block({"video": False}) == RecordingOptions(video=False)
assert RecordingOptions.from_block({"codec": "avc1"}) == RecordingOptions(codec="avc1")
def test_make_record_config_maps_every_option_and_keeps_the_trial_window():
options = RecordingOptions(video=False, codec="avc1", lossless_frames=True)
config = make_record_config(options, 12.5)
assert config.video is False
assert config.codec == "avc1"
assert config.lossless_frames is True
assert config.fps == 12.5
# One continuous recording per trial: the recorder is never segmented.
assert config.segment_seconds == 24 * 3600.0
assert config.preserve_all_frames is True
def test_the_default_options_read_the_preference_block_with_fallbacks():
assert default_options({}) == RecordingOptions()
assert default_options(None) == RecordingOptions()
assert default_options({"recording": {"video": False, "codec": "avc1",
"lossless_frames": True}}) == RecordingOptions(
video=False, codec="avc1", lossless_frames=True)
# The upgrade path: a block saved before a key existed still gets its default.
assert default_options({"recording": {"video": False}}) == RecordingOptions(video=False)
def test_probing_reports_every_candidate_and_leaves_no_probe_files(tmp_path):
scratch = tmp_path / "scratch"
availability = probe_codecs(scratch)
assert set(availability) == set(CODEC_CANDIDATES)
# The suite writes real mp4v clips everywhere else, so this one must be found.
assert availability["mp4v"] is True
assert all(isinstance(value, bool) for value in availability.values())
assert list(scratch.rglob("*.mp4")) == []
def test_an_unavailable_codec_is_refused_before_recording():
availability = {"mp4v": True, "avc1": False, "H264": False}
with pytest.raises(ValueError, match="avc1"):
require_codec_available(RecordingOptions(codec="avc1"), availability)
# The probes that passed are accepted, and a data-only recording needs no codec.
require_codec_available(RecordingOptions(codec="mp4v"), availability)
require_codec_available(RecordingOptions(video=False, codec="avc1"), availability)
def test_an_unprobed_codec_is_left_to_the_caller():
"""Availability is probed once at start-up; without a result nothing is assumed."""
require_codec_available(RecordingOptions(codec="avc1"), None)
+94
View File
@@ -0,0 +1,94 @@
"""Opening a session folder in the platform file manager.
The operator needs the folder the recording is in, not a dialog that only shows a
path string. The launch is mocked here, because the test must assert the exact
argument the platform file manager receives — ``explorer /select,<file>`` is a
single argument whose comma and quoting are the part that breaks silently.
"""
from __future__ import annotations
import pytest
from orisys_autocal.ui import reveal as reveal_module
def _capture_launch(monkeypatch) -> list:
calls: list = []
monkeypatch.setattr(reveal_module.subprocess, "Popen", lambda command: calls.append(command))
return calls
@pytest.mark.parametrize(
("platform", "executable"),
[("win32", "explorer"), ("linux", "xdg-open"), ("darwin", "open")],
)
def test_revealing_a_folder_uses_the_platform_file_manager(
tmp_path, monkeypatch, platform, executable
):
folder = tmp_path / "session"
folder.mkdir()
monkeypatch.setattr(reveal_module.sys, "platform", platform)
calls = _capture_launch(monkeypatch)
assert reveal_module.reveal(folder) is True
assert calls == [[executable, str(folder)]]
def test_a_selected_file_is_highlighted_with_explorer(tmp_path, monkeypatch):
folder = tmp_path / "session"
folder.mkdir()
video = folder / "raw_0001.mp4"
video.write_bytes(b"video")
monkeypatch.setattr(reveal_module.sys, "platform", "win32")
calls = _capture_launch(monkeypatch)
assert reveal_module.reveal(folder, select=video) is True
# One command-line string: the comma belongs to the switch and the path is
# quoted, which is what explorer expects for a path with spaces.
assert calls == [f'explorer /select,"{video}"']
def test_a_selected_file_that_is_gone_falls_back_to_the_folder(tmp_path, monkeypatch):
folder = tmp_path / "session"
folder.mkdir()
missing = folder / "raw_0001.mp4"
monkeypatch.setattr(reveal_module.sys, "platform", "win32")
calls = _capture_launch(monkeypatch)
assert reveal_module.reveal(folder, select=missing) is True
assert calls == [["explorer", str(folder)]]
def test_a_missing_path_is_not_opened(tmp_path, monkeypatch):
monkeypatch.setattr(reveal_module.sys, "platform", "win32")
calls = _capture_launch(monkeypatch)
assert reveal_module.reveal(tmp_path / "gone") is False
assert calls == []
def test_no_path_is_not_opened(monkeypatch):
calls = _capture_launch(monkeypatch)
assert reveal_module.reveal(None) is False
assert reveal_module.reveal("") is False
assert calls == []
def test_a_failed_launch_is_reported_instead_of_raised(tmp_path, monkeypatch):
folder = tmp_path / "session"
folder.mkdir()
monkeypatch.setattr(reveal_module.sys, "platform", "win32")
def fail(command):
raise OSError("no file manager")
monkeypatch.setattr(reveal_module.subprocess, "Popen", fail)
assert reveal_module.reveal(folder) is False
+42
View File
@@ -263,3 +263,45 @@ def test_an_axis_without_samples_is_unavailable_without_blocking_the_others(tmp_
assert by_axis["FY"]["R2"] is None
assert by_axis["FY"]["slope_raw_per_N"] == pytest.approx(0.0)
assert summary["windows"][0]["axis_samples"]["FX"] == 0
#: The block a session recorded without video writes (interfaces.md 2.2).
VIDEO_OFF = {"video": False, "codec": "mp4v", "lossless_frames": False}
def test_a_session_recorded_without_video_is_still_eligible(tmp_path):
"""The operator chose data only: the force rows are the evidence, not a video."""
root = multipoint_session(
tmp_path / "rec", sites=("P1",), repeats=1, levels=_levels(),
video=False, recording_block=VIDEO_OFF, capture_status="complete",
)
summary = build_summary(root)
assert summary["windows"]
assert all(row["capture_status"] == "complete" for row in summary["windows"])
assert all(row["eligible"] for row in summary["windows"])
assert all(row["FN_mean"] is not None for row in summary["windows"])
assert summary["sensitivity"]
assert summary["missing_reasons"] == []
def test_a_session_without_a_recording_block_still_expects_video(tmp_path):
"""Old sessions have no block, so the video stays required exactly as before."""
root = multipoint_session(
tmp_path / "rec", sites=("P1",), repeats=1, levels=_levels(),
video=False, # data rows only, and no block saying that was intended
)
from orisys_autocal.recording_options import RecordingOptions
from orisys_autocal.session_review import review_session
review = review_session(root)
assert review["recording"] == RecordingOptions().as_block()
assert RecordingOptions.from_block(review.get("recording")).video is True
summary = build_summary(root)
assert all(row["eligible"] is False for row in summary["windows"])
assert all(row["excluded_reason"] == REASON_NO_FRAME_SOURCE
for row in summary["windows"])
assert REASON_NO_FRAME_SOURCE in summary["missing_reasons"]
+89
View File
@@ -3,7 +3,9 @@
from __future__ import annotations
import csv
import hashlib
import json
import logging
import pytest
@@ -14,11 +16,16 @@ from orisys_autocal.summary_export import (
POINTER_NAME,
SummaryExportError,
load_summary,
published_summary_directory,
publish_summary,
render_html,
)
def _sha256(path):
return hashlib.sha256(path.read_bytes()).hexdigest()
@pytest.fixture
def session(tmp_path):
root = multipoint_session(
@@ -173,3 +180,85 @@ def test_the_sensitivity_csv_has_one_row_per_point_trial_axis(session):
assert {row["review_status"] for row in rows} == {"candidate"}
fn = [row for row in rows if row["component"] == "FN"]
assert all(float(row["slope_raw_per_N"]) == pytest.approx(40.0) for row in fn)
def test_the_published_files_are_mirrored_under_the_session_root(session):
"""The folder the operator opens is the byte-identical copy the pointer names."""
summary = build_summary(session)
published = publish_summary(session, summary)
pointer = _pointer(session)
flat = session / "summary"
assert flat.is_dir()
assert sorted(path.name for path in flat.iterdir()) == sorted(pointer["files"])
for name, digest in pointer["files"].items():
assert _sha256(flat / name) == digest, name
assert (flat / name).read_bytes() == (published / name).read_bytes(), name
# The readable copy is not a second pointer: the authoritative path is derived/.
assert pointer["path"] == f"derived/{published.name}"
assert published_summary_directory(session) == flat
assert load_summary(session)[1]["summary_id"] == summary["summary_id"]
def test_a_failed_copy_reports_without_moving_the_pointer_back(session, monkeypatch, caplog):
"""Losing the readable copy must not lose the published evidence or the pointer."""
import orisys_autocal.summary_export as module
first = build_summary(session)
publish_summary(session, first)
flat = session / "summary"
before = {path.name: _sha256(path) for path in flat.iterdir()}
first_pointer = _pointer(session)
second = build_summary(session)
second["generated_at"] = "2999-01-01T00:00:00+00:00"
reports: list[str] = []
def explode(*args, **kwargs):
raise OSError("no space left on device")
monkeypatch.setattr(module, "_write_flat_summary", explode)
with caplog.at_level(logging.WARNING, logger="orisys_autocal.summary_export"):
published = publish_summary(session, second, on_warning=reports.append)
pointer = _pointer(session)
assert pointer["summary_id"] == second["summary_id"]
assert pointer["path"] == f"derived/{published.name}"
assert first_pointer["path"] != pointer["path"]
for name, digest in pointer["files"].items():
assert _sha256(published / name) == digest, name
# The previous readable copy still stands, unchanged.
assert {path.name: _sha256(path) for path in flat.iterdir()} == before
assert reports and "no space left on device" in reports[0]
assert any("no space left on device" in record.message for record in caplog.records)
def test_a_stale_readable_copy_is_not_offered_as_the_current_summary(session, monkeypatch):
"""A failed mirror keeps the older copy; opening it must not pass it off as current.
The pointer has moved on, so ``summary/`` describes a superseded version. The
operator must be sent to the directory the pointer names, not to numbers the
pointer no longer claims.
"""
import orisys_autocal.summary_export as module
first = build_summary(session)
publish_summary(session, first)
second = build_summary(session)
second["generated_at"] = "2999-01-01T00:00:00+00:00"
def explode(*args, **kwargs):
raise OSError("no space left on device")
monkeypatch.setattr(module, "_write_flat_summary", explode)
published = publish_summary(session, second)
pointer = _pointer(session)
assert pointer["summary_id"] == second["summary_id"]
opened = published_summary_directory(session)
assert opened is not None
assert opened == published, "the stale summary/ was offered as the current summary"
stored = json.loads((opened / "summary.json").read_text(encoding="utf-8"))
assert stored["summary_id"] == second["summary_id"]
+18 -10
View File
@@ -32,11 +32,19 @@ def _finish_simulated_campaign(campaign) -> None:
def _acquired_hashes(root) -> dict[str, str]:
"""Hashes of the recorded evidence only: derived results are not evidence."""
"""Hashes of the recorded evidence only: export artifacts are not evidence.
The published summary lives in ``derived/`` and its readable mirror in
``summary/``; both are written by the export, so neither is part of the
acquisition and neither may appear in this comparison.
"""
return {
str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest()
for path in root.rglob("*")
if path.is_file() and "derived" not in path.parts and path.name != POINTER_NAME
if path.is_file()
and "derived" not in path.parts
and "summary" not in path.parts
and path.name != POINTER_NAME
}
@@ -78,8 +86,8 @@ def test_a_failed_export_is_reported_and_can_be_retried(
_pump(panel)
assert not (root / POINTER_NAME).exists()
assert "disk full" in panel.result_label.text()
assert panel.trial.status == "idle" # the acquisition result is untouched
assert "disk full" in panel.message.text()
assert panel._directory() == "" # the acquisition result is untouched
monkeypatch.undo()
assert panel.publish_summary_for(root)
@@ -127,16 +135,16 @@ def test_a_published_candidate_is_republished_when_the_review_is_saved(qt_app, t
def test_the_multipoint_end_of_campaign_starts_the_export(make_window, tmp_path):
window = make_window()
panel = window.calibration_panel
multipoint = panel.multipoint
multipoint.campaign.add_site("P1", 0, 0, 1, [5.0], 1, 1, max_force_n=20)
directory = multipoint.campaign.start(
campaign = panel.campaign
campaign.add_site("P1", 0, 0, 1, [5.0], 1, 1, max_force_n=20)
directory = campaign.start(
tmp_path, sensor_model="finger", sensor_sn="SN1",
coordinate_frame="sensor", max_force_n=20, simulate=True,
)
_finish_simulated_campaign(multipoint.campaign)
assert multipoint.campaign.phase == "done"
_finish_simulated_campaign(campaign)
assert campaign.phase == "done"
multipoint._poll()
panel._poll()
_pump(panel)
assert (directory / POINTER_NAME).is_file()