ykcheungab
343 lines
13 KiB
Python
343 lines
13 KiB
Python
"""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_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")
|
|
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")
|
|
panel.sensor_model.setText("finger")
|
|
panel.levels.setText("5, 10")
|
|
panel.start_button.click()
|
|
assert panel.trial.status == "recording"
|
|
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()
|