ykcheungab
06e06126ca
选点、对准、加载与录像分散在两个面板，状态靠隐藏相位判断。现改由不依赖 Qt 的采集控制器驱动，并加入录像选项、汇总平铺副本、保持到时自动结束、会话结束回到空闲，以及无设备时的模拟采集。已运行 Autocal 非 Qt 测试并通过范围内门禁；界面测试待 Windows 复核，未做实机验收。
1778 lines
65 KiB
Python
1778 lines
65 KiB
Python
"""Table-driven checks for the headless capture controller.
|
|
|
|
The controller is what the new capture page renders, so every test here drives a
|
|
real :class:`~orisys_autocal.multipoint.Campaign` through its steps: no Qt window,
|
|
no camera, no platform. The campaigns run ``simulate=True``, the only mode the
|
|
AutoCal suite may use (see ``conftest.py``); a faked ``capture_state`` is used
|
|
only where the test is about the recording guard itself.
|
|
|
|
Acceptance coverage (plan P3): the complete load ladder, liftoff confirmation,
|
|
point change, skip (the Campaign advance, its counter rollback and its guard),
|
|
remeasure, reacquire, stop from every state, alignment results reported by the
|
|
window (a request alone is never a finished move), a reason for every disabled
|
|
action, and the same ``session.json`` / ``reference.csv`` structure whether the
|
|
campaign is driven directly or through the controller.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import csv
|
|
import json
|
|
import time
|
|
import types
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from orisys_autocal import capture_controller
|
|
from orisys_autocal.capture_controller import (
|
|
READINESS_ITEMS,
|
|
READINESS_KEYS,
|
|
ActionBlocked,
|
|
CaptureController,
|
|
ReadinessCheck,
|
|
)
|
|
from orisys_autocal.multipoint import Campaign
|
|
|
|
HOLD_S = 0.02
|
|
MAX_FORCE = 20.0
|
|
|
|
#: A pose the campaign may command; it is what the finger page would have stored.
|
|
POSE = {
|
|
"x": 10.0,
|
|
"y": 20.0,
|
|
"z": 30.0,
|
|
"yaw": 5.0,
|
|
"pitch": 10.0,
|
|
"residual_deg": 0.01,
|
|
"upright": True,
|
|
"in_range": True,
|
|
"ok": True,
|
|
"normal_aligned_to": "tool_axis:z",
|
|
}
|
|
|
|
#: Every primary key the frozen interface allows.
|
|
PRIMARY_KEYS = frozenset(
|
|
{
|
|
"align_point",
|
|
"start_session",
|
|
"confirm_target",
|
|
"confirm_hold",
|
|
"confirm_liftoff",
|
|
"finish_session",
|
|
"closing",
|
|
}
|
|
)
|
|
SECONDARY_KEYS = ("remeasure", "skip", "stop")
|
|
|
|
LOCALES = Path(__file__).resolve().parents[1] / "config" / "locales"
|
|
|
|
_OMITTED = object()
|
|
_SOLVER = object()
|
|
|
|
|
|
# --------------------------------------------------------------------------- setup
|
|
|
|
|
|
def _campaign(*, levels=(5.0,), hold_s=HOLD_S, repeats=1, sites=("P1",)) -> Campaign:
|
|
"""A campaign whose points are solved, as the finger page would have left them."""
|
|
campaign = Campaign()
|
|
for site_id in sites:
|
|
campaign.add_solved_site(
|
|
site_id,
|
|
1.0,
|
|
2.0,
|
|
3.0,
|
|
(0.0, 0.0, 1.0),
|
|
dict(POSE),
|
|
list(levels),
|
|
hold_s,
|
|
repeats,
|
|
max_force_n=MAX_FORCE,
|
|
)
|
|
return campaign
|
|
|
|
|
|
def _start_arguments(root):
|
|
"""The window state ``Campaign.start`` needs, as the page supplies it."""
|
|
|
|
def arguments():
|
|
return {
|
|
"root": root,
|
|
"sensor_model": "finger",
|
|
"sensor_sn": "SN1",
|
|
"coordinate_frame": "sensor",
|
|
"max_force_n": MAX_FORCE,
|
|
"simulate": True,
|
|
}
|
|
|
|
return arguments
|
|
|
|
|
|
def _controller(
|
|
campaign,
|
|
root,
|
|
*,
|
|
solver=_OMITTED,
|
|
readiness=None,
|
|
max_force_n=MAX_FORCE,
|
|
start_arguments=_OMITTED,
|
|
) -> CaptureController:
|
|
return CaptureController(
|
|
campaign,
|
|
solver=_SOLVER if solver is _OMITTED else solver,
|
|
readiness=readiness,
|
|
max_force_n=max_force_n,
|
|
start_arguments=(
|
|
_start_arguments(root) if start_arguments is _OMITTED else start_arguments
|
|
),
|
|
)
|
|
|
|
|
|
def _sleep_hold() -> None:
|
|
"""Wait past the configured hold, which ``complete_hold`` rejects before."""
|
|
time.sleep(HOLD_S + 0.01)
|
|
|
|
|
|
class _FakeRecorder:
|
|
"""The writer interface ``Campaign`` reads through its guard and summary.
|
|
|
|
Only the attributes used by the skip guard and ``capture_summary`` exist: the
|
|
test is about refusing to skip, not about recording frames.
|
|
"""
|
|
|
|
is_recording = True
|
|
|
|
def stats(self) -> types.SimpleNamespace:
|
|
return types.SimpleNamespace()
|
|
|
|
|
|
def _aligned_and_started(controller: CaptureController, campaign: Campaign) -> None:
|
|
"""Walk the idle rung: request the move, confirm it, then start the session."""
|
|
controller.perform("align_point")
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned()
|
|
controller.perform("start_session")
|
|
assert campaign.phase == "load"
|
|
|
|
|
|
def _drive_step(controller: CaptureController, campaign: Campaign) -> None:
|
|
"""Confirm the target and finish the hold of the step the campaign is on."""
|
|
controller.perform("confirm_target", force_n=float(campaign.progress()["target_n"]))
|
|
assert campaign.phase == "hold"
|
|
_sleep_hold()
|
|
controller.perform("confirm_hold")
|
|
|
|
|
|
def _drive_repeat(controller: CaptureController, campaign: Campaign) -> None:
|
|
"""Drive the open repeat(s) of the point in focus, stopping at liftoff or done."""
|
|
while campaign.phase == "load":
|
|
_drive_step(controller, campaign)
|
|
|
|
|
|
def _assert_action_reasons(controller: CaptureController) -> None:
|
|
"""The P3 exit rule: a disabled action always names a locale key, a ready one never."""
|
|
actions = [controller.primary_action(), *controller.secondary_actions()]
|
|
for action in actions:
|
|
assert action.key in PRIMARY_KEYS | set(SECONDARY_KEYS), action
|
|
if action.enabled:
|
|
assert action.blocked_reason == "", action
|
|
else:
|
|
assert action.blocked_reason, action
|
|
assert action.blocked_reason.startswith("cap."), action
|
|
|
|
|
|
def _manifest(campaign: Campaign) -> dict:
|
|
return json.loads((campaign.directory / "session.json").read_text(encoding="utf-8"))
|
|
|
|
|
|
def _reference_rows(campaign: Campaign) -> dict[str, list[dict]]:
|
|
rows: dict[str, list[dict]] = {}
|
|
for trial in _manifest(campaign)["trials"]:
|
|
path = campaign.directory / trial["directory"] / "reference.csv"
|
|
with path.open(encoding="utf-8-sig", newline="") as handle:
|
|
rows[trial["directory"]] = list(csv.DictReader(handle))
|
|
return rows
|
|
|
|
|
|
def _campaign_events(campaign: Campaign) -> list[dict]:
|
|
"""The session-level events, one row per event."""
|
|
path = campaign.directory / "events.csv"
|
|
with path.open(encoding="utf-8-sig", newline="") as handle:
|
|
return list(csv.DictReader(handle))
|
|
|
|
|
|
def _flat_locale(name: str) -> set[str]:
|
|
data = json.loads((LOCALES / f"{name}.json").read_text(encoding="utf-8"))
|
|
keys: set[str] = set()
|
|
|
|
def walk(node, prefix=""):
|
|
for key, value in node.items():
|
|
path = f"{prefix}{key}"
|
|
if isinstance(value, dict):
|
|
walk(value, f"{path}.")
|
|
else:
|
|
keys.add(path)
|
|
|
|
walk(data)
|
|
return keys
|
|
|
|
|
|
# --------------------------------------------------------------------------- readiness
|
|
|
|
|
|
def test_readiness_reports_all_six_rows_in_the_frozen_order(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
|
|
checks = controller.readiness()
|
|
|
|
assert [check.key for check in checks] == list(READINESS_KEYS)
|
|
assert list(READINESS_KEYS) == [
|
|
"calibration_bundle",
|
|
"input_frames",
|
|
"bus_ready",
|
|
"output_writable",
|
|
"sensor_sn",
|
|
"max_force",
|
|
]
|
|
assert [check.label_key for check in checks] == [
|
|
label for _key, label in READINESS_ITEMS
|
|
]
|
|
assert all(check.ok for check in checks)
|
|
assert controller.ready
|
|
|
|
|
|
def test_a_failing_check_keeps_all_six_rows_and_blocks_the_start(tmp_dir):
|
|
def probe():
|
|
return [
|
|
ReadinessCheck(
|
|
"calibration_bundle", False, "cap.check.bundle", "bundle missing"
|
|
)
|
|
]
|
|
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out", readiness=probe)
|
|
checks = {check.key: check for check in controller.readiness()}
|
|
|
|
assert [check.key for check in controller.readiness()] == list(READINESS_KEYS)
|
|
assert checks["calibration_bundle"].ok is False
|
|
assert checks["calibration_bundle"].detail == "bundle missing"
|
|
# An unreported fact is not a satisfied one.
|
|
for key in ("input_frames", "bus_ready", "output_writable", "sensor_sn"):
|
|
assert checks[key].ok is False
|
|
assert checks[key].detail == "the readiness probe did not report this check"
|
|
assert controller.ready is False
|
|
|
|
controller.perform("align_point")
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned()
|
|
action = controller.primary_action()
|
|
assert action.key == "start_session"
|
|
assert action.enabled is False
|
|
assert action.blocked_reason == "cap.block.not_ready"
|
|
with pytest.raises(ActionBlocked) as info:
|
|
controller.perform("start_session")
|
|
assert info.value.reason == "cap.block.not_ready"
|
|
|
|
|
|
def test_a_probe_that_raises_does_not_break_the_page(tmp_dir):
|
|
def probe():
|
|
raise RuntimeError("probe exploded")
|
|
|
|
controller = _controller(_campaign(), tmp_dir / "out", readiness=probe)
|
|
checks = {check.key: check for check in controller.readiness()}
|
|
|
|
assert checks["calibration_bundle"].ok is False
|
|
assert checks["calibration_bundle"].detail == "probe exploded"
|
|
assert checks["max_force"].ok is True # the controller owns that fact
|
|
assert controller.ready is False
|
|
|
|
|
|
@pytest.mark.parametrize("max_force", [None, 0.0, -1.0])
|
|
def test_max_force_is_the_controllers_own_fact(tmp_dir, max_force):
|
|
def probe():
|
|
# Even a probe that claims the force is fine cannot set this fact.
|
|
return [ReadinessCheck(key, True, label) for key, label in READINESS_ITEMS]
|
|
|
|
controller = _controller(
|
|
_campaign(), tmp_dir / "out", readiness=probe, max_force_n=max_force
|
|
)
|
|
checks = {check.key: check for check in controller.readiness()}
|
|
|
|
assert checks["max_force"].ok is False
|
|
assert checks["max_force"].detail == "max_force_n is not set"
|
|
assert controller.ready is False
|
|
|
|
|
|
def test_a_max_force_above_zero_passes_the_sixth_check(tmp_dir):
|
|
controller = _controller(_campaign(), tmp_dir / "out", max_force_n=0.001)
|
|
checks = {check.key: check for check in controller.readiness()}
|
|
assert checks["max_force"].ok is True
|
|
assert checks["max_force"].detail == ""
|
|
|
|
|
|
# --------------------------------------------------------------------------- the ladder
|
|
|
|
|
|
def test_the_primary_button_walks_the_whole_load_ladder(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
observed: list[tuple[str, str, bool]] = []
|
|
|
|
def snapshot(label: str):
|
|
action = controller.primary_action()
|
|
_assert_action_reasons(controller)
|
|
observed.append((label, action.key, action.enabled))
|
|
return action
|
|
|
|
action = snapshot("idle")
|
|
assert (action.key, action.enabled) == ("align_point", True)
|
|
controller.perform("align_point")
|
|
request = controller.pending_alignment
|
|
assert request is not None
|
|
assert request.site_id == "P1"
|
|
assert request.targets == {
|
|
key: POSE[key] for key in ("x", "y", "yaw", "pitch", "z")
|
|
}
|
|
controller.pending_alignment = None
|
|
assert controller.pending_alignment is None
|
|
# The request alone did not move anything; the window confirms the move.
|
|
controller.mark_aligned()
|
|
|
|
action = snapshot("aligned")
|
|
assert (action.key, action.enabled) == ("start_session", True)
|
|
controller.perform("start_session")
|
|
assert campaign.phase == "load"
|
|
|
|
for index in range(3):
|
|
action = snapshot(f"load{index}")
|
|
assert (action.key, action.enabled) == ("confirm_target", True)
|
|
# Simulated capture never records, so the alignment guard is not involved.
|
|
assert campaign.capture_state == "idle"
|
|
controller.perform(
|
|
"confirm_target", force_n=float(campaign.progress()["target_n"])
|
|
)
|
|
action = snapshot(f"hold{index}-waiting")
|
|
assert action.key == "confirm_hold"
|
|
assert action.enabled is False
|
|
assert action.blocked_reason == "cap.block.hold_running"
|
|
_sleep_hold()
|
|
action = snapshot(f"hold{index}")
|
|
assert (action.key, action.enabled) == ("confirm_hold", True)
|
|
controller.perform("confirm_hold")
|
|
|
|
action = snapshot("done")
|
|
assert (action.key, action.enabled) == ("finish_session", True)
|
|
controller.perform("finish_session")
|
|
# Finishing closes the session out; the files stay as they were.
|
|
assert campaign.phase == "idle"
|
|
|
|
assert observed == [
|
|
("idle", "align_point", True),
|
|
("aligned", "start_session", True),
|
|
("load0", "confirm_target", True),
|
|
("hold0-waiting", "confirm_hold", False),
|
|
("hold0", "confirm_hold", True),
|
|
("load1", "confirm_target", True),
|
|
("hold1-waiting", "confirm_hold", False),
|
|
("hold1", "confirm_hold", True),
|
|
("load2", "confirm_target", True),
|
|
("hold2-waiting", "confirm_hold", False),
|
|
("hold2", "confirm_hold", True),
|
|
("done", "finish_session", True),
|
|
]
|
|
assert _manifest(campaign)["status"] == "complete"
|
|
rows = _reference_rows(campaign)["sites/P1/repeat-001"]
|
|
assert [row["branch"] for row in rows] == ["baseline", "load", "unload"]
|
|
assert [row["hold_complete"] for row in rows] == ["1", "1", "1"]
|
|
|
|
|
|
def test_the_next_repeat_of_a_point_needs_no_alignment(tmp_dir):
|
|
campaign = _campaign(repeats=2)
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
|
|
for _step in range(3): # one repeat: baseline, load, unload
|
|
_drive_step(controller, campaign)
|
|
|
|
assert campaign.phase == "load" # the second repeat opened directly
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("confirm_target", True)
|
|
|
|
for _step in range(3):
|
|
_drive_step(controller, campaign)
|
|
assert campaign.phase == "done"
|
|
assert _manifest(campaign)["status"] == "complete"
|
|
assert len(_reference_rows(campaign)) == 2 # two repeats recorded
|
|
|
|
|
|
def test_the_next_point_is_aligned_during_liftoff(tmp_dir):
|
|
campaign = _campaign(sites=("P1", "P2"))
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
_drive_repeat(controller, campaign)
|
|
|
|
assert campaign.phase == "liftoff"
|
|
state = controller.poll()
|
|
assert state["pending_alignment_site"] == "P2"
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("align_point", True)
|
|
controller.perform("align_point")
|
|
request = controller.pending_alignment
|
|
assert request is not None and request.site_id == "P2"
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned()
|
|
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("confirm_liftoff", True)
|
|
controller.perform("confirm_liftoff")
|
|
assert campaign.phase == "load"
|
|
assert campaign.progress()["site_id"] == "P2"
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("confirm_target", True)
|
|
|
|
_drive_repeat(controller, campaign)
|
|
assert campaign.phase == "done"
|
|
assert [row.site_id for row in controller.model().sites] == ["P1", "P2"]
|
|
|
|
|
|
def test_without_a_solver_the_liftoff_still_advances(tmp_dir):
|
|
campaign = _campaign(sites=("P1", "P2"))
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
_drive_repeat(controller, campaign)
|
|
|
|
# The solver goes away after the session started: aligning the next point is
|
|
# impossible, so the ladder must still offer the liftoff confirmation rather
|
|
# than leave the operator stuck.
|
|
controller.solver = None
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("confirm_liftoff", True)
|
|
controller.perform("confirm_liftoff")
|
|
assert campaign.phase == "load"
|
|
assert campaign.progress()["site_id"] == "P2"
|
|
|
|
|
|
# --------------------------------------------------------------------------- alignment confirmation
|
|
|
|
|
|
def test_an_alignment_request_alone_does_not_make_the_point_ready(tmp_dir):
|
|
"""A requested move can still be refused, cancelled or fail: no auto-confirm."""
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "idle")
|
|
|
|
controller.perform("align_point")
|
|
|
|
assert controller.pending_alignment is not None # the request is out
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("align_point", True)
|
|
# Start is not even offered: an unconfirmed pose must never begin a session.
|
|
with pytest.raises(ValueError):
|
|
controller.perform("start_session")
|
|
|
|
campaign = _campaign(sites=("P1", "P2"))
|
|
controller = _controller(campaign, tmp_dir / "liftoff")
|
|
_aligned_and_started(controller, campaign)
|
|
_drive_repeat(controller, campaign)
|
|
assert campaign.phase == "liftoff"
|
|
|
|
controller.perform("align_point")
|
|
assert controller.pending_alignment is not None
|
|
assert controller.primary_action().key == "align_point"
|
|
with pytest.raises(ValueError):
|
|
controller.perform("confirm_liftoff")
|
|
|
|
|
|
def test_marking_the_move_on_offers_the_next_step(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "idle")
|
|
controller.perform("align_point")
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned()
|
|
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("start_session", True)
|
|
|
|
campaign = _campaign(sites=("P1", "P2"))
|
|
controller = _controller(campaign, tmp_dir / "liftoff")
|
|
_aligned_and_started(controller, campaign)
|
|
_drive_repeat(controller, campaign)
|
|
controller.perform("align_point")
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned()
|
|
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("confirm_liftoff", True)
|
|
|
|
|
|
def test_a_failed_move_leaves_the_point_unaligned_and_retryable(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
controller.perform("align_point")
|
|
controller.pending_alignment = None
|
|
|
|
controller.mark_alignment_failed() # refused, cancelled or failed on the way
|
|
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("align_point", True)
|
|
with pytest.raises(ValueError):
|
|
controller.perform("start_session")
|
|
|
|
# The retry path is still available, and an explicit site id is accepted.
|
|
controller.perform("align_point")
|
|
assert controller.pending_alignment is not None
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned("P1")
|
|
assert controller.primary_action().key == "start_session"
|
|
|
|
controller.mark_alignment_failed("P1")
|
|
assert controller.primary_action().key == "align_point"
|
|
|
|
|
|
def test_the_result_report_tolerates_a_queue_with_no_point_in_focus(tmp_dir):
|
|
controller = _controller(_campaign(sites=()), tmp_dir / "empty")
|
|
|
|
controller.mark_alignment_failed() # an error path must never raise
|
|
with pytest.raises(ValueError):
|
|
controller.mark_aligned() # a confirmation with no point is a caller bug
|
|
|
|
|
|
# --------------------------------------------------------------------------- disabled actions
|
|
|
|
|
|
def test_start_session_reports_each_missing_precondition(tmp_dir):
|
|
empty = _campaign(sites=())
|
|
action = _controller(empty, tmp_dir / "empty").primary_action()
|
|
assert (action.key, action.enabled) == ("start_session", False)
|
|
assert action.blocked_reason == "cap.block.no_sites"
|
|
|
|
no_arguments = _campaign()
|
|
controller = _controller(no_arguments, tmp_dir / "arguments", start_arguments=None)
|
|
controller.perform("align_point")
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned()
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("start_session", False)
|
|
assert action.blocked_reason == "cap.block.no_session"
|
|
with pytest.raises(ActionBlocked) as info:
|
|
controller.perform("start_session")
|
|
assert info.value.reason == "cap.block.no_session"
|
|
|
|
|
|
def test_align_point_reports_whether_the_point_can_be_commanded(tmp_dir):
|
|
no_solver = _controller(_campaign(), tmp_dir / "solver", solver=None)
|
|
action = no_solver.primary_action()
|
|
assert (action.key, action.enabled) == ("align_point", False)
|
|
assert action.blocked_reason == "cap.block.no_solver"
|
|
with pytest.raises(ActionBlocked) as info:
|
|
no_solver.perform("align_point")
|
|
assert info.value.reason == "cap.block.no_solver"
|
|
|
|
unposed = Campaign()
|
|
unposed.add_site("P1", 1.0, 2.0, 3.0, [5.0], HOLD_S, 1, max_force_n=MAX_FORCE)
|
|
controller = _controller(unposed, tmp_dir / "pose")
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("align_point", False)
|
|
assert action.blocked_reason == "cap.block.no_pose"
|
|
with pytest.raises(ActionBlocked) as info:
|
|
controller.perform("align_point")
|
|
assert info.value.reason == "cap.block.no_pose"
|
|
|
|
|
|
def test_a_move_in_the_middle_of_a_recording_is_refused(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
# What ``campaign.attach_capture`` does once the page has a camera frame.
|
|
campaign.capture_state = "recording"
|
|
|
|
# The normal ladder never offers a move here; the guard is for a direct call.
|
|
assert controller.primary_action().key == "confirm_target"
|
|
with pytest.raises(ActionBlocked) as info:
|
|
controller.perform("align_point")
|
|
assert info.value.reason == "cap.block.align_while_recording"
|
|
assert info.value.label_key == "cap.action.align"
|
|
|
|
controller.perform("confirm_target", force_n=0.0)
|
|
assert campaign.phase == "hold"
|
|
with pytest.raises(ActionBlocked) as info:
|
|
controller.perform("align_point")
|
|
assert info.value.reason == "cap.block.align_while_recording"
|
|
|
|
|
|
def _sweep_states(tmp_dir):
|
|
"""Every distinct controller state a disabled reason can come from."""
|
|
|
|
def failing_probe():
|
|
return [ReadinessCheck("bus_ready", False, "cap.check.bus")]
|
|
|
|
cases = []
|
|
|
|
campaign = _campaign(sites=())
|
|
cases.append(("empty", _controller(campaign, tmp_dir / "empty"), campaign))
|
|
|
|
campaign = _campaign()
|
|
cases.append(("idle", _controller(campaign, tmp_dir / "idle"), campaign))
|
|
|
|
campaign = _campaign()
|
|
cases.append(
|
|
(
|
|
"idle-no-solver",
|
|
_controller(campaign, tmp_dir / "nossolver", solver=None),
|
|
campaign,
|
|
)
|
|
)
|
|
|
|
unposed = Campaign()
|
|
unposed.add_site("P1", 1.0, 2.0, 3.0, [5.0], HOLD_S, 1, max_force_n=MAX_FORCE)
|
|
cases.append(("idle-no-pose", _controller(unposed, tmp_dir / "nopose"), unposed))
|
|
|
|
campaign = _campaign()
|
|
cases.append(
|
|
(
|
|
"idle-not-ready",
|
|
_controller(campaign, tmp_dir / "notready", readiness=failing_probe),
|
|
campaign,
|
|
)
|
|
)
|
|
|
|
campaign = _campaign()
|
|
cases.append(
|
|
(
|
|
"idle-no-session",
|
|
_controller(campaign, tmp_dir / "nosession", start_arguments=None),
|
|
campaign,
|
|
)
|
|
)
|
|
|
|
campaign = _campaign(sites=("P1", "P2"))
|
|
controller = _controller(campaign, tmp_dir / "load")
|
|
_aligned_and_started(controller, campaign)
|
|
cases.append(("load", controller, campaign))
|
|
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "hold")
|
|
_aligned_and_started(controller, campaign)
|
|
controller.perform("confirm_target", force_n=0.0)
|
|
cases.append(("hold", controller, campaign))
|
|
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "recording")
|
|
_aligned_and_started(controller, campaign)
|
|
campaign.capture_state = "recording"
|
|
cases.append(("recording", controller, campaign))
|
|
|
|
campaign = _campaign(sites=("P1", "P2"))
|
|
controller = _controller(campaign, tmp_dir / "liftoff")
|
|
_aligned_and_started(controller, campaign)
|
|
_drive_repeat(controller, campaign)
|
|
cases.append(("liftoff", controller, campaign))
|
|
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "closing")
|
|
campaign.capture_state = "closing"
|
|
cases.append(("closing", controller, campaign))
|
|
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "done")
|
|
_aligned_and_started(controller, campaign)
|
|
_drive_repeat(controller, campaign)
|
|
cases.append(("done", controller, campaign))
|
|
|
|
return cases
|
|
|
|
|
|
def test_every_disabled_action_names_a_locale_reason_and_the_key_exists(tmp_dir):
|
|
referenced: set[str] = set()
|
|
for name, controller, _campaign_ in _sweep_states(tmp_dir):
|
|
actions = [controller.primary_action(), *controller.secondary_actions()]
|
|
for action in actions:
|
|
assert action.key in PRIMARY_KEYS | set(SECONDARY_KEYS), (name, action)
|
|
referenced.add(action.label_key)
|
|
if action.enabled:
|
|
assert action.blocked_reason == "", (name, action)
|
|
else:
|
|
assert action.blocked_reason, (name, action)
|
|
assert action.blocked_reason.startswith("cap."), (name, action)
|
|
referenced.add(action.blocked_reason)
|
|
|
|
# A reason is a locale key: a key the pages cannot translate is a blank label.
|
|
for language in ("zh_CN", "en_US"):
|
|
missing = referenced - _flat_locale(language)
|
|
assert not missing, (language, sorted(missing))
|
|
|
|
|
|
# --------------------------------------------------------------------------- skip
|
|
|
|
|
|
def test_skip_is_offered_while_a_point_is_open_and_not_recording(tmp_dir):
|
|
campaign = _campaign(sites=("P1", "P2"))
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
|
|
action = {item.key: item for item in controller.secondary_actions()}["skip"]
|
|
assert action.key == "skip"
|
|
assert action.enabled is True
|
|
assert action.blocked_reason == ""
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("sites", "expected_phase", "expected_status"),
|
|
[
|
|
(("P1", "P2"), "liftoff", "in_progress"),
|
|
(("P1",), "done", "incomplete"),
|
|
],
|
|
)
|
|
def test_skip_advances_the_campaign_past_the_current_point(
|
|
tmp_dir, sites, expected_phase, expected_status
|
|
):
|
|
campaign = _campaign(sites=sites)
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
if len(sites) > 1:
|
|
_drive_step(controller, campaign) # leaves one finished hold to roll back
|
|
assert campaign._completed_holds == 1 # the counter is the claim
|
|
|
|
controller.perform("skip")
|
|
|
|
assert campaign.phase == expected_phase
|
|
assert campaign._completed_holds == 0
|
|
manifest = _manifest(campaign)
|
|
assert manifest["status"] == expected_status
|
|
assert manifest["trials"] == [] # the unfinished record is removed, not re-statused
|
|
assert not (campaign.directory / "sites" / "P1" / "repeat-001").exists()
|
|
assert [site.site_id for site in campaign.sites] == list(sites)
|
|
|
|
skipped = [
|
|
row for row in _campaign_events(campaign) if row["kind"] == "site_skipped"
|
|
]
|
|
assert skipped
|
|
assert json.loads(skipped[-1]["detail"])["site_id"] == "P1"
|
|
|
|
if expected_phase == "liftoff":
|
|
assert campaign.pending_alignment_site == "P2"
|
|
else:
|
|
assert controller.primary_action().key == "finish_session"
|
|
|
|
|
|
def test_skip_never_touches_a_recording_repeat(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
# What the page looks like once ``campaign.attach_capture`` owns the repeat.
|
|
campaign._recorder = _FakeRecorder() # the writer guard is the claim
|
|
campaign.capture_state = "recording"
|
|
|
|
with pytest.raises(RuntimeError, match="stop the recording before skipping"):
|
|
campaign.skip_current_point()
|
|
|
|
action = {item.key: item for item in controller.secondary_actions()}["skip"]
|
|
assert action.enabled is False
|
|
assert action.blocked_reason == "cap.block.skip_live"
|
|
with pytest.raises(ActionBlocked) as info:
|
|
controller.perform("skip")
|
|
assert info.value.reason == "cap.block.skip_live"
|
|
|
|
|
|
@pytest.mark.parametrize("state", ["idle", "done", "closing"])
|
|
def test_skip_is_busy_when_no_repeat_is_open(tmp_dir, state):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / state)
|
|
if state == "done":
|
|
_aligned_and_started(controller, campaign)
|
|
_drive_repeat(controller, campaign)
|
|
assert campaign.phase == "done"
|
|
elif state == "closing":
|
|
campaign.capture_state = "closing"
|
|
|
|
action = {item.key: item for item in controller.secondary_actions()}["skip"]
|
|
assert action.enabled is False
|
|
assert action.blocked_reason == "cap.block.busy"
|
|
with pytest.raises(ActionBlocked) as info:
|
|
controller.perform("skip")
|
|
assert info.value.reason == "cap.block.busy"
|
|
|
|
|
|
# --------------------------------------------------------------------------- stop / closing
|
|
|
|
|
|
def _state_for(root: Path, state: str) -> tuple[CaptureController, Campaign]:
|
|
if state == "idle":
|
|
campaign = _campaign()
|
|
return _controller(campaign, root / "idle"), campaign
|
|
campaign = _campaign(sites=("P1", "P2")) if state == "liftoff" else _campaign()
|
|
controller = _controller(campaign, root / state)
|
|
_aligned_and_started(controller, campaign)
|
|
if state == "hold":
|
|
controller.perform("confirm_target", force_n=0.0)
|
|
elif state in {"liftoff", "done"}:
|
|
_drive_repeat(controller, campaign)
|
|
return controller, campaign
|
|
|
|
|
|
@pytest.mark.parametrize("state", ["idle", "load", "hold", "liftoff", "done"])
|
|
def test_stop_is_offered_exactly_while_a_session_is_running(tmp_dir, state):
|
|
controller, campaign = _state_for(tmp_dir, state)
|
|
action = {item.key: item for item in controller.secondary_actions()}["stop"]
|
|
|
|
if state in {"idle", "done"}:
|
|
assert action.enabled is False
|
|
assert action.blocked_reason == "cap.block.stop_unavailable"
|
|
with pytest.raises(ActionBlocked) as info:
|
|
controller.perform("stop")
|
|
assert info.value.reason == "cap.block.stop_unavailable"
|
|
return
|
|
|
|
assert action.enabled is True
|
|
controller.perform("stop")
|
|
assert campaign.phase == "aborted"
|
|
assert campaign.is_active is False
|
|
manifest = _manifest(campaign)
|
|
assert manifest["status"] == "incomplete"
|
|
assert manifest["reason"] == "operator stopped"
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("finish_session", True)
|
|
|
|
|
|
def test_stop_can_carry_the_shutdown_reason(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
|
|
controller.perform("stop", reason="application closed")
|
|
|
|
assert campaign.phase == "aborted"
|
|
assert _manifest(campaign)["reason"] == "application closed"
|
|
|
|
|
|
def test_a_blank_stop_reason_is_refused_before_stopping(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
|
|
with pytest.raises(ValueError):
|
|
controller.perform("stop", reason=" ")
|
|
|
|
# The refusal happens before the campaign is touched: the session still runs.
|
|
assert campaign.phase == "load"
|
|
assert campaign.is_active is True
|
|
|
|
controller.perform("stop", reason=None) # the default wording still applies
|
|
assert campaign.phase == "aborted"
|
|
assert _manifest(campaign)["reason"] == "operator stopped"
|
|
|
|
|
|
def test_closing_is_a_read_only_primary_action(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
campaign.capture_state = "closing"
|
|
|
|
action = controller.primary_action()
|
|
assert (action.key, action.label_key) == ("closing", "cap.block.closing")
|
|
assert action.enabled is False
|
|
assert action.blocked_reason == "cap.block.closing"
|
|
with pytest.raises(ActionBlocked) as info:
|
|
controller.perform("closing")
|
|
assert info.value.reason == "cap.block.closing"
|
|
assert info.value.label_key == "cap.block.closing"
|
|
|
|
# poll() is what finalises the close; the page must call it every tick.
|
|
state = controller.poll()
|
|
assert state["capture_state"] != "closing"
|
|
|
|
|
|
# --------------------------------------------------------------------------- queue editing
|
|
|
|
|
|
def _six_axis_rotation() -> list[float]:
|
|
return [1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 1.0]
|
|
|
|
|
|
def test_the_model_carries_every_field_the_queue_renders(tmp_dir):
|
|
campaign = Campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
controller.add_point(
|
|
"P1",
|
|
1.0,
|
|
2.0,
|
|
3.0,
|
|
(0.0, 0.0, 1.0),
|
|
dict(POSE),
|
|
[5.0, 10.0],
|
|
HOLD_S,
|
|
2,
|
|
uv_px=(11.0, 22.0),
|
|
)
|
|
|
|
row = controller.model().sites[0]
|
|
site = campaign.sites[0]
|
|
assert row.site_id == site.site_id == "P1"
|
|
assert row.position == {"x": 1.0, "y": 2.0, "z": 3.0}
|
|
assert row.repeats_planned == site.repeats == 2
|
|
assert row.repeats_complete == 0
|
|
assert row.complete is False
|
|
assert row.reachable is True
|
|
assert row.levels == tuple(site.levels_n) == (5.0, 10.0)
|
|
assert row.hold_s == site.hold_s == HOLD_S
|
|
assert row.uv_px == site.uv_px == (11.0, 22.0)
|
|
assert row.normal == site.normal == (0.0, 0.0, 1.0)
|
|
assert row.pose == site.pose
|
|
assert row.pose["x"] == POSE["x"]
|
|
assert row.pose is not site.pose # a snapshot, not a handle into the campaign
|
|
assert row.position_frame == site.position_frame == "sensor"
|
|
assert row.skipped is False
|
|
|
|
# A point without a solved pose still renders: the optional fields are None.
|
|
plain = Campaign()
|
|
plain.add_site("P9", 1.0, 2.0, 3.0, [5.0], HOLD_S, 1, max_force_n=MAX_FORCE)
|
|
blank = _controller(plain, tmp_dir / "plain").model().sites[0]
|
|
assert (blank.normal, blank.pose, blank.uv_px) == (None, None, None)
|
|
assert blank.position_frame == ""
|
|
|
|
|
|
def test_the_queue_can_be_edited_while_idle(tmp_dir):
|
|
campaign = Campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
controller.add_point(
|
|
"P1", 1.0, 2.0, 3.0, (0.0, 0.0, 1.0), dict(POSE), [5.0], HOLD_S, 1
|
|
)
|
|
controller.add_point(
|
|
"P2", 4.0, 5.0, 6.0, (0.0, 0.0, 1.0), dict(POSE), [10.0], HOLD_S, 2
|
|
)
|
|
assert [site.site_id for site in campaign.sites] == ["P1", "P2"]
|
|
|
|
controller.update_point("P2", 7.0, 8.0, 9.0, [10.0, 15.0], 0.5, 3, new_id="P2b")
|
|
|
|
updated = campaign.sites[1]
|
|
assert updated.site_id == "P2b"
|
|
assert (updated.x, updated.y, updated.z) == (7.0, 8.0, 9.0)
|
|
assert tuple(updated.levels_n) == (10.0, 15.0)
|
|
assert updated.hold_s == 0.5
|
|
assert updated.repeats == 3
|
|
|
|
controller.move_point("P2b", -1)
|
|
assert [site.site_id for site in campaign.sites] == ["P2b", "P1"]
|
|
|
|
controller.set_force_rotation(_six_axis_rotation())
|
|
assert campaign._rotation is not None # the rotation the campaign will use
|
|
|
|
controller.remove_point("P1")
|
|
assert [site.site_id for site in campaign.sites] == ["P2b"]
|
|
|
|
|
|
@pytest.mark.parametrize("state", ["load", "hold", "liftoff", "done"])
|
|
def test_the_queue_cannot_be_edited_outside_idle(tmp_dir, state):
|
|
controller, campaign = _state_for(tmp_dir, state)
|
|
before = [(site.site_id, site.x, site.y, site.z) for site in campaign.sites]
|
|
calls = (
|
|
lambda: controller.add_point(
|
|
"PX", 1.0, 2.0, 3.0, (0.0, 0.0, 1.0), dict(POSE), [5.0], HOLD_S, 1
|
|
),
|
|
lambda: controller.update_point("P1", 1.0, 2.0, 3.0, [5.0], HOLD_S, 1),
|
|
lambda: controller.remove_point("P1"),
|
|
lambda: controller.move_point("P1", 1),
|
|
lambda: controller.set_force_rotation(_six_axis_rotation()),
|
|
lambda: controller.store_solved_pose("P1", dict(POSE)),
|
|
)
|
|
for call in calls:
|
|
with pytest.raises(ActionBlocked) as info:
|
|
call()
|
|
assert info.value.reason == "cap.block.queue_busy"
|
|
assert [(site.site_id, site.x, site.y, site.z) for site in campaign.sites] == before
|
|
|
|
|
|
def test_adding_or_updating_a_point_needs_a_maximum_force(tmp_dir):
|
|
campaign = Campaign()
|
|
controller = _controller(campaign, tmp_dir / "out", max_force_n=None)
|
|
|
|
with pytest.raises(ActionBlocked) as info:
|
|
controller.add_point(
|
|
"P1", 1.0, 2.0, 3.0, (0.0, 0.0, 1.0), dict(POSE), [5.0], HOLD_S, 1
|
|
)
|
|
assert info.value.reason == "cap.block.no_max_force"
|
|
with pytest.raises(ActionBlocked) as info:
|
|
controller.update_point("P1", 1.0, 2.0, 3.0, [5.0], HOLD_S, 1)
|
|
assert info.value.reason == "cap.block.no_max_force"
|
|
assert campaign.sites == []
|
|
|
|
# Both reasons are locale keys the pages can translate.
|
|
for language in ("zh_CN", "en_US"):
|
|
keys = _flat_locale(language)
|
|
assert {"cap.block.queue_busy", "cap.block.no_max_force"} <= keys
|
|
|
|
|
|
def test_storing_a_re_solved_pose_reaches_the_model_as_a_copy(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
before = controller.model().sites[0].pose
|
|
assert before is not None
|
|
|
|
solved = dict(POSE, x=42.0, y=-3.0, z=17.5, residual_deg=0.25)
|
|
controller.store_solved_pose("P1", solved)
|
|
|
|
row = controller.model().sites[0]
|
|
site = campaign.sites[0]
|
|
assert row.pose == site.pose
|
|
assert row.pose["x"] == 42.0
|
|
assert row.pose["residual_deg"] == 0.25
|
|
assert row.pose is not site.pose # the row is a snapshot, not the record
|
|
assert row.normal == site.normal # a pose store does not touch the normal
|
|
|
|
# A re-solve drops the "this point is aligned" flag: the old move aimed elsewhere.
|
|
controller.mark_aligned()
|
|
assert controller.primary_action().key == "start_session"
|
|
controller.store_solved_pose("P1", dict(POSE, residual_deg=0.5))
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("align_point", True)
|
|
|
|
|
|
def test_setting_the_maximum_force_reopens_the_queue(tmp_dir):
|
|
campaign = Campaign()
|
|
controller = _controller(campaign, tmp_dir / "out", max_force_n=None)
|
|
assert {check.key: check for check in controller.readiness()}[
|
|
"max_force"
|
|
].ok is False
|
|
with pytest.raises(ActionBlocked) as info:
|
|
controller.add_point(
|
|
"P1", 1.0, 2.0, 3.0, (0.0, 0.0, 1.0), dict(POSE), [5.0], HOLD_S, 1
|
|
)
|
|
assert info.value.reason == "cap.block.no_max_force"
|
|
|
|
controller.set_max_force(25.0)
|
|
|
|
checks = {check.key: check for check in controller.readiness()}
|
|
assert checks["max_force"].ok is True
|
|
assert controller.ready is True
|
|
controller.add_point(
|
|
"P1", 1.0, 2.0, 3.0, (0.0, 0.0, 1.0), dict(POSE), [20.0], HOLD_S, 1
|
|
)
|
|
assert [site.site_id for site in campaign.sites] == ["P1"]
|
|
# The new value is what a queue edit is validated against.
|
|
with pytest.raises(ValueError):
|
|
controller.add_point(
|
|
"P2", 1.0, 2.0, 3.0, (0.0, 0.0, 1.0), dict(POSE), [30.0], HOLD_S, 1
|
|
)
|
|
controller.set_max_force(40.0)
|
|
controller.add_point(
|
|
"P2", 1.0, 2.0, 3.0, (0.0, 0.0, 1.0), dict(POSE), [30.0], HOLD_S, 1
|
|
)
|
|
assert [site.site_id for site in campaign.sites] == ["P1", "P2"]
|
|
|
|
controller.set_max_force(None)
|
|
assert {check.key: check for check in controller.readiness()}[
|
|
"max_force"
|
|
].ok is False
|
|
with pytest.raises(ActionBlocked) as info:
|
|
controller.add_point(
|
|
"P3", 1.0, 2.0, 3.0, (0.0, 0.0, 1.0), dict(POSE), [5.0], HOLD_S, 1
|
|
)
|
|
assert info.value.reason == "cap.block.no_max_force"
|
|
|
|
|
|
def test_a_skipped_point_is_marked_in_the_queue(tmp_dir):
|
|
campaign = _campaign(sites=("P1", "P2"))
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
assert [row.skipped for row in controller.model().sites] == [False, False]
|
|
|
|
controller.perform("skip")
|
|
|
|
assert [row.skipped for row in controller.model().sites] == [True, False]
|
|
assert [row.site_id for row in controller.model().sites] == ["P1", "P2"]
|
|
|
|
|
|
def test_starting_a_session_clears_stale_skip_marks(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
controller._skipped.add("P1") # a mark a finished session would leave behind
|
|
assert controller.model().sites[0].skipped is True
|
|
|
|
controller.perform("align_point")
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned()
|
|
controller.perform("start_session")
|
|
|
|
assert controller.model().sites[0].skipped is False
|
|
|
|
|
|
def test_removing_a_point_forgets_its_skip_mark(tmp_dir):
|
|
campaign = _campaign(sites=("P1", "P2"))
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
controller._skipped.add("P2")
|
|
assert controller.model().sites[1].skipped is True
|
|
|
|
controller.remove_point("P2")
|
|
assert [row.site_id for row in controller.model().sites] == ["P1"]
|
|
|
|
controller.add_point(
|
|
"P2", 1.0, 2.0, 3.0, (0.0, 0.0, 1.0), dict(POSE), [5.0], HOLD_S, 1
|
|
)
|
|
assert controller.model().sites[1].skipped is False
|
|
|
|
|
|
# --------------------------------------------------------------------------- poll / model / perform
|
|
|
|
|
|
def test_poll_advances_closing_and_reports_the_frozen_contract(tmp_dir, monkeypatch):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
controller.perform("confirm_target", force_n=0.0)
|
|
assert campaign.phase == "hold"
|
|
|
|
calls: list[bool] = []
|
|
original = campaign.poll
|
|
|
|
def spy():
|
|
calls.append(True)
|
|
return original()
|
|
|
|
monkeypatch.setattr(campaign, "poll", spy)
|
|
state = controller.poll()
|
|
|
|
assert calls == [True]
|
|
assert set(state) == {
|
|
"phase",
|
|
"capture_state",
|
|
"capture_reason",
|
|
"remaining_s",
|
|
"pending_alignment_site",
|
|
}
|
|
assert state["phase"] == "hold"
|
|
assert state["remaining_s"] > 0
|
|
assert state["pending_alignment_site"] == ""
|
|
|
|
|
|
def test_the_model_reports_the_queue_and_follows_the_campaign(tmp_dir):
|
|
campaign = _campaign(sites=("P1", "P2"), repeats=2)
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
|
|
model = controller.model()
|
|
assert [row.site_id for row in model.sites] == ["P1", "P2"]
|
|
assert model.sites[0].position == {"x": 1.0, "y": 2.0, "z": 3.0}
|
|
assert model.sites[0].repeats_planned == 2
|
|
assert model.sites[0].repeats_complete == 0
|
|
assert model.sites[0].complete is False
|
|
assert all(row.reachable for row in model.sites)
|
|
assert model.current_index == 0
|
|
assert set(model.progress) == set(campaign.progress())
|
|
|
|
_aligned_and_started(controller, campaign)
|
|
for _step in range(3):
|
|
_drive_step(controller, campaign)
|
|
assert controller.model().sites[0].repeats_complete == 1
|
|
assert controller.model().current_index == 0
|
|
|
|
for _step in range(3): # finish P1's two repeats, then liftoff to P2
|
|
_drive_step(controller, campaign)
|
|
model = controller.model()
|
|
assert model.current_index == 1
|
|
assert model.sites[0].complete is True
|
|
assert model.sites[0].repeats_complete == 2
|
|
|
|
controller.perform("align_point")
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned()
|
|
controller.perform("confirm_liftoff")
|
|
assert controller.model().current_index == 1
|
|
|
|
|
|
def test_the_model_marks_an_unreachable_pose(tmp_dir):
|
|
campaign = Campaign()
|
|
bad_pose = dict(POSE, in_range=False, ok=False, reason="out of travel")
|
|
campaign.add_solved_site(
|
|
"P1",
|
|
1.0,
|
|
2.0,
|
|
3.0,
|
|
(0.0, 0.0, 1.0),
|
|
bad_pose,
|
|
[5.0],
|
|
HOLD_S,
|
|
1,
|
|
max_force_n=MAX_FORCE,
|
|
)
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
|
|
model = controller.model()
|
|
assert model.sites[0].reachable is False
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("align_point", False)
|
|
assert action.blocked_reason == "cap.block.no_pose"
|
|
|
|
|
|
def test_perform_rejects_keys_and_inputs_the_contract_does_not_offer(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
|
|
with pytest.raises(ValueError):
|
|
controller.perform("nonsense")
|
|
# confirm_target is not the current action while the page is idle.
|
|
with pytest.raises(ValueError):
|
|
controller.perform("confirm_target", force_n=1.0)
|
|
|
|
controller.perform("align_point")
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned()
|
|
controller.perform("start_session")
|
|
|
|
with pytest.raises(ValueError):
|
|
controller.perform("confirm_target") # neither force nor wrench
|
|
with pytest.raises(ValueError):
|
|
controller.perform("confirm_target", force_n=1.0, wrench=(1.0, 2.0, 3.0))
|
|
with pytest.raises(ValueError):
|
|
controller.perform("confirm_target", force_n=1.0, extra=True)
|
|
|
|
|
|
def test_a_six_axis_reference_carries_its_rotation(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
rotation = [0.0, -1.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0]
|
|
|
|
# A rotation belongs to the six-axis reading, never to a weight reading.
|
|
with pytest.raises(ValueError):
|
|
controller.perform("confirm_target", force_n=1.0, rotation=rotation)
|
|
with pytest.raises(ValueError):
|
|
controller.perform("confirm_target", rotation=rotation) # no wrench either
|
|
|
|
controller.perform("confirm_target", wrench=(1.0, 2.0, 5.0), rotation=rotation)
|
|
|
|
assert campaign.phase == "hold"
|
|
manifest = _manifest(campaign)
|
|
assert manifest["force_to_tool_rotation"] == rotation
|
|
_sleep_hold()
|
|
controller.perform("confirm_hold")
|
|
|
|
first = _reference_rows(campaign)["sites/P1/repeat-001"][0]
|
|
assert first["step_index"] == "0"
|
|
assert first["reference_source"] == "wrench"
|
|
assert (float(first["fx_N"]), float(first["fy_N"]), float(first["fz_N"])) == (
|
|
1.0,
|
|
2.0,
|
|
5.0,
|
|
)
|
|
# The projection is (R @ F) · tool_axis; it is not a raw axis.
|
|
assert float(first["F_ref_normal_N"]) == pytest.approx(5.0)
|
|
assert first["force_to_tool_rotation"] == ("0.0 -1.0 0.0 1.0 0.0 0.0 0.0 0.0 1.0")
|
|
assert first["hold_complete"] == "1"
|
|
|
|
# The rotation is optional: the next confirmation reuses the one just entered.
|
|
controller.perform("confirm_target", wrench=(0.0, 0.0, 4.0))
|
|
assert campaign.phase == "hold"
|
|
_sleep_hold()
|
|
controller.perform("confirm_hold")
|
|
second = _reference_rows(campaign)["sites/P1/repeat-001"][1]
|
|
assert second["force_to_tool_rotation"] == first["force_to_tool_rotation"]
|
|
assert float(second["F_ref_normal_N"]) == pytest.approx(4.0)
|
|
|
|
|
|
# --------------------------------------------------------------------------- remeasure / reacquire
|
|
|
|
|
|
def test_remeasure_discards_a_hold_in_progress(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
controller.perform("confirm_target", force_n=0.0)
|
|
|
|
actions = {item.key: item for item in controller.secondary_actions()}
|
|
assert actions["remeasure"].enabled is True
|
|
controller.perform("remeasure")
|
|
|
|
assert campaign.phase == "load"
|
|
assert campaign.progress()["target_n"] == 0.0 # the same step is waiting again
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("confirm_target", True)
|
|
|
|
controller.perform("confirm_target", force_n=0.0)
|
|
_sleep_hold()
|
|
controller.perform("confirm_hold")
|
|
rows = _reference_rows(campaign)["sites/P1/repeat-001"]
|
|
assert [row["hold_complete"] for row in rows] == ["0", "1"]
|
|
assert {row["step_index"] for row in rows} == {"0"}
|
|
|
|
|
|
def test_remeasure_reopens_the_step_that_just_finished(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
_drive_step(controller, campaign)
|
|
assert campaign.phase == "load"
|
|
assert campaign.progress()["target_n"] == 5.0
|
|
|
|
actions = {item.key: item for item in controller.secondary_actions()}
|
|
assert actions["remeasure"].enabled is True
|
|
controller.perform("remeasure")
|
|
|
|
assert campaign.phase == "load"
|
|
assert campaign.progress()["target_n"] == 0.0 # the finished step is open again
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("confirm_target", True)
|
|
|
|
|
|
def test_reacquire_follows_the_campaign_and_keeps_the_point_aligned(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
_drive_repeat(controller, campaign) # single point: the session finishes
|
|
assert campaign.phase == "done"
|
|
assert controller.primary_action().key == "finish_session"
|
|
|
|
campaign.reacquire("P1")
|
|
|
|
state = controller.poll()
|
|
assert state["phase"] == "load"
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("confirm_target", True) # no second move
|
|
model = controller.model()
|
|
assert model.current_index == 0
|
|
assert model.sites[0].repeats_complete == 1
|
|
|
|
_drive_repeat(controller, campaign) # the reacquired repeat
|
|
assert campaign.phase == "liftoff" # reacquire reopens the point by itself
|
|
controller.perform("confirm_liftoff")
|
|
assert campaign.phase == "done"
|
|
model = controller.model()
|
|
assert model.sites[0].repeats_complete == 2
|
|
assert model.sites[0].complete is True
|
|
|
|
|
|
# --------------------------------------------------------------------------- purity and structure
|
|
|
|
|
|
def test_the_controller_module_imports_no_qt_and_no_motion_service():
|
|
path = Path(capture_controller.__file__)
|
|
tree = ast.parse(path.read_text(encoding="utf-8"))
|
|
imported: set[str] = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
imported.update(alias.name for alias in node.names)
|
|
elif isinstance(node, ast.ImportFrom):
|
|
module = node.module or ""
|
|
if node.level:
|
|
module = "." * node.level + module
|
|
imported.add(module)
|
|
for module in imported:
|
|
assert not module.startswith(("PySide", "PyQt")), module
|
|
assert module not in {"orisys_autocal.motion.service", ".motion.service"}, (
|
|
module
|
|
)
|
|
assert ".multipoint" in imported # it wraps a Campaign and nothing lower
|
|
|
|
|
|
VOLATILE_SESSION_KEYS = {"session_id", "started_at", "ended_at"}
|
|
VOLATILE_REFERENCE_COLUMNS = {"t_start_host_mono_ns", "t_end_host_mono_ns"}
|
|
VOLATILE_EVENT_COLUMNS = {"t_host_mono_ns", "t_host_utc_ns"}
|
|
|
|
|
|
def _masked(node, volatile: set[str]):
|
|
if isinstance(node, dict):
|
|
return {
|
|
key: ("<masked>" if key in volatile else _masked(value, volatile))
|
|
for key, value in node.items()
|
|
}
|
|
if isinstance(node, list):
|
|
return [_masked(item, volatile) for item in node]
|
|
return node
|
|
|
|
|
|
def _masked_rows(rows: list[dict], volatile: set[str]) -> list[dict]:
|
|
return [
|
|
{key: ("<masked>" if key in volatile else value) for key, value in row.items()}
|
|
for row in rows
|
|
]
|
|
|
|
|
|
def _masked_events(path: Path) -> list[dict]:
|
|
with path.open(encoding="utf-8-sig", newline="") as handle:
|
|
return _masked_rows(list(csv.DictReader(handle)), VOLATILE_EVENT_COLUMNS)
|
|
|
|
|
|
def _tree(root: Path) -> list[str]:
|
|
return sorted(
|
|
path.relative_to(root).as_posix() for path in root.rglob("*") if path.is_file()
|
|
)
|
|
|
|
|
|
def _drive_direct(campaign: Campaign, root: Path) -> Campaign:
|
|
"""Drive the campaign without the controller, as the old page did."""
|
|
campaign.start(
|
|
root,
|
|
sensor_model="finger",
|
|
sensor_sn="SN1",
|
|
coordinate_frame="sensor",
|
|
max_force_n=MAX_FORCE,
|
|
simulate=True,
|
|
)
|
|
for _tick in range(200):
|
|
phase = campaign.phase
|
|
if phase == "done":
|
|
return campaign
|
|
if phase == "load":
|
|
campaign.confirm_reference(float(campaign.progress()["target_n"]))
|
|
elif phase == "hold":
|
|
_sleep_hold()
|
|
campaign.complete_hold()
|
|
elif phase == "liftoff":
|
|
campaign.confirm_liftoff()
|
|
elif phase == "closing":
|
|
campaign.poll()
|
|
else:
|
|
raise AssertionError(f"unexpected phase {phase}")
|
|
raise AssertionError("the directly driven campaign never finished")
|
|
|
|
|
|
def _drive_through_controller(
|
|
controller: CaptureController, campaign: Campaign
|
|
) -> None:
|
|
"""Drive the same campaign through the primary button only."""
|
|
for _tick in range(200):
|
|
if campaign.phase == "done":
|
|
return
|
|
action = controller.primary_action()
|
|
if action.key == "align_point":
|
|
controller.perform("align_point")
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned() # the window says the planned move finished
|
|
elif action.key == "start_session":
|
|
controller.perform("start_session")
|
|
elif action.key == "confirm_target":
|
|
controller.perform(
|
|
"confirm_target", force_n=float(campaign.progress()["target_n"])
|
|
)
|
|
elif action.key == "confirm_hold":
|
|
if not action.enabled:
|
|
_sleep_hold()
|
|
continue
|
|
controller.perform("confirm_hold")
|
|
elif action.key in {"confirm_liftoff", "finish_session"}:
|
|
controller.perform(action.key)
|
|
elif action.key == "closing":
|
|
controller.poll()
|
|
else:
|
|
raise AssertionError(f"unexpected primary action {action}")
|
|
raise AssertionError("the controller-driven campaign never finished")
|
|
|
|
|
|
def test_the_controller_writes_the_same_records_as_the_campaign(tmp_dir):
|
|
"""The page must not produce a different session format than a direct campaign."""
|
|
|
|
def build() -> Campaign:
|
|
return _campaign(levels=(5.0,), repeats=1, sites=("P1", "P2"))
|
|
|
|
direct = _drive_direct(build(), tmp_dir / "direct")
|
|
campaign = build()
|
|
controller = _controller(campaign, tmp_dir / "controller")
|
|
_drive_through_controller(controller, campaign)
|
|
assert direct.phase == campaign.phase == "done"
|
|
|
|
assert direct.directory is not None and campaign.directory is not None
|
|
assert _tree(direct.directory) == _tree(campaign.directory)
|
|
assert _masked(_manifest(direct), VOLATILE_SESSION_KEYS) == _masked(
|
|
_manifest(campaign), VOLATILE_SESSION_KEYS
|
|
)
|
|
|
|
direct_rows = _reference_rows(direct)
|
|
controller_rows = _reference_rows(campaign)
|
|
assert (
|
|
sorted(direct_rows)
|
|
== sorted(controller_rows)
|
|
== [
|
|
"sites/P1/repeat-001",
|
|
"sites/P2/repeat-001",
|
|
]
|
|
)
|
|
for directory, rows in direct_rows.items():
|
|
assert list(rows[0]) == list(controller_rows[directory][0]) # header order
|
|
assert _masked_rows(rows, VOLATILE_REFERENCE_COLUMNS) == _masked_rows(
|
|
controller_rows[directory], VOLATILE_REFERENCE_COLUMNS
|
|
)
|
|
|
|
assert _masked_events(direct.directory / "events.csv") == _masked_events(
|
|
campaign.directory / "events.csv"
|
|
)
|
|
for trial_dir in ("sites/P1/repeat-001", "sites/P2/repeat-001"):
|
|
assert _masked_events(direct.directory / trial_dir / "events.csv") == (
|
|
_masked_events(campaign.directory / trial_dir / "events.csv")
|
|
)
|
|
assert (direct.directory / "stable_means.csv").read_bytes() == (
|
|
(campaign.directory / "stable_means.csv").read_bytes()
|
|
)
|
|
|
|
|
|
# ------------------------------------------------- closing out, auto hold, no hardware
|
|
|
|
|
|
def _auto_step(
|
|
controller: CaptureController, campaign: Campaign, hold_s: float = HOLD_S
|
|
) -> None:
|
|
"""Confirm the target, then let the planned hold time run out on its own."""
|
|
controller.perform("confirm_target", force_n=float(campaign.progress()["target_n"]))
|
|
assert campaign.phase == "hold"
|
|
time.sleep(hold_s + 0.03)
|
|
controller.poll()
|
|
|
|
|
|
def _walk_to_the_end(
|
|
controller: CaptureController, campaign: Campaign, hold_s: float = HOLD_S
|
|
) -> None:
|
|
"""Drive a started session to ``done`` using only the page's own calls."""
|
|
for _ in range(60):
|
|
action = controller.primary_action()
|
|
if campaign.phase in {"done", "aborted"}:
|
|
return
|
|
if action.key == "confirm_target":
|
|
_auto_step(controller, campaign, hold_s)
|
|
elif action.key == "confirm_liftoff":
|
|
controller.perform("confirm_liftoff")
|
|
elif action.key == "align_point":
|
|
controller.perform("align_point")
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned()
|
|
else: # pragma: no cover - a stuck walk is the failure to report
|
|
raise AssertionError(f"stuck at {action}")
|
|
raise AssertionError("the session never finished") # pragma: no cover
|
|
|
|
|
|
def test_finishing_a_session_returns_to_idle_and_a_new_one_can_start(tmp_dir):
|
|
"""F1: the finish button must lead somewhere, not offer itself forever."""
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
_walk_to_the_end(controller, campaign)
|
|
first = campaign.directory
|
|
assert controller.primary_action().key == "finish_session"
|
|
manifest_before = (first / "session.json").read_bytes()
|
|
|
|
controller.perform("finish_session")
|
|
|
|
assert campaign.phase == "idle"
|
|
# The finished session's evidence is exactly as it was.
|
|
assert campaign.directory == first
|
|
assert (first / "session.json").read_bytes() == manifest_before
|
|
# The point is no longer aligned: the platform may have moved since.
|
|
action = controller.primary_action()
|
|
assert action.key != "finish_session"
|
|
assert (action.key, action.enabled) == ("align_point", True)
|
|
# The queue is editable again, and a second session can start and finish.
|
|
controller.add_point("P2", 4.0, 5.0, 6.0, (0.0, 0.0, 1.0), dict(POSE), [5.0], HOLD_S, 1)
|
|
controller.perform("align_point")
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned("P1")
|
|
controller.perform("start_session")
|
|
assert campaign.phase == "load"
|
|
assert campaign.directory != first
|
|
_walk_to_the_end(controller, campaign)
|
|
assert campaign.phase == "done"
|
|
|
|
|
|
def test_a_stopped_session_can_also_be_closed_out(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
controller.perform("stop")
|
|
assert campaign.phase == "aborted"
|
|
assert controller.primary_action().key == "finish_session"
|
|
controller.perform("finish_session")
|
|
assert campaign.phase == "idle"
|
|
assert controller.primary_action().key in {"align_point", "start_session"}
|
|
|
|
|
|
@pytest.mark.parametrize("state", ["idle", "load"])
|
|
def test_only_a_finished_session_can_be_closed_out(tmp_dir, state):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
if state == "load":
|
|
_aligned_and_started(controller, campaign)
|
|
with pytest.raises(RuntimeError):
|
|
campaign.end_session()
|
|
assert campaign.phase == state
|
|
|
|
|
|
def test_a_hold_ends_by_itself_at_the_planned_time(tmp_dir):
|
|
"""F3: the recorded window is the plan, however late the page noticed."""
|
|
hold_s = 0.2
|
|
campaign = _campaign(hold_s=hold_s)
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
|
|
controller.perform(
|
|
"confirm_target", force_n=float(campaign.progress()["target_n"])
|
|
)
|
|
# Before the time is up the hold stays open and the button says why.
|
|
controller.poll()
|
|
assert campaign.phase == "hold"
|
|
assert controller.primary_action().blocked_reason == "cap.block.hold_running"
|
|
|
|
# The operator is busy for far longer than the plan, then the page ticks.
|
|
time.sleep(hold_s + 0.5)
|
|
state = controller.poll()
|
|
assert state["phase"] == "load"
|
|
assert controller.primary_action().key == "confirm_target"
|
|
|
|
_walk_to_the_end(controller, campaign, hold_s)
|
|
rows = _reference_rows(campaign)["sites/P1/repeat-001"]
|
|
assert [row["hold_complete"] for row in rows] == ["1", "1", "1"]
|
|
for row in rows:
|
|
length = int(row["t_end_host_mono_ns"]) - int(row["t_start_host_mono_ns"])
|
|
assert length == int(hold_s * 1e9), row
|
|
|
|
|
|
def test_auto_complete_hold_does_nothing_outside_a_running_hold(tmp_dir):
|
|
campaign = _campaign()
|
|
assert campaign.auto_complete_hold() is False
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
assert campaign.phase == "load"
|
|
assert campaign.auto_complete_hold() is False
|
|
|
|
|
|
def _offline_controller(campaign, root, *, sites_solver=None, probe=lambda: False):
|
|
"""A controller as the page builds it for a simulation with nothing connected."""
|
|
return CaptureController(
|
|
campaign,
|
|
solver=sites_solver,
|
|
readiness=lambda: [
|
|
ReadinessCheck(key, True, label)
|
|
for key, label in READINESS_ITEMS
|
|
if key != "max_force"
|
|
],
|
|
max_force_n=MAX_FORCE,
|
|
start_arguments=_start_arguments(root),
|
|
alignment_required=probe,
|
|
)
|
|
|
|
|
|
def test_a_simulation_with_nothing_connected_runs_without_alignment(tmp_dir):
|
|
"""F5: no platform, no solver and no move: start, record and finish."""
|
|
campaign = _campaign(sites=("P1", "P2"))
|
|
controller = _offline_controller(campaign, tmp_dir / "out", sites_solver=None)
|
|
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("start_session", True)
|
|
controller.perform("start_session")
|
|
assert controller.pending_alignment is None
|
|
|
|
seen: set[str] = set()
|
|
for _ in range(60):
|
|
action = controller.primary_action()
|
|
seen.add(action.key)
|
|
if campaign.phase == "done":
|
|
break
|
|
if action.key == "confirm_target":
|
|
_auto_step(controller, campaign)
|
|
elif action.key == "confirm_liftoff":
|
|
controller.perform("confirm_liftoff")
|
|
else: # pragma: no cover
|
|
raise AssertionError(f"stuck at {action}")
|
|
assert campaign.phase == "done"
|
|
assert "align_point" not in seen
|
|
assert "confirm_liftoff" in seen
|
|
# Nothing in the run asked the window to move an axis.
|
|
assert controller.pending_alignment is None
|
|
assert _manifest(campaign)["status"] == "complete"
|
|
assert _manifest(campaign)["simulate"] is True
|
|
|
|
controller.perform("finish_session")
|
|
assert campaign.phase == "idle"
|
|
# Closing out still leaves a simulation ready to run again without a move.
|
|
assert controller.primary_action().key == "start_session"
|
|
|
|
|
|
def test_alignment_stays_required_unless_the_page_says_otherwise(tmp_dir):
|
|
campaign = _campaign()
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
assert controller.primary_action().key == "align_point"
|
|
|
|
def broken() -> bool:
|
|
raise RuntimeError("the page is not ready")
|
|
|
|
# A probe that fails is the strict answer, never the permissive one.
|
|
controller = _offline_controller(campaign, tmp_dir / "out", probe=broken)
|
|
assert controller.primary_action().key == "align_point"
|
|
|
|
|
|
# ------------------------------------------------ stale alignment, failed skip
|
|
|
|
|
|
def test_an_earlier_alignment_is_forgotten_when_the_platform_moved_elsewhere(tmp_dir):
|
|
"""F4: a jog or a move from another page makes "aligned" untrue."""
|
|
campaign = _campaign(sites=("P1", "P2"))
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
controller.perform("align_point")
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned()
|
|
assert controller.primary_action().key == "start_session"
|
|
|
|
controller.invalidate_alignment()
|
|
|
|
action = controller.primary_action()
|
|
assert (action.key, action.enabled) == ("align_point", True)
|
|
with pytest.raises(ValueError, match="not an offered"):
|
|
controller.perform("start_session")
|
|
|
|
|
|
def test_the_next_point_must_be_aligned_again_after_an_outside_move(tmp_dir):
|
|
campaign = _campaign(sites=("P1", "P2"))
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
_drive_repeat(controller, campaign)
|
|
assert campaign.phase == "liftoff"
|
|
controller.perform("align_point")
|
|
controller.pending_alignment = None
|
|
controller.mark_aligned("P2")
|
|
assert controller.primary_action().key == "confirm_liftoff"
|
|
|
|
controller.invalidate_alignment()
|
|
|
|
assert controller.primary_action().key == "align_point"
|
|
|
|
|
|
def test_a_skip_that_cannot_move_the_folder_leaves_the_repeat_untouched(
|
|
tmp_dir, monkeypatch
|
|
):
|
|
"""F6: a locked folder must not leave a half-skipped repeat behind."""
|
|
campaign = _campaign(sites=("P1", "P2"))
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
directory = campaign.directory / "sites" / "P1" / "repeat-001"
|
|
trials_before = len(campaign._document["trials"])
|
|
|
|
real_replace = Path.replace
|
|
|
|
def locked(self, target):
|
|
if self == directory:
|
|
raise PermissionError(13, "the folder is in use by another process")
|
|
return real_replace(self, target)
|
|
|
|
monkeypatch.setattr(Path, "replace", locked)
|
|
with pytest.raises(RuntimeError, match="could not be set aside"):
|
|
controller.perform("skip")
|
|
monkeypatch.setattr(Path, "replace", real_replace)
|
|
|
|
# Nothing about the campaign changed, and the point was not marked skipped.
|
|
assert campaign.phase == "load"
|
|
assert len(campaign._document["trials"]) == trials_before
|
|
assert directory.is_dir()
|
|
assert controller.model().sites[0].skipped is False
|
|
# The repeat is still usable: its files were reattached, header intact.
|
|
_drive_repeat(controller, campaign)
|
|
rows = _reference_rows(campaign)["sites/P1/repeat-001"]
|
|
assert [row["hold_complete"] for row in rows] == ["1", "1", "1"]
|
|
|
|
|
|
def test_a_skip_sets_the_folder_aside_and_moves_on(tmp_dir):
|
|
campaign = _campaign(sites=("P1", "P2"))
|
|
controller = _controller(campaign, tmp_dir / "out")
|
|
_aligned_and_started(controller, campaign)
|
|
directory = campaign.directory / "sites" / "P1" / "repeat-001"
|
|
|
|
controller.perform("skip")
|
|
|
|
assert not directory.exists()
|
|
assert campaign.phase == "liftoff"
|
|
assert [t["site_id"] for t in campaign._document["trials"]] == []
|
|
assert controller.model().sites[0].skipped is True
|