实现 finger 单测点手动试验采集

将精确 UV 选点、实时预览与逐档人工确认接入同一页面,并保存每次试验的视频、事件和帧采集时间。通过离线视频、页面及记录回归测试。
This commit is contained in:
ykcheungab
2026-10-11 15:55:45 +08:00
parent 32dcecb270
commit dc7e290489
12 changed files with 896 additions and 15 deletions
+263
View File
@@ -0,0 +1,263 @@
"""One manually confirmed finger load/unload trial.
The acquisition worker owns frames; the shared recorder owns disk I/O. This
module owns only the site, planned steps and operator-confirmed event order.
"""
from __future__ import annotations
import hashlib
import json
import time
from dataclasses import dataclass
from pathlib import Path
import numpy as np
from orisys_app_shared.record import ExperimentRecorder, RecordConfig
@dataclass(frozen=True)
class Site:
u_px: int
v_px: int
position_sensor_mm: tuple[float, float, float]
face_index: int
raster_xy_px: tuple[int, int]
uv_origin_raster_xy_px: tuple[int, int]
def uv_origin(flat) -> tuple[int, int]:
"""Lowest valid pixel in the valid column nearest the image midpoint."""
rows, columns = np.nonzero(flat.face_index >= 0)
if not len(rows):
raise ValueError("UV map has no valid sensor pixels")
middle = (flat.width - 1) / 2
column = int(columns[np.argmin(np.abs(columns - middle))])
row = int(rows[columns == column].max())
return column, row
def site_at(
flat, u_px: int, v_px: int, *, origin: tuple[int, int] | None = None
) -> Site:
"""Resolve a precise raster pixel; empty mask pixels cannot become sites."""
if flat is None:
raise ValueError("UV map is not loaded")
origin_x, origin_y = origin if origin is not None else uv_origin(flat)
raster_x, raster_y = origin_x + int(u_px), origin_y - int(v_px)
sample = flat.sample(raster_x, raster_y)
if sample is None:
raise ValueError("UV pixel is outside the valid sensor surface")
point, _normal, face = sample
return Site(
int(u_px),
int(v_px),
tuple(float(v) for v in point),
int(face),
(raster_x, raster_y),
(origin_x, origin_y),
)
def load_path(levels: list[float]) -> list[tuple[str, float]]:
"""One continuous ascent and descent, ending at zero."""
values = [float(value) for value in levels]
if not values or any(not 0.0 < value <= 30.0 for value in values):
raise ValueError("load levels must be in (0, 30] N")
if values != sorted(set(values)):
raise ValueError("load levels must be strictly increasing")
return [("load", value) for value in values] + [
("unload", value) for value in [*reversed(values[:-1]), 0.0]
]
class ManualTrial:
"""Connect a running finger source to one independent recording directory."""
def __init__(self, worker) -> None:
self.worker = worker
self.recorder: ExperimentRecorder | None = None
self.directory: Path | None = None
self.steps: list[tuple[str, float]] = []
self.step_index = 0
self.status = "idle"
self.aborted = False
self.reference_alignment_possible = False
self._last_confirm_ns: int | None = None
def start(
self,
*,
root: Path,
site: Site,
levels: list[float],
hold_s: float,
sensor_sn: str,
bundle_path: str,
source: dict,
config_ref: str,
fps: float,
) -> Path:
if self.status in {"recording", "closing"}:
raise RuntimeError("a trial is already active")
if not self.worker.is_running or self.worker.latest() is None:
raise RuntimeError("start the finger input and wait for its first frame")
if not sensor_sn.strip():
raise ValueError("sensor SN is required")
if not float(fps) > 0.0:
raise ValueError("input frame rate must be positive")
self.steps = load_path(levels)
if not 10.0 <= float(hold_s) <= 30.0:
raise ValueError("hold time must be between 10 and 30 seconds")
packet = self.worker.latest()
if packet.capture_mono_ns is None:
raise RuntimeError("the input has no frame-bound host timestamp")
mesh_path = Path(bundle_path) / "mesh.npz"
config_path = Path(config_ref)
if not mesh_path.is_file() or not config_path.is_file():
raise ValueError("sensor mesh and configuration files are required")
mesh_sha256 = hashlib.sha256(mesh_path.read_bytes()).hexdigest()
config_sha256 = hashlib.sha256(config_path.read_bytes()).hexdigest()
self.reference_alignment_possible = packet.time_basis == "host_camera"
metadata = {
"purpose": "autocal_finger_manual_trial",
"sensor_sn": sensor_sn.strip(),
"site": {
"uv_xy_px": [site.u_px, site.v_px],
"position_sensor_mm": list(site.position_sensor_mm),
"face_index": site.face_index,
"raster_xy_px": list(site.raster_xy_px),
"uv_origin_raster_xy_px": list(site.uv_origin_raster_xy_px),
"bundle_path": bundle_path,
"mesh_sha256": mesh_sha256,
},
"plan": {"levels_n": levels, "hold_s": float(hold_s)},
"input_source": source,
"config_ref": config_ref,
"config_sha256": config_sha256,
"time_basis": packet.time_basis,
"reference_alignment_possible": self.reference_alignment_possible,
"reference_source": "manual_operator_confirmation",
}
recorder = ExperimentRecorder(
RecordConfig(fps=float(fps), segment_seconds=24 * 3600.0)
)
directory = recorder.start(
root=Path(root), session_id=packet.session_id, metadata=metadata
)
self.recorder = recorder
self.directory = directory
self.step_index = 0
self.aborted = False
self._last_confirm_ns = None
self.status = "recording"
self.worker.add_sink(recorder.on_packet)
self._event("trial_started", {"site_uv_xy_px": [site.u_px, site.v_px]})
return directory
@property
def next_step(self) -> tuple[str, float] | None:
return (
self.steps[self.step_index] if self.step_index < len(self.steps) else None
)
def _event(self, kind: str, detail: dict) -> None:
recorder = self.recorder
if recorder is None:
return
now_mono = time.perf_counter_ns()
packet = self.worker.latest()
if packet is not None and packet.capture_mono_ns is not None:
event_mono_t = (
float(packet.mono_t or 0.0) + (now_mono - packet.capture_mono_ns) / 1e9
)
frame_id = packet.frame_id
else:
event_mono_t = None
frame_id = -1
recorder.log_event(
kind,
json.dumps(
{**detail, "t_host_mono_ns": now_mono, "t_host_utc_ns": time.time_ns()},
ensure_ascii=False,
),
mono_t=event_mono_t,
frame_id=frame_id,
)
def confirm(self, reference_n: float) -> tuple[str, float]:
if self.status != "recording" or self.next_step is None:
raise RuntimeError("no load step is waiting for confirmation")
branch, target = self.next_step
reference = float(reference_n)
if not 0.0 <= reference <= 30.0:
raise ValueError("reference load must be between 0 and 30 N")
now_ns = time.perf_counter_ns()
if self._last_confirm_ns is not None:
self._event(
"step_ended",
{
"step_index": self.step_index - 1,
"held_s": (now_ns - self._last_confirm_ns) / 1e9,
},
)
self._event(
"step_reached",
{
"step_index": self.step_index,
"branch": branch,
"target_load_n": target,
"reference_load_n": reference,
"reference_kind": "manual",
},
)
self.step_index += 1
self._last_confirm_ns = now_ns
return branch, target
@property
def held_s(self) -> float | None:
if self._last_confirm_ns is None or self.status != "recording":
return None
return (time.perf_counter_ns() - self._last_confirm_ns) / 1e9
def finish(self, *, aborted: bool = False, reason: str = "") -> None:
if self.status != "recording" or self.recorder is None:
raise RuntimeError("no active trial")
if not aborted and self.next_step is not None:
raise RuntimeError("complete the load/unload path before finishing")
if self._last_confirm_ns is not None:
self._event(
"step_ended",
{"step_index": self.step_index - 1, "held_s": self.held_s},
)
self._event(
"trial_aborted" if aborted else "trial_finished", {"reason": reason}
)
self.aborted = aborted
self.worker.remove_sink(self.recorder.on_packet)
self.recorder.stop(wait=False)
self.status = "closing"
def poll(self) -> str:
if self.status == "recording" and self.recorder is not None:
if not self.worker.is_running or not self.recorder.is_recording:
self.finish(aborted=True, reason="input or writer stopped")
if self.status == "closing" and self.recorder is not None:
report = self.recorder.report
if report is not None:
self.status = (
"complete"
if (
report.ok
and report.stats.frames > 0
and report.stats.video_frames > 0
and report.stats.segments == 1
and report.stats.dropped_data == 0
and report.stats.dropped_video == 0
and not self.aborted
and self.next_step is None
)
else "incomplete"
)
return self.status
@@ -0,0 +1,351 @@
"""Single-site manual finger trial: precise UV site, live view and recording."""
from __future__ import annotations
from pathlib import Path
from orisys_app_shared.i18n import t
from orisys_app_shared.paths import outputs_dir
from PySide6.QtCore import QTimer, Signal
from PySide6.QtWidgets import (
QDoubleSpinBox,
QFileDialog,
QFormLayout,
QGridLayout,
QGroupBox,
QHBoxLayout,
QLabel,
QLineEdit,
QPushButton,
QSpinBox,
QTabWidget,
QVBoxLayout,
QWidget,
)
from ..trial import ManualTrial, site_at, uv_origin
from .sensor_panel import FlatSurfaceView
class CalibrationPanel(QWidget):
"""One operator page. No motion commands are issued by this widget."""
loadBundleRequested = Signal()
def __init__(self, source_panel: QWidget, parent: QWidget | None = None) -> None:
super().__init__(parent)
self.source_panel = source_panel
self.trial = ManualTrial(source_panel.worker)
self._flat = None
self._origin: tuple[int, int] | None = None
self._bundle_path = ""
self._site = None
outer = QVBoxLayout(self)
header = QHBoxLayout()
self.sn_label = QLabel(self)
self.sensor_sn = QLineEdit(self)
self.sensor_sn.setObjectName("calibrationSensorSn")
self.output_label = QLabel(self)
self.output_path = QLineEdit(str(Path(outputs_dir()) / "calibrations"), self)
self.output_path.setObjectName("calibrationOutputPath")
self.output_browse = QPushButton(self)
header.addWidget(self.sn_label)
header.addWidget(self.sensor_sn)
header.addWidget(self.output_label)
header.addWidget(self.output_path, 1)
header.addWidget(self.output_browse)
outer.addLayout(header)
self.source_state = QLabel(self)
outer.addWidget(self.source_state)
body = QHBoxLayout()
self.views = QTabWidget(self)
self.views.setObjectName("calibrationViews")
self.site_view = QWidget(self.views)
site_layout = QVBoxLayout(self.site_view)
self.flat_view = FlatSurfaceView(self.site_view)
self.flat_view.pixelPicked.connect(self._picked)
site_layout.addWidget(self.flat_view, 1)
self.bundle_button = QPushButton(self.site_view)
self.bundle_button.clicked.connect(self.loadBundleRequested)
site_layout.addWidget(self.bundle_button)
self.views.addTab(self.site_view, "")
self.views.addTab(self.source_panel, "")
body.addWidget(self.views, 3)
controls = QGroupBox(self)
controls.setObjectName("groupBox")
form = QFormLayout(controls)
self.u_spin = QSpinBox(controls)
self.u_spin.setObjectName("calibrationU")
self.v_spin = QSpinBox(controls)
self.v_spin.setObjectName("calibrationV")
coords = QWidget(controls)
coord_row = QHBoxLayout(coords)
coord_row.setContentsMargins(0, 0, 0, 0)
coord_row.addWidget(self.u_spin)
coord_row.addWidget(self.v_spin)
self.coords_label = QLabel(controls)
form.addRow(self.coords_label, coords)
self.position_label = QLabel(controls)
self.position_value = QLabel("—", controls)
form.addRow(self.position_label, self.position_value)
self.levels_label = QLabel(controls)
self.levels = QLineEdit("5, 10, 20, 30", controls)
self.levels.setObjectName("calibrationLevels")
form.addRow(self.levels_label, self.levels)
self.hold_label = QLabel(controls)
self.hold = QDoubleSpinBox(controls)
self.hold.setRange(10, 30)
self.hold.setValue(20)
self.hold.setSuffix(" s")
form.addRow(self.hold_label, self.hold)
self.target_label = QLabel(controls)
form.addRow(self.target_label)
self.reference_label = QLabel(controls)
self.reference = QLineEdit(controls)
self.reference.setObjectName("calibrationReference")
form.addRow(self.reference_label, self.reference)
actions = QWidget(controls)
action_grid = QGridLayout(actions)
action_grid.setContentsMargins(0, 0, 0, 0)
self.start_button = QPushButton(actions)
self.confirm_button = QPushButton(actions)
self.finish_button = QPushButton(actions)
self.abort_button = QPushButton(actions)
action_grid.addWidget(self.start_button, 0, 0)
action_grid.addWidget(self.confirm_button, 0, 1)
action_grid.addWidget(self.finish_button, 1, 0)
action_grid.addWidget(self.abort_button, 1, 1)
form.addRow(actions)
self.progress_label = QLabel(controls)
self.progress_label.setWordWrap(True)
form.addRow(self.progress_label)
self.result_label = QLabel(controls)
self.result_label.setWordWrap(True)
form.addRow(self.result_label)
body.addWidget(controls, 2)
outer.addLayout(body, 1)
self.output_browse.clicked.connect(self._choose_output)
self.u_spin.valueChanged.connect(self._site_changed)
self.v_spin.valueChanged.connect(self._site_changed)
self.start_button.clicked.connect(self._start)
self.confirm_button.clicked.connect(self._confirm)
self.finish_button.clicked.connect(self._finish)
self.abort_button.clicked.connect(self._abort)
self._timer = QTimer(self)
self._timer.setInterval(250)
self._timer.timeout.connect(self._poll)
self._timer.start()
self.retranslate()
self._sync()
def set_flat(self, flat, bundle_path: str) -> None:
self._flat = flat
self._bundle_path = bundle_path
self.flat_view.set_flat(flat)
origin_x, origin_y = uv_origin(flat)
self._origin = (origin_x, origin_y)
self.u_spin.setRange(-origin_x, flat.width - 1 - origin_x)
self.v_spin.setRange(origin_y - flat.height + 1, origin_y)
self._site_changed()
def _picked(self, u: int, v: int) -> None:
origin_x, origin_y = self._origin
self.u_spin.setValue(u - origin_x)
self.v_spin.setValue(origin_y - v)
self._site_changed()
def _site_changed(self) -> None:
try:
self._site = site_at(
self._flat,
self.u_spin.value(),
self.v_spin.value(),
origin=self._origin,
)
except ValueError:
self._site = None
origin = self._origin
self.flat_view.set_marker(
(origin[0] + self.u_spin.value(), origin[1] - self.v_spin.value())
if origin is not None
else None
)
self.flat_view.set_selected(
self._site.raster_xy_px if self._site is not None else None
)
self.position_value.setText(
"—"
if self._site is None
else ", ".join(f"{v:.3f}" for v in self._site.position_sensor_mm) + " mm"
)
self._sync()
def _choose_output(self) -> None:
path = QFileDialog.getExistingDirectory(
self, t("calibration.output"), self.output_path.text()
)
if path:
self.output_path.setText(path)
def _start(self) -> None:
try:
if self._site is None:
raise ValueError(t("calibration.invalid_site"))
if not self.output_path.text().strip():
raise ValueError(t("calibration.output_required"))
levels = [float(value.strip()) for value in self.levels.text().split(",")]
source = {
"kind": self.source_panel.source_kind.currentData(),
"value": (
self.source_panel.video_path.text().strip()
if self.source_panel.source_kind.currentData() == "video"
else self.source_panel.camera_index.value()
),
}
mode = self.source_panel.mode
if mode is None:
raise RuntimeError(t("calibration.start_input"))
directory = self.trial.start(
root=Path(self.output_path.text().strip()),
site=self._site,
levels=levels,
hold_s=self.hold.value(),
sensor_sn=self.sensor_sn.text(),
bundle_path=self._bundle_path,
source=source,
config_ref=self.source_panel.config_ref.text().strip(),
fps=mode.fps,
)
self.views.setCurrentWidget(self.source_panel)
self.result_label.setText(
str(directory)
+ (
"\n" + t("calibration.replay_note")
if not self.trial.reference_alignment_possible
else ""
)
)
except (OSError, RuntimeError, ValueError) as exc:
self.result_label.setText(t("calibration.error").format(detail=exc))
self._sync()
def _confirm(self) -> None:
try:
if not self.reference.text().strip():
raise ValueError(t("calibration.reference_required"))
self.trial.confirm(float(self.reference.text()))
self.reference.clear()
except (RuntimeError, ValueError) as exc:
self.result_label.setText(t("calibration.error").format(detail=exc))
self._sync()
def _finish(self) -> None:
try:
self.trial.finish()
except RuntimeError as exc:
self.result_label.setText(t("calibration.error").format(detail=exc))
self._sync()
def _abort(self) -> None:
try:
self.trial.finish(aborted=True, reason="operator aborted")
except RuntimeError as exc:
self.result_label.setText(t("calibration.error").format(detail=exc))
self._sync()
def _poll(self) -> None:
previous = self.trial.status
current = self.trial.poll()
if current != previous and current in {"complete", "incomplete"}:
self.result_label.setText(
t("calibration.result").format(
status=t(f"calibration.{current}"), path=self.trial.directory
)
+ (
"\n" + t("calibration.replay_note")
if not self.trial.reference_alignment_possible
else ""
)
)
self._sync()
def shutdown(self) -> None:
self._timer.stop()
if self.trial.status == "recording":
self.trial.finish(aborted=True, reason="application closed")
if self.trial.status == "closing" and self.trial.recorder is not None:
self.trial.recorder.stop(wait=True)
def _sync(self) -> None:
recording = self.trial.status == "recording"
locked = recording or self.trial.status == "closing"
for widget in (
self.u_spin,
self.v_spin,
self.levels,
self.hold,
self.sensor_sn,
self.output_path,
self.output_browse,
self.bundle_button,
):
widget.setEnabled(not locked)
self.start_button.setEnabled(not locked and self._site is not None)
self.confirm_button.setEnabled(recording and self.trial.next_step is not None)
self.finish_button.setEnabled(recording and self.trial.next_step is None)
self.abort_button.setEnabled(recording)
step = self.trial.next_step
self.target_label.setText(
t("calibration.next_step").format(
branch=t(f"calibration.{step[0]}"), load=step[1]
)
if recording and step is not None
else t("calibration.no_step")
)
self.progress_label.setText(
t("calibration.progress").format(
done=self.trial.step_index,
total=len(self.trial.steps),
status=t(f"calibration.{self.trial.status}"),
)
+ (
t("calibration.held").format(seconds=self.trial.held_s)
if self.trial.held_s is not None
else ""
)
)
worker = self.source_panel.worker
packet = worker.latest()
self.source_state.setText(
t("calibration.source_state").format(
status=t("source.running")
if worker.is_running
else t("source.stopped"),
frame=packet.frame_id if packet is not None else "—",
dropped=getattr(packet, "capture_dropped", "—")
if packet is not None
else "—",
)
)
def retranslate(self) -> None:
self.sn_label.setText(t("calibration.sensor_sn"))
self.output_label.setText(t("calibration.output"))
self.output_browse.setText(t("source.browse"))
self.views.setTabText(0, t("calibration.site_view"))
self.views.setTabText(1, t("calibration.live_view"))
self.bundle_button.setText(t("finger.load"))
self.coords_label.setText(t("calibration.uv"))
self.position_label.setText(t("calibration.position"))
self.levels_label.setText(t("calibration.levels"))
self.hold_label.setText(t("calibration.hold"))
self.reference_label.setText(t("calibration.reference"))
self.start_button.setText(t("calibration.start"))
self.confirm_button.setText(t("calibration.confirm"))
self.finish_button.setText(t("calibration.finish"))
self.abort_button.setText(t("calibration.abort"))
self._sync()
@@ -49,6 +49,7 @@ class FingerSourcePanel(QWidget):
self._active = False
self._error = ""
self._status_key = "source.idle"
self.mode = None
layout = QVBoxLayout(self)
fields = QGridLayout()
@@ -180,7 +181,7 @@ class FingerSourcePanel(QWidget):
self.preview.setText(t("source.no_frame"))
self.frame_label.clear()
try:
self.worker.start(source, config_ref=str(config))
self.mode = self.worker.start(source, config_ref=str(config))
except Exception as exc: # noqa: BLE001 - show source/config errors to operator
self._show_error(str(exc))
return
+7 -1
View File
@@ -64,6 +64,7 @@ from .axis_readout import AxisReadoutPanel
from .brand_bar import BrandBar
from .bus_data_panel import BusDataPanel
from .bus_panel import default_bas_dir
from .calibration_panel import CalibrationPanel
from .device_setup import DeviceSetupDialog
from .finger_source_panel import FingerSourcePanel
from .sensor_panel import FingerPanel
@@ -295,12 +296,14 @@ class AutoCalMainWindow(QMainWindow):
source_kind=str(self.prefs.get("finger_source_kind") or "video"),
camera_index=self.prefs.get("finger_camera_index") or 0,
)
self.calibration_panel = CalibrationPanel(self.finger_source_panel, body)
self.calibration_panel.loadBundleRequested.connect(self._on_load_bundle)
self.workspace = WorkspaceTabs(
self.axis_panel,
self.sequence_panel,
self.finger_panel,
body,
sensor_panel=self.finger_source_panel,
sensor_panel=self.calibration_panel,
)
self.main_splitter = QSplitter(Qt.Orientation.Horizontal, body)
@@ -413,6 +416,7 @@ class AutoCalMainWindow(QMainWindow):
texture_path, texture = self._resolve_texture(bundle)
flat = rasterise(bundle, flat_size, texture=texture)
self.finger_panel.set_bundle(bundle, flat)
self.calibration_panel.set_flat(flat, str(path))
self.finger_panel.set_texture_state(textured=flat.textured, path=str(texture_path or ""))
self.prefs["finger_bundle_path"] = str(path)
info = bundle.describe()
@@ -1286,6 +1290,7 @@ class AutoCalMainWindow(QMainWindow):
self.sequence_panel.retranslate()
self.finger_panel.retranslate()
self.finger_source_panel.retranslate()
self.calibration_panel.retranslate()
self.workspace.retranslate()
self.menu_file.setTitle(t("menu.file"))
@@ -1381,6 +1386,7 @@ class AutoCalMainWindow(QMainWindow):
def closeEvent(self, event: QCloseEvent) -> None: # noqa: N802 - Qt override
"""Stop the platform, close the controller, then persist and leave."""
self.calibration_panel.shutdown()
self.finger_source_panel.shutdown()
logger.info("Closing: stopping axes and shutting the poller down")
try: