补齐 Autocal 测点质量与独立 trial 验证

按已确认窗口比较重复性、加载卸载差异和零载恢复,并逐 trial 留出验证。55 项定向测试、已录视频复核闭环及代码文档检查通过。
This commit is contained in:
ykcheungab
2026-10-11 18:27:35 +08:00
parent 3304b4eda9
commit 161e64278b
7 changed files with 352 additions and 4 deletions
@@ -38,6 +38,78 @@ def _fit(windows: list[dict]) -> dict | None:
}
def _quality(windows: list[dict]) -> dict:
"""Compare confirmed window means, never random frames from the same trial."""
confirmed = [w for w in windows if w["stability_verified"] and w["status"] == "candidate"]
trial_ids = sorted({w["trial_id"] for w in confirmed})
groups = {}
for window in confirmed:
key = (window["branch"], window["reference_n"])
groups.setdefault(key, {}).setdefault(window["trial_id"], []).append(window["raw_mean"])
repeatability = []
for (branch, load), trials in sorted(groups.items()):
means = {trial: float(np.mean(values)) for trial, values in trials.items()}
values = list(means.values())
repeatability.append({
"branch": branch, "reference_n": load, "trials": len(means),
"trial_means_raw": means,
"raw_range": float(np.ptp(values)) if len(values) >= 2 else None,
"raw_std_between_trials": float(np.std(values, ddof=1)) if len(values) >= 2 else None,
})
hysteresis, zero_return, folds = [], [], []
errors = []
for trial_id in trial_ids:
own = [w for w in confirmed if w["trial_id"] == trial_id]
# Exact declared reference forces only; no nearest-load interpolation.
branches = {}
for window in own:
branches.setdefault((window["branch"], window["reference_n"]), []).append(window["raw_mean"])
branch_means = {key: float(np.mean(values)) for key, values in branches.items()}
for (branch, load), raw in sorted(branch_means.items()):
if branch == "load" and load > 0 and ("unload", load) in branch_means:
unload = branch_means[("unload", load)]
hysteresis.append({"trial_id": trial_id, "reference_n": load,
"load_raw": raw, "unload_raw": unload,
"unload_minus_load_raw": unload - raw})
if ("zero", 0.0) in branch_means and ("unload", 0.0) in branch_means:
zero_return.append({"trial_id": trial_id,
"initial_zero_raw": branch_means[("zero", 0.0)],
"final_zero_raw": branch_means[("unload", 0.0)],
"final_minus_initial_raw": branch_means[("unload", 0.0)] - branch_means[("zero", 0.0)]})
training = [w for w in confirmed if w["trial_id"] != trial_id and w["branch"] in {"zero", "load"}]
fit = _fit(training)
fold = {"held_out_trial": trial_id,
"training_trials": sorted({w["trial_id"] for w in training}),
"fit": fit, "windows": [], "status": "insufficient_training"}
if fit is not None:
low, high = min(w["raw_mean"] for w in training), max(w["raw_mean"] for w in training)
for window in own:
predicted = fit["gain_n_per_raw"] * window["raw_mean"] + fit["offset_n"]
error = predicted - window["reference_n"]
fold["windows"].append({
"step_index": window["step_index"], "branch": window["branch"],
"reference_n": window["reference_n"], "predicted_n": predicted,
"error_n": error, "outside_training_raw_range": not low <= window["raw_mean"] <= high,
})
errors.append(error)
fold["status"] = "available"
folds.append(fold)
return {
"basis": "confirmed_window_means", "status": "exploratory",
"reference_alignment_possible": bool(confirmed) and all(w["reference_alignment_possible"] for w in confirmed),
"repeatability": repeatability, "hysteresis": hysteresis, "zero_return": zero_return,
"validation": {
"method": "leave_one_trial_out", "folds": folds,
"status": "available" if errors else "insufficient_data",
"evaluated_windows": len(errors),
"rmse_n": float(np.sqrt(np.mean(np.square(errors)))) if errors else None,
"mae_n": float(np.mean(np.abs(errors))) if errors else None,
"max_abs_error_n": float(np.max(np.abs(errors))) if errors else None,
"outside_training_raw_range": sum(w["outside_training_raw_range"] for f in folds for w in f["windows"]),
},
}
def review_session(directory: Path, *, selections: dict | None = None) -> dict:
"""Join frame identity to host timestamps and inspect only complete windows.
@@ -144,7 +216,8 @@ def review_session(directory: Path, *, selections: dict | None = None) -> dict:
fit = _fit(training)
confirmed = [w for w in training if w["stability_verified"]]
confirmed_fit = _fit(confirmed)
sites.append({**site, "windows": selected, "fit": fit, "confirmed_fit": confirmed_fit})
sites.append({**site, "windows": selected, "fit": fit,
"confirmed_fit": confirmed_fit, "quality": _quality(selected)})
return {
"format_version": "autocal-review-1.0", "session_id": manifest["session_id"],
"source_directory": str(root), "session_status": manifest["status"],
@@ -0,0 +1,78 @@
"""Per-site trial comparisons and held-out validation from reviewed data."""
from orisys_app_shared.i18n import t
from PySide6.QtWidgets import (
QDialog, QHeaderView, QLabel, QTableWidget, QTableWidgetItem,
QTabWidget, QVBoxLayout, QWidget,
)
def _number(value):
return "—" if value is None else f"{value:.6g}"
class QualityReviewDialog(QDialog):
def __init__(self, site: dict, parent=None):
super().__init__(parent)
self.setWindowTitle(t("quality.title") + " · " + site["site_id"])
self.resize(1060, 560)
layout = QVBoxLayout(self)
notice = QLabel(t("quality.notice"), self)
notice.setWordWrap(True)
layout.addWidget(notice)
self.tabs = QTabWidget(self)
layout.addWidget(self.tabs, 1)
quality = site["quality"]
self._table("repeatability", ("branch", "load", "trials", "range", "std"), [
[t("calibration." + row["branch"]), _number(row["reference_n"]), str(row["trials"]),
_number(row["raw_range"]), _number(row["raw_std_between_trials"])]
for row in quality["repeatability"]])
self._table("hysteresis", ("trial", "load", "load_raw", "unload_raw", "delta_raw"), [
[row["trial_id"], _number(row["reference_n"]), _number(row["load_raw"]),
_number(row["unload_raw"]), _number(row["unload_minus_load_raw"])]
for row in quality["hysteresis"]])
self._table("zero_return", ("trial", "initial_zero", "final_zero", "delta_raw"), [
[row["trial_id"], _number(row["initial_zero_raw"]), _number(row["final_zero_raw"]),
_number(row["final_minus_initial_raw"])] for row in quality["zero_return"]])
validation = quality["validation"]
rows = []
for fold in validation["folds"]:
prefix = [fold["held_out_trial"], ", ".join(fold["training_trials"]) or "—"]
if not fold["windows"]:
rows.append(prefix + ["—"] * 4 + [t("quality.insufficient_training")])
for window in fold["windows"]:
rows.append(prefix + [t("calibration." + window["branch"]),
_number(window["reference_n"]), _number(window["predicted_n"]),
_number(window["error_n"]),
t("quality.outside_range" if window["outside_training_raw_range"] else "quality.in_range")])
summary = t("quality.validation_summary").format(
count=validation["evaluated_windows"], rmse=_number(validation["rmse_n"]),
mae=_number(validation["mae_n"]), maximum=_number(validation["max_abs_error_n"]),
outside=validation["outside_training_raw_range"],
)
self._table("validation", ("held_out", "training", "branch", "load", "predicted", "error", "range_status"),
rows, summary)
def _table(self, key, columns, rows, summary=""):
page = QWidget(self.tabs)
layout = QVBoxLayout(page)
note = QLabel(t("quality." + key + "_note"), page)
note.setWordWrap(True)
layout.addWidget(note)
if summary:
label = QLabel(summary, page)
label.setWordWrap(True)
layout.addWidget(label)
if not rows:
layout.addWidget(QLabel(t("quality.no_data"), page))
table = QTableWidget(len(rows), len(columns), page)
table.setObjectName("quality_" + key)
table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers)
table.setHorizontalHeaderLabels([t("quality.col_" + column) for column in columns])
table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
table.horizontalHeader().setStretchLastSection(True)
for row, values in enumerate(rows):
for column, value in enumerate(values):
table.setItem(row, column, QTableWidgetItem(value))
layout.addWidget(table, 1)
self.tabs.addTab(page, t("quality." + key))
@@ -43,7 +43,13 @@ class SessionReviewDialog(QDialog):
layout.addLayout(row)
self.summary = QLabel(self)
self.summary.setWordWrap(True)
layout.addWidget(self.summary)
summary_row = QHBoxLayout()
summary_row.addWidget(self.summary, 1)
self.quality_button = QPushButton(t("quality.open"), self)
self.quality_button.setObjectName("reviewQuality")
self.quality_button.clicked.connect(self._quality)
summary_row.addWidget(self.quality_button)
layout.addLayout(summary_row)
self.chart = QChart()
self.chart_view = QChartView(self.chart, self)
self.chart_view.setRenderHint(QPainter.RenderHint.Antialiasing)
@@ -95,6 +101,7 @@ class SessionReviewDialog(QDialog):
def _show_site(self) -> None:
index = self.sites.currentIndex()
self.quality_button.setEnabled(index >= 0)
self.chart.removeAllSeries()
for axis in self.chart.axes():
self.chart.removeAxis(axis)
@@ -150,6 +157,13 @@ class SessionReviewDialog(QDialog):
self.table.selectRow(0)
self._show_window()
def _quality(self):
from .quality_review_dialog import QualityReviewDialog
index = self.sites.currentIndex()
if index >= 0:
QualityReviewDialog(self.review["sites"][index], self).exec()
def _current_window(self):
site, row = self.sites.currentIndex(), self.table.currentRow()
if site < 0 or row < 0 or row >= len(self.review["sites"][site]["windows"]):