"""A manually confirmed site produces one traceable trial directory.""" from __future__ import annotations import csv import json import time from dataclasses import replace from types import SimpleNamespace import numpy as np import pytest from orisys_app_shared.acquisition.models import SdkPacket from orisys_autocal.trial import ( REFERENCE_HEADER, ManualTrial, load_path, ) from orisys_autocal.sensor.uv_flat import FlatSurface, site_at, uv_origin class Worker: def __init__(self) -> None: self.is_running = True self.sinks = [] self.packet = SdkPacket( raw_frame=np.zeros((32, 32, 3), dtype=np.uint8), processed_frame=np.zeros((32, 32, 3), dtype=np.uint8), timestamp=time.time(), fps=30, process_ms=1, session_id=2, frame_id=1, mono_t=1.0, capture_mono_ns=time.perf_counter_ns(), capture_utc_ns=time.time_ns(), source_frame_id=7, time_basis="host_camera", normal_force_raw=1.25, ) def latest(self): return self.packet def add_sink(self, sink): self.sinks.append(sink) def remove_sink(self, sink): self.sinks.remove(sink) def emit(self): for sink in self.sinks: sink(self.packet) def stop(self, *, wait=True): self.is_running = False @pytest.fixture def flat(): mask = np.full((5, 7), -1, dtype=np.int32) mask[1:4, 2:5] = 2 mask[4, 2] = 2 # A valid pixel may be below the centre-column origin. positions = np.zeros((5, 7, 3), dtype=np.float64) for y in range(5): for x in range(7): positions[y, x] = (x, y, 3) normals = np.zeros_like(positions) normals[:, :, 2] = 1 return FlatSurface((5, 7), np.zeros((5, 7, 3), dtype=np.uint8), positions, normals, mask, (0, 0), (1, 1)) @pytest.fixture def clock(monkeypatch): from orisys_autocal import trial as trial_module class Clock: now = time.perf_counter_ns() def advance(self, seconds): self.now += round(seconds * 1e9) value = Clock() monkeypatch.setattr(trial_module, "time", SimpleNamespace( perf_counter_ns=lambda: value.now, time_ns=time.time_ns)) return value @pytest.fixture def trial_setup(tmp_path, flat): bundle = tmp_path / "bundle" bundle.mkdir() (bundle / "mesh.npz").write_bytes(b"mesh") config = tmp_path / "sensor.json" config.write_text("{}", encoding="utf-8") return dict(root=tmp_path, site=site_at(flat, -1, 2), levels=[5, 10], hold_s=20, sensor_model="Finger D40", sensor_sn="F-001", bundle_path=str(bundle), source={"kind": "camera", "value": 0}, config_ref=str(config), fps=30) def test_uv_coordinates_are_centered_and_masked(flat): assert uv_origin(flat) == (3, 3) site = site_at(flat, -1, 2) assert site.raster_xy_px == (2, 1) assert site.position_sensor_mm == (2.0, 1.0, 3.0) assert site_at(flat, -1, -1).raster_xy_px == (2, 4) with pytest.raises(ValueError): site_at(flat, -3, 0) def test_trial_persists_colour_region_outline_and_rejects_mismatched_mapping(trial_setup, flat): from orisys_autocal.sensor.uv_outline import outline_snapshot flat.textured = True flat.image_rgb[:] = 255 flat.image_rgb[2:4, 2:5] = [20, 80, 200] snapshot = outline_snapshot(flat) worker = Worker() trial = ManualTrial(worker) with pytest.raises(ValueError, match="outline"): trial.start(**trial_setup, uv_outline={**snapshot, "mapping_sha256": "wrong"}) directory = trial.start(**trial_setup, uv_outline=snapshot) worker.emit() trial.finish(aborted=True, reason="test ended") trial.recorder.stop(wait=True) trial.poll() manifest = json.loads((directory.parent.parent / "session.json").read_text(encoding="utf-8")) assert manifest["uv_outline"] == snapshot and snapshot["patch"] trial.finish_session() def test_manual_trial_records_video_stamps_and_load_unload_events(trial_setup, clock): worker = Worker() trial = ManualTrial(worker) assert load_path([5, 10]) == [ ("zero", 0), ("load", 5), ("load", 10), ("unload", 5), ("unload", 0)] directory = trial.start(**trial_setup) worker.emit() for index, value in enumerate((0, 5, 10, 5, 0)): assert trial.step_index == index trial.confirm(value) assert trial.step_index == index with pytest.raises(RuntimeError, match="hold"): trial.confirm(value) with pytest.raises(RuntimeError): trial.finish() clock.advance(19) trial.poll() assert trial.holding assert trial.remaining_s == pytest.approx(1) clock.advance(3) # Late UI polling must not extend the reference window. trial.poll() assert trial.step_index == index + 1 assert not trial.holding clock.advance(4) # Adjustment time must have no constant-force label. trial.finish() assert trial.recorder.stop(wait=True).ok assert trial.poll() == "complete" assert directory.name == "trial-001" assert "Finger-D40_F-001" in directory.parent.parent.name assert len(list(directory.glob("raw_*.mp4"))) == 1 with (directory / "frames.csv").open(encoding="utf-8", newline="") as handle: row = next(csv.DictReader(handle)) assert row["t_host_mono_ns"] == str(worker.packet.capture_mono_ns) assert row["source_frame_id"] == "7" assert row["time_basis"] == "host_camera" with (directory / "reference.csv").open(encoding="utf-8", newline="") as handle: reader = csv.DictReader(handle) assert tuple(reader.fieldnames) == REFERENCE_HEADER references = list(reader) assert [row["branch"] for row in references] == ["zero", "load", "load", "unload", "unload"] assert [float(row["F_ref_normal_N"]) for row in references] == [0, 5, 10, 5, 0] assert all(row["F_ref_x_N"] == row["F_ref_y_N"] == row["F_ref_z_N"] == "" for row in references) assert all(row["stability_verified"] == "0" and row["hold_complete"] == "1" for row in references) assert all(int(r["t_end_host_mono_ns"]) - int(r["t_start_host_mono_ns"]) == 20_000_000_000 for r in references) assert all(int(a["t_end_host_mono_ns"]) < int(b["t_start_host_mono_ns"]) for a, b in zip(references, references[1:])) with (directory / "session.json").open(encoding="utf-8") as handle: manifest = json.load(handle) assert manifest["trial_status"] == "complete" assert manifest["site"]["uv_xy_px"] == [-1, 2] assert len(manifest["steps"]) == 5 assert manifest["uv_definition"]["mapping_sha256"] == trial_setup["site"].mapping_sha256 session_path = directory.parent.parent / "session.json" root = json.loads(session_path.read_text(encoding="utf-8")) assert root["status"] == "in_progress" assert root["trials"][0]["status"] == "complete" assert trial.finish_session()[1] == "complete" assert json.loads(session_path.read_text(encoding="utf-8"))["status"] == "complete" def test_retries_and_other_sites_share_one_session(trial_setup, clock): worker = Worker() trial = ManualTrial(worker) directories = [] for site in (trial_setup["site"], trial_setup["site"], replace(trial_setup["site"], u_px=0, raster_xy_px=(3, 1), position_sensor_mm=(3, 1, 3))): directory = trial.start(**{**trial_setup, "site": site}) directories.append(directory) worker.emit() trial.confirm(0) clock.advance(3) trial.finish(aborted=True, reason="operator ended early") trial.recorder.stop(wait=True) assert trial.poll() == "incomplete" rows = list(csv.DictReader((directory / "reference.csv").open(encoding="utf-8"))) assert rows[0]["hold_complete"] == "0" assert len({d.parent.parent for d in directories}) == 1 root = trial.session.document assert len(root["sites"]) == 2 assert [t["repeat_index"] for t in root["trials"]] == [1, 2, 1] assert [t["site_id"] for t in root["trials"]] == ["site-001", "site-001", "site-002"] with pytest.raises(ValueError, match="finish this session"): trial.start(**{**trial_setup, "sensor_sn": "F-002"}) with pytest.raises(ValueError, match="finish this session"): trial.start(**{**trial_setup, "site": replace(site, mapping_sha256="other")}) assert trial.finish_session()[1] == "incomplete" new_directory = trial.start(**trial_setup) assert new_directory.parent.parent != directories[0].parent.parent trial.finish(aborted=True) trial.recorder.stop(wait=True) trial.poll() trial.finish_session() def test_input_end_aborts_active_hold(trial_setup, clock): worker = Worker() worker.packet = replace(worker.packet, time_basis="host_replay") trial = ManualTrial(worker) directory = trial.start(**trial_setup) assert not trial.reference_alignment_possible worker.emit() trial.confirm(0) clock.advance(2) worker.is_running = False trial.poll() trial.recorder.stop(wait=True) assert trial.poll() == "incomplete" assert trial.entry["reason"] == "input or writer stopped" assert json.loads((directory / "session.json").read_text(encoding="utf-8"))["steps"][0]["status"] == "incomplete" 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 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"] 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() window.close() def test_calibration_page_keeps_precise_uv_site_when_switching_to_live_view( make_window, ): from orisys_app_shared.i18n import t from orisys_autocal.sensor.uv_flat import FlatSurface window = make_window() mask = np.full((5, 7), -1, dtype=np.int32) mask[1:4, 2:5] = 2 positions = np.zeros((5, 7, 3), dtype=np.float64) positions[1, 2] = (2, 1, 3) flat = FlatSurface( size_hw=(5, 7), image_rgb=np.zeros((5, 7, 3), dtype=np.uint8), positions_mm=positions, normals=np.zeros_like(positions), face_index=mask, 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) window.close()