diff --git a/apps/studio/docs/record-channel.md b/apps/studio/docs/record-channel.md new file mode 100644 index 0000000..45541f8 --- /dev/null +++ b/apps/studio/docs/record-channel.md @@ -0,0 +1,131 @@ +# 记录通道与数据可靠性 + +本文记录 Studio 在"完整记录、数据可靠优先"这轮改动中的架构决定，以及针对架构评审 +第 8 节所列缺陷的修法与验证方式。 + +核心结论是：**标定采样、全程历史、无效样本隔离之所以会出问题，根因是同一处架构缺口 +—— 采集数据只有一个出口，就是 16 ms 的 UI 定时器。** + +--- + +## 1. 改动前的问题链 + +``` +worker 线程 ──> self._latest（单槽，后写覆盖） + │ + UI 定时器 _on_run_tick（16 ms）──┬──> 曲线历史（只保留窗口内样本） + ├──> 标定采样（且被 if self._show_cal 守卫） + └──> 箭头/深度渲染 +``` + +由此派生出评审指出的多个缺陷，它们不是彼此独立的问题： + +| 现象 | 机制 | +|---|---| +| 标定依赖面板可见 | `app_window.py` 中 `push_raw` 被 `if self._show_cal:` 守卫，而它是唯一检查录制截止时间的地方 | +| 标定录制中无法停止 | `push_raw` 永不执行 → `_finish_recording` 永不调用 → `_recording` 恒为真 → 而 `_recording` 又让录制按钮禁用。**两次点击都无法脱困** | +| 历史被窗口裁掉 | `MetricHistory._prune()` 主动丢弃 `t < t_end - window_sec` 的样本 | +| 面板未刷新时丢数据 | 单槽 `_latest` 后写覆盖，UI 降帧即丢样本 | + +--- + +## 2. 现在的结构 + +``` +worker 线程 ──> self._latest（显示，可有损） + │ + └──> _emit(packet) ──> 记录 sink（记录，不可丢） + │ + app_window._on_packet_recorded（运行在采集线程，不碰 Qt） + ├──> SessionStore 全程存储 + └──> CalibrationRecorder.feed() 标定采样缓冲 + +UI 线程 _on_run_tick ──> 曲线显示（仍可降帧） + └──> CalibrationRecorder.tick() 到点后结算（Qt 回调留在 UI 线程） +``` + +分工原则： + +- **`feed()` 在采集线程调用**，只往线程安全的缓冲里追加，绝不触碰 Qt 控件。因此 + 隐藏标定面板、曲线重绘变慢、窗口后台化都不会中断记录。 +- **`tick()` 在 UI 线程调用**，负责判定截止时间、计算统计量并触发回调 —— Qt 信号与 + 控件更新必须留在 UI 线程。 +- **sink 是普通 callable 列表**，某个 sink 抛异常会被记录并只影响该 packet，不会杀掉采集。 + +### 关键文件 + +| 文件 | 职责 | +|---|---| +| `shared/.../acquisition/sinks.py` | `RecordSinkMixin`：sink 注册与扇出，两个 worker 共用 | +| `shared/.../force/session.py` | `SessionStore`：全程存储，内存有界，双单位 + 有效性 + 帧号 | +| `shared/.../force/calibration_recorder.py` | 标定采样状态机，与 Qt 无关 | + +--- + +## 3. SessionStore 的三条保证 + +1. **不是"把窗口调大"**：按定长 numpy 块追加，O(1) 写入；达到 `max_samples` 时释放最旧 + 的整块并累加 `dropped`。`dropped` 由界面显式提示（`status.samples_dropped`）， + 不允许静默丢数据。 +2. **同时保存 raw 与标定后数值**：每个样本存 `(fn,fx,fy)_raw` 与 `(fn,fx,fy)_N` 两套。 + 切换显示单位时曲线从 store 重建，而不是把新单位的值追加到旧缓冲上。 +3. **无效样本标记而不是归零**：读取失败时存 NaN 且 `valid=False`；`iter_valid()` 只产出 + 有效样本，因此标定统计不可能把一个失败读数当作 0 平均进去。 + +--- + +## 4. 针对评审第 8 节的修法 + +| # | 问题 | 修法 | +|---|---|---| +| 1 | 标定中无法点击停止 | `_refresh_enabled` 改为 `can_record = self._running`，录制中按钮保持可用并显示「停止」 | +| 2 | 切换 raw/N 后历史混用单位 | 新增 `ForcePlotsPanel.set_units(unit, store)`：清空缓冲并从 `SessionStore` 按目标单位重建；时间原点保留，X 轴不会跳回 0 | +| 3 | 异常被当成正常数据 | `read_forces_from_sensor` 不再吞异常：返回 `valid=False` + `invalid_reason` + NaN；`SdkPacket` 新增 `valid` / `invalid_reason` / `raw_available`；无效样本不进曲线、不进标定 | +| 4 | 标定依赖面板可见 | 采样改由 worker sink 驱动；`_show_cal` 只影响可见性，不再作为采样入口 | +| 5 | 历史 FN 被截断 | 去掉写入前的 `max(0.0, fn)` 夹值与显示前的 `np.maximum`；负值原样保留，越界由坐标轴范围裁剪（只影响绘制，不影响数据） | +| 6 | 导出覆盖并自动应用标定 | 导出只写用户标定目录（`local_bundle_dir=None`），不再改写 `config/sensors//`；应用前弹窗显示目标设备类型与配置，且只下发给匹配的 worker | +| 7 | 深度显示误导 | `depth_map_to_bgr` 支持 `value_range` 固定色标与 `invalid_mask`；色标在会话内锁定到首个有效深度帧并在面板头部显示数值范围；无深度时显示明确提示，**不再回退成普通形变图** | +| 8 | 历史只保留窗口内样本 | 见 §3；`SessionStore` 与显示用的 `MetricHistory` 分离 | + +另外修正了评审未列出的一处：RAW 通道不可用时，原代码把标定后的 N 值复制进 raw 字段 +（`fn_raw = fn`）。现改为 `raw_available=False` 且 raw 保持 NaN，避免"原始值"实际上是 +标定值。 + +--- + +## 5. 验证 + +```powershell +$py = "D:\orisys_ws\orisys-app\.venv-build\Scripts\python.exe" +$env:PYTHONPATH = "D:\orisys_ws\orisys-app\shared\src;D:\orisys_ws\orisys-app\apps\studio\src" +cd D:\orisys_ws\orisys-app\apps\studio +& $py tools/ui_check.py # 含 "record channel" 段落 +``` + +- `shared/tests/test_record_channel.py`：11 项，覆盖内存上界、双单位、无效隔离、 + 跨线程 feed、截止时间结算、取消语义。 +- `shared/tests/test_worker_cancellation.py`：4 项，覆盖非阻塞 stop、线程所有权、 + 会话交接。 +- `tools/ui_check.py`：新增 `record channel` 段落，断言记录通道在**完全没有 UI 定时器 + 触发**的情况下仍收满样本、隐藏标定面板仍能采样、录制按钮是可用取消入口、 + 深度缺失不回退、色标锁帧。 +- `apps/studio/tests/test_ui_smoke.py`：上述不变量各有对应 pytest 用例，复用 + `ui_check._StuckWorker` 与 `ui_check._fake_packet`，保持单一来源。 + +### 已知环境限制 + +本机未安装 pytest，且沙箱拒绝写入工作区之外的路径，`conftest.py` 的 `tmp_path` 因此 +写 prefs 失败。上述 pytest 用例是用等价驱动脚本执行的（参数化用例除外）；`ui_check.py` +覆盖同样的不变量且可完整运行。 + +--- + +## 6. 尚未完成（第二阶段及以后） + +- MP4 分段录制、`frames.csv` 帧索引、`session.json` 元数据（评审第一阶段剩余部分）。 +- CSV 导出（`session_id/frame_id/t_rel_s/FN_raw/…/valid/calibration_id`）——数据模型 + `SessionSample` 已按这些字段设计，`SessionStore.iter_valid()` 已可直接供导出使用。 +- 全程曲线浏览（跟随/历史、框选缩放、降采样保峰值）。 +- 标定诊断扩展：稳定后采样、异常点排除/恢复、拟合残差、独立验证点。当前 + `CalibrationPoint` 已记录三轴均值/标准差/有效样本数、方向、轮次与实际采样时长。 +- PyOpenGL 三维显示（`mesh3d_panel.py` 目前仍是占位符，不是已有实现）。 diff --git a/apps/studio/src/orisys_studio/app_window.py b/apps/studio/src/orisys_studio/app_window.py index ae292e1..ab418d8 100644 --- a/apps/studio/src/orisys_studio/app_window.py +++ b/apps/studio/src/orisys_studio/app_window.py @@ -29,6 +29,8 @@ from PyQt5.QtWidgets import ( ) from orisys_app_shared.force import select_display_forces +from orisys_app_shared.force.calibration_recorder import CalibrationRecorder +from orisys_app_shared.force.session import SessionSample, SessionStore from orisys_app_shared.i18n import set_locale, t from orisys_app_shared.logging import report_error from orisys_app_shared.paths import logo_path, sensor_json_path @@ -115,6 +117,19 @@ class StudioMainWindow(QMainWindow): self._last_plot_packet_key: tuple[int, float] | None = None self._last_rendered_view_mode: str | None = None self._run_source_label = "" + # ---- record channel ------------------------------------------------- + # Fed from the acquisition thread via the worker's sink list, so it does + # not depend on the UI refresh rate or on which panels are visible. + self.session_store = SessionStore() + self.cal_recorder = CalibrationRecorder( + on_finished=self._on_calibration_point, + on_cancelled=self._on_calibration_cancelled, + ) + self._record_error_count = 0 + self._dropped_reported = 0 + self._active_calibration_id = "" + self._depth_range_value: tuple[float, float] | None = None + self._last_depth_frame = None icon = QIcon(str(logo_path())) if not icon.isNull(): @@ -169,6 +184,7 @@ class StudioMainWindow(QMainWindow): self._apply_panel_visibility() self._restore_splitter_sizes() self._update_run_enabled() + self.cal_panel.bind_recorder(self.cal_recorder) self.cal_panel.set_meta_context( sensor_type=self.sensor_picker.sensor_type(), config_ref=self._resolve_config_ref(), @@ -204,6 +220,40 @@ class StudioMainWindow(QMainWindow): on_error=self._on_worker_error, on_status=self._on_worker_status, ) + # Both workers publish into the same record channel. + for worker in (self._box_worker, self._finger_worker): + worker.add_sink(self._on_packet_recorded) + + # ------------------------------------------------------- record channel + def _on_packet_recorded(self, packet) -> None: + """Runs on the **acquisition thread**. Must stay cheap and Qt-free. + + The plots are allowed to drop frames; this path is not. It is also the + reason calibration sampling no longer depends on the calibration panel + being visible. + """ + sample = SessionSample( + frame_id=int(getattr(packet, "frame_id", 0)), + t_rel_s=float(packet.timestamp), + fn_raw=float(packet.normal_force_raw), + fx_raw=float(packet.shear_x_raw), + fy_raw=float(packet.shear_y_raw), + fn_n=float(packet.normal_force), + fx_n=float(packet.shear_x), + fy_n=float(packet.shear_y), + valid=bool(getattr(packet, "valid", True)), + calibration_id=self._calibration_id(), + error=str(getattr(packet, "invalid_reason", "") or ""), + ) + self.session_store.append(sample) + self.cal_recorder.feed(sample) + if not sample.valid: + self._record_error_count += 1 + + def _calibration_id(self) -> str: + """Cached per session — this runs for every acquired sample.""" + return self._active_calibration_id + def _wire_toolbar(self) -> None: self.toolbar.search_clicked.connect(self._on_search_cameras) @@ -511,6 +561,12 @@ class StudioMainWindow(QMainWindow): def _on_units_changed(self, units: str) -> None: self.prefs["units"] = units self.force_plots.set_fixed_newton(str(units) == "newton") + # Rebuild the visible curves from the whole-session store, which keeps + # both triplets, instead of appending new-unit values onto old ones. + self.force_plots.set_units( + "N" if str(units) == "newton" else "raw", + store=self.session_store, + ) self._persist_prefs() def _on_image_view_mode_changed(self, mode: str) -> None: @@ -567,11 +623,45 @@ class StudioMainWindow(QMainWindow): self._persist_prefs() def _on_calibration_exported(self, path: str) -> None: - """Apply newly written force_calibration.json to the running Sensor.""" + """Apply a freshly exported calibration — only on explicit confirmation. + + The export no longer overwrites the device bundle, and it no longer + silently pushes the coefficients into both workers. The user is shown + the exact target device first, and only the matching worker is updated. + """ + sensor_type = self.sensor_picker.sensor_type() + config_ref = self._resolve_config_ref() or "" + answer = QMessageBox.question( + self, + t("cal.apply_title"), + t("cal.apply_confirm", sensor_type=sensor_type, config=config_ref), + QMessageBox.Yes | QMessageBox.No, + QMessageBox.No, + ) + if answer != QMessageBox.Yes: + self.statusBar().showMessage(t("cal.apply_skipped")) + return + worker = self._worker_for(sensor_type) + if worker is None or not hasattr(worker, "set_pending_calibration"): + self.statusBar().showMessage(t("cal.apply_no_target")) + return + worker.set_pending_calibration(path) + self.statusBar().showMessage( + t("cal.apply_done", sensor_type=sensor_type) + ) + + def _worker_for(self, sensor_type: str): self._ensure_workers() - for worker in (self._box_worker, self._finger_worker): - if worker is not None and hasattr(worker, "set_pending_calibration"): - worker.set_pending_calibration(path) + if sensor_type == "finger": + return self._finger_worker + return self._box_worker + + def _on_calibration_point(self, point) -> None: + """Called on the UI thread when a calibration point finishes.""" + self.cal_panel.on_record_finished(point) + + def _on_calibration_cancelled(self) -> None: + self.cal_panel.on_record_cancelled() def _on_finger_3d_toggled(self, visible: bool) -> None: if self.sensor_picker.sensor_type() != "finger": @@ -806,10 +896,22 @@ class StudioMainWindow(QMainWindow): self._last_session_error = "" self._run_source_label = str(source) self._active_sensor_type = sensor_type + # Fresh record channel for the new session. + self._active_calibration_id = str(config_ref) + self._record_error_count = 0 + self._dropped_reported = 0 + self._depth_range_value = None + self._last_depth_frame = None + self.session_store.reset(calibration_id=str(config_ref)) + self.cal_recorder.reset_session() self._ensure_workers() self._active_worker = ( self._finger_worker if sensor_type == "finger" else self._box_worker ) + # Register the record channel on the worker we are about to start, so + # it is wired even if the worker was installed by something other than + # ``_ensure_workers`` (``add_sink`` is idempotent). + self._active_worker.add_sink(self._on_packet_recorded) try: self._active_worker.start(source, config_ref=config_ref) except Exception as exc: @@ -948,32 +1050,43 @@ class StudioMainWindow(QMainWindow): if is_new: self._last_plot_packet_key = packet_key - display = select_display_forces( - units_pref=self.toolbar.units(), - force_calibrated=packet.force_calibrated, - fn=packet.normal_force, - fx=packet.shear_x, - fy=packet.shear_y, - fn_raw=packet.normal_force_raw, - fx_raw=packet.shear_x_raw, - fy_raw=packet.shear_y_raw, - ) - if display.using_raw_fallback: - self.statusBar().showMessage(t("status.units_raw_fallback")) - self.force_plots.push( - packet.timestamp, - display.fn, - display.fx, - display.fy, - unit=display.unit, - ) - if self._show_cal: - self.cal_panel.push_raw( - packet.normal_force_raw, - packet.shear_x_raw, - packet.shear_y_raw, - packet.timestamp, + # Invalid readouts must not be drawn as if they were data, and they + # must not be written into the plot buffer: NaN would poison the + # auto-range and a zero would be indistinguishable from a real one. + if packet.valid: + display = select_display_forces( + units_pref=self.toolbar.units(), + force_calibrated=packet.force_calibrated, + fn=packet.normal_force, + fx=packet.shear_x, + fy=packet.shear_y, + fn_raw=packet.normal_force_raw, + fx_raw=packet.shear_x_raw, + fy_raw=packet.shear_y_raw, ) + if display.using_raw_fallback: + self.statusBar().showMessage(t("status.units_raw_fallback")) + self.force_plots.push( + packet.timestamp, + display.fn, + display.fx, + display.fy, + unit=display.unit, + ) + else: + self._report_invalid_sample(packet.invalid_reason) + # The calibration panel is a *view*: sampling is driven by the + # record channel, so hiding the panel can no longer stall it. + self.cal_panel.set_live( + packet.normal_force_raw, + packet.shear_x_raw, + packet.shear_y_raw, + ) + + # Finalise calibration sampling on the UI thread once its deadline passes. + self.cal_recorder.tick() + self.cal_panel.poll_recorder() + self._report_dropped_samples() mode = self.arrow_panel.view_mode() if is_new or mode != self._last_rendered_view_mode: @@ -981,6 +1094,52 @@ class StudioMainWindow(QMainWindow): if self._show_fps and is_new: self.arrow_panel.set_fps(packet.fps) + def _report_invalid_sample(self, reason: str) -> None: + """Surface failed readouts instead of letting them pass as data.""" + if self._session_phase == "running": + self._set_phase("error") + self.statusBar().showMessage( + t("status.sample_invalid", count=self._record_error_count, reason=reason) + ) + + def _report_dropped_samples(self) -> None: + """Never let the memory bound discard data silently.""" + dropped = self.session_store.dropped + if dropped > self._dropped_reported: + self._dropped_reported = dropped + self.statusBar().showMessage(t("status.samples_dropped", count=dropped)) + + # ------------------------------------------------------------ depth scale + def _depth_range(self) -> tuple[float, float] | None: + """The colour scale, latched to the first frame that carries depth. + + Re-fitting the range on every frame (the old behaviour) made colours + drift, so two frames could look identical while holding different + depths. Latching keeps frames comparable for the whole session; the + numeric range is shown in the panel header. + """ + if self._depth_range_value is not None: + return self._depth_range_value + depth = self._last_depth_frame + if depth is None: + return None + arr = np.asarray(depth, dtype=np.float64) + finite = np.isfinite(arr) + if arr.size == 0 or not np.any(finite): + return None + lo = float(np.nanmin(arr[finite])) + hi = float(np.nanmax(arr[finite])) + if hi <= lo: + hi = lo + 1.0 + self._depth_range_value = (lo, hi) + return self._depth_range_value + + def _depth_scale_text(self) -> str: + span = self._depth_range() + if span is None: + return "" + return t("panels.depth_scale", low=f"{span[0]:.3f}", high=f"{span[1]:.3f}") + def _render_run_image(self, packet) -> None: if packet is None: return @@ -996,8 +1155,20 @@ class StudioMainWindow(QMainWindow): draw_style = "finger" if self._active_sensor_type == "finger" else "box" colormap = flow_colormap(self._theme) if mode == "depth": - depth_bgr = depth_map_to_bgr(packet.depth) - frame = depth_bgr if depth_bgr is not None else packet.processed_frame + self._last_depth_frame = packet.depth + depth_bgr = depth_map_to_bgr( + packet.depth, + value_range=self._depth_range(), + invalid_mask=packet.valid_mask, + ) + if depth_bgr is None: + # Never substitute the processed frame here: that made a missing + # depth map look like a depth reading. Say so instead. + self.arrow_panel.clear_preview(t("panels.depth_unavailable")) + self.arrow_panel.set_depth_scale(None) + return + self.arrow_panel.set_depth_scale(self._depth_scale_text()) + frame = depth_bgr elif mode == "arrows_stitched": frame = draw_arrow_display( packet.processed_frame, @@ -1056,6 +1227,12 @@ class StudioMainWindow(QMainWindow): def closeEvent(self, event) -> None: # noqa: N802 self._stop_preview(announce=False) + # Detach the record channel first: a worker that is still winding down + # must not keep a reference to this window (and through it to widgets + # that Qt is about to destroy). + for worker in (self._box_worker, self._finger_worker): + if worker is not None and hasattr(worker, "remove_sink"): + worker.remove_sink(self._on_packet_recorded) # Blocking: on the way out we must not leave a device handle behind. self._stop_run(reason="user_stop", wait=True) self._persist_prefs() diff --git a/shared/src/orisys_app_shared/force/__init__.py b/shared/src/orisys_app_shared/force/__init__.py index ed73b8c..f1e8f37 100644 --- a/shared/src/orisys_app_shared/force/__init__.py +++ b/shared/src/orisys_app_shared/force/__init__.py @@ -1,4 +1,4 @@ -"""Force history and unit helpers.""" +"""Force history, session storage, calibration sampling and unit helpers.""" from orisys_app_shared.force.calibration_io import ( CHANNEL_KEYS, @@ -8,19 +8,39 @@ from orisys_app_shared.force.calibration_io import ( export_force_calibration_v2, meta_from_sensor_json, ) +from orisys_app_shared.force.calibration_recorder import ( + AxisStats, + CalibrationPoint, + CalibrationRecorder, +) from orisys_app_shared.force.history import MetricHistory, PlotState, lowpass_filter +from orisys_app_shared.force.session import ( + NA, + SessionSample, + SessionSlice, + SessionStore, + summarise, +) from orisys_app_shared.force.units import DisplayForces, select_display_forces __all__ = [ "CHANNEL_KEYS", + "NA", + "AxisStats", "CalSample", "CalSessionStore", + "CalibrationPoint", + "CalibrationRecorder", "DisplayForces", "MetricHistory", "PlotState", + "SessionSample", + "SessionSlice", + "SessionStore", "build_force_calibration_v2", "export_force_calibration_v2", "lowpass_filter", "meta_from_sensor_json", "select_display_forces", + "summarise", ] diff --git a/shared/src/orisys_app_shared/force/calibration_recorder.py b/shared/src/orisys_app_shared/force/calibration_recorder.py new file mode 100644 index 0000000..70bb2b9 --- /dev/null +++ b/shared/src/orisys_app_shared/force/calibration_recorder.py @@ -0,0 +1,284 @@ +"""Calibration sampling state machine, independent of any Qt widget. + +Two responsibilities are deliberately split across two threads: + +``feed(sample)`` + Called from the acquisition thread (via the worker's sink list) for **every** + acquired sample. It only appends to a thread-safe buffer, so recording keeps + running when the calibration panel is hidden, when the plots are redrawing + slowly, or when the UI is busy. It never touches Qt. + +``tick()`` + Called from the UI thread. It evaluates the sampling deadline and turns the + buffered samples into a finished :class:`CalibrationPoint`, invoking the + callbacks on the UI thread where it is safe to touch widgets. + +Each point stores all three axes rather than only the loaded one, so crosstalk +between axes is observable from the recorded data. The non-loaded axes carry raw +statistics but **no** reference load: their true load is unknown during a +single-axis calibration, so it is left empty rather than filled with zero. +""" + +from __future__ import annotations + +import threading +import time +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Callable, Literal, Optional + +import numpy as np + +from orisys_app_shared.force.session import NA, SessionSample, summarise + +ChannelKey = Literal["fn", "fx", "fy"] +CHANNEL_KEYS: tuple[ChannelKey, ...] = ("fn", "fx", "fy") + +STATUS_OK = "ok" +STATUS_FEW_SAMPLES = "few_samples" +STATUS_CANCELLED = "cancelled" + + +@dataclass(frozen=True) +class AxisStats: + """Mean / standard deviation / valid count for one axis.""" + + mean: float = NA + std: float = NA + n: int = 0 + + def as_dict(self) -> dict: + return {"mean": self.mean, "std": self.std, "n": self.n} + + +@dataclass +class CalibrationPoint: + """One recorded calibration point, with all three axes captured.""" + + channel: ChannelKey + reference_load_n: float + direction: str = "load" + round_index: int = 0 + note: str = "" + started_at: str = "" + ended_at: str = "" + duration_s: float = 0.0 + requested_duration_s: float = 0.0 + n_total: int = 0 + n_valid: int = 0 + raw: dict[str, AxisStats] = field(default_factory=dict) + newtons: dict[str, AxisStats] = field(default_factory=dict) + status: str = STATUS_OK + + def loaded_stats(self) -> AxisStats: + return self.raw.get(self.channel, AxisStats()) + + def as_dict(self) -> dict: + return { + "channel": self.channel, + "reference_load_n": self.reference_load_n, + # Only the loaded axis has a known reference load; the other two are + # intentionally absent rather than zero. + "direction": self.direction, + "round_index": self.round_index, + "note": self.note, + "started_at": self.started_at, + "ended_at": self.ended_at, + "duration_s": self.duration_s, + "requested_duration_s": self.requested_duration_s, + "n_total": self.n_total, + "n_valid": self.n_valid, + "status": self.status, + "raw": {key: stats.as_dict() for key, stats in self.raw.items()}, + "newtons": {key: stats.as_dict() for key, stats in self.newtons.items()}, + } + + +class CalibrationRecorder: + """Owns the calibration sampling lifecycle.""" + + def __init__( + self, + *, + on_finished: Optional[Callable[[CalibrationPoint], None]] = None, + on_cancelled: Optional[Callable[[], None]] = None, + clock: Callable[[], float] = time.perf_counter, + wall_clock: Callable[[], datetime] = lambda: datetime.now(timezone.utc), + ) -> None: + self._on_finished = on_finished + self._on_cancelled = on_cancelled + self._clock = clock + self._wall_clock = wall_clock + self._lock = threading.Lock() + self._recording = False + self._channel: ChannelKey = "fn" + self._reference_load_n = 0.0 + self._direction = "load" + self._note = "" + self._started_perf = 0.0 + self._duration_s = 0.0 + self._started_at = "" + self._samples: list[tuple[float, float, float, float, float, float, float, bool]] = [] + self._rounds: dict[ChannelKey, int] = {} + self.points: list[CalibrationPoint] = [] + + # ------------------------------------------------------------------ state + @property + def is_recording(self) -> bool: + with self._lock: + return self._recording + + @property + def channel(self) -> ChannelKey: + with self._lock: + return self._channel + + @property + def requested_duration_s(self) -> float: + with self._lock: + return self._duration_s + + def elapsed_s(self) -> float: + with self._lock: + if not self._recording: + return 0.0 + return max(0.0, self._clock() - self._started_perf) + + def progress(self) -> float: + elapsed = self.elapsed_s() + duration = self.requested_duration_s + if duration <= 0: + return 0.0 + return min(1.0, elapsed / duration) + + def note(self) -> str: + with self._lock: + return self._note + + # ------------------------------------------------------------- lifecycle + def reset_session(self) -> None: + """Clear per-session state; called when a new acquisition starts.""" + with self._lock: + self._recording = False + self._samples = [] + self._rounds = {} + self.points = [] + + def start( + self, + *, + channel: ChannelKey, + reference_load_n: float, + duration_s: float, + direction: str = "load", + note: str = "", + ) -> None: + with self._lock: + self._channel = channel if channel in CHANNEL_KEYS else "fn" + self._reference_load_n = float(reference_load_n) + self._duration_s = max(0.05, float(duration_s)) + self._direction = str(direction or "load") + self._note = str(note or "") + self._started_perf = self._clock() + self._started_at = self._wall_clock().isoformat(timespec="seconds") + self._samples = [] + self._recording = True + + def cancel(self) -> None: + with self._lock: + if not self._recording: + return + self._recording = False + self._samples = [] + if self._on_cancelled is not None: + self._on_cancelled() + + def feed(self, sample: SessionSample) -> None: + """Buffer one sample. Safe to call from the acquisition thread.""" + with self._lock: + if not self._recording: + return + self._samples.append( + ( + float(sample.t_rel_s), + float(sample.fn_raw), + float(sample.fx_raw), + float(sample.fy_raw), + float(sample.fn_n), + float(sample.fx_n), + float(sample.fy_n), + bool(sample.valid), + ) + ) + + def tick(self) -> Optional[CalibrationPoint]: + """Finish on the UI thread once the deadline has passed.""" + with self._lock: + if not self._recording: + return None + elapsed = self._clock() - self._started_perf + if elapsed < self._duration_s: + return None + return self._finish() + + def finish_now(self) -> Optional[CalibrationPoint]: + """Stop early and keep whatever has been captured so far.""" + with self._lock: + if not self._recording: + return None + return self._finish() + + # --------------------------------------------------------------- internals + def _finish(self) -> Optional[CalibrationPoint]: + with self._lock: + samples = self._samples + self._samples = [] + self._recording = False + channel = self._channel + reference = self._reference_load_n + direction = self._direction + note = self._note + started_at = self._started_at + duration = self._duration_s + ended_at = self._wall_clock().isoformat(timespec="seconds") + actual_duration = 0.0 + if samples: + actual_duration = max(0.0, samples[-1][0] - samples[0][0]) + + block = np.asarray(samples, dtype=np.float64) if samples else np.zeros((0, 8)) + valid_mask = block[:, 7] > 0.5 if block.size else np.zeros(0, dtype=bool) + valid_block = block[valid_mask] if block.size else block + + def axis_stats(column: int) -> AxisStats: + if valid_block.size == 0: + return AxisStats() + mean, std, count = summarise(valid_block[:, column]) + return AxisStats(mean=mean, std=std, n=count) + + point = CalibrationPoint( + channel=channel, + reference_load_n=reference, + direction=direction, + round_index=self._next_round(channel), + note=note, + started_at=started_at, + ended_at=ended_at, + duration_s=actual_duration, + requested_duration_s=duration, + n_total=int(block.shape[0]), + n_valid=int(valid_block.shape[0]), + raw={"fn": axis_stats(1), "fx": axis_stats(2), "fy": axis_stats(3)}, + newtons={"fn": axis_stats(4), "fx": axis_stats(5), "fy": axis_stats(6)}, + status=STATUS_OK if valid_block.shape[0] >= 2 else STATUS_FEW_SAMPLES, + ) + with self._lock: + self.points.append(point) + if self._on_finished is not None: + self._on_finished(point) + return point + + def _next_round(self, channel: ChannelKey) -> int: + with self._lock: + index = self._rounds.get(channel, 0) + self._rounds[channel] = index + 1 + return index diff --git a/shared/src/orisys_app_shared/force/history.py b/shared/src/orisys_app_shared/force/history.py index 8bda6cb..367a3ee 100644 --- a/shared/src/orisys_app_shared/force/history.py +++ b/shared/src/orisys_app_shared/force/history.py @@ -102,6 +102,16 @@ class MetricHistory: None for _ in range(self.channels) ] + def clear(self, *, keep_origin: bool = True) -> None: + """Drop every sample. + + Unlike :meth:`reset` this keeps the time origin by default, so a buffer + rebuilt after a display-unit switch does not restart the X axis at zero. + """ + origin = self._t0 if keep_origin else None + self.reset() + self._t0 = origin + def _ensure_capacity(self, needed: int) -> None: needed = max(needed, self.max_points, 2) if needed <= self._capacity: diff --git a/shared/src/orisys_app_shared/force/session.py b/shared/src/orisys_app_shared/force/session.py new file mode 100644 index 0000000..cad584c --- /dev/null +++ b/shared/src/orisys_app_shared/force/session.py @@ -0,0 +1,304 @@ +"""Full-session force sample store. + +The visible plot buffer (:class:`~orisys_app_shared.force.history.MetricHistory`) +deliberately drops every sample outside its time window, so it can never be the +source of truth for an export, a calibration point or an audit trail. + +This store keeps the whole session. It is fed from the acquisition thread via +the worker's sink list, so what gets recorded does not depend on the UI refresh +rate, on which panel happens to be visible, or on how fast the plots redraw. + +Design notes +------------ +* Samples are appended into fixed-size numpy blocks, so ``append`` is O(1) and + memory stays bounded. When ``max_samples`` is reached the oldest block is + released and ``dropped`` is incremented — callers must surface that counter + rather than let data vanish silently. +* Both the raw and the calibrated triplet are stored per sample, so switching + the display unit never has to reinterpret old data. +* Invalid samples (a failed readout) are stored with ``valid=False`` and NaN + values. They are never silently rewritten to zero, and consumers that build + calibration points or fits must filter on ``valid``. +""" + +from __future__ import annotations + +import threading +from dataclasses import dataclass +from typing import Iterable, Iterator, Optional + +import numpy as np + +#: Stands in for "this value is not available / not valid". +NA = float("nan") + +DEFAULT_BLOCK_SIZE = 8192 +#: ~4.5 h at 30 Hz of three raw + three calibrated float64 channels. +DEFAULT_MAX_SAMPLES = 500_000 + +# Column layout of the packed store. +_COL_FRAME_ID = 0 +_COL_T = 1 +_COL_FN_RAW = 2 +_COL_FX_RAW = 3 +_COL_FY_RAW = 4 +_COL_FN_N = 5 +_COL_FX_N = 6 +_COL_FY_N = 7 +_COL_VALID = 8 +_COLUMNS = 9 + + +@dataclass(frozen=True) +class SessionSample: + """One acquisition step, as handed to the record channel.""" + + frame_id: int + t_rel_s: float + fn_raw: float = NA + fx_raw: float = NA + fy_raw: float = NA + fn_n: float = NA + fx_n: float = NA + fy_n: float = NA + valid: bool = True + calibration_id: str = "" + error: str = "" + + def raw_triplet(self) -> tuple[float, float, float]: + return (self.fn_raw, self.fx_raw, self.fy_raw) + + def newton_triplet(self) -> tuple[float, float, float]: + return (self.fn_n, self.fx_n, self.fy_n) + + def has_newton(self) -> bool: + return all(np.isfinite(v) for v in self.newton_triplet()) + + def has_raw(self) -> bool: + return all(np.isfinite(v) for v in self.raw_triplet()) + + +@dataclass(frozen=True) +class SessionSlice: + """A copy of a time range, ready for plotting or analysis.""" + + times: np.ndarray + raw: np.ndarray + newtons: np.ndarray + valid: np.ndarray + frame_ids: np.ndarray + truncated: bool = False + + def __len__(self) -> int: + return int(self.times.size) + + def value_for_unit(self, unit: str) -> np.ndarray: + """Return the ``(n, 3)`` block matching a display unit.""" + if str(unit).upper().startswith("N"): + return self.newtons + return self.raw + + +class SessionStore: + """Thread-safe, memory-bounded, whole-session sample store.""" + + def __init__( + self, + *, + max_samples: int = DEFAULT_MAX_SAMPLES, + block_size: int = DEFAULT_BLOCK_SIZE, + ) -> None: + self._lock = threading.RLock() + self.max_samples = max(2, int(max_samples)) + self.block_size = max(16, int(block_size)) + self._blocks: list[np.ndarray] = [] + self._count = 0 + self._dropped = 0 + self._session_id = 0 + self._calibration_id = "" + self._t0: Optional[float] = None + self._last_t: Optional[float] = None + self._invalid = 0 + + # ------------------------------------------------------------- lifecycle + def reset(self, *, session_id: int = 0, calibration_id: str = "") -> None: + with self._lock: + self._blocks = [] + self._count = 0 + self._dropped = 0 + self._invalid = 0 + self._session_id = int(session_id) + self._calibration_id = str(calibration_id or "") + self._t0 = None + self._last_t = None + + def set_calibration_id(self, calibration_id: str) -> None: + with self._lock: + self._calibration_id = str(calibration_id or "") + + # ---------------------------------------------------------------- writing + def append(self, sample: SessionSample) -> None: + """Record one sample. Never raises on bad values — it marks them invalid.""" + with self._lock: + if self._t0 is None: + self._t0 = float(sample.t_rel_s) + t = float(sample.t_rel_s) - self._t0 + valid = 1.0 if sample.valid else 0.0 + if not sample.valid: + self._invalid += 1 + row = ( + int(sample.frame_id), + t, + float(sample.fn_raw), + float(sample.fx_raw), + float(sample.fy_raw), + float(sample.fn_n), + float(sample.fx_n), + float(sample.fy_n), + valid, + ) + self._write_row(row) + self._last_t = float(sample.t_rel_s) + + def _write_row(self, row: tuple) -> None: + if self._count % self.block_size == 0: + self._blocks.append( + np.full((self.block_size, _COLUMNS), np.nan, dtype=np.float64) + ) + block = self._blocks[-1] + offset = self._count % self.block_size + block[offset, :] = row + self._count += 1 + if self._count > self.max_samples: + self._release_oldest() + + def _release_oldest(self) -> None: + """Drop whole blocks from the front so memory stays bounded. + + The block currently being written is never released, so the most recent + ``block_size`` samples always survive. + """ + while self._count > self.max_samples and len(self._blocks) > 1: + self._blocks.pop(0) + self._count -= self.block_size + self._dropped += self.block_size + if self._count < 0: + self._count = 0 + + # ---------------------------------------------------------------- reading + @property + def count(self) -> int: + with self._lock: + return self._count + + @property + def dropped(self) -> int: + """Samples evicted by the memory bound — must be surfaced, never hidden.""" + with self._lock: + return self._dropped + + @property + def invalid(self) -> int: + with self._lock: + return self._invalid + + @property + def session_id(self) -> int: + with self._lock: + return self._session_id + + @property + def calibration_id(self) -> str: + with self._lock: + return self._calibration_id + + @property + def duration_s(self) -> float: + with self._lock: + if self._last_t is None or self._t0 is None: + return 0.0 + return float(self._last_t - self._t0) + + def _packed(self) -> np.ndarray: + if self._count == 0: + return np.empty((0, _COLUMNS), dtype=np.float64) + blocks: list[np.ndarray] = [] + remaining = self._count + for block in self._blocks: + take = min(remaining, block.shape[0]) + blocks.append(block[:take]) + remaining -= take + if remaining <= 0: + break + return np.vstack(blocks) if len(blocks) > 1 else blocks[0] + + def data(self) -> SessionSlice: + """The whole stored session (used by export).""" + with self._lock: + return self._slice_from(self._packed()) + + def tail(self, seconds: float) -> SessionSlice: + """The last ``seconds`` of the session.""" + with self._lock: + packed = self._packed() + if packed.shape[0] == 0: + return self._slice_from(packed) + t_end = float(packed[-1, _COL_T]) + mask = packed[:, _COL_T] >= (t_end - max(0.0, float(seconds))) + return self._slice_from(packed[mask]) + + def window(self, t0: float, t1: float) -> SessionSlice: + with self._lock: + packed = self._packed() + if packed.shape[0] == 0: + return self._slice_from(packed) + mask = (packed[:, _COL_T] >= float(t0)) & (packed[:, _COL_T] <= float(t1)) + return self._slice_from(packed[mask]) + + def _slice_from(self, packed: np.ndarray) -> SessionSlice: + if packed.shape[0] == 0: + empty = np.empty(0, dtype=np.float64) + return SessionSlice( + times=empty, + raw=np.empty((0, 3)), + newtons=np.empty((0, 3)), + valid=np.empty(0, dtype=bool), + frame_ids=np.empty(0, dtype=np.int64), + ) + return SessionSlice( + times=packed[:, _COL_T].copy(), + raw=packed[:, [_COL_FN_RAW, _COL_FX_RAW, _COL_FY_RAW]].copy(), + newtons=packed[:, [_COL_FN_N, _COL_FX_N, _COL_FY_N]].copy(), + valid=packed[:, _COL_VALID] > 0.5, + frame_ids=packed[:, _COL_FRAME_ID].astype(np.int64, copy=True), + truncated=self._dropped > 0, + ) + + def iter_valid(self, unit: str = "raw") -> Iterator[tuple[float, float, float, float]]: + """Yield ``(t_rel_s, v0, v1, v2)`` for valid samples only. + + Calibration fitting uses this so a failed readout can never be averaged + in as a zero. + """ + slice_ = self.data() + values = slice_.value_for_unit(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 + yield ( + float(slice_.times[index]), + float(row[0]), + float(row[1]), + float(row[2]), + ) + + +def summarise(values: Iterable[float]) -> tuple[float, float, int]: + """Return ``(mean, std, n)`` over the finite values, NaN when empty.""" + block = np.asarray([v for v in values if np.isfinite(v)], dtype=np.float64) + if block.size == 0: + return (NA, NA, 0) + return (float(block.mean()), float(block.std()), int(block.size))