subject: fix calibration cancel, unit mixing, FN clamping, export/apply and depth scale

2026-09-13 12:35:05 +08:00
parent db228b6f57
commit f501fa93f3
9 changed files with 725 additions and 78 deletions
@@ -43,6 +43,11 @@ class ArrowPanel(QFrame):
        self.fps_label = QLabel()
        self.fps_label.setObjectName("panelMeta")
        self.fps_label.setVisible(False)
+        # Shows the fixed depth colour-scale range, so two frames can be
+        # compared instead of the scale silently re-fitting every frame.
+        self.depth_scale_label = QLabel()
+        self.depth_scale_label.setObjectName("panelMeta")
+        self.depth_scale_label.setVisible(False)
        self._fps_visible = False
        self._fps_value: float | None = None
        self._rotation_deg = 90
@@ -53,6 +58,7 @@ class ArrowPanel(QFrame):
        self.view_combo.currentIndexChanged.connect(self._on_view_index_changed)
        header.addWidget(self.title)
        header.addStretch(1)
+        header.addWidget(self.depth_scale_label)
        header.addWidget(self.fps_label)
        header.addWidget(self.view_combo)
        layout.addLayout(header)
@@ -111,6 +117,16 @@ class ArrowPanel(QFrame):
        self._fps_value = None if fps is None else float(fps)
        self._refresh_fps_label()

+    def set_depth_scale(self, text: str | None) -> None:
+        """Show the fixed depth colour-scale range (``None`` hides it)."""
+        if not text:
+            self.depth_scale_label.clear()
+            self.depth_scale_label.setVisible(False)
+            return
+        self.depth_scale_label.setText(str(text))
+        self.depth_scale_label.setToolTip(t("panels.depth_scale_tip"))
+        self.depth_scale_label.setVisible(True)
+
    def _refresh_fps_label(self) -> None:
        if not self._fps_visible or self._fps_value is None:
            self.fps_label.clear()
@@ -2,9 +2,6 @@

 from __future__ import annotations

