diff --git a/apps/studio/config/compound_eye.png b/apps/studio/config/compound_eye.png index 7ea4e12..4759fff 100644 Binary files a/apps/studio/config/compound_eye.png and b/apps/studio/config/compound_eye.png differ diff --git a/apps/studio/config/locales/en_US.json b/apps/studio/config/locales/en_US.json index ddb44c4..2a8245f 100644 --- a/apps/studio/config/locales/en_US.json +++ b/apps/studio/config/locales/en_US.json @@ -102,7 +102,24 @@ "fn": "FN", "fx": "FX", "fy": "FY", - "no_data": "—" + "no_data": "—", + "mode_follow": "Following live", + "mode_browse": "Browsing history (display paused; capture and recording continue)", + "return_latest": "Back to latest", + "return_latest_tip": "Leave history browsing and follow the newest data again", + "overview": "Whole session", + "overview_tip": "Fit the entire session; drag the overview strip to move the window", + "box_zoom": "Box zoom", + "box_zoom_tip": "Switch to rubber-band zoom (drag a region on the chart)", + "selection": "Selection {duration}s / {samples} samples — {detail}", + "selection_empty": "No valid samples in the selection", + "browse_only_tip": "Drag or scroll on the chart to browse history" + }, + "layout": { + "default": "Free layout", + "preset_2d_name": "2D mode (depth + charts)", + "preset_2d": "2D mode: depth map on the left, force charts on the right, one shared time axis", + "preset_default": "Free layout restored" }, "record": { "label": "Record", @@ -135,10 +152,40 @@ "sn_placeholder": "Serial number (optional)", "live_raw_label": "Live raw", "live_raw": "Live raw FN={fn} FX={fx} FY={fy}", - "col_weight": "Weight (N)", + "col_weight": "Reference load", + "col_fn": "FN mean", + "col_fx": "FX mean", + "col_fy": "FY mean", + "col_stability": "Stability", + "col_state": "State", "col_raw": "Raw mean", "col_n": "n", "col_std": "σ", + "state_ok": "OK", + "state_excluded": "Excluded", + "state_validation": "Validation", + "state_exclude": "Exclude", + "state_exclude_tip": "Drop the selected point from the fit and the export (kept on record, reversible)", + "state_restore": "Restore", + "mark_validation": "Mark validation", + "mark_validation_tip": "Hold the selected point out of the fit and use it to test the fit", + "excluded": "Point excluded from the fit", + "restored": "Point restored", + "marked_validation": "Point marked as a validation point", + "unmarked_validation": "Validation mark removed", + "fit_insufficient": "A fit needs at least 2 usable points (currently {count})", + "fit_summary": "Fit {channel}: N = {gain} × raw + {offset} ({count} points, max residual {residual} N)", + "fit_validation": "Independent validation on {count} point(s), max deviation {error} N", + "point_title": "Calibration point {channel} @ {load} N", + "point_channel": "Loaded axis", + "point_direction": "Direction", + "point_round": "Round", + "point_duration": "Actual sampling time", + "point_duration_value": "{actual} s (requested {requested} s)", + "point_samples": "Samples", + "point_samples_value": "{valid} valid / {total} total", + "point_axes": "Three-axis mean ± std", + "point_residual": "Fit residual", "col_actions": "", "recording": "Recording calibration samples…", "recording_stop_tip": "Click to cancel this calibration recording (always available)", diff --git a/apps/studio/config/locales/zh_CN.json b/apps/studio/config/locales/zh_CN.json index 652f51d..7cf5bc9 100644 --- a/apps/studio/config/locales/zh_CN.json +++ b/apps/studio/config/locales/zh_CN.json @@ -102,10 +102,26 @@ "fn": "FN", "fx": "FX", "fy": "FY", - "no_data": "—" + "no_data": "—", + "mode_follow": "跟随实时", + "mode_browse": "浏览历史（显示暂停，采集与记录继续）", + "return_latest": "返回最新", + "return_latest_tip": "退出历史浏览，恢复跟随最新数据", + "overview": "全程总览", + "overview_tip": "把整个会话缩放到一屏；拖动下方总览条可移动视窗", + "box_zoom": "框选", + "box_zoom_tip": "切换为框选放大（拖动图表框选区域）", + "selection": "选区 {duration}s / {samples} 点 — {detail}", + "selection_empty": "选区无有效样本", + "browse_only_tip": "拖动或滚轮即可进入历史浏览" }, - "record": { - "label": "记录", + "layout": { + "default": "自由布局", + "preset_2d_name": "二维模式（深度图 + 曲线）", + "preset_2d": "已切换到二维模式：左侧深度图、右侧三轴曲线，共用时间轴", + "preset_default": "已恢复自由布局" + }, + "record": { "label": "记录", "mode_video_data": "视频 + 数据", "mode_data_only": "仅数据", "mode_tip": "视频记录的是处理前的相机输入图像（不含箭头与文字叠加）。MP4 是有损封装，不等于无损原始数据。", @@ -136,10 +152,40 @@ "sn_placeholder": "序列号（可选）", "live_raw_label": "实时原始值", "live_raw": "实时原始值 FN={fn} FX={fx} FY={fy}", - "col_weight": "重量 (N)", + "col_weight": "参考载荷", + "col_fn": "FN 均值", + "col_fx": "FX 均值", + "col_fy": "FY 均值", + "col_stability": "稳定性", + "col_state": "状态", "col_raw": "原始均值", "col_n": "n", "col_std": "σ", + "state_ok": "正常", + "state_excluded": "已排除", + "state_validation": "验证点", + "state_exclude": "排除", + "state_exclude_tip": "把选中点排除出拟合与导出（记录保留，可恢复）", + "state_restore": "恢复", + "mark_validation": "设为验证点", + "mark_validation_tip": "把选中点留作独立验证点：不参与拟合，用于检验拟合", + "excluded": "该点已排除出拟合", + "restored": "该点已恢复", + "marked_validation": "该点已标记为验证点", + "unmarked_validation": "该点已取消验证点标记", + "fit_insufficient": "拟合需要至少 2 个有效点（当前 {count} 个）", + "fit_summary": "拟合 {channel}: N = {gain} × raw + {offset}（{count} 点，最大残差 {residual} N）", + "fit_validation": "独立验证 {count} 点，最大偏差 {error} N", + "point_title": "标定点 {channel} @ {load} N", + "point_channel": "加载轴", + "point_direction": "方向", + "point_round": "轮次", + "point_duration": "实际采样时长", + "point_duration_value": "{actual} s（请求 {requested} s）", + "point_samples": "样本数", + "point_samples_value": "有效 {valid} / 共 {total}", + "point_axes": "三轴均值 ± 标准差", + "point_residual": "拟合残差", "col_actions": "", "recording": "正在采集标定样本…", "recording_stop_tip": "点击可取消本次标定采集（随时可用）", diff --git a/apps/studio/src/orisys_studio/app_window.py b/apps/studio/src/orisys_studio/app_window.py index 22f690b..a6b9c5d 100644 --- a/apps/studio/src/orisys_studio/app_window.py +++ b/apps/studio/src/orisys_studio/app_window.py @@ -102,6 +102,7 @@ class StudioMainWindow(QMainWindow): self._theme = str(self.prefs.get("theme") or "dark") self._language = str(self.prefs.get("language") or "zh_CN") self._show_cal = bool(self.prefs.get("show_cal_panel", False)) + self._layout_preset = str(self.prefs.get("layout_preset") or "default") self._show_finger_3d = bool(self.prefs.get("show_finger_3d", True)) self._show_fps = bool(self.prefs.get("show_fps", False)) self._session_phase = "idle" @@ -174,6 +175,8 @@ class StudioMainWindow(QMainWindow): self.mesh3d_panel = Mesh3dPanel() self.arrow_panel = ArrowPanel() self.force_plots = ForcePlotsPanel(theme=self._theme) + # The charts browse the whole session, not just their rolling buffer. + self.force_plots.set_store(self.session_store) self.cal_panel = CalPanel(theme=self._theme) self.content_splitter.addWidget(self.mesh3d_panel) self.content_splitter.addWidget(self.arrow_panel) @@ -500,6 +503,23 @@ class StudioMainWindow(QMainWindow): self.act_cal_panel.toggled.connect(self._on_cal_panel_toggled) self.view_menu.addAction(self.act_cal_panel) + self.layout_group = QActionGroup(self) + self.layout_group.setExclusive(True) + self.act_layout_default = QAction(self) + self.act_layout_default.setCheckable(True) + self.act_layout_default.setChecked(self._layout_preset != "2d") + self.act_layout_default.triggered.connect( + lambda: self._on_layout_preset("default") + ) + self.act_layout_2d = QAction(self) + self.act_layout_2d.setCheckable(True) + self.act_layout_2d.setChecked(self._layout_preset == "2d") + self.act_layout_2d.triggered.connect(lambda: self._on_layout_preset("2d")) + self.layout_group.addAction(self.act_layout_default) + self.layout_group.addAction(self.act_layout_2d) + self.view_menu.addAction(self.act_layout_default) + self.view_menu.addAction(self.act_layout_2d) + self.act_finger_3d = QAction(self) self.act_finger_3d.setCheckable(True) self.act_finger_3d.setChecked(self._show_finger_3d) @@ -586,36 +606,62 @@ class StudioMainWindow(QMainWindow): self.act_cal_panel.setChecked(self._show_cal) self.act_cal_panel.blockSignals(False) is_finger = self.sensor_picker.sensor_type() == "finger" - self.act_finger_3d.setEnabled(is_finger) + self.act_finger_3d.setEnabled(is_finger and self._layout_preset != "2d") self.act_finger_3d.blockSignals(True) self.act_finger_3d.setChecked(bool(self._show_finger_3d) if is_finger else False) self.act_finger_3d.blockSignals(False) + self.act_layout_default.blockSignals(True) + self.act_layout_default.setChecked(self._layout_preset != "2d") + self.act_layout_default.blockSignals(False) + self.act_layout_2d.blockSignals(True) + self.act_layout_2d.setChecked(self._layout_preset == "2d") + self.act_layout_2d.blockSignals(False) self._apply_arrow_rotation_for_sensor() def _apply_panel_visibility(self) -> None: is_finger = self.sensor_picker.sensor_type() == "finger" - show_3d = is_finger and bool(self._show_finger_3d) + show_3d = is_finger and bool(self._show_finger_3d) and self._layout_preset != "2d" self.mesh3d_panel.setVisible(show_3d) self.cal_panel.setVisible(self._show_cal) self.mesh3d_panel.setMinimumWidth(MIN_PANEL_WIDTHS["mesh3d"]) self.arrow_panel.setMinimumWidth(MIN_PANEL_WIDTHS["arrow"]) self.force_plots.setMinimumWidth(MIN_PANEL_WIDTHS["plots"]) self.cal_panel.setMinimumWidth(MIN_PANEL_WIDTHS["cal"]) - self.act_finger_3d.setEnabled(is_finger) + self.act_finger_3d.setEnabled(is_finger and self._layout_preset != "2d") self.act_finger_3d.blockSignals(True) self.act_finger_3d.setChecked(show_3d) self.act_finger_3d.blockSignals(False) + # The 2D preset is the reviewed layout: depth map and force charts side + # by side, sharing one time axis, with calibration folded away. + arrow_weight = 3 if self._layout_preset != "2d" else 5 + plots_weight = 4 if self._layout_preset != "2d" else 6 for i in range(self.content_splitter.count()): widget = self.content_splitter.widget(i) if widget is self.mesh3d_panel: self.content_splitter.setStretchFactor(i, 2 if show_3d else 0) elif widget is self.arrow_panel: - self.content_splitter.setStretchFactor(i, 3) + self.content_splitter.setStretchFactor(i, arrow_weight) elif widget is self.force_plots: - self.content_splitter.setStretchFactor(i, 4) + self.content_splitter.setStretchFactor(i, plots_weight) elif widget is self.cal_panel: self.content_splitter.setStretchFactor(i, 3 if self._show_cal else 0) + def _on_layout_preset(self, preset: str) -> None: + """Switch between the free layout and the reviewed 2D preset.""" + self._layout_preset = "2d" if preset == "2d" else "default" + self.prefs["layout_preset"] = self._layout_preset + if self._layout_preset == "2d": + # Depth map on the left, force charts on the right, one timeline. + self.arrow_panel.set_view_mode("depth") + self._show_cal = False + self.prefs["show_cal_panel"] = False + self._apply_panel_visibility() + self._sync_menu_checks() + self._persist_prefs() + self.statusBar().showMessage( + t("layout.preset_2d" if self._layout_preset == "2d" else "layout.preset_default") + ) + def _update_run_enabled(self) -> None: sensor_ok = self.sensor_picker.sensor_type() in RUNNABLE_TYPES source_ok = True @@ -641,6 +687,8 @@ class StudioMainWindow(QMainWindow): self.act_reveal_config.setText(t("menu.reveal_config")) self.view_menu.setTitle(t("menu.view")) self.act_cal_panel.setText(t("menu.calibration_panel")) + self.act_layout_default.setText(t("layout.default")) + self.act_layout_2d.setText(t("layout.preset_2d_name")) self.act_finger_3d.setText(t("menu.finger_3d")) self.act_show_fps.setText(t("menu.show_fps")) self.act_plot_settings.setText(t("menu.plot_settings")) @@ -1066,6 +1114,7 @@ class StudioMainWindow(QMainWindow): self._dropped_reported = 0 self._depth_range_value = None self._last_depth_frame = None + self.force_plots.set_store(self.session_store) # A new session invalidates the previous recording. self._stop_recording(wait=True) self._session_counter += 1 diff --git a/apps/studio/src/orisys_studio/ui/cal_panel.py b/apps/studio/src/orisys_studio/ui/cal_panel.py index 4ae27a3..9569f9e 100644 --- a/apps/studio/src/orisys_studio/ui/cal_panel.py +++ b/apps/studio/src/orisys_studio/ui/cal_panel.py @@ -2,8 +2,9 @@ from __future__ import annotations +import numpy as np from PyQt5.QtCore import Qt, pyqtSignal -from PyQt5.QtGui import QFont, QFontDatabase +from PyQt5.QtGui import QColor, QFont, QFontDatabase from PyQt5.QtWidgets import ( QAbstractItemView, QButtonGroup, @@ -35,8 +36,15 @@ from orisys_app_shared.i18n import t from orisys_app_shared.paths import calibration_export_dir from orisys_app_shared.theme import color +from orisys_studio.ui.cal_point_dialog import CalibrationPointDialog + CHANNEL_LABELS = ("fn", "fx", "fy") +#: Diagnostic columns, in display order: reference load, the three axis means, +#: a stability measure and the point's state. +DIAGNOSTIC_COLUMNS = ("load", "fn", "fx", "fy", "stability", "state") +DIAGNOSTIC_COMPACT_COLUMNS = (0, 4, 5) + def _mono_font() -> str: families = set(QFontDatabase().families()) @@ -143,7 +151,9 @@ class CalPanel(QFrame): self.empty_hint.setWordWrap(True) root.addWidget(self.empty_hint) - self.table = QTableWidget(0, 4) + # Diagnostic table: the loaded axis plus the two crosstalk axes, a + # stability measure and the point's state. + self.table = QTableWidget(0, len(DIAGNOSTIC_COLUMNS)) self.table.setSelectionBehavior(QAbstractItemView.SelectRows) self.table.setSelectionMode(QAbstractItemView.SingleSelection) self.table.setEditTriggers(QAbstractItemView.NoEditTriggers) @@ -151,11 +161,19 @@ class CalPanel(QFrame): self.table.setShowGrid(False) header = self.table.horizontalHeader() header.setSectionResizeMode(QHeaderView.Stretch) - header.setSectionResizeMode(2, QHeaderView.ResizeToContents) + for column in DIAGNOSTIC_COMPACT_COLUMNS: + header.setSectionResizeMode(column, QHeaderView.ResizeToContents) self.table.verticalHeader().setVisible(False) self.table.verticalHeader().setDefaultSectionSize(26) + self.table.itemSelectionChanged.connect(self._on_row_selected) + self.table.doubleClicked.connect(self._on_row_activated) root.addWidget(self.table, 1) + self.fit_label = QLabel() + self.fit_label.setObjectName("hintText") + self.fit_label.setWordWrap(True) + root.addWidget(self.fit_label) + sn_row = QHBoxLayout() self._sn_label = QLabel() self._sn_label.setObjectName("fieldLabel") @@ -170,16 +188,28 @@ class CalPanel(QFrame): self.btn_delete = QPushButton() self.btn_clear_channel = QPushButton() self.btn_clear_all = QPushButton() + self.btn_toggle_excluded = QPushButton() + self.btn_toggle_validation = QPushButton() self.btn_export = QPushButton() self.btn_export.setObjectName("primaryButton") - for button in (self.btn_delete, self.btn_clear_channel, self.btn_clear_all): + for button in ( + self.btn_delete, + self.btn_clear_channel, + self.btn_clear_all, + self.btn_toggle_excluded, + self.btn_toggle_validation, + ): # The action row must survive a narrow three-panel layout. button.setProperty("compact", "true") button.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed) + self.btn_toggle_excluded.clicked.connect(self._on_toggle_excluded) + self.btn_toggle_validation.clicked.connect(self._on_toggle_validation) self.btn_delete.clicked.connect(self._on_delete_row) self.btn_clear_channel.clicked.connect(self._on_clear_channel) self.btn_clear_all.clicked.connect(self._on_clear_all) self.btn_export.clicked.connect(self._on_export) + actions.addWidget(self.btn_toggle_excluded) + actions.addWidget(self.btn_toggle_validation) actions.addWidget(self.btn_delete) actions.addWidget(self.btn_clear_channel) actions.addWidget(self.btn_clear_all) @@ -230,20 +260,14 @@ class CalPanel(QFrame): 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._rebuild_store() self._refresh_table() + stats = point.loaded_stats() if point.status == "few_samples": self.status_message.emit(t("cal.record_failed_few")) return self.status_message.emit( - t("cal.record_ok", weight=f"{sample.weight_n:g}", raw=f"{sample.raw_mean:.3f}") + t("cal.record_ok", weight=f"{point.reference_load_n:g}", raw=f"{stats.mean:.3f}") ) def push_raw(self, fn: float, fx: float, fy: float, _timestamp: float | None = None) -> None: @@ -261,14 +285,21 @@ class CalPanel(QFrame): self.btn_clear_channel.setText(t("cal.clear_channel")) self.btn_clear_all.setText(t("cal.clear_all")) self.btn_export.setText(t("cal.export")) + self.btn_export.setText(t("cal.export")) + self.btn_toggle_excluded.setText(t("cal.state_exclude")) + self.btn_toggle_validation.setText(t("cal.mark_validation")) + self.btn_toggle_excluded.setToolTip(t("cal.state_exclude_tip")) + self.btn_toggle_validation.setToolTip(t("cal.mark_validation_tip")) self._sn_label.setText(t("cal.sn")) self.sn_edit.setPlaceholderText(t("cal.sn_placeholder")) self.empty_hint.setText(t("cal.empty_hint")) headers = ( t("cal.col_weight"), - t("cal.col_raw"), - t("cal.col_n"), - t("cal.col_std"), + t("cal.col_fn"), + t("cal.col_fx"), + t("cal.col_fy"), + t("cal.col_stability"), + t("cal.col_state"), ) for i, text in enumerate(headers): item = self.table.horizontalHeaderItem(i) @@ -280,6 +311,7 @@ class CalPanel(QFrame): self._refresh_channel_badge() self._render_live_label() self._refresh_enabled() + self._refresh_table() # ------------------------------------------------------------- internals def _refresh_channel_badge(self) -> None: @@ -393,39 +425,173 @@ class CalPanel(QFrame): self._recorder.requested_duration_s, ) + # ------------------------------------------------------------ diagnostics + def _diagnostic_points(self): + """The recorder's points for the active channel, or the export store.""" + if self._recorder is not None: + return self._recorder.points_for(self._channel) + return [] + + def _current_point(self): + row = self.table.currentRow() + points = self._diagnostic_points() + if 0 <= row < len(points): + return points[row] + return None + def _refresh_table(self) -> None: - rows = self._store.channel(self._channel) - self.table.setRowCount(len(rows)) - for r, sample in enumerate(rows): + points = self._diagnostic_points() + self.table.setRowCount(len(points)) + for row, point in enumerate(points): values = ( - f"{sample.weight_n:g}", - f"{sample.raw_mean:.4f}", - str(sample.n), - f"{sample.std:.4f}", + f"{point.reference_load_n:g}", + self._axis_cell(point, "fn"), + self._axis_cell(point, "fx"), + self._axis_cell(point, "fy"), + self._stability_cell(point), + t(point.state_key()), ) - for c, text in enumerate(values): + for column, text in enumerate(values): item = QTableWidgetItem(text) item.setFlags(item.flags() & ~Qt.ItemIsEditable) - if c != 2: + if column in (0, 1, 2, 3): item.setTextAlignment(int(Qt.AlignRight | Qt.AlignVCenter)) else: item.setTextAlignment(int(Qt.AlignCenter)) - self.table.setItem(r, c, item) - self.empty_hint.setVisible(not rows) + if point.excluded: + item.setForeground(QColor(color(self._theme, "text_muted"))) + elif point.is_validation: + item.setForeground(QColor(color(self._theme, "info"))) + self.table.setItem(row, column, item) + self.empty_hint.setVisible(not points) + self._refresh_fit_label() + + @staticmethod + def _axis_cell(point, channel: str) -> str: + value = point.axis_mean(channel) + if not np.isfinite(value): + return "—" + if channel == point.channel and np.isfinite(point.residual_n): + # Residual of the fit on the same cell as the value it came from. + return f"{value:.4f} ({point.residual_n:+.4f})" + return f"{value:.4f}" + + @staticmethod + def _stability_cell(point) -> str: + stats = point.loaded_stats() + if not np.isfinite(stats.std): + return "—" + return f"σ {stats.std:.4f} / n {stats.n}" + + def _refresh_fit_label(self) -> None: + if self._recorder is None: + self.fit_label.setText("") + return + fit = self._recorder.fit(self._channel) + if not fit.usable: + self.fit_label.setText(t("cal.fit_insufficient", count=fit.n_points)) + return + text = t( + "cal.fit_summary", + channel=self._channel.upper(), + gain=f"{fit.gain:g}", + offset=f"{fit.offset:g}", + count=fit.n_points, + residual=f"{fit.max_residual_n:.4f}", + ) + if fit.validation_points: + text += " " + t( + "cal.fit_validation", + error=f"{fit.validation_error_n:.4f}", + count=fit.validation_points, + ) + self.fit_label.setText(text) + + def _on_row_selected(self) -> None: + point = self._current_point() + self.btn_toggle_excluded.setEnabled(point is not None) + self.btn_toggle_validation.setEnabled(point is not None) + if point is not None: + state = t("cal.state_restore") if point.excluded else t("cal.state_exclude") + self.btn_toggle_excluded.setText(state) + + def _on_row_activated(self) -> None: + """Double click a row to inspect that point's three-axis curve.""" + point = self._current_point() + if point is None or not point.trace: + return + CalibrationPointDialog(point, theme=self._theme, parent=self).exec_() + + def _on_toggle_excluded(self) -> None: + row = self.table.currentRow() + point = self._current_point() + if point is None: + return + self._recorder.set_excluded(row, not point.excluded) + self._rebuild_store() + self._refresh_table() + self.status_message.emit( + t("cal.excluded" if point.excluded else "cal.restored") + ) + + def _on_toggle_validation(self) -> None: + row = self.table.currentRow() + point = self._current_point() + if point is None: + return + self._recorder.set_validation(row, not point.is_validation) + self._rebuild_store() + self._refresh_table() + self.status_message.emit( + t("cal.marked_validation" if point.is_validation else "cal.unmarked_validation") + ) + + def _rebuild_store(self) -> None: + """Rebuild the export store from the points that are still in the fit.""" + self._store.clear() + if self._recorder is None: + return + for point in self._recorder.points: + if point.excluded: + continue + stats = point.loaded_stats() + if not np.isfinite(stats.mean): + continue + self._store.upsert( + point.channel, + CalSample( + weight_n=float(point.reference_load_n), + raw_mean=float(stats.mean), + n=int(stats.n), + std=float(stats.std), + ), + ) def _on_delete_row(self) -> None: row = self.table.currentRow() - if row < 0: + point = self._current_point() + if point is None or self._recorder is None: return - self._store.remove_at(self._channel, row) + # Deleting is an exclusion: the recorded evidence stays, but it leaves + # the fit and the export. + self._recorder.set_excluded(row, True) + self._rebuild_store() self._refresh_table() def _on_clear_channel(self) -> None: self._store.clear(self._channel) + if self._recorder is not None: + with self._recorder._lock: # noqa: SLF001 - panel owns the view + self._recorder.points = [ + p for p in self._recorder.points if p.channel != self._channel + ] self._refresh_table() def _on_clear_all(self) -> None: self._store.clear() + if self._recorder is not None: + with self._recorder._lock: # noqa: SLF001 + self._recorder.points = [] self._refresh_table() def _on_export(self) -> None: diff --git a/apps/studio/src/orisys_studio/ui/cal_point_dialog.py b/apps/studio/src/orisys_studio/ui/cal_point_dialog.py new file mode 100644 index 0000000..a73b2c6 --- /dev/null +++ b/apps/studio/src/orisys_studio/ui/cal_point_dialog.py @@ -0,0 +1,147 @@ +"""Inspect one calibration point: its three-axis trace and its diagnostics. + +Kept in a dialog rather than inline so the calibration panel can stay narrow in +the three-panel layout, while a single point can still be examined in full. +""" + +from __future__ import annotations + +import numpy as np +from PyQt5.QtCore import Qt +from PyQt5.QtGui import QFont, QFontDatabase +from PyQt5.QtWidgets import ( + QDialog, + QDialogButtonBox, + QFormLayout, + QLabel, + QVBoxLayout, +) + +from orisys_app_shared.force import NA +from orisys_app_shared.i18n import t +from orisys_app_shared.theme import color + +try: + import pyqtgraph as pg +except ImportError as exc: # pragma: no cover + raise ImportError("pyqtgraph is required for the calibration point dialog") from exc + +TRACE_TOKENS = ("plot_fn", "plot_fx", "plot_fy") +TRACE_KEYS = ("fn", "fx", "fy") + + +def _mono_font(point_size: int = 9) -> QFont: + families = set(QFontDatabase().families()) + chosen = "monospace" + for name in ("Consolas", "Cascadia Mono", "DejaVu Sans Mono", "Courier New"): + if name in families: + chosen = name + break + return QFont(chosen, point_size) + + +class CalibrationPointDialog(QDialog): + """One recorded point: all three raw traces plus its fit residual.""" + + def __init__(self, point, *, theme: str = "dark", parent=None) -> None: + super().__init__(parent) + self._point = point + self._theme = theme + self.setWindowTitle( + t( + "cal.point_title", + channel=point.channel.upper(), + load=f"{point.reference_load_n:g}", + ) + ) + self.resize(560, 420) + + root = QVBoxLayout(self) + root.setSpacing(8) + + form = QFormLayout() + form.setHorizontalSpacing(10) + form.setVerticalSpacing(4) + self._add_row(form, "cal.point_channel", point.channel.upper()) + self._add_row(form, "cal.point_direction", point.direction) + self._add_row(form, "cal.point_round", str(point.round_index)) + self._add_row( + form, + "cal.point_duration", + t( + "cal.point_duration_value", + actual=f"{point.duration_s:.2f}", + requested=f"{point.requested_duration_s:.2f}", + ), + ) + self._add_row( + form, + "cal.point_samples", + t("cal.point_samples_value", valid=point.n_valid, total=point.n_total), + ) + stats = [] + for key in TRACE_KEYS: + axis = point.raw.get(key) + if axis is None or not np.isfinite(axis.mean): + stats.append(f"{key.upper()} —") + continue + stats.append(f"{key.upper()} {axis.mean:.4f} ± {axis.std:.4f}") + value = QLabel(" ".join(stats)) + value.setFont(_mono_font()) + self._add_row(form, "cal.point_axes", "") + form.addRow(value) + residual = point.residual_n + self._add_row( + form, + "cal.point_residual", + "—" if not np.isfinite(residual) else f"{residual:+.4f}", + ) + root.addLayout(form) + + self.plot_widget = pg.GraphicsLayoutWidget() + self.plot_widget.setBackground(color(theme, "plot_bg")) + root.addWidget(self.plot_widget, 1) + self._build_traces() + + buttons = QDialogButtonBox(QDialogButtonBox.Close) + buttons.rejected.connect(self.reject) + buttons.accepted.connect(self.accept) + root.addWidget(buttons) + + def _add_row(self, form: QFormLayout, key: str, value: str) -> None: + label = QLabel(t(key)) + label.setObjectName("fieldLabel") + field = QLabel(str(value)) + form.addRow(label, field) + + def _build_traces(self) -> None: + """One row per axis, stacked, sharing the sample axis.""" + pen_axis = pg.mkPen(color(self._theme, "text_secondary")) + first = None + for row, key in enumerate(TRACE_KEYS): + plot = self.plot_widget.addPlot(row=row, col=0) + plot.showGrid(x=True, y=True, alpha=0.25) + plot.setMouseEnabled(x=True, y=False) + plot.getAxis("left").setPen(pen_axis) + plot.getAxis("left").setTextPen(pen_axis) + for axis_name in ("left", "bottom"): + plot.getAxis(axis_name).setPen(pen_axis) + plot.getAxis(axis_name).setTextPen(pen_axis) + plot.setLabel( + "left", key.upper(), color=color(self._theme, TRACE_TOKENS[row]) + ) + if first is None: + first = plot + else: + plot.setXLink(first) + self.plot_widget.ci.layout.setRowStretchFactor(row, 1) + values = self._point.trace.get(key) + if not values: + continue + step = self._point.trace_dt_s or 0.01 + times = np.arange(len(values), dtype=float) * step + plot.plot( + times, + np.asarray(values, dtype=float), + pen=pg.mkPen(color(self._theme, TRACE_TOKENS[row]), width=2), + ) diff --git a/apps/studio/src/orisys_studio/ui/force_plots.py b/apps/studio/src/orisys_studio/ui/force_plots.py index 06c2baa..4adbe2a 100644 --- a/apps/studio/src/orisys_studio/ui/force_plots.py +++ b/apps/studio/src/orisys_studio/ui/force_plots.py @@ -2,10 +2,18 @@ from __future__ import annotations +import time + import numpy as np -from PyQt5.QtCore import Qt +from PyQt5.QtCore import Qt, pyqtSignal from PyQt5.QtGui import QColor, QFont, QFontDatabase -from PyQt5.QtWidgets import QFrame, QHBoxLayout, QLabel, QVBoxLayout +from PyQt5.QtWidgets import ( + QFrame, + QHBoxLayout, + QLabel, + QPushButton, + QVBoxLayout, +) from orisys_app_shared.acquisition.constants import FORCE_SAMPLE_RATE_HZ, LOWPASS_CUTOFF_HZ from orisys_app_shared.force import MetricHistory @@ -23,6 +31,8 @@ CHANNEL_KEYS = ("fn", "fx", "fy") CHANNEL_LABELS = ("FN", "FX", "FY") # Theme token that colors each channel curve / label, per index. CHANNEL_TOKENS = ("plot_fn", "plot_fx", "plot_fy") +#: How often the whole-session overview strip is rebuilt while following live. +OVERVIEW_REFRESH_S = 1.0 def _nice_ticks(lo: float, hi: float, count: int = 5) -> list[tuple[float, str]]: @@ -56,7 +66,29 @@ def _mono_font(point_size: int = 9, *, bold: bool = False) -> QFont: return font +class _Sync: + """Marks a block of range changes as programmatic (not a user gesture).""" + + def __init__(self, panel: "ForcePlotsPanel") -> None: + self._panel = panel + self._previous = panel._syncing + + def __enter__(self) -> "_Sync": + self._panel._syncing = True + return self + + def __exit__(self, *_exc) -> None: + self._panel._syncing = self._previous + + class ForcePlotsPanel(QFrame): + #: The visible time window changed while browsing history. + window_changed = pyqtSignal(float, float) + #: The shared cursor moved; other views sync their playback position to it. + cursor_moved = pyqtSignal(float) + #: "follow" (live) or "browse" (user-controlled window). + mode_changed = pyqtSignal(str) + def __init__(self, parent=None, *, theme: str = "dark") -> None: super().__init__(parent) self.setObjectName("contentPanel") @@ -68,6 +100,15 @@ class ForcePlotsPanel(QFrame): self._fixed_newton = True self._unit = "raw" self._last_values: tuple[float, float, float] = (0.0, 0.0, 0.0) + # -- browsing state ------------------------------------------------- + self._store = None + self._mode = "follow" + self._window: tuple[float, float] = (0.0, 0.0) + self._cursor_t = 0.0 + self._syncing = False + self._overview_dirty = True + self._last_overview_at = 0.0 + self._max_draw_points = 2000 self.history = MetricHistory( channels=3, window_sec=4.0, @@ -86,7 +127,23 @@ class ForcePlotsPanel(QFrame): self.title = QLabel() self.title.setObjectName("panelTitle") header.addWidget(self.title) + self.mode_label = QLabel() + self.mode_label.setObjectName("hintText") + header.addWidget(self.mode_label) header.addStretch(1) + self.btn_latest = QPushButton() + self.btn_latest.setProperty("compact", "true") + self.btn_latest.clicked.connect(self.follow_latest) + header.addWidget(self.btn_latest) + self.btn_overview = QPushButton() + self.btn_overview.setProperty("compact", "true") + self.btn_overview.clicked.connect(self.show_all) + header.addWidget(self.btn_overview) + self.btn_box_zoom = QPushButton() + self.btn_box_zoom.setProperty("compact", "true") + self.btn_box_zoom.setCheckable(True) + self.btn_box_zoom.toggled.connect(self._on_box_zoom_toggled) + header.addWidget(self.btn_box_zoom) self.unit_badge = QLabel() self.unit_badge.setObjectName("unitBadge") header.addWidget(self.unit_badge) @@ -103,6 +160,7 @@ class ForcePlotsPanel(QFrame): self.plots = [] self.curves = [] + self.cursors = [] pen = pg.mkPen(color(theme, CHANNEL_TOKENS[0]), width=2) for i, _name in enumerate(CHANNEL_LABELS): plot = self.plot_widget.addPlot(row=i, col=0) @@ -111,8 +169,40 @@ class ForcePlotsPanel(QFrame): if i > 0: plot.setXLink(self.plots[0]) curve = plot.plot(pen=pen) + # One shared time cursor across all three channels. + cursor = pg.InfiniteLine( + pos=0.0, angle=90, movable=False, pen=pg.mkPen("#888888", width=1) + ) + plot.addItem(cursor, ignoreBounds=True) + plot.getViewBox().sigXRangeChanged.connect(self._on_x_range_changed) self.plots.append(plot) self.curves.append(curve) + self.cursors.append(cursor) + + # Whole-session overview: the full envelope plus the current window. + self.overview = self.plot_widget.addPlot(row=len(CHANNEL_LABELS), col=0) + self.overview.setMouseEnabled(x=False, y=False) + self.overview.showGrid(x=True, y=False, alpha=0.2) + self.overview.setMaximumHeight(46) + self.overview.getAxis("left").setStyle(showValues=False) + self.overview_curve = self.overview.plot(pen=pg.mkPen("#8A93A6", width=1)) + self.overview_region = pg.LinearRegionItem( + values=(0.0, 1.0), movable=False, brush=(90, 110, 190, 60) + ) + self.overview_region.setZValue(-10) + self.overview.addItem(self.overview_region) + self.overview_cursor = pg.InfiniteLine( + pos=0.0, angle=90, movable=False, pen=pg.mkPen("#888888", width=1) + ) + self.overview.addItem(self.overview_cursor, ignoreBounds=True) + for row in range(len(CHANNEL_LABELS)): + self.plot_widget.ci.layout.setRowStretchFactor(row, 3) + self.plot_widget.ci.layout.setRowStretchFactor(len(CHANNEL_LABELS), 1) + + self.stats_label = QLabel() + self.stats_label.setObjectName("panelMeta") + self.stats_label.setTextInteractionFlags(Qt.TextSelectableByMouse) + root.addWidget(self.stats_label) values_row = QHBoxLayout() values_row.setSpacing(6) @@ -130,6 +220,7 @@ class ForcePlotsPanel(QFrame): root.addLayout(values_row) self.set_theme(theme) + self._interactive(False) self._apply_axis_mode() self.retranslate() @@ -223,8 +314,17 @@ class ForcePlotsPanel(QFrame): self._clear_curves() self._last_values = (0.0, 0.0, 0.0) self._unit = "raw" + self._mode = "follow" + self._window = (0.0, 0.0) + self._cursor_t = 0.0 + self._overview_dirty = True + self._interactive(False) + self.set_cursor(0.0) + self.overview_curve.setData([], []) + self.stats_label.setText("") self._render_value_cards() self._apply_axis_mode() + self._refresh_mode_label() def push( self, @@ -241,26 +341,51 @@ class ForcePlotsPanel(QFrame): # store; a bare change here would silently reinterpret old samples. self._unit = unit self._last_values = (float(fn), float(fx), float(fy)) + self._render_value_cards() + if self._mode == "browse": + # Browsing is a display pause, not a capture pause: the record + # channel keeps running and the view simply stops following. + self._overview_dirty = True + return use_fixed = self._uses_fixed_y() # 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() - for i, curve in enumerate(self.curves): - curve.setData(state.times, values[:, i]) - self.plots[i].setXRange(*state.x_range, padding=0.0) - if use_fixed: - ymin, ymax = self._fixed_y_range(i) - self.plots[i].setYRange(ymin, ymax, padding=0.0) - self.plots[i].getAxis("left").setTicks([_nice_ticks(ymin, ymax, count=5)]) - else: - self.plots[i].getAxis("left").setTicks(None) - self.plots[i].setYRange(*state.y_ranges[i], padding=0.0) + self._syncing = True + try: + values = state.values.copy() + for i, curve in enumerate(self.curves): + curve.setData(state.times, values[:, i]) + self.plots[i].setXRange(*state.x_range, padding=0.0) + if use_fixed: + ymin, ymax = self._fixed_y_range(i) + self.plots[i].setYRange(ymin, ymax, padding=0.0) + self.plots[i].getAxis("left").setTicks( + [_nice_ticks(ymin, ymax, count=5)] + ) + else: + self.plots[i].getAxis("left").setTicks(None) + self.plots[i].setYRange(*state.y_ranges[i], padding=0.0) + finally: + self._syncing = False + self.set_cursor(float(timestamp) - (self._time_origin() or 0.0)) + # The overview is a full-store read, so refresh it on a timer rather + # than on every frame. + now = time.perf_counter() + if self._overview_dirty or (now - self._last_overview_at) >= OVERVIEW_REFRESH_S: + self._overview_dirty = False + self._last_overview_at = now + self._sync_overview() + + def _time_origin(self) -> float: + """The store's time origin, so the cursor matches the overview axis.""" + if self._store is None: + return 0.0 + return self._store.origin_time() def set_units(self, unit: str, store=None) -> None: """Switch display units without mixing units in the curve buffer. @@ -275,11 +400,216 @@ class ForcePlotsPanel(QFrame): self._unit = unit self.history.clear() self._clear_curves() + self._overview_dirty = True if store is not None: - self._reload_from_store(store) + self._store = store + if self._mode == "browse": + self.refresh_from_store() + else: + self._reload_from_store(store) self._render_value_cards() self._apply_axis_mode() + # ---------------------------------------------------------------- browsing + def set_store(self, store) -> None: + """Attach the whole-session store the browser reads from.""" + self._store = store + self._overview_dirty = True + self._refresh_mode_label() + + def mode(self) -> str: + return self._mode + + def cursor_time(self) -> float: + return self._cursor_t + + def window(self) -> tuple[float, float]: + if self._mode == "browse": + return self._window + if self._store is not None: + return self._store.extent() + return (0.0, 0.0) + + def follow_latest(self) -> None: + """Leave history browsing and resume the live rolling window.""" + self._set_mode("follow") + self._apply_axis_mode() + + def show_all(self) -> None: + """Overview: fit the entire session, in browse mode.""" + if self._store is None: + return + first, last = self._store.extent() + if last <= first: + return + self._set_mode("browse") + self.set_window(first, last) + + def set_window(self, t0: float, t1: float) -> None: + """Show ``[t0, t1]`` from the whole-session store.""" + if self._store is None: + return + first, last = self._store.extent() + if t1 <= t0: + t1 = t0 + max(0.001, self.history.window_sec) + t0 = max(first, float(t0)) + t1 = min(max(last, first + 0.001), float(t1)) + self._set_mode("browse") + self._window = (t0, t1) + self.refresh_from_store() + + def refresh_from_store(self) -> None: + """Redraw the visible window. Called on pan, zoom and data changes.""" + if self._store is None: + return + t0, t1 = self._window + slice_ = self._store.visible(t0, t1, max_points=self._max_draw_points) + self._syncing = True + try: + if len(slice_) == 0: + for curve in self.curves: + curve.setData([], []) + else: + values = slice_.value_for_unit(self._unit) + for index, curve in enumerate(self.curves): + column = values[:, index] + finite = np.isfinite(column) + curve.setData(slice_.times[finite], column[finite]) + for plot in self.plots: + plot.setXRange(t0, t1, padding=0.0) + self._apply_axis_mode() + finally: + self._syncing = False + self._update_selection_readout() + self._sync_overview() + self.window_changed.emit(t0, t1) + + def _set_mode(self, mode: str) -> None: + mode = "browse" if str(mode) == "browse" else "follow" + if mode == self._mode: + return + self._mode = mode + self._interactive(mode == "browse") + self._refresh_mode_label() + self._sync_overview() + self.mode_changed.emit(mode) + + def _interactive(self, enabled: bool) -> None: + """Pan/zoom is only meaningful over a stored window.""" + for plot in self.plots: + plot.setMouseEnabled(x=bool(enabled), y=False) + if not enabled: + plot.getViewBox().setMouseMode(pg.ViewBox.PanMode) + + def _on_box_zoom_toggled(self, checked: bool) -> None: + mode = pg.ViewBox.RectMode if checked else pg.ViewBox.PanMode + if not checked or self._mode != "browse": + self._set_mode("browse" if checked else self._mode) + for plot in self.plots: + plot.getViewBox().setMouseMode(mode) + + def _on_x_range_changed(self, _viewbox, x_range) -> None: + """User scrolled or dragged: leave live mode and read from the store. + + Programmatic range changes are ignored via ``self._syncing`` — without + that, redrawing a window would feed back in as if the user had dragged, + and "back to latest" could never take effect. + """ + if self._syncing or self._store is None: + return + x0, x1 = float(x_range[0]), float(x_range[1]) + if not (x1 > x0): + return + first, last = self._store.extent() + if last <= first: + return + # Keep the window inside the recorded session. + x0 = max(first, x0) + x1 = min(last, x1) + if not (x1 > x0): + return + if self._mode != "browse": + self._set_mode("browse") + self._window = (x0, x1) + self.refresh_from_store() + + def _guard(self) -> "_Sync": + """Context manager marking range changes as programmatic.""" + return _Sync(self) + + def set_cursor(self, t: float, *, emit: bool = False) -> None: + """Move the shared time cursor without redrawing the data.""" + self._cursor_t = float(t) + for cursor in self.cursors: + cursor.setValue(self._cursor_t) + self.overview_cursor.setValue(self._cursor_t) + if emit: + self.cursor_moved.emit(self._cursor_t) + + def _update_selection_readout(self) -> None: + if self._store is None: + self.stats_label.setText("") + return + t0, t1 = self._window + stats = self._store.statistics(t0, t1, unit=self._unit) + if stats.samples == 0: + self.stats_label.setText(t("chart.selection_empty")) + return + unit = unit_display(self._unit) + parts = [] + for index, label in enumerate(CHANNEL_LABELS): + peak = stats.peaks[index] + mean = stats.means[index] + std = stats.stds[index] + if not np.isfinite(peak): + continue + parts.append( + f"{label} 峰 {peak:.3g}{unit} · 均 {mean:.3g} · σ {std:.3g}" + ) + self.stats_label.setText( + t( + "chart.selection", + duration=f"{stats.duration_s:.2f}", + samples=stats.samples, + detail=" ".join(parts), + ) + ) + + def _sync_overview(self) -> None: + if self._store is None: + return + first, last = self._store.extent() + if last <= first: + self.overview_curve.setData([], []) + return + # The overview is a full-store read, so only rebuild it when it can have + # changed (window/unit/session) rather than on every pan. + slice_ = self._store.visible(first, last, max_points=self._overview_points()) + values = slice_.value_for_unit(self._unit) + column = values[:, 0] + finite = np.isfinite(column) + self.overview_curve.setData(slice_.times[finite], column[finite]) + self.overview.setXRange(first, last, padding=0.0) + if self._mode == "browse": + self.overview_region.setRegion(self._window) + self.overview_region.show() + else: + self.overview_region.hide() + self.overview.setYRange( + float(np.nanmin(column[finite])) if finite.any() else 0.0, + float(np.nanmax(column[finite])) if finite.any() else 1.0, + padding=0.05, + ) + + def _overview_points(self) -> int: + return 1200 + + def _refresh_mode_label(self) -> None: + key = "chart.mode_browse" if self._mode == "browse" else "chart.mode_follow" + self.mode_label.setText(t(key)) + self.btn_latest.setEnabled(self._mode == "browse") + self.btn_overview.setEnabled(self._store is not None) + def _reload_from_store(self, store) -> None: """Re-populate the visible buffer from the whole-session store.""" window = self.history.window_sec @@ -337,15 +667,17 @@ class ForcePlotsPanel(QFrame): def _apply_axis_mode(self) -> None: use_fixed = self._preview_fixed_y() - for i, plot in enumerate(self.plots): - if use_fixed: - ymin, ymax = self._fixed_y_range(i) - plot.setYRange(ymin, ymax, padding=0.0) - plot.getAxis("left").setTicks([_nice_ticks(ymin, ymax, count=5)]) - else: - plot.getAxis("left").setTicks(None) - for plot in self.plots: - plot.setXRange(0.0, self.history.window_sec, padding=0.0) + with self._guard(): + for i, plot in enumerate(self.plots): + if use_fixed: + ymin, ymax = self._fixed_y_range(i) + plot.setYRange(ymin, ymax, padding=0.0) + plot.getAxis("left").setTicks([_nice_ticks(ymin, ymax, count=5)]) + else: + plot.getAxis("left").setTicks(None) + if self._mode == "follow": + for plot in self.plots: + plot.setXRange(0.0, self.history.window_sec, padding=0.0) def _clear_curves(self) -> None: for curve in self.curves: @@ -364,4 +696,11 @@ class ForcePlotsPanel(QFrame): self.title.setText(t("panels.plots")) for badge, label in zip(self.channel_badges, CHANNEL_LABELS): badge.setText(label) + self.btn_latest.setText(t("chart.return_latest")) + self.btn_latest.setToolTip(t("chart.return_latest_tip")) + self.btn_overview.setText(t("chart.overview")) + self.btn_overview.setToolTip(t("chart.overview_tip")) + self.btn_box_zoom.setText(t("chart.box_zoom")) + self.btn_box_zoom.setToolTip(t("chart.box_zoom_tip")) + self._refresh_mode_label() self._render_value_cards() diff --git a/apps/studio/tests/test_ui_smoke.py b/apps/studio/tests/test_ui_smoke.py index 7329e16..f5aee07 100644 --- a/apps/studio/tests/test_ui_smoke.py +++ b/apps/studio/tests/test_ui_smoke.py @@ -1,9 +1,11 @@ -"""Offscreen UI smoke tests: construction, layout adaptation, theme tokens.""" +"""Offscreen UI smoke tests: construction, layout adaptation, theme tokens.""" from __future__ import annotations from pathlib import Path +import numpy as np + import pytest import ui_check @@ -199,7 +201,7 @@ def test_toolbar_run_button_is_labelled(make_window): window = _show(make_window(), (1400, 880)) label = window.toolbar.btn_run.text() assert label.strip() - assert label != "â–¶" + assert label != "â–¶" window.toolbar.set_run_checked(True) assert window.toolbar.btn_run.text() != label @@ -264,14 +266,92 @@ def test_cal_panel_records_and_lists_samples(make_window): cal.set_running(True) assert cal.btn_record.isEnabled() assert cal.btn_record.text() + assert cal.table.columnCount() == 6, "the diagnostic columns changed" - from orisys_app_shared.force.calibration_io import CalSample - - cal._store.upsert("fn", CalSample(weight_n=1.0, raw_mean=9.7, n=120, std=0.02)) - cal._refresh_table() - assert cal.table.rowCount() == 1 + _add_calibration_point(window, load=1.0, raw=100.2) + _add_calibration_point(window, load=2.0, raw=400.0) # off-trend + _add_calibration_point(window, load=3.0, raw=300.6) + assert cal.table.rowCount() == 3 assert not cal.empty_hint.isVisible() + headers = [ + cal.table.horizontalHeaderItem(i).text() for i in range(cal.table.columnCount()) + ] + from orisys_app_shared.i18n import t + + assert headers == [ + t("cal.col_weight"), + t("cal.col_fn"), + t("cal.col_fx"), + t("cal.col_fy"), + t("cal.col_stability"), + t("cal.col_state"), + ] + + +def _add_calibration_point(window, *, load: float, raw: float, samples: int = 40) -> None: + """Drive one point through the recorder and into the panel's table.""" + from orisys_app_shared.force.session import SessionSample + + window.cal_recorder.start(channel="fn", reference_load_n=load, duration_s=1.0) + for index in range(samples): + window.cal_recorder.feed( + SessionSample( + frame_id=index, + t_rel_s=index / 100.0, + fn_raw=raw, + fx_raw=0.4, + fy_raw=-0.2, + valid=True, + ) + ) + window.cal_panel.on_record_finished(window.cal_recorder.finish_now()) + + +def test_calibration_diagnostics_drive_the_fit(make_window): + window = _show(make_window(), (1400, 880)) + window.cal_panel.set_running(True) + _add_calibration_point(window, load=1.0, raw=100.2) + _add_calibration_point(window, load=2.0, raw=400.0) + _add_calibration_point(window, load=3.0, raw=300.6) + _add_calibration_point(window, load=5.0, raw=500.9) + + recorder = window.cal_recorder + fit = recorder.fit("fn") + assert fit.usable and fit.n_points == 4 + worst = fit.max_residual_n + assert worst > 0.05, "the off-trend point produced no residual" + assert window.cal_panel.fit_label.text() + + window.cal_panel.table.setCurrentCell(1, 0) + window.cal_panel._on_toggle_excluded() + assert recorder.fit("fn").max_residual_n < worst, "exclusion did not tighten the fit" + assert recorder.points[1].excluded + # An excluded point leaves the export store but stays on record. + assert len(window.cal_panel._store.channel("fn")) == 3 + + window.cal_panel.table.setCurrentCell(2, 0) + window.cal_panel._on_toggle_validation() + held_out = recorder.fit("fn") + assert held_out.validation_points == 1 + assert recorder.points[2].is_validation + assert np.isnan(recorder.points[2].residual_n), "validation point entered the fit" + + +def test_calibration_point_dialog_shows_three_axes(make_window): + window = _show(make_window(), (1400, 880)) + _add_calibration_point(window, load=1.0, raw=100.2) + point = window.cal_recorder.points[0] + assert set(point.trace) == {"fn", "fx", "fy"} + assert point.trace_dt_s > 0 + + from orisys_studio.ui.cal_point_dialog import CalibrationPointDialog + + dialog = CalibrationPointDialog(point, theme="dark") + assert dialog.windowTitle() + assert len(dialog.plot_widget.ci.items) == 3, "not one trace per axis" + dialog.close() + def test_force_plots_render_demo_data(make_window): window = _show(make_window(), (1440, 880)) @@ -516,3 +596,4 @@ def test_recording_leaves_unavailable_values_empty(make_window, record_root): _finish(window, stuck) + diff --git a/apps/studio/tools/ui_check.py b/apps/studio/tools/ui_check.py index ad7ea91..80f4b33 100644 --- a/apps/studio/tools/ui_check.py +++ b/apps/studio/tools/ui_check.py @@ -61,6 +61,23 @@ def section(name: str) -> None: print(f"\n== {name} ==") +def _calibration_sample(index: int, *, fn_raw: float, fx_raw: float = 0.4, fy_raw: float = -0.2): + """A sample shaped for the calibration recorder.""" + from orisys_app_shared.force.session import SessionSample + + return SessionSample( + frame_id=index, + t_rel_s=index / 100.0, + fn_raw=fn_raw, + fx_raw=fx_raw, + fy_raw=fy_raw, + fn_n=fn_raw / 100.0, + fx_n=fx_raw / 100.0, + fy_n=fy_raw / 100.0, + valid=True, + ) + + def _fake_packet( frame_id: int, *, @@ -75,6 +92,8 @@ def _fake_packet( nan = float("nan") blank = np.zeros((8, 8, 3), np.uint8) + # A single tall sample makes it easy to prove decimation keeps peaks. + fn_raw = 99.0 if frame_id == 700 else (9.0 + frame_id * 0.01) return SdkPacket( raw_frame=blank, processed_frame=blank, @@ -495,15 +514,71 @@ def main() -> int: window.cal_panel.set_running(True) check(window.cal_panel.btn_record.isEnabled(), "record button stays disabled") - window.cal_panel._store.upsert( - "fn", CalSample(weight_n=1.0, raw_mean=9.7, n=120, std=0.02) + check( + window.cal_panel.table.columnCount() == 6, + "the calibration table is no longer the six-column diagnostic view", ) - window.cal_panel._refresh_table() + + # Drive a real point through the recorder: the table is a view over + # the recorded points, not over a separate sample list. + window.cal_recorder.start( + channel="fn", reference_load_n=1.0, duration_s=1.0 + ) + for index in range(40): + window.cal_recorder.feed( + _calibration_sample(index, fn_raw=100.0 + index * 0.01) + ) + point = window.cal_recorder.finish_now() + window.cal_panel.on_record_finished(point) + app.processEvents() check(window.cal_panel.table.rowCount() == 1, "calibration row missing") check( not window.cal_panel.empty_hint.isVisible(), "empty hint still visible with rows", ) + check( + window.cal_panel.fit_label.text() != "", + "the fit summary is empty", + ) + + # Three more points, one of them off-trend, so the fit has a residual + # that excluding the outlier must reduce and enough left over to hold + # one out as a validation point. + for load, raw in ((2.0, 400.0), (3.0, 300.6), (5.0, 500.9)): + window.cal_recorder.start( + channel="fn", reference_load_n=load, duration_s=1.0 + ) + for index in range(40): + window.cal_recorder.feed(_calibration_sample(index, fn_raw=raw)) + window.cal_panel.on_record_finished(window.cal_recorder.finish_now()) + app.processEvents() + check(window.cal_panel.table.rowCount() == 4, "calibration rows missing") + + worst = window.cal_recorder.fit("fn").max_residual_n + check(worst > 0.05, f"the off-trend point produced no residual: {worst}") + window.cal_panel.table.setCurrentCell(1, 0) + window.cal_panel._on_toggle_excluded() + app.processEvents() + check( + window.cal_recorder.fit("fn").max_residual_n < worst, + "excluding an outlier did not tighten the fit", + ) + check( + window.cal_panel.table.item(1, 5).text() == t("cal.state_excluded"), + "the excluded row does not show its state", + ) + + # A validation point is held out of the fit and reported separately. + window.cal_panel.table.setCurrentCell(2, 0) + window.cal_panel._on_toggle_validation() + app.processEvents() + check( + t("cal.fit_validation").split("{")[0].strip() + in window.cal_panel.fit_label.text(), + "no validation summary after marking a validation point", + ) + window.cal_recorder.reset_session() + window.cal_panel._refresh_table() window._set_phase("running") check("status_running" not in window.status_chip.text(), "chip not localized") @@ -843,6 +918,135 @@ def main() -> int: stuck.release() _shutil.rmtree(scratch, ignore_errors=True) + section("session browsing") + import numpy as np + + stuck = _StuckWorker() + window = StudioMainWindow( + prefs={ + "language": "zh_CN", + "theme": "dark", + "sensor_type": "box", + "source_mode": "camera", + } + ) + window.resize(1440, 880) + window.show() + window._ensure_workers = lambda: None + window._box_worker = stuck + window._finger_worker = stuck + window._start_run() + window._run_timer.stop() + window._set_phase("running") + for index in range(1500): + stuck.emit(_fake_packet(index)) + app.processEvents() + + plots = window.force_plots + check(plots.mode() == "follow", "the charts do not start in live mode") + check( + plots.stats_label.text() == "", + "the selection readout is not blank while following live", + ) + + first, last = window.session_store.extent() + check(last > first, "the session store has no extent after recording") + + plots.show_all() + check(plots.mode() == "browse", "the overview did not enter browse mode") + check( + all(plot.getViewBox().state["mouseEnabled"][0] for plot in plots.plots), + "browsing does not enable panning", + ) + window_texts = plots.stats_label.text() + check(window_texts != "", "no selection statistics in browse mode") + check("峰" in window_texts, f"the peak is missing from the readout: {window_texts!r}") + + # Decimation must keep the spike while bounding the point count. + drawn = max(int(curve.getData()[0].size) for curve in plots.curves) + check(drawn <= 2 * plots._max_draw_points, f"overview drew {drawn} points") + spike = float(np.nanmax(plots.curves[0].getData()[1])) + check(spike > 90.0, f"the overview lost the spike: {spike}") + + plots.set_window(10.0, 20.0) + check(plots.window() == (10.0, 20.0), f"set_window did not apply: {plots.window()}") + check( + plots.overview_curve.getData()[0].size > 0, + "the whole-session overview strip is empty", + ) + + # A user gesture enters browse; "back to latest" must actually leave it. + plots._syncing = False + plots.plots[0].setXRange(5.0, 8.0, padding=0.0) + app.processEvents() + check(plots.mode() == "browse", "a user pan did not enter browse mode") + plots.follow_latest() + check(plots.mode() == "follow", "back-to-latest did not restore live mode") + + plots.set_cursor(3.5, emit=True) + check( + [round(cursor.value(), 3) for cursor in plots.cursors] == [3.5, 3.5, 3.5], + "the time cursor is not shared across the three channels", + ) + + # History browsing is a display pause: capture and recording keep going. + plots.show_all() + before = window.session_store.count + for index in range(1500, 1560): + stuck.emit(_fake_packet(index)) + check( + window.session_store.count > before, + "browsing history stopped the record channel", + ) + + window.close() + app.processEvents() + stuck.release() + + section("layout preset") + stuck = _StuckWorker() + window = StudioMainWindow( + prefs={ + "language": "zh_CN", + "theme": "dark", + "sensor_type": "finger", + "source_mode": "camera", + "show_finger_3d": True, + } + ) + window.resize(1600, 900) + window.show() + window._ensure_workers = lambda: None + window._box_worker = stuck + window._finger_worker = stuck + app.processEvents() + + check(window._layout_preset == "default", "the layout preset did not load") + check(window.mesh3d_panel.isVisible(), "the 3D panel is hidden in the free layout") + + window._on_layout_preset("2d") + app.processEvents() + check(window._layout_preset == "2d", "the 2D preset did not apply") + check(not window.mesh3d_panel.isVisible(), "the 2D preset keeps the 3D panel") + check( + window.arrow_panel.view_mode() == "depth", + "the 2D preset does not show the depth map", + ) + check(not window.cal_panel.isVisible(), "the 2D preset keeps calibration expanded") + check(not window.act_finger_3d.isEnabled(), "the 3D toggle is live in the 2D preset") + check(window.act_layout_2d.isChecked(), "the 2D menu entry is not checked") + + window._on_layout_preset("default") + app.processEvents() + check(window.mesh3d_panel.isVisible(), "the free layout did not restore the 3D panel") + check(window.act_finger_3d.isEnabled(), "the 3D toggle stayed disabled") + window._persist_prefs() + check(window.prefs.get("layout_preset") == "default", "the preset was not persisted") + + window.close() + app.processEvents() + stuck.release() + print(f"\nscenes: {scene_count} checks: {CHECKS} failures: {len(FAILURES)}") for failure in FAILURES: print(f" FAIL: {failure}") diff --git a/shared/src/orisys_app_shared/force/__init__.py b/shared/src/orisys_app_shared/force/__init__.py index f1e8f37..485afea 100644 --- a/shared/src/orisys_app_shared/force/__init__.py +++ b/shared/src/orisys_app_shared/force/__init__.py @@ -16,6 +16,7 @@ from orisys_app_shared.force.calibration_recorder import ( from orisys_app_shared.force.history import MetricHistory, PlotState, lowpass_filter from orisys_app_shared.force.session import ( NA, + SelectionStats, SessionSample, SessionSlice, SessionStore, @@ -34,6 +35,7 @@ __all__ = [ "DisplayForces", "MetricHistory", "PlotState", + "SelectionStats", "SessionSample", "SessionSlice", "SessionStore", diff --git a/shared/src/orisys_app_shared/force/calibration_recorder.py b/shared/src/orisys_app_shared/force/calibration_recorder.py index 70bb2b9..532f06d 100644 --- a/shared/src/orisys_app_shared/force/calibration_recorder.py +++ b/shared/src/orisys_app_shared/force/calibration_recorder.py @@ -69,10 +69,30 @@ class CalibrationPoint: raw: dict[str, AxisStats] = field(default_factory=dict) newtons: dict[str, AxisStats] = field(default_factory=dict) status: str = STATUS_OK + #: Downsampled per-axis raw trace, so a row can be inspected after the fact. + trace: dict[str, list[float]] = field(default_factory=dict) + trace_dt_s: float = 0.0 + #: Diagnostics, all operator-controlled. Excluded points leave the fit; + #: validation points are held out of it and used to test it. + excluded: bool = False + is_validation: bool = False + #: Filled in by :meth:`CalibrationRecorder.fit`. + residual_n: float = NA def loaded_stats(self) -> AxisStats: return self.raw.get(self.channel, AxisStats()) + def axis_mean(self, channel: str) -> float: + stats = self.raw.get(channel) + return stats.mean if stats is not None else NA + + def state_key(self) -> str: + if self.excluded: + return "cal.state_excluded" + if self.is_validation: + return "cal.state_validation" + return "cal.state_ok" + def as_dict(self) -> dict: return { "channel": self.channel, @@ -89,11 +109,51 @@ class CalibrationPoint: "n_total": self.n_total, "n_valid": self.n_valid, "status": self.status, + "excluded": self.excluded, + "is_validation": self.is_validation, + "residual_n": self.residual_n, "raw": {key: stats.as_dict() for key, stats in self.raw.items()}, "newtons": {key: stats.as_dict() for key, stats in self.newtons.items()}, } +@dataclass(frozen=True) +class FitResult: + """Least-squares result for one channel.""" + + channel: ChannelKey + gain: float = NA + offset: float = NA + n_points: int = 0 + max_residual_n: float = NA + validation_error_n: float = NA + validation_points: int = 0 + + @property + def usable(self) -> bool: + return self.n_points >= 2 and np.isfinite(self.gain) and np.isfinite(self.offset) + + def predict(self, raw_value: float) -> float: + if not self.usable: + return NA + return float(self.gain) * float(raw_value) + float(self.offset) + + def as_dict(self) -> dict: + return { + "channel": self.channel, + "gain_n_per_raw": self.gain, + "offset_n": self.offset, + "points_in_fit": self.n_points, + "max_residual_n": self.max_residual_n, + "validation_error_n": self.validation_error_n, + "validation_points": self.validation_points, + } + + +#: Cap on stored trace samples per axis; peaks are preserved when reducing. +MAX_TRACE_POINTS = 400 + + class CalibrationRecorder: """Owns the calibration sampling lifecycle.""" @@ -270,6 +330,8 @@ class CalibrationRecorder: 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, + trace=_trace_from(valid_block), + trace_dt_s=_trace_dt(valid_block), ) with self._lock: self.points.append(point) @@ -277,8 +339,119 @@ class CalibrationRecorder: self._on_finished(point) return point + # ------------------------------------------------------------ diagnostics + def set_excluded(self, index: int, excluded: bool) -> None: + """Remove (or restore) a point from the fit and from the export.""" + with self._lock: + if not 0 <= index < len(self.points): + return + self.points[index].excluded = bool(excluded) + if excluded: + self.points[index].is_validation = False + + def set_validation(self, index: int, validation: bool) -> None: + """Hold a point out of the fit and use it to test the fit instead.""" + with self._lock: + if not 0 <= index < len(self.points): + return + self.points[index].is_validation = bool(validation) + if validation: + self.points[index].excluded = False + + def points_for(self, channel: ChannelKey) -> list[CalibrationPoint]: + with self._lock: + return [point for point in self.points if point.channel == channel] + + def fit(self, channel: ChannelKey) -> FitResult: + """Least-squares gain/offset from the raw mean to the reference load. + + Excluded points are dropped, validation points are held out, and every + point gets its residual filled in so the table can show the misfit. + """ + points = self.points_for(channel) + usable = [ + point + for point in points + if not point.excluded and not point.is_validation + and np.isfinite(point.reference_load_n) + and np.isfinite(point.loaded_stats().mean) + ] + for point in points: + point.residual_n = NA + if len(usable) < 2: + return FitResult(channel=channel, n_points=len(usable)) + + raws = np.asarray([point.loaded_stats().mean for point in usable], dtype=float) + loads = np.asarray([point.reference_load_n for point in usable], dtype=float) + if np.allclose(raws, raws[0]): + return FitResult(channel=channel, n_points=len(usable)) + gain, offset = np.polyfit(raws, loads, 1) + residuals = (gain * raws + offset) - loads + for point, residual in zip(usable, residuals): + point.residual_n = float(residual) + + validation = [ + point + for point in points + if not point.excluded and point.is_validation + and np.isfinite(point.reference_load_n) + and np.isfinite(point.loaded_stats().mean) + ] + validation_error = NA + if validation: + errors = [ + abs(gain * point.loaded_stats().mean + offset - point.reference_load_n) + for point in validation + ] + validation_error = float(max(errors)) + + return FitResult( + channel=channel, + gain=float(gain), + offset=float(offset), + n_points=len(usable), + max_residual_n=float(np.max(np.abs(residuals))), + validation_error_n=validation_error, + validation_points=len(validation), + ) + def _next_round(self, channel: ChannelKey) -> int: with self._lock: index = self._rounds.get(channel, 0) self._rounds[channel] = index + 1 return index + + +def _trace_from(valid_block: np.ndarray) -> dict[str, list[float]]: + """Downsampled raw trace per axis, bounded by ``MAX_TRACE_POINTS``.""" + if valid_block.size == 0: + return {} + traces: dict[str, list[float]] = {} + for key, column in (("fn", 1), ("fx", 2), ("fy", 3)): + values = valid_block[:, column] + values = values[np.isfinite(values)] + if values.size == 0: + continue + if values.size > MAX_TRACE_POINTS: + # Keep the shape of the trace, including its extremes. + edges = np.linspace(0, values.size, MAX_TRACE_POINTS + 1, dtype=int) + reduced = [ + values[start:max(start + 1, stop)] + for start, stop in zip(edges[:-1], edges[1:]) + ] + values = np.asarray( + [ + bucket[int(np.argmax(np.abs(bucket)))] + for bucket in reduced + if bucket.size + ] + ) + traces[key] = [float(value) for value in values] + return traces + + +def _trace_dt(valid_block: np.ndarray) -> float: + if valid_block.shape[0] < 2: + return 0.0 + span = float(valid_block[-1, 0] - valid_block[0, 0]) + return span / max(1, valid_block.shape[0] - 1) diff --git a/shared/src/orisys_app_shared/force/session.py b/shared/src/orisys_app_shared/force/session.py index cad584c..8cab55c 100644 --- a/shared/src/orisys_app_shared/force/session.py +++ b/shared/src/orisys_app_shared/force/session.py @@ -24,7 +24,7 @@ Design notes from __future__ import annotations import threading -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import Iterable, Iterator, Optional import numpy as np @@ -99,6 +99,55 @@ class SessionSlice: return self.raw +@dataclass(frozen=True) +class SelectionStats: + """What the browser shows for the current selection.""" + + duration_s: float = 0.0 + samples: int = 0 + peaks: tuple[float, float, float] = (NA, NA, NA) + means: tuple[float, float, float] = (NA, NA, NA) + stds: tuple[float, float, float] = (NA, NA, NA) + + +def _decimate_preserving_peaks(rows: np.ndarray, max_points: int) -> np.ndarray: + """Reduce ``rows`` to about ``max_points`` while keeping every spike. + + Each bucket contributes its minimum and its maximum sample (in time order), + so the envelope of the signal survives decimation instead of being averaged + flat. + """ + total = rows.shape[0] + if max_points < 4 or total <= max_points: + return rows + bucket = max(2, total // max(2, max_points // 2)) + usable = (total // bucket) * bucket + head = rows[:usable].reshape(-1, bucket, rows.shape[1]) + tail = rows[usable:] + # Index of the extreme value on the first channel; ties go to the earlier one. + values = head[:, :, _COL_FN_RAW] + finite = np.isfinite(values) + values = np.where(finite, values, -np.inf) + low = np.argmin(values, axis=1) + high = np.argmax(values, axis=1) + first = np.minimum(low, high) + second = np.maximum(low, high) + index = np.arange(head.shape[0]) + picked = np.concatenate( + [ + head[index, first], + head[index, second], + ], + axis=0, + ) + # ``concatenate`` above groups all lows then all highs; restore time order. + order = np.argsort(picked[:, _COL_FRAME_ID], kind="stable") + picked = picked[order] + if tail.shape[0]: + picked = np.vstack([picked, tail]) + return picked + + class SessionStore: """Thread-safe, memory-bounded, whole-session sample store.""" @@ -219,6 +268,11 @@ class SessionStore: return 0.0 return float(self._last_t - self._t0) + def origin_time(self) -> float: + """Absolute acquisition timestamp that ``t_rel_s == 0`` refers to.""" + with self._lock: + return float(self._t0) if self._t0 is not None else 0.0 + def _packed(self) -> np.ndarray: if self._count == 0: return np.empty((0, _COLUMNS), dtype=np.float64) @@ -255,6 +309,102 @@ class SessionStore: mask = (packed[:, _COL_T] >= float(t0)) & (packed[:, _COL_T] <= float(t1)) return self._slice_from(packed[mask]) + # -------------------------------------------------------------- browsing + def extent(self) -> tuple[float, float]: + """``(first_t, last_t)`` of the stored session; ``(0, 0)`` when empty.""" + with self._lock: + if self._count == 0 or self._t0 is None or self._last_t is None: + return (0.0, 0.0) + first = self._blocks[0][0, _COL_T] if self._blocks else 0.0 + return (float(first), float(self._last_t - self._t0)) + + def _block_span(self, block: np.ndarray, rows: int) -> tuple[float, float]: + return (float(block[0, _COL_T]), float(block[rows - 1, _COL_T])) + + def visible(self, t0: float, t1: float, max_points: int = 2000) -> SessionSlice: + """Decimated slice for drawing, keeping peaks. + + Only the blocks that overlap ``[t0, t1]`` are touched, so browsing a + long session does not copy the whole thing, and the decimation emits the + min and the max of each bucket so a spike can never be smoothed away. + """ + t0 = float(t0) + t1 = float(t1) + if t1 < t0: + t0, t1 = t1, t0 + with self._lock: + rows = self._packed_window(t0, t1) + truncated = self._dropped > 0 + if max_points and rows.shape[0] > max_points: + rows = _decimate_preserving_peaks(rows, max_points) + return replace(self._slice_from(rows), truncated=truncated) + + def _packed_window(self, t0: float, t1: float) -> np.ndarray: + """Concatenate only the rows inside the window. Caller holds the lock.""" + if self._count == 0: + return np.empty((0, _COLUMNS), dtype=np.float64) + selected: list[np.ndarray] = [] + remaining = self._count + for block in self._blocks: + rows = min(remaining, block.shape[0]) + remaining -= rows + if rows <= 0: + break + span = self._block_span(block, rows) + if span[1] < t0 or span[0] > t1: + continue # blocks are time-ordered: skip whole blocks + times = block[:rows, _COL_T] + start = int(np.searchsorted(times, t0, side="left")) + stop = int(np.searchsorted(times, t1, side="right")) + if stop > start: + selected.append(block[start:stop]) + if not selected: + return np.empty((0, _COLUMNS), dtype=np.float64) + return np.vstack(selected) if len(selected) > 1 else selected[0] + + def statistics(self, t0: float, t1: float, unit: str = "raw") -> "SelectionStats": + """Duration, peak, mean and std over a selection, per channel.""" + t0 = float(t0) + t1 = float(t1) + if t1 < t0: + t0, t1 = t1, t0 + with self._lock: + rows = self._packed_window(t0, t1) + if rows.shape[0] == 0: + return SelectionStats(duration_s=0.0, samples=0) + mask = rows[:, _COL_VALID] > 0.5 + valid_rows = rows[mask] + times = rows[:, _COL_T] + duration = float(times[-1] - times[0]) if times.size else 0.0 + if valid_rows.shape[0] == 0: + return SelectionStats(duration_s=duration, samples=0) + columns = ( + (_COL_FN_N, _COL_FX_N, _COL_FY_N) + if str(unit).upper().startswith("N") + else (_COL_FN_RAW, _COL_FX_RAW, _COL_FY_RAW) + ) + peaks: list[float] = [] + means: list[float] = [] + stds: list[float] = [] + for column in columns: + values = valid_rows[:, column] + values = values[np.isfinite(values)] + if values.size == 0: + peaks.append(NA) + means.append(NA) + stds.append(NA) + continue + peaks.append(float(values[np.argmax(np.abs(values))])) + means.append(float(values.mean())) + stds.append(float(values.std())) + return SelectionStats( + duration_s=duration, + samples=int(valid_rows.shape[0]), + peaks=tuple(peaks), + means=tuple(means), + stds=tuple(stds), + ) + def _slice_from(self, packed: np.ndarray) -> SessionSlice: if packed.shape[0] == 0: empty = np.empty(0, dtype=np.float64) diff --git a/shared/tests/test_record_channel.py b/shared/tests/test_record_channel.py index 8eafbb9..8061078 100644 --- a/shared/tests/test_record_channel.py +++ b/shared/tests/test_record_channel.py @@ -16,6 +16,7 @@ import math import threading import time +import numpy as np import pytest from orisys_app_shared.force.calibration_recorder import ( @@ -100,6 +101,102 @@ def test_store_window_queries_are_relative_to_the_first_sample(): assert float(first_second.times[0]) == 0.0 +# ------------------------------------------------------------- session browsing + + +def _ramp_store(count: int = 3000, *, spike_at: int = 1500) -> SessionStore: + store = SessionStore(max_samples=20_000, block_size=64) + store.reset(session_id=1) + for index in range(count): + fn = index * 0.01 + if index == spike_at: + fn = 99.0 + store.append( + SessionSample( + frame_id=index, + t_rel_s=index / 30.0, + fn_raw=fn, + fx_raw=1.0, + fy_raw=-1.0, + fn_n=fn / 10.0, + fx_n=0.1, + fy_n=-0.1, + valid=(index % 10) != 3, + ) + ) + return store + + +def test_extent_covers_the_stored_session(): + store = _ramp_store(600) + first, last = store.extent() + assert first == 0.0 + assert last == pytest.approx(599 / 30.0, abs=1e-6) + assert SessionStore().extent() == (0.0, 0.0) + + +def test_visible_decimates_but_keeps_the_peak(): + store = _ramp_store() + full = store.data() + window = store.visible(0.0, 100.0, max_points=200) + + assert len(window) <= 400, "decimation did not bound the point count" + assert len(window) < len(full), "a long window should be decimated" + assert float(window.value_for_unit("raw")[:, 0].max()) == 99.0, ( + "decimation averaged away the spike" + ) + assert bool((np.diff(window.times) >= 0).all()), "decimation broke time order" + + +def test_visible_reads_only_the_requested_window(): + store = _ramp_store() + window = store.visible(10.0, 20.0, max_points=100_000) + assert float(window.times.min()) == pytest.approx(10.0, abs=1 / 30.0) + assert float(window.times.max()) == pytest.approx(20.0, abs=1 / 30.0) + assert len(window) == pytest.approx(301, abs=1) + + +def test_visible_survives_an_empty_or_inverted_window(): + store = _ramp_store(120) + assert len(store.visible(500.0, 600.0)) == 0 + # An inverted window is normalised rather than returning nothing. + assert len(store.visible(20.0, 10.0)) == len(store.visible(10.0, 20.0)) + + +def test_statistics_report_duration_peak_mean_and_std(): + store = _ramp_store(600) + stats = store.statistics(0.0, 20.0, unit="raw") + + assert stats.samples == 540, "invalid samples must not be counted" + assert stats.duration_s == pytest.approx(20.0, abs=0.05) + # Last sample inside the window is index 599, so the peak is index * 0.01. + assert stats.peaks[0] == pytest.approx(5.99, abs=0.02) + assert stats.means[1] == pytest.approx(1.0) + assert stats.stds[1] == pytest.approx(0.0, abs=1e-9) + + +def test_statistics_follow_the_display_unit(): + store = _ramp_store(600) + raw = store.statistics(0.0, 20.0, unit="raw") + newtons = store.statistics(0.0, 20.0, unit="N") + assert newtons.peaks[0] == pytest.approx(raw.peaks[0] / 10.0, abs=0.01) + + +def test_statistics_of_an_empty_selection_are_blank_not_zero(): + store = _ramp_store(120) + stats = store.statistics(900.0, 1000.0) + assert stats.samples == 0 + assert all(math.isnan(value) for value in stats.means) + + +def test_origin_time_matches_the_first_sample(): + store = _ramp_store(10) + assert store.origin_time() == pytest.approx(0.0) # _sample starts at t=0 + store.reset() + store.append(SessionSample(frame_id=0, t_rel_s=5000.0, fn_raw=1.0)) + assert store.origin_time() == pytest.approx(5000.0) + + # ------------------------------------------------------- CalibrationRecorder