orisys-tech/orisys-app
3
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

完善 Autocal 多测点标定流程及会话复核

Browse Source 
统一测点坐标、保持窗口和试验目录，补齐逐点初步分析。定向测试与真实视频闭环通过；两项轮询失败在修改前版本同样复现。
This commit is contained in:
ykcheungab
2026-10-11 17:43:16 +08:00
parent 7828bc1fe9
commit 04adf17f60
17 changed files with 1347 additions and 403 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/autocal/tests/test_manual_trial.py
+217 -103
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import csv
import json
import time
from dataclasses import replace
from types import SimpleNamespace
import numpy as np
@@ -14,10 +15,10 @@ from orisys_autocal.trial import (
REFERENCE_HEADER,
ManualTrial,
load_path,
site_at,
uv_origin,
)
from orisys_autocal.sensor.uv_flat import FlatSurface, site_at, uv_origin
class Worker:
def __init__(self) -> None:
@@ -52,140 +53,253 @@ class Worker:
for sink in self.sinks:
sink(self.packet)
def stop(self, *, wait=True):
self.is_running = False
def test_uv_coordinates_are_centered_and_masked():
@pytest.fixture
def flat():
mask = np.full((5, 7), -1, dtype=np.int32)
mask[1:4, 2:5] = 2
flat = SimpleNamespace(
width=7,
height=5,
face_index=mask,
sample=lambda x, y: (
(np.array([x, y, 3.0]), np.array([0, 0, 1]), 2)
if 0 <= x < 7 and 0 <= y < 5 and mask[y, x] >= 0
else None
),
)
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(tmp_dir):
def test_manual_trial_records_video_stamps_and_load_unload_events(trial_setup, clock):
worker = Worker()
trial = ManualTrial(worker)
bundle = tmp_dir / "bundle"
bundle.mkdir()
(bundle / "mesh.npz").write_bytes(b"mesh")
config = tmp_dir / "sensor.json"
config.write_text("{}", encoding="utf-8")
site = SimpleNamespace(
u_px=0,
v_px=1,
position_sensor_mm=(1.0, 2.0, 3.0),
face_index=2,
raster_xy_px=(3, 2),
uv_origin_raster_xy_px=(3, 3),
)
assert load_path([5, 10]) == [
("load", 5),
("load", 10),
("unload", 5),
("unload", 0),
]
directory = trial.start(
root=tmp_dir,
site=site,
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,
)
("zero", 0), ("load", 5), ("load", 10), ("unload", 5), ("unload", 0)]
directory = trial.start(**trial_setup)
worker.emit()
for value in (5, 10, 5, 0):
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()
report = trial.recorder.stop(wait=True)
assert report.ok
assert trial.recorder.stop(wait=True).ok
assert trial.poll() == "complete"
assert "Finger-D40_F-001" in directory.name
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"
assert row["capture_dropped"] == "0"
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] == ["load", "load", "unload", "unload"]
assert [row["F_ref_normal_N"] for row in references] == [
"5.0",
"10.0",
"5.0",
"0.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" for row in references)
assert all(
int(row["t_start_host_mono_ns"]) < int(row["t_end_host_mono_ns"])
for row in references
)
assert all(
previous["t_end_host_mono_ns"] == following["t_start_host_mono_ns"]
for previous, following in zip(references, references[1:])
)
with (directory / "events.csv").open(encoding="utf-8", newline="") as handle:
events = list(csv.DictReader(handle))
reached = [event for event in events if event["kind"] == "step_reached"]
assert len(reached) == 4
assert [json.loads(event["detail"])["branch"] for event in reached] == [
"load",
"load",
"unload",
"unload",
]
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:
session = json.load(handle)
assert session["site"]["uv_xy_px"] == [0, 1]
assert session["sensor_model"] == "Finger D40"
assert session["reference_file"] == "reference.csv"
assert session["site"]["mesh_sha256"]
assert len(list(tmp_dir.glob("*/session.json"))) == 1
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"
second_directory = trial.start(
root=tmp_dir,
site=site,
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_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(5)
trial.finish(aborted=True, reason="operator aborted")
trial.confirm(0)
clock.advance(2)
worker.is_running = False
trial.poll()
trial.recorder.stop(wait=True)
assert trial.poll() == "incomplete"
assert len(list(tmp_dir.glob("*/session.json"))) == 2
assert second_directory != directory
with (second_directory / "reference.csv").open(encoding="utf-8", newline="") as handle:
assert [row["F_ref_normal_N"] for row in csv.DictReader(handle)] == ["5.0"]
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(