-import math
-import time
-
 from PyQt5.QtCore import Qt, pyqtSignal
 from PyQt5.QtGui import QFont, QFontDatabase
 from PyQt5.QtWidgets import (
@@ -35,7 +32,7 @@ from orisys_app_shared.force.calibration_io import (
    meta_from_sensor_json,
 )
 from orisys_app_shared.i18n import t
-from orisys_app_shared.paths import calibration_export_dir, sensors_dir
+from orisys_app_shared.paths import calibration_export_dir
 from orisys_app_shared.theme import color

 CHANNEL_LABELS = ("fn", "fx", "fy")
@@ -51,7 +48,8 @@ def _mono_font() -> str:

 class CalPanel(QFrame):
    status_message = pyqtSignal(str)
-    calibration_exported = pyqtSignal(str)  # path applied to running sensor
+    #: Coefficients were written; the window asks the user before applying them.
+    calibration_exported = pyqtSignal(str)

    def __init__(self, parent=None, *, theme: str = "dark") -> None:
        super().__init__(parent)
@@ -60,16 +58,14 @@ class CalPanel(QFrame):
        self._store = CalSessionStore()
        self._channel: ChannelKey = "fn"
        self._running = False
-        self._recording = False
-        self._record_deadline = 0.0
-        self._record_started = 0.0
-        self._record_samples: list[float] = []
+        #: Set by :meth:`bind_recorder`. Sampling lives there, not here, so the
+        #: lifecycle no longer depends on this panel being visible.
+        self._recorder = None
        self._live_fn = 0.0
        self._live_fx = 0.0
        self._live_fy = 0.0
        self._sensor_type = "box"
        self._config_ref: str | None = None
-        self._record_duration_s = 7.0
        self._mono = _mono_font()
        self._mono_font = QFont(self._mono, 10)

@@ -201,32 +197,58 @@ class CalPanel(QFrame):
        self._theme = theme
        self._render_live_label()

+    # --------------------------------------------------------- recorder link
+    def bind_recorder(self, recorder) -> None:
+        """Attach the sampling state machine that owns the recording lifecycle."""
+        self._recorder = recorder
+
+    @property
+    def _recording(self) -> bool:
+        return bool(self._recorder is not None and self._recorder.is_recording)
+
    def set_running(self, running: bool) -> None:
        self._running = bool(running)
        if not self._running and self._recording:
-            self._cancel_recording()
+            self._recorder.cancel()
        self._refresh_enabled()

    def set_meta_context(self, *, sensor_type: str, config_ref: str | None) -> None:
        self._sensor_type = str(sensor_type or "box")
        self._config_ref = config_ref

-    def push_raw(self, fn: float, fx: float, fy: float, _timestamp: float | None = None) -> None:
+    def set_live(self, fn: float, fx: float, fy: float) -> None:
+        """Update the live readout. Sampling is handled by the recorder."""
        self._live_fn = float(fn)
        self._live_fx = float(fx)
        self._live_fy = float(fy)
        self._render_live_label()
-        if not self._recording:
+
+    def set_progress(self, fraction: float, elapsed_s: float, total_s: float) -> None:
+        self.progress.setVisible(True)
+        self.progress.setValue(max(0, min(1000, int(fraction * 1000))))
+        self.progress.setFormat(f"{elapsed_s:.1f} / {total_s:.1f} s")
+
+    def add_point(self, point) -> None:
+        """Record a finished calibration point (three axes captured)."""
+        stats = point.loaded_stats()
+        sample = CalSample(
+            weight_n=float(point.reference_load_n),
+            raw_mean=float(stats.mean),
+            n=int(stats.n),
+            std=float(stats.std),
+        )
+        self._store.upsert(point.channel, sample)
+        self._refresh_table()
+        if point.status == "few_samples":
+            self.status_message.emit(t("cal.record_failed_few"))
            return
-        now = time.perf_counter()
-        value = {"fn": self._live_fn, "fx": self._live_fx, "fy": self._live_fy}[self._channel]
-        self._record_samples.append(float(value))
-        span = max(1e-6, self._record_deadline - self._record_started)
-        frac = min(1.0, max(0.0, (now - self._record_started) / span))
-        self.progress.setValue(int(frac * 1000))
-        self.progress.setFormat(f"{now - self._record_started:.1f} / {span:.1f} s")
-        if now >= self._record_deadline:
-            self._finish_recording()
+        self.status_message.emit(
+            t("cal.record_ok", weight=f"{sample.weight_n:g}", raw=f"{sample.raw_mean:.3f}")
+        )
+
+    def push_raw(self, fn: float, fx: float, fy: float, _timestamp: float | None = None) -> None:
+        """Back-compat shim: live readout only, sampling lives in the recorder."""
+        self.set_live(fn, fx, fy)

    def retranslate(self) -> None:
        self.title.setText(t("panels.cal"))
@@ -277,14 +299,19 @@ class CalPanel(QFrame):
        if key is None:
            return
        if self._recording:
-            self._cancel_recording()
+            self._recorder.cancel()
        self._channel = key  # type: ignore[assignment]
        self._refresh_channel_badge()
        self._refresh_table()

    def _refresh_enabled(self) -> None:
-        can_record = self._running and not self._recording
+        # While recording the button must stay **enabled**: it is the cancel
+        # affordance. Disabling it here was why a recording could not be stopped.
+        can_record = self._running
        self.btn_record.setEnabled(can_record)
+        self.btn_record.setText(
+            t("cal.recording_stop") if self._recording else t("cal.record")
+        )
        self.btn_record.setObjectName("dangerButton" if self._recording else "primaryButton")
        self.btn_record.style().unpolish(self.btn_record)
        self.btn_record.style().polish(self.btn_record)
@@ -298,6 +325,8 @@ class CalPanel(QFrame):
        record_hint = t("cal.record_disabled") if not self._running else ""
        if not self._recording:
            self.btn_record.setToolTip(record_hint)
+        else:
+            self.btn_record.setToolTip(t("cal.recording_stop_tip"))

    def _render_live_label(self) -> None:
        colors = {
@@ -321,53 +350,47 @@ class CalPanel(QFrame):
        )

    def _on_record_clicked(self) -> None:
+        if self._recorder is None:
+            return
        if self._recording:
-            self._cancel_recording()
-            self.status_message.emit(t("cal.record_cancelled"))
+            self._recorder.cancel()
            return
        if not self._running:
            return
-        self._recording = True
-        self._record_samples = []
-        self._record_duration_s = float(self.duration_spin.value())
-        self._record_started = time.perf_counter()
-        self._record_deadline = self._record_started + self._record_duration_s
+        duration = float(self.duration_spin.value())
+        self._recorder.start(
+            channel=self._channel,
+            reference_load_n=float(self.weight_spin.value()),
+            duration_s=duration,
+            direction="load",
+        )
        self.progress.setVisible(True)
        self.progress.setValue(0)
-        self.progress.setFormat(f"0.0 / {self._record_duration_s:.1f} s")
+        self.progress.setFormat(f"0.0 / {duration:.1f} s")
        self._refresh_enabled()
        self.status_message.emit(t("cal.recording"))

-    def _cancel_recording(self) -> None:
-        self._recording = False
-        self._record_samples = []
+    def on_record_cancelled(self) -> None:
+        """Called by the window when the recorder was cancelled."""
        self.progress.setValue(0)
        self.progress.setFormat("")
        self._refresh_enabled()
+        self.status_message.emit(t("cal.record_cancelled"))

-    def _finish_recording(self) -> None:
-        samples = list(self._record_samples)
-        self._recording = False
-        self._record_samples = []
+    def on_record_finished(self, point) -> None:
        self.progress.setValue(1000)
        self.progress.setFormat("")
        self._refresh_enabled()
-        if len(samples) < 2:
-            self.status_message.emit(t("cal.record_failed_few"))
+        self.add_point(point)
+
+    def poll_recorder(self) -> None:
+        """Mirror the recorder's progress; the recorder owns the deadline."""
+        if not self._recording:
            return
-        mean = sum(samples) / len(samples)
-        var = sum((x - mean) ** 2 for x in samples) / len(samples)
-        std = math.sqrt(var)
-        sample = CalSample(
-            weight_n=float(self.weight_spin.value()),
-            raw_mean=mean,
-            n=len(samples),
-            std=std,
-        )
-        self._store.upsert(self._channel, sample)
-        self._refresh_table()
-        self.status_message.emit(
-            t("cal.record_ok", weight=f"{sample.weight_n:g}", raw=f"{sample.raw_mean:.3f}")
+        self.set_progress(
+            self._recorder.progress(),
+            self._recorder.elapsed_s(),
+            self._recorder.requested_duration_s,
        )

    def _refresh_table(self) -> None:
@@ -410,16 +433,17 @@ class CalPanel(QFrame):
        meta["sensor_type"] = self._sensor_type
        meta["sn"] = self.sn_edit.text().strip()
        meta["record_duration_s"] = float(self.duration_spin.value())
-        local_bundle = sensors_dir() / self._sensor_type
-        if not local_bundle.is_dir():
-            local_bundle = None
        try:
-            primary, local_path, warnings = export_force_calibration_v2(
+            # Write into the user's calibration directory only. Overwriting the
+            # device bundle under config/sensors/<type>/ is now an explicit,
+            # separate step so that exporting cannot silently change what the
+            # next Sensor() will load.
+            primary, _local_path, warnings = export_force_calibration_v2(
                self._store,
                export_dir=calibration_export_dir(),
                sensor_type=self._sensor_type,
                meta=meta,
-                local_bundle_dir=local_bundle,
+                local_bundle_dir=None,
            )
        except ValueError as exc:
            QMessageBox.warning(self, t("panels.cal"), str(exc))
@@ -429,12 +453,9 @@ class CalPanel(QFrame):
            return

        msg = t("cal.export_ok", path=str(primary))
-        if local_path is not None:
-            msg += "\n" + t("cal.export_local", path=str(local_path))
        if warnings:
            msg += "\n" + t("cal.export_warnings", detail="; ".join(warnings))
        self.status_message.emit(msg)
-        # Prefer the bundle file for live apply so the next Sensor() also sees it.
-        apply_path = local_path if local_path is not None else primary
-        self.calibration_exported.emit(str(apply_path))
+        # Ask the window to confirm the target device before applying.
+        self.calibration_exported.emit(str(primary))
        QMessageBox.information(self, t("panels.cal"), msg)
@@ -235,18 +235,22 @@ class ForcePlotsPanel(QFrame):
        *,
        unit: str = "raw",
    ) -> None:
-        self._unit = unit
+        if unit != self._unit:
+            # Never mix units in one buffer. Callers switch units through
+            # ``set_units``/``set_fixed_newton`` which rebuild from the session
+            # store; a bare change here would silently reinterpret old samples.
+            self._unit = unit
        self._last_values = (float(fn), float(fx), float(fy))
        use_fixed = self._uses_fixed_y()
-        plot_fn = max(0.0, float(fn)) if use_fixed else float(fn)
-        self.history.push(timestamp, plot_fn, float(fx), float(fy))
+        # Store the true value. Clamping here (the old ``max(0.0, fn)``) erased
+        # negative FN from the buffer and hid zero drift; out-of-range values are
+        # now handled by the axis range, which only affects what is drawn.
+        self.history.push(timestamp, float(fn), float(fx), float(fy))
        self._render_value_cards()
        state = self.history.state()
        if state is None:
            return
        values = state.values.copy()
-        if use_fixed:
-            values[:, 0] = np.maximum(values[:, 0], 0.0)
        for i, curve in enumerate(self.curves):
            curve.setData(state.times, values[:, i])
            self.plots[i].setXRange(*state.x_range, padding=0.0)
@@ -258,6 +262,50 @@ class ForcePlotsPanel(QFrame):
                self.plots[i].getAxis("left").setTicks(None)
                self.plots[i].setYRange(*state.y_ranges[i], padding=0.0)

+    def set_units(self, unit: str, store=None) -> None:
+        """Switch display units without mixing units in the curve buffer.
+
+        When a session store is supplied the visible window is rebuilt from it,
+        because it keeps both the raw and the calibrated triplet for every
+        sample. Otherwise the buffer is cleared rather than reinterpreted.
+        """
+        unit = "N" if str(unit).upper().startswith("N") else "raw"
+        if unit == self._unit and store is None:
+            return
+        self._unit = unit
+        self.history.clear()
+        self._clear_curves()
+        if store is not None:
+            self._reload_from_store(store)
+        self._render_value_cards()
+        self._apply_axis_mode()
+
+    def _reload_from_store(self, store) -> None:
+        """Re-populate the visible buffer from the whole-session store."""
+        window = self.history.window_sec
+        slice_ = store.tail(window) if window > 0 else store.data()
+        if len(slice_) == 0:
+            return
+        values = slice_.value_for_unit(self._unit)
+        for index in range(len(slice_)):
+            if not bool(slice_.valid[index]):
+                continue
+            row = values[index]
+            if not all(np.isfinite(v) for v in row):
+                continue
+            self.history.push(
+                float(slice_.times[index]),
+                float(row[0]),
+                float(row[1]),
+                float(row[2]),
+            )
+        state = self.history.state()
+        if state is None:
+            return
+        for i, curve in enumerate(self.curves):
+            curve.setData(state.times, state.values[:, i])
+            self.plots[i].setXRange(*state.x_range, padding=0.0)
+
    def _uses_fixed_y(self) -> bool:
        """Fixed Y whenever we have limits for the active unit.