feat(autocal): 添加标定记录总览与测点计划对照

集中显示采集覆盖、加载卸载响应与已确认法向灵敏度,并保存计划和复核快照。通过44项相关测试、已有视频会话检查及代码和文档检查。
This commit is contained in:
ykcheungab
2026-10-11 19:03:40 +08:00
parent 161e64278b
commit b200f3d55c
10 changed files with 663 additions and 11 deletions
+38
View File
@@ -490,6 +490,44 @@
"depth_missing": "No valid depth/deformation for this frame",
"flow_missing": "No valid displacement for this frame"
},
"overview": {
"title": "Calibration record overview",
"notice": "Inspect acquired sites and normal response; click UV points or table rows. Circles: confirmed windows; squares: candidates. Trial shades differ. The fit uses all eligible loading data for the site and is not refitted by the trial filter.",
"coverage": "Acquisition status",
"sensitivity": "Normal sensitivity",
"import_plan": "Import planned-point CSV…",
"clear_plan": "Clear plan comparison",
"load": "Load saved review…",
"save": "Save new overview and review",
"summary": "Acquired {collected}; planned {planned}; missing {missing}; incomplete attempts {incomplete}; pending review {pending}; reviewed {reviewed}; outside plan {extra}.",
"no_plan": "Without a plan, missing sites cannot be determined. Plan order is separate from acquisition site_id.",
"uncollected": "Not acquired",
"incomplete": "Has incomplete attempts",
"pending": "Pending review",
"reviewed": "Reviewed",
"confirmed": "Confirmed",
"candidate": "Candidate",
"coverage_map": "Red: incomplete attempts; orange: pending; green: reviewed; grey: not acquired",
"sensitivity_map": "Confirmed sensitivity: blue→red {low:.4g}→{high:.4g} raw/N<br/>Grey: no confirmed fit",
"details": "Inspect signal / review windows…",
"trial_filter": "Displayed trial",
"all_trials": "All trials",
"site_fit": "Site loading fit (all eligible trials)",
"response_title": "Normal response: reference force → raw signal",
"selected_fit": "Site exploratory sensitivity {sensitivity} raw/N ({basis}); fit range and trial count are in the table. Physical signal/reference validation is still required.",
"no_windows": "No complete hold windows for this site/trial. Incomplete trials are excluded from fits; raw files are retained.",
"col_order": "Plan order",
"col_site": "Acquired site",
"col_uv": "U / V (px)",
"col_xyz": "Sensor XYZ (mm)",
"col_status": "Status",
"col_trials": "Complete / all trials",
"col_windows": "Confirmed / complete windows",
"col_levels": "Confirmed loading levels (N)",
"col_sensitivity": "Sensitivity (raw/N) / basis",
"col_range": "Fit range (N)",
"col_fit_trials": "Fit trials"
},
"quality": {
"open": "Repeatability and validation…",
"title": "Site quality analysis",
+38
View File
@@ -490,6 +490,44 @@
"depth_missing": "本帧无有效深度/形变信号",
"flow_missing": "本帧无有效位移信号"
},
"overview": {
"title": "标定记录总览",
"notice": "查看实际采集覆盖与法向响应;点击 UV 点或表格联动。圆点为已确认窗口,方点为候选窗口;各 trial 用不同明暗度。拟合线为本测点全部符合条件的加载数据,筛选 trial 不会重新拟合。",
"coverage": "采集状态",
"sensitivity": "法向灵敏度",
"import_plan": "导入计划测点 CSV…",
"clear_plan": "清除计划对照",
"load": "载入已保存复核…",
"save": "另存总览与复核",
"summary": "已采集 {collected} 点;计划 {planned} 点;未采 {missing};有未完成试验 {incomplete};待复核 {pending};已复核 {reviewed};计划外 {extra}。",
"no_plan": "未导入计划表,无法判断漏采点;序号仅用于计划对照,不作为采集 site_id。",
"uncollected": "尚未采集",
"incomplete": "有未完成试验",
"pending": "待复核",
"reviewed": "已复核",
"confirmed": "已确认",
"candidate": "候选",
"coverage_map": "红:有未完成试验;橙:待复核;绿:已复核;灰:未采",
"sensitivity_map": "已确认灵敏度:蓝→红 {low:.4g}→{high:.4g} raw/N<br/>灰:无已确认拟合",
"details": "查看信号/复核窗口…",
"trial_filter": "显示试验",
"all_trials": "全部 trial",
"site_fit": "测点加载拟合(全部符合条件的 trial)",
"response_title": "法向响应:参考力 → 原始信号",
"selected_fit": "本测点初步灵敏度 {sensitivity} raw/N({basis});拟合范围和 trial 数见表格。原始信号与实际参考力仍须实机验证。",
"no_windows": "所选点/trial 无完整保持窗口;未完成试验不参与拟合,原始文件仍保留。",
"col_order": "计划序号",
"col_site": "采集测点",
"col_uv": "U / V (px)",
"col_xyz": "传感器 XYZ (mm)",
"col_status": "状态",
"col_trials": "完成/全部 trial",
"col_windows": "确认/完整窗口",
"col_levels": "已确认加载档位 (N)",
"col_sensitivity": "灵敏度 (raw/N)/依据",
"col_range": "拟合范围 (N)",
"col_fit_trials": "拟合 trial 数"
},
"quality": {
"open": "重复性与验证…",
"title": "测点质量分析",
@@ -1,6 +1,6 @@
# Autocal 标定操作与数据格式草稿
状态:2026-10-11 修订,待实机验收。阶段一仅针对 finger。多测点会话、手动加载/卸载保持、视频和相机输入、模拟页测点操作表导出已接通;已可重开会话查看保持窗口、加载/卸载曲线及 UV 测点初步灵敏度;FT 同步、图像基准快照及正式灵敏度结果仍待实现。软件检查不代表设备或标定精度通过。
状态:2026-10-11 修订,阶段一仅针对 finger,待实机验收。已接通手动加载/卸载、多测点采集、记录总览和窗口复核;FT 同步及正式灵敏度结果仍待实现。
## 范围与坐标
@@ -34,11 +34,11 @@
“提前结束(保留未完成数据)”立即停止本次试验;不删除数据,不续录旧 trial。视频播放结束或写入停止会中止当前试验。关闭程序时结束当前试验并关闭会话;异常退出留下的 `in_progress` 不代表已完成。目前不能恢复旧会话继续采集。
保持倒计时只结束档位窗口,整段录像到完成或提前结束才停止。到时不等于稳定;`stability_verified=0`,正式分析仍须复核窗口。初始和最终零载当前使用相同保持时间,后续可根据恢复实测调整。
倒计时只结束档位窗口,完成或提前结束才停录像。到时不等于稳定;`stability_verified=0`,仍须复核。初始和最终零载保持时间相同,后续按恢复实测调整。
## 模拟页与操作表
模拟页和标定页共用 UV 映射。在模拟页选点、调整列表顺序后点击“导出测点操作表…”,得到 UTF-8 CSV。操作员按表中 U/V 手动复现;当前不提供序列导入。三维位置被手工改动而与原像素不一致时拒绝导出,须重新在图上选点。
模拟页和标定页共用 UV 映射。选点、排序后“导出测点操作表…”得到 UTF-8 CSV;操作员按 U/V 手动复现,不自动执行序列。三维位置与原像素不一致时拒绝导出,须重新选点。
| 操作表字段 | 含义 |
|---|---|
@@ -69,7 +69,7 @@ finger_F001_<时间>_<标识>/
...
```
10 个点、每点 3 次正常循环共 30 个 trial 目录;中止和重试各占一次,档位数不增加目录数。
10 点各做 3 次共 30 个 trial;中止和重试各占一次,档位数不增加目录数。
| 文件 | 内容及写入时机 |
|---|---|
@@ -80,6 +80,7 @@ finger_F001_<时间>_<标识>/
| `forces.csv` | 图像算法原始及处理后输出;不是独立真实力 |
| `reference.csv` | 每个实际开始的保持窗口一行;完整窗口在计时终点关闭,中断窗口在停止时关闭 |
| `derived/review_<时间>_<标识>/review.json` | 另存初步分析,包含窗口均值、标准差、拟合和质量说明;不覆盖采集数据 |
| 同目录 `overview.json` | 总览版本、session_id、计划 CSV 内容及来源哈希、测点覆盖和拟合摘要;载入时按原始数据及复核窗口重算 |
| `derived/review_<时间>_<标识>/baseline_<trial>.png` | 已确认初始零载窗口的第一张有效录像帧;作为检查基准,不是跟踪器内部参考状态;无视频时在分析结果中标记 unavailable |
| `events.csv` | `trial_started`、`step_reached`、`hold_started`、`hold_ended`、`trial_finished`/`trial_aborted`;保存电脑事件时间及 step 信息 |
@@ -107,11 +108,21 @@ finger_F001_<时间>_<标识>/
主时间是电脑取得该帧时绑定的单调时间,UTC 用于追溯;处理完成时间不能代替采集时间。FT/ECAT 后续接入同源接收时间,按时间关联,不按行号配对;低延迟不等于零延迟。旧视频的 `host_replay` 是本次解码时间,只能调试,不能恢复原实验与参考设备的物理同步。
点击“查看标定数据…”选择会话根目录;在独立窗口切换测点,检查每档有效样本、均值、标准差和质量说明。初步拟合仅使用初始零载及加载分支的窗口均值,各窗口等权;卸载点保留用于对照,不混入加载拟合。缺样本、档位或信号变化时不生成系数。UV 图仅显示已测点,不外推整面;拟合 RMSE 不是独立验证误差。
点击“查看标定数据…”选会话目录,进入记录总览;点击 UV 点或表格,联动法向响应和 trial 筛选,再“查看信号/复核窗口…”。
采集页持续显示 FN/FX/FY 原始信号及最近 60 秒曲线;已有标定的轴另显示 N,未标定的轴不将 raw 转称为 N。无效帧在曲线中断开;停止或超过 1 秒没有新帧时明确提示。此处没有实时 FT 参考力,人工填写的参考力仍是独立声明。
| 总览 | 规则 |
|---|---|
| 计划对照 | 导入模拟页 CSV;核对映射、原点、轴向、尺寸及三维位置,按 U/V 匹配;标记漏采和计划外点,同点重复序号合并 |
| 采集状态 | 未采、有未完成试验、待复核、已复核;完成次数及已确认窗口数分别列出,不把未完成 trial 用于拟合 |
| 法向响应 | 横轴实际参考 N,纵轴 FN_raw 均值;区分加载/卸载及候选/已确认,可筛选 trial;拟合仍基于本点全部合格加载窗口 |
| 灵敏度图 | 仅已确认拟合按 raw/N 着色,其余灰色;不插值,表中另列拟合范围及 trial 数 |
| 保存恢复 | “另存总览与复核”保存计划快照及窗口;载入 review.json 同时恢复总览,无 overview.json 时仅恢复复核 |
实时 UV 图有“真实背景”“位移箭头”“深度”三个独立开关;取消真实背景显示纯色画布,深度颜色可叠加在两种背景上。深度色阶按当前帧有效 UV 区域内的有限值最小/最大值自动调整,并显示范围;同一颜色在不同帧不代表相同数值,算法单位不代表毫米。缺失或过期字段不绘制,显示原因。显示切换不改变数据和录像,也不旋转 UV 坐标。
拟合使用初始零载及加载窗口均值,各窗口等权;卸载仅作对照。缺样本、档位或信号变化时不生成系数。拟合 RMSE 不是独立验证误差。
采集页显示 FN/FX/FY raw 及最近 60 秒曲线;已有标定的轴另显示 N。无效帧断开,停止或超过 1 秒无新帧时提示。人工参考力不是实时 FT。
实时 UV 图有真实背景、位移箭头、深度三个开关;关闭背景显示纯色画布。深度按当前帧有效 UV 有限值范围自适应,显示范围;跨帧同色不代表同值,算法单位不是毫米。缺失或过期字段不绘制并提示。显示不改变录像及坐标。
复核时选中表格中的保持段,检查下方原始样本图;起点、终点以该段保持开始为 0 秒,只能在原保持段内调整。勾选“已检查此段稳定”,点击“应用窗口并重算”,最后“另存分析结果”。至少有两个不同载荷的已确认加载/零载窗口时,显示已确认窗口拟合;不足时明确显示候选拟合。人工确认只表示看过稳定性,不证明参考力准确或物理同步。
@@ -0,0 +1,106 @@
"""Coverage and planned-point comparison over the existing review contract."""
import csv
import hashlib
import json
import math
from pathlib import Path
from .calibration_session import write_json
from .session_review import load_review, save_review
def plan_points(plan: dict, definition: dict) -> list[dict]:
points = {}
origin_x, origin_y = definition["origin_raster_xy_px"]
height, width = definition["raster_size_hw"]
orders = set()
for row in plan["rows"]:
if (row["mapping_sha256"] != definition["mapping_sha256"]
or row["u_positive"] != definition["u_positive"]
or row["v_positive"] != definition["v_positive"]
or (int(row["origin_raster_x_px"]), int(row["origin_raster_y_px"])) != (origin_x, origin_y)
or (int(row["raster_height_px"]), int(row["raster_width_px"])) != (height, width)):
raise ValueError("planned points use a different UV mapping, origin or axes")
u, v, order = int(row["u_px"]), int(row["v_px"]), int(row["order"])
position = [float(row[f"{axis}_sensor_mm"]) for axis in "xyz"]
if (not 0 <= origin_x + u < width or not 0 <= origin_y - v < height
or not all(math.isfinite(x) for x in position) or order <= 0 or order in orders):
raise ValueError("invalid planned point or duplicate order")
orders.add(order)
point = points.setdefault((u, v), {"uv_xy_px": [u, v], "position_sensor_mm": position, "orders": []})
if point["position_sensor_mm"] != position:
raise ValueError("one UV point has inconsistent planned 3D coordinates")
point["orders"].append(order)
if not points:
raise ValueError("the planned-point file contains no points")
return sorted(points.values(), key=lambda p: min(p["orders"]))
def read_plan(path: Path, definition: dict) -> dict:
with path.open(encoding="utf-8-sig", newline="") as handle:
rows = list(csv.DictReader(handle))
plan = {"source": str(path), "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), "rows": rows}
plan_points(plan, definition)
return plan
def overview_rows(review: dict, plan: dict | None = None) -> list[dict]:
planned = plan_points(plan, review["uv_definition"]) if plan is not None else []
planned_uv = {tuple(p["uv_xy_px"]) for p in planned}
actual = {tuple(site["uv_xy_px"]): site for site in review["sites"]}
rows = []
for point in planned + [s for uv, s in actual.items() if uv not in planned_uv]:
site = actual.get(tuple(point["uv_xy_px"]))
if site is not None and point.get("orders") and site.get("position_sensor_mm"):
if not all(math.isclose(a, b, rel_tol=0, abs_tol=1e-6)
for a, b in zip(point["position_sensor_mm"], site["position_sensor_mm"])):
raise ValueError("planned and recorded 3D coordinates do not match")
trials = [t for t in review["trials"] if site is not None and t.get("site_id") == site["site_id"]]
complete = [t for t in trials if t["status"] == "complete"]
windows = site["windows"] if site else []
confirmed = [w for w in windows if w["stability_verified"] and w["status"] == "candidate"]
all_reviewed = bool(complete) and all(
sum(w["trial_id"] == t["trial_id"] for w in confirmed)
== len(t.get("plan", {}).get("steps", [w for w in windows if w["trial_id"] == t["trial_id"]]))
and any(w["trial_id"] == t["trial_id"] for w in confirmed) for t in complete)
state = ("uncollected" if not trials else "incomplete" if len(complete) != len(trials)
else "reviewed" if all_reviewed else "pending")
fit = (site["confirmed_fit"] or site["fit"]) if site else None
training = [w for w in windows if w["branch"] in {"zero", "load"} and w["status"] == "candidate"
and (not site["confirmed_fit"] or w["stability_verified"])] if site else []
rows.append({
"site_id": site["site_id"] if site else None, "uv_xy_px": point["uv_xy_px"],
"position_sensor_mm": site.get("position_sensor_mm") if site else point["position_sensor_mm"],
"orders": point.get("orders", []), "planned": bool(point.get("orders")),
"status": state, "trials": len(trials), "complete_trials": len(complete),
"confirmed_windows": len(confirmed), "windows": len(windows),
"confirmed_levels_n": sorted({w["reference_n"] for w in confirmed if w["branch"] in {"zero", "load"}}),
"fit": fit, "fit_basis": ("confirmed" if site["confirmed_fit"] else "candidate") if fit else "none",
"fit_trials": len({w["trial_id"] for w in training}) if fit else 0,
"fit_range_n": [min(w["reference_n"] for w in training), max(w["reference_n"] for w in training)] if fit else None,
})
return rows
def save_overview(review: dict, plan: dict | None) -> Path:
rows = overview_rows(review, plan)
path = save_review(review)
write_json(path.parent / "overview.json", {
"format_version": "autocal-overview-1.0", "session_id": review["session_id"],
"plan": plan, "sites": rows,
})
return path
def load_overview(directory: Path, path: Path) -> tuple[dict, dict | None]:
review = load_review(directory, path)
overview_path = path.parent / "overview.json"
plan = None
if overview_path.is_file():
saved = json.loads(overview_path.read_text(encoding="utf-8"))
if saved["format_version"] != "autocal-overview-1.0" or saved["session_id"] != review["session_id"]:
raise ValueError("overview identity mismatch")
plan = saved["plan"]
overview_rows(review, plan)
return review, plan
@@ -223,6 +223,7 @@ def review_session(directory: Path, *, selections: dict | None = None) -> dict:
"source_directory": str(root), "session_status": manifest["status"],
"status": "exploratory", "sites": sites, "issues": issues,
"selections": selections, "input_sha256": inputs, "analysis_inputs": analysis_inputs,
"trials": manifest["trials"], "uv_definition": manifest["uv_definition"],
}
@@ -351,7 +351,7 @@ class CalibrationPanel(QWidget):
self._sync()
def _review(self) -> None:
from .session_review_dialog import SessionReviewDialog
from .record_overview_dialog import RecordOverviewDialog
session = self.trial.session
directory = QFileDialog.getExistingDirectory(
@@ -360,7 +360,7 @@ class CalibrationPanel(QWidget):
)
if directory:
try:
dialog = SessionReviewDialog(Path(directory), self)
dialog = RecordOverviewDialog(Path(directory), self)
dialog.exec()
except (OSError, ValueError, KeyError, TypeError) as exc:
self.result_label.setText(t("calibration.error").format(detail=exc))
@@ -0,0 +1,278 @@
"""Recorded-site coverage and normal response, with optional operator CSV plan."""
from pathlib import Path
from orisys_app_shared.i18n import t
from PySide6.QtCharts import QChart, QChartView, QLineSeries, QScatterSeries, QValueAxis
from PySide6.QtCore import Qt
from PySide6.QtGui import QColor, QPainter
from PySide6.QtWidgets import (
QComboBox, QDialog, QFileDialog, QHBoxLayout, QHeaderView, QLabel,
QPushButton, QTableWidget, QTableWidgetItem, QVBoxLayout,
)
from ..record_overview import load_overview, overview_rows, read_plan, save_overview
from ..session_review import review_session
from .session_review_dialog import SessionReviewDialog
class RecordOverviewDialog(QDialog):
def __init__(self, directory: Path, parent=None):
super().__init__(parent)
self.directory = directory
self.review = review_session(directory)
self.plan = None
self.rows = []
self.setWindowTitle(t("overview.title"))
self.resize(1240, 850)
layout = QVBoxLayout(self)
self.notice = QLabel(t("overview.notice"), self)
self.notice.setWordWrap(True)
layout.addWidget(self.notice)
toolbar = QHBoxLayout()
self.map_mode = QComboBox(self)
self.map_mode.addItems([t("overview.coverage"), t("overview.sensitivity")])
toolbar.addWidget(self.map_mode)
for name, slot in (("import_plan", self._import_plan), ("clear_plan", self._clear_plan),
("load", self._load), ("save", self._save)):
button = QPushButton(t("overview." + name), self)
button.setObjectName("overview_" + name)
button.clicked.connect(slot)
toolbar.addWidget(button)
layout.addLayout(toolbar)
self.summary = QLabel(self)
self.summary.setWordWrap(True)
layout.addWidget(self.summary)
charts = QHBoxLayout()
self.uv_chart, self.response_chart = QChart(), QChart()
self.uv_view, self.response_view = QChartView(self.uv_chart), QChartView(self.response_chart)
for view in (self.uv_view, self.response_view):
view.setRenderHint(QPainter.RenderHint.Antialiasing)
charts.addWidget(self.uv_view, 1)
charts.addWidget(self.response_view, 2)
layout.addLayout(charts, 2)
self.table = QTableWidget(0, 11, self)
self.table.setObjectName("overviewSites")
self.table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers)
self.table.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows)
self.table.setSelectionMode(QTableWidget.SelectionMode.SingleSelection)
self.table.setHorizontalHeaderLabels([t("overview.col_" + key) for key in (
"order", "site", "uv", "xyz", "status", "trials", "windows", "levels", "sensitivity", "range", "fit_trials")])
self.table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
self.table.horizontalHeader().setStretchLastSection(True)
layout.addWidget(self.table, 1)
selection = QHBoxLayout()
self.trials = QComboBox(self)
self.trials.setObjectName("overviewTrial")
self.details_button = QPushButton(t("overview.details"), self)
selection.addWidget(QLabel(t("overview.trial_filter"), self))
selection.addWidget(self.trials, 1)
selection.addWidget(self.details_button)
layout.addLayout(selection)
self.selected_summary = QLabel(self)
self.selected_summary.setWordWrap(True)
layout.addWidget(self.selected_summary)
self.message = QLabel(self)
self.message.setWordWrap(True)
layout.addWidget(self.message)
self.table.itemSelectionChanged.connect(self._select_site)
self.map_mode.currentIndexChanged.connect(self._show_uv)
self.trials.currentIndexChanged.connect(self._show_response)
self.details_button.clicked.connect(self._details)
self._refresh()
def _refresh(self):
selected = self.table.currentRow()
self.rows = overview_rows(self.review, self.plan)
self.table.blockSignals(True)
self.table.setRowCount(len(self.rows))
for index, row in enumerate(self.rows):
fit = row["fit"]
sensitivity = (f"{fit['sensitivity_raw_per_n']:.6g} · " + t("overview." + row["fit_basis"])) if fit else "—"
values = [", ".join(map(str, row["orders"])) or "—", row["site_id"] or "—",
", ".join(map(str, row["uv_xy_px"])),
", ".join(f"{x:.3f}" for x in row["position_sensor_mm"]) if row["position_sensor_mm"] else "—",
t("overview." + row["status"]), f"{row['complete_trials']}/{row['trials']}",
f"{row['confirmed_windows']}/{row['windows']}",
", ".join(f"{v:g}" for v in row["confirmed_levels_n"]) or "—", sensitivity,
"–".join(f"{v:g}" for v in row["fit_range_n"]) if fit else "—", str(row["fit_trials"])]
for column, value in enumerate(values):
self.table.setItem(index, column, QTableWidgetItem(value))
self.table.blockSignals(False)
collected = sum(row["trials"] > 0 for row in self.rows)
self.summary.setText(t("overview.summary").format(
collected=collected, planned=sum(r["planned"] for r in self.rows) if self.plan else "—",
missing=sum(r["status"] == "uncollected" for r in self.rows),
incomplete=sum(r["status"] == "incomplete" for r in self.rows),
pending=sum(r["status"] == "pending" for r in self.rows),
reviewed=sum(r["status"] == "reviewed" for r in self.rows),
extra=sum(not r["planned"] and r["trials"] > 0 for r in self.rows) if self.plan else "—",
) + ("\n" + t("overview.no_plan") if self.plan is None else ""))
self.message.setText(t("review.issues").format(count=len(self.review["issues"])))
if self.rows:
self.table.selectRow(min(max(selected, 0), len(self.rows) - 1))
self._select_site()
def _select_site(self):
index = self.table.currentRow()
self.trials.blockSignals(True)
self.trials.clear()
self.trials.addItem(t("overview.all_trials"), None)
if index >= 0:
site_id = self.rows[index]["site_id"]
for trial in self.review["trials"]:
if site_id is not None and trial.get("site_id") == site_id:
self.trials.addItem(trial["trial_id"] + " · " + t("calibration." + trial["status"]), trial["trial_id"])
self.trials.blockSignals(False)
self._show_uv()
self._show_response()
@staticmethod
def _clear_chart(chart):
chart.removeAllSeries()
for axis in chart.axes():
axis.hide()
chart.removeAxis(axis)
axis.deleteLater()
@staticmethod
def _axes(chart):
points = [point for series in chart.series() for point in series.points()]
for horizontal in (True, False):
axis = QValueAxis()
values = [p.x() if horizontal else p.y() for p in points]
if values:
low, high = min(values), max(values)
margin = max((high - low) * .1, .1)
axis.setRange(low - margin, high + margin)
chart.addAxis(axis, Qt.AlignmentFlag.AlignBottom if horizontal else Qt.AlignmentFlag.AlignLeft)
for series in chart.series():
series.attachAxis(axis)
def _show_uv(self):
self._clear_chart(self.uv_chart)
sensitivity = self.map_mode.currentIndex() == 1
values = [s["confirmed_fit"]["sensitivity_raw_per_n"] for s in self.review["sites"] if s["confirmed_fit"]]
low, high = (min(values), max(values)) if values else (0, 0)
colors = {"uncollected": "#929aa7", "incomplete": "#ca4c46", "pending": "#d8911d", "reviewed": "#29835f"}
for index, row in enumerate(self.rows):
color = QColor(colors[row["status"]])
if sensitivity:
color = QColor("#929aa7")
if row["fit_basis"] == "confirmed":
value = row["fit"]["sensitivity_raw_per_n"]
ratio = (value - low) / (high - low) if high > low else .5
color = QColor.fromHsvF((1 - ratio) * .65, .8, .85)
series = QScatterSeries()
series.setColor(color)
series.setBorderColor(QColor("#151515") if index == self.table.currentRow() else color)
series.setMarkerSize(15 if index == self.table.currentRow() else 10)
series.append(*row["uv_xy_px"])
series.clicked.connect(lambda _point, target=index: self.table.selectRow(target))
self.uv_chart.addSeries(series)
self._axes(self.uv_chart)
definition = self.review["uv_definition"]
origin = definition.get("origin_raster_xy_px")
shape = definition.get("raster_size_hw")
for axis in self.uv_chart.axes(Qt.Orientation.Horizontal):
axis.setTitleText("U (px)")
if origin and shape:
axis.setRange(-origin[0] - .5, shape[1] - origin[0] - .5)
for axis in self.uv_chart.axes(Qt.Orientation.Vertical):
axis.setTitleText("V (px)")
if origin and shape:
axis.setRange(origin[1] - shape[0] + .5, origin[1] + .5)
self.uv_chart.legend().hide()
self.uv_chart.setTitle(t("overview.sensitivity_map").format(low=low, high=high)
if sensitivity else t("overview.coverage_map"))
def _show_response(self):
self._clear_chart(self.response_chart)
index = self.table.currentRow()
site_id = self.rows[index]["site_id"] if index >= 0 else None
site = next((s for s in self.review["sites"] if s["site_id"] == site_id), None)
trial_id = self.trials.currentData()
windows = [w for w in site["windows"] if trial_id is None or w["trial_id"] == trial_id] if site else []
self.details_button.setEnabled(site is not None and bool(windows))
for trial_index, trial in enumerate(sorted({w["trial_id"] for w in windows})):
for branch, color in (("zero", "#657388"), ("load", "#167ac6"), ("unload", "#df8b13")):
for confirmed in (True, False):
points = [w for w in windows if w["trial_id"] == trial and w["branch"] == branch
and w["status"] == "candidate" and w["stability_verified"] == confirmed]
if not points:
continue
series = QScatterSeries()
series.setName(trial + " · " + t("calibration." + branch) + " · " + t("overview.confirmed" if confirmed else "overview.candidate"))
series.setMarkerShape(QScatterSeries.MarkerShape.MarkerShapeCircle if confirmed else QScatterSeries.MarkerShape.MarkerShapeRectangle)
series.setColor(QColor(color).lighter(100 + 20 * (trial_index % 4)))
series.setMarkerSize(10 if confirmed else 7)
for window in points:
series.append(window["reference_n"], window["raw_mean"])
self.response_chart.addSeries(series)
fit = (site["confirmed_fit"] or site["fit"]) if site else None
if fit and windows:
training = [w for w in site["windows"] if w["branch"] in {"zero", "load"}
and w["status"] == "candidate" and (not site["confirmed_fit"] or w["stability_verified"])]
line = QLineSeries()
line.setName(t("overview.site_fit"))
for load in (min(w["reference_n"] for w in training), max(w["reference_n"] for w in training)):
line.append(load, (load - fit["offset_n"]) / fit["gain_n_per_raw"])
self.response_chart.addSeries(line)
self._axes(self.response_chart)
for axis in self.response_chart.axes(Qt.Orientation.Horizontal):
axis.setTitleText(t("review.force_axis"))
for axis in self.response_chart.axes(Qt.Orientation.Vertical):
axis.setTitleText(t("review.raw_axis"))
self.response_chart.setTitle(t("overview.response_title"))
self.response_chart.legend().setAlignment(Qt.AlignmentFlag.AlignRight)
self.selected_summary.setText(t("overview.selected_fit").format(
sensitivity=f"{fit['sensitivity_raw_per_n']:.6g}" if fit else "—",
basis=t("overview.confirmed" if site and site["confirmed_fit"] else "overview.candidate") if fit else "—",
) if windows else t("overview.no_windows"))
def _import_plan(self):
path, _ = QFileDialog.getOpenFileName(self, t("overview.import_plan"), str(self.directory), "CSV (*.csv)")
if path:
try:
plan = read_plan(Path(path), self.review["uv_definition"])
overview_rows(self.review, plan)
self.plan = plan
self._refresh()
except (OSError, ValueError, KeyError, TypeError) as exc:
self.message.setText(t("calibration.error").format(detail=exc))
def _clear_plan(self):
self.plan = None
self._refresh()
def _details(self):
row = self.rows[self.table.currentRow()]
index = next(i for i, site in enumerate(self.review["sites"]) if site["site_id"] == row["site_id"])
dialog = SessionReviewDialog(self.directory, self, review=self.review)
dialog.sites.setCurrentIndex(index)
trial = self.trials.currentData()
if trial:
windows = dialog.review["sites"][index]["windows"]
match = next((i for i, w in enumerate(windows) if w["trial_id"] == trial), None)
if match is not None:
dialog.table.selectRow(match)
dialog.exec()
self.review = dialog.review
self._refresh()
def _save(self):
try:
path = save_overview(self.review, self.plan)
self.message.setText(t("review.saved").format(path=path))
except (OSError, ValueError, KeyError, TypeError) as exc:
self.message.setText(t("calibration.error").format(detail=exc))
def _load(self):
path, _ = QFileDialog.getOpenFileName(self, t("overview.load"), str(self.directory / "derived"), "JSON (*.json)")
if path:
try:
review, plan = load_overview(self.directory, Path(path))
self.review, self.plan = review, plan
self._refresh()
except (OSError, ValueError, KeyError, TypeError) as exc:
self.message.setText(t("calibration.error").format(detail=exc))
@@ -19,11 +19,11 @@ from ..session_review import review_session, save_review, load_review
class SessionReviewDialog(QDialog):
"""Open persisted data independently of the acquisition page's state."""
def __init__(self, directory: Path, parent=None) -> None:
def __init__(self, directory: Path, parent=None, *, review: dict | None = None) -> None:
super().__init__(parent)
self.setWindowTitle(t("review.title"))
self.resize(1080, 760)
self.review = review_session(directory)
self.review = review if review is not None else review_session(directory)
self.directory = directory
layout = QVBoxLayout(self)
self.notice = QLabel(t("review.notice"), self)
+7
View File
@@ -238,6 +238,13 @@ def test_operator_export_reproduces_uv_sites_and_rejects_edited_xyz(qt_app, flat
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)]
+173
View File
@@ -0,0 +1,173 @@
"""Recorded coverage, plan identity and the normal-response operator journey."""
import csv
import json
from copy import deepcopy
import pytest
from PySide6.QtWidgets import QPushButton
from test_session_review import repeated_session as repeated_session, session as session
from orisys_autocal.calibration_session import write_json
from orisys_autocal.record_overview import load_overview, overview_rows, read_plan, save_overview
from orisys_autocal.session_review import review_session
@pytest.fixture
def overview_session(repeated_session):
root, selections = repeated_session
manifest = json.loads((root / "session.json").read_text(encoding="utf-8"))
definition = {"mapping_sha256": "test-map", "origin_raster_xy_px": [5, 8],
"raster_size_hw": [12, 12], "u_positive": "right", "v_positive": "up", "unit": "px"}
manifest["uv_definition"] = definition
for index, site in enumerate(manifest["sites"]):
site["position_sensor_mm"] = [index + 1., 2., 3.]
for trial in manifest["trials"]:
if trial["status"] == "complete":
path = root / trial["directory"] / "session.json"
metadata = json.loads(path.read_text(encoding="utf-8"))
metadata["uv_definition"] = definition
write_json(path, metadata)
trial["plan"] = {"steps": list(range(5))}
manifest["sites"].append({"site_id": "site-003", "uv_xy_px": [3, 2], "position_sensor_mm": [3., 2., 3.]})
manifest["trials"].append({"trial_id": "trial-005", "site_id": "site-003", "status": "incomplete"})
write_json(root / "session.json", manifest)
rows = []
for order, (u, v), position in ((1, (-3, 4), [2., 2., 3.]),
(2, (1, -2), [1., 2., 3.]),
(3, (-1, 1), [4., 2., 3.])):
rows.append({"order": order, "site_id": f"site-{order:03d}", "u_px": u, "v_px": v,
"x_sensor_mm": position[0], "y_sensor_mm": position[1], "z_sensor_mm": position[2],
"origin_raster_x_px": 5, "origin_raster_y_px": 8,
"raster_width_px": 12, "raster_height_px": 12, "mapping_sha256": "test-map",
"u_positive": "right", "v_positive": "up"})
plan_path = root / "operator-plan.csv"
with plan_path.open("w", newline="", encoding="utf-8-sig") as handle:
writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
writer.writeheader()
writer.writerows(rows)
return root, selections, plan_path
def test_overview_distinguishes_coverage_without_matching_plan_ids(overview_session):
root, selections, plan_path = overview_session
review = review_session(root, selections=selections)
plan = read_plan(plan_path, review["uv_definition"])
rows = overview_rows(review, plan)
assert [r["site_id"] for r in rows] == ["site-002", "site-001", None, "site-003"]
assert [r["status"] for r in rows] == ["reviewed", "reviewed", "uncollected", "incomplete"]
assert rows[1]["complete_trials"] == 2 and rows[1]["fit_trials"] == 2
assert rows[1]["confirmed_windows"] == 10 and rows[1]["fit_range_n"] == [0, 10]
assert rows[2]["fit"] is None and rows[3]["windows"] == 0
assert overview_rows(review_session(root), plan)[0]["status"] == "pending"
saved = save_overview(review, plan)
restored, stored_plan = load_overview(root, saved)
assert overview_rows(restored, stored_plan) == rows
assert stored_plan["sha256"] == plan["sha256"]
@pytest.mark.parametrize("key,value", [("mapping_sha256", "other"), ("origin_raster_x_px", 6),
("v_positive", "down"), ("u_px", 999), ("x_sensor_mm", "nan")])
def test_plan_refuses_mapping_or_coordinate_mismatch(overview_session, key, value):
root, selections, plan_path = overview_session
review = review_session(root, selections=selections)
plan = read_plan(plan_path, review["uv_definition"])
plan["rows"][0][key] = value
with pytest.raises(ValueError):
overview_rows(review, plan)
def test_plan_merges_repeated_uv_points_and_refuses_conflicting_positions(overview_session):
root, selections, plan_path = overview_session
review = review_session(root, selections=selections)
plan = read_plan(plan_path, review["uv_definition"])
plan["rows"].append({**plan["rows"][0], "order": "4"})
assert overview_rows(review, plan)[0]["orders"] == [1, 4]
plan["rows"][0]["x_sensor_mm"] = "9"
with pytest.raises(ValueError):
overview_rows(review, plan)
plan["rows"].pop()
with pytest.raises(ValueError, match="3D"):
overview_rows(review, plan)
def test_overview_requires_all_planned_hold_windows_for_reviewed_status(overview_session):
root, selections, _ = overview_session
review = review_session(root, selections=selections)
altered = deepcopy(review)
altered["trials"][0]["plan"]["steps"].append(5)
assert overview_rows(altered)[0]["status"] == "pending"
assert overview_rows(review)[0]["status"] == "reviewed"
def test_overview_gui_plan_click_trial_filter_and_saved_roundtrip(qt_app, overview_session, monkeypatch, tmp_path):
from orisys_autocal.ui import record_overview_dialog
from orisys_autocal.ui.session_review_dialog import SessionReviewDialog
root, selections, plan_path = overview_session
dialog = record_overview_dialog.RecordOverviewDialog(root)
dialog.review = review_session(root, selections=selections)
dialog._refresh()
monkeypatch.setattr(record_overview_dialog.QFileDialog, "getOpenFileName", lambda *args: (str(plan_path), ""))
dialog.findChild(QPushButton, "overview_import_plan").click()
assert dialog.table.rowCount() == 4
assert dialog.table.item(0, 1).text() == "site-002"
dialog.uv_chart.series()[1].clicked.emit(dialog.uv_chart.series()[1].at(0))
assert dialog.table.currentRow() == 1
assert len(dialog.uv_chart.axes()) == 2
assert all(len(s.attachedAxes()) == 2 for s in dialog.uv_chart.series())
assert dialog.trials.count() == 3
dialog.trials.setCurrentIndex(1)
names = [s.name() for s in dialog.response_chart.series()]
assert any("trial-001" in name for name in names)
assert not any("trial-004" in name for name in names)
assert len(dialog.response_chart.axes()) == 2
assert all(len(s.attachedAxes()) == 2 for s in dialog.response_chart.series())
load_series = next(s for s in dialog.response_chart.series() if s.count() == 2 and "trial-001" in s.name())
assert [(load_series.at(i).x(), load_series.at(i).y()) for i in range(2)] == [(5, 11), (10, 21)]
opened = []
monkeypatch.setattr(SessionReviewDialog, "exec", lambda child: opened.append(child))
dialog.details_button.click()
assert opened[0].sites.currentIndex() == 0
assert opened[0].review["selections"] == selections
dialog.table.selectRow(2)
assert not dialog.details_button.isEnabled()
dialog.table.selectRow(3)
dialog.trials.setCurrentIndex(1)
assert not dialog.response_chart.series()
assert not dialog.details_button.isEnabled()
dialog.table.selectRow(2)
dialog.map_mode.setCurrentIndex(1)
assert len(dialog.uv_chart.series()) == 4
assert dialog.uv_chart.series()[0].color().name() != "#929aa7"
assert dialog.uv_chart.series()[2].color().name() == "#929aa7"
assert dialog.uv_chart.series()[3].color().name() == "#929aa7"
dialog.show()
qt_app.processEvents()
assert dialog.grab().save(str(tmp_path / "overview-sensitivity.png"))
dialog.table.selectRow(1)
dialog.map_mode.setCurrentIndex(0)
qt_app.processEvents()
assert dialog.grab().save(str(tmp_path / "overview-response.png"))
dialog.findChild(QPushButton, "overview_save").click()
saved = next(root.glob("derived/*/review.json"))
dialog.findChild(QPushButton, "overview_clear_plan").click()
assert dialog.table.rowCount() == 3
monkeypatch.setattr(record_overview_dialog.QFileDialog, "getOpenFileName", lambda *args: (str(saved), ""))
dialog.findChild(QPushButton, "overview_load").click()
assert dialog.table.rowCount() == 4
dialog.close()
def test_calibration_review_action_opens_record_overview(make_window, overview_session, monkeypatch):
from orisys_autocal.ui.record_overview_dialog import RecordOverviewDialog
from orisys_autocal.ui.calibration_panel import QFileDialog
root, _, _ = overview_session
opened = []
monkeypatch.setattr(QFileDialog, "getExistingDirectory", lambda *args: str(root))
monkeypatch.setattr(RecordOverviewDialog, "exec", lambda dialog: opened.append(dialog))
window = make_window()
window.calibration_panel.review_button.click()
assert len(opened) == 1 and len(opened[0].rows) == 3
window.close()