orisys-tech/orisys-app
3
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

统一力值阈值处理并保留原始采样

Browse Source 
移除自动显示基线，使逐轴阈值统一作用于曲线、读数与记录输出，标定拟合继续使用原始采样。
共享层全量测试和相关 Studio 测试通过。
This commit is contained in:
ykcheungab
2026-10-07 12:25:12 +08:00
parent 6194778a6f
commit 86996694b4
27 changed files with 507 additions and 1512 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/studio/config/locales/en_US.json
+7 -10
View File
@@ -32,10 +32,6 @@
"scene3d_settings": "3D display settings…",
"show_fps": "Show FPS",
"show_diagnostics": "Diagnostics",
"raw_auto_zero": "Finger raw auto-zero at start",
"raw_auto_zero_tip": "Value cards only: estimate an idle bias and deadband after start. Curves, statistics and CSV keep the measured values.",
"raw_reestimate": "Re-estimate raw baseline…",
"raw_reestimate_tip": "Drop the current display baseline and re-estimate from the next quiet samples (value cards only).",
"settings": "Settings area",
"plot_settings": "Plot settings…",
"rotate_arrows": "Rotate arrows 90°",
@@ -71,9 +67,10 @@
"plot_shear_raw_max": "Shear raw max",
"plot_shear_raw_max_tip": "Shear-force axis ±max for FX/FY in raw (uncalibrated) mode. 0 = auto range",
"plot_deadband_n": "Zero threshold (N)",
"plot_deadband_n_tip": "Display deadband floor in Newton mode: a reading inside the threshold shows as 0 and the curve flattens on the zero line. 0 = auto (the noise estimated from the quiet startup window; Newton mode has no fixed floor). Display only — the record and forces.csv keep the measured values",
"plot_deadband_n_tip": "Absolute per-axis zero threshold in N. Values within the threshold become 0; 0 disables it. Recordings retain acquired and processed values",
"plot_deadband_raw": "Zero threshold (raw)",
"plot_deadband_raw_tip": "Display deadband floor in raw mode (raw counts). 0 = auto (the built-in 50-count floor). Raise it by hand when the calibration data is suspect; display only",
"plot_deadband_raw_tip": "Absolute per-axis zero threshold in raw counts. Values within the threshold become 0; 0 disables it. Calibration fitting still uses acquired samples",
"plot_value_decimals": "Value decimals",
"reset_defaults": "Reset to defaults",
"plot_orientation_group": "Finger 2D presentation (2D only; the 3D view keeps the design frame)",
"plot_orientation": "Orientation",
@@ -137,7 +134,7 @@
"playback_needs_recording": "This video has no recording beside it (no forces.csv): replay needs a recording; to just watch the video open it as a video, or use View ▸ Analyze video",
"playback_video_only": "Play this video (no recording beside it: the picture plays, the force charts are empty)",
"reset_sensor": "Reset sensor",
"reset_sensor_tip": "Zero the sensor tracker (the device's own zero). Zeroing it also restarts the display baseline, so the curves and the value cards re-learn their zero from the next quiet samples; the curve history is kept — View ▸ Reset views (F8) clears the views.",
"reset_sensor_tip": "Reset the sensor tracker and its reference frame. Force thresholds remain as configured; chart history is kept.",
"reset_sensor_unavailable": "Zero the sensor tracker — available while a session is running (start a run first).",
"reset_display": "Reset display",
"reset_display_tip": "Clear the accumulated display: the curves, cursor and value cards go back to their defaults and the replay returns to the start of the recording. There is no device session in this view, so nothing of the device is touched; the 3D camera is View ▸ Reset views (F8).",
@@ -286,7 +283,7 @@
"value_unavailable": "n/a",
"value_unavailable_tip": "No curve is loaded for {channel}, so it has no value in this unit (raw counts are never shown in its place)",
"zero_zone": "zero zone",
"zero_zone_tip": "Display deadband after auto baseline; curves, statistics and CSV keep the measured values",
"zero_zone_tip": "This value is within the configured threshold; curves and output columns also show zero. Original samples are retained.",
"unit_toggle": "Switch unit",
"unit_toggle_tip": "Click to switch the three curves and readouts between newton and raw units",
"browse_only_tip": "Drag or scroll on the chart to browse history"
@@ -482,8 +479,8 @@
"capture_slow": "The camera delivers less than the configured frame rate: {detail}. The recording keeps the real rate. This is a capture-side (camera/driver) matter that software cannot change: use a camera that really does deliver 30 fps — after checking that the cable or port is not slowing this one down — and measure again.",
"video_ended": "Video ended",
"camera_disconnected": "Camera disconnected",
"reset_ok": "Sensor tracker reset (the display baseline restarts)",
"reset_pending": "Sensor zero requested; the display baseline restarts once the acquisition thread confirms…",
"reset_ok": "Sensor tracker reset",
"reset_pending": "Sensor tracker reset requested; waiting for acquisition-thread confirmation…",
"reset_display": "Display reset: curves back to their defaults, replay at the start",
"view_reset": "Views reset",
"paused": "Paused: nothing is computed, the device stays connected",
apps/studio/config/locales/zh_CN.json
+7 -10
View File
@@ -32,10 +32,6 @@
"scene3d_settings": "三维显示设置…",
"show_fps": "显示 FPS",
"show_diagnostics": "诊断信息",
"raw_auto_zero": "启动时 raw 自动归零",
"raw_auto_zero_tip": "仅影响数字卡展示：启动后自动估计无载偏置与死区。曲线、统计与 CSV 仍为测量值。",
"raw_reestimate": "重新估计 raw 基线…",
"raw_reestimate_tip": "丢弃当前展示基线，从接下来的无载样本重新估计（仅数字卡）。",
"settings": "设置区",
"plot_settings": "绘图设置…",
"rotate_arrows": "旋转箭头图 90°",
@@ -71,9 +67,10 @@
"plot_shear_raw_max": "原始剪切上限",
"plot_shear_raw_max_tip": "未标定（原始）模式下的剪切力 FX/FY 纵轴 ±上限。0 = 自动量程",
"plot_deadband_n": "归零阈值（N）",
"plot_deadband_n_tip": "牛顿模式下的显示死区下限：读数在阈值内即显示为 0，曲线压平在零线上。0 = 自动（用启动静默窗口估计的噪声，牛顿模式无固定下限）。只影响显示，记录与 forces.csv 仍保留测量值",
"plot_deadband_n_tip": "牛顿输出的绝对归零阈值，逐轴生效。|值| 不超过阈值时输出 0；0 表示关闭。实验记录保留原值及处理后值",
"plot_deadband_raw": "归零阈值（原始）",
"plot_deadband_raw_tip": "原始模式下的显示死区下限（原始计数）。0 = 自动（默认下限 50 计数）。标定数据可疑时可手动放大，只影响显示",
"plot_deadband_raw_tip": "原始计数的绝对归零阈值，逐轴生效。|值| 不超过阈值时输出 0；0 表示关闭。标定拟合仍使用原始采样",
"plot_value_decimals": "数值小数位",
"reset_defaults": "恢复默认",
"plot_orientation_group": "手指 2D 显示方向（只影响 2D 画面；3D 视图仍用设计坐标系）",
"plot_orientation": "方向",
@@ -137,7 +134,7 @@
"playback_needs_recording": "这段视频旁边没有记录（缺 forces.csv）：回放需要一条记录；只想看视频请直接打开视频，或用 视图 ▸ 分析视频",
"playback_video_only": "播放这段视频（没有配套记录：可以看画面，力曲线为空）",
"reset_sensor": "重置传感器",
"reset_sensor_tip": "把传感器追踪器归零（清零），这是设备自己的零点。归零同时会重新估计显示基线：曲线与数字卡按之后的静默样本重新学习零点；曲线历史保留 —— 要清空视图请用「视图 → 复位视图」（F8）。",
"reset_sensor_tip": "重置传感器追踪器及参考帧；力值阈值保持设定值，曲线历史保留。",
"reset_sensor_unavailable": "把传感器追踪器归零 —— 仅在采集会话运行时可用（先点「运行」）。",
"reset_display": "复位显示",
"reset_display_tip": "清空累积的显示：曲线、游标与数字卡恢复默认，回放回到录像开头。本视图没有设备会话，不会改动设备；三维视角请用「视图 → 复位视图」（F8）。",
@@ -286,7 +283,7 @@
"value_unavailable": "不可用",
"value_unavailable_tip": "{channel} 轴未加载标定曲线，因此没有该单位的数值（不会以原始值替代）",
"zero_zone": "零区",
"zero_zone_tip": "展示归零后的死区内读数；曲线、统计与 CSV 仍为原始测量值",
"zero_zone_tip": "读数位于设定阈值内；曲线与输出列同样归零，原始采样另行保留。",
"unit_toggle": "切换单位",
"unit_toggle_tip": "点击在「牛顿」与「原始单位」之间切换三轴曲线与读数",
"browse_only_tip": "拖动或滚轮即可进入历史浏览"
@@ -482,8 +479,8 @@
"capture_slow": "摄像头交付的帧率低于配置值：{detail}。记录会按实际帧率保存。这是采集端（摄像头/驱动）的问题，软件改不了：换一个真正能达到 30 fps 的摄像头，或先确认线材/端口没有把它降速，然后再测。",
"video_ended": "视频已结束",
"camera_disconnected": "摄像机已断开",
"reset_ok": "传感器追踪器已重置（显示基线重新估计）",
"reset_pending": "已请求传感器归零，采集线程确认后显示基线重新估计…",
"reset_ok": "传感器追踪器已重置",
"reset_pending": "已请求重置传感器追踪器，等待采集线程确认…",
"reset_display": "显示已复位：曲线恢复默认，回放回到开头",
"view_reset": "视图已复位",
"paused": "已暂停：不再计算，设备保持连接",
apps/studio/config/preferences.default.json
+3 -2
View File
@@ -10,7 +10,6 @@
"show_cal_panel": false,
"show_finger_3d": true,
"show_fps": false,
"raw_auto_zero": true,
"gl_profile": "desktop33",
"scene3d_package": "",
"scene3d_display": {
@@ -30,7 +29,9 @@
"plot_fn_raw_max": 1000000.0,
"plot_shear_raw_max": 500000.0,
"plot_deadband_n": 0.0,
"plot_deadband_raw": 0.0,
"plot_deadband_raw": 50.0,
"plot_newton_decimals": 2,
"plot_raw_decimals": 0,
"image_view_mode": "arrows_pure",
"arrow_rotation_deg": 90,
"finger_2d_orientation": "rotate180",
apps/studio/docs/ui-elements.json
-4
View File
@@ -182,10 +182,6 @@
"purpose": "view menu: show the frame-rate readout" },
{ "id": "act_show_diagnostics", "path": "act_show_diagnostics", "objectName": null, "class": "QAction",
"purpose": "view menu: how much the panels explain themselves around their pictures" },
{ "id": "act_raw_auto_zero", "path": "act_raw_auto_zero", "objectName": null, "class": "QAction",
"purpose": "view menu: the display baseline (auto-zero) master switch" },
{ "id": "act_raw_reestimate", "path": "act_raw_reestimate", "objectName": null, "class": "QAction",
"purpose": "view menu: restart the idle baseline window" },
{ "id": "act_settings", "path": "act_settings", "objectName": null, "class": "QAction",
"purpose": "view menu: the rail's settings area, the same state as its toggle button" },
{ "id": "act_plot_settings", "path": "act_plot_settings", "objectName": null, "class": "QAction",
apps/studio/src/orisys_studio/acquire_presenter.py
+12 -41
View File
@@ -11,18 +11,13 @@ of the 2D panel is decided here once (``replay_owns_image``) and nowhere else.
from __future__ import annotations
import logging
from typing import Any, Protocol
import numpy as np
from orisys_app_shared.force import select_display_forces
from orisys_app_shared.i18n import t
from orisys_app_shared.record.sample import to_record_sample
from orisys_app_shared.record.sample import process_record_sample, to_record_sample
from PySide6.QtCore import QObject
logger = logging.getLogger(__name__)
class AcquirePresentHost(Protocol):
"""What the presenter needs from the shell around a live packet."""
@@ -104,12 +99,9 @@ class AcquirePresenter(QObject):
if is_new and not display_paused:
self._last_plot_packet_key = packet_key
self._push_forces(packet)
self._cal_panel.set_live(
packet.normal_force_raw,
packet.shear_x_raw,
packet.shear_y_raw,
)
record = self._push_forces(packet)
if record is not None:
self._cal_panel.set_live(*record.processed_triplet("raw"))
# Finalise calibration sampling on the UI thread once its deadline passes.
self._cal_recorder.tick()
@@ -139,14 +131,19 @@ class AcquirePresenter(QObject):
self._host.refresh_status_rail()
self._host.refresh_record_state()
def _push_forces(self, packet: Any) -> None:
def _push_forces(self, packet: Any):
record = to_record_sample(packet)
if not record.valid:
self._host.session.last_sample_valid = False
self._host.show_status(
self._host.session.invalid_sample_message(record.invalid_reason)
)
return
return None
record = process_record_sample(
record,
raw_threshold=self._charts.presentation.min_deadband_for("raw"),
newton_threshold=self._charts.presentation.min_deadband_for("N"),
)
display = select_display_forces(
units_pref=self._host.toolbar.units(),
sample=record,
@@ -160,9 +157,6 @@ class AcquirePresenter(QObject):
self._host.session.last_sample_valid = True
self.last_display_fallback = bool(display.using_raw_fallback)
self.last_display_available = tuple(bool(flag) for flag in display.available)
flow_p95 = _flow_p95(packet)
contact = getattr(packet, "contact", None)
contact_found = bool(getattr(contact, "found", False)) if contact else False
self._charts.push(
self._host.session_time(packet),
display.values[0],
@@ -170,29 +164,6 @@ class AcquirePresenter(QObject):
display.values[2],
unit=display.unit,
available=display.available,
flow_p95=flow_p95,
contact_found=contact_found,
forces_valid=bool(getattr(packet, "forces_valid", True)),
)
def _flow_p95(packet: Any) -> float | None:
flow = getattr(packet, "flow", None)
if flow is None:
return None
try:
arr = np.asarray(flow, dtype=np.float64)
if arr.ndim != 3 or arr.shape[-1] != 2:
return None
mag = np.linalg.norm(arr, axis=-1)
mask = getattr(packet, "valid_mask", None)
if mask is not None:
m = np.asarray(mask, dtype=bool)
vals = mag[m & np.isfinite(mag)] if m.shape == mag.shape else mag[np.isfinite(mag)]
else:
vals = mag[np.isfinite(mag)]
if vals.size:
return float(np.percentile(vals, 95))
except Exception: # noqa: BLE001
logger.debug("flow p95 skipped", exc_info=True)
return None
return record
apps/studio/src/orisys_studio/app_window.py
+8 -61
View File
@@ -592,7 +592,7 @@ class StudioMainWindow(QMainWindow):
self._render_run_image(packet)
def replay_owns_image(self) -> bool:
return self._replay_owns_image()
return self.replay_ctl.owns_image()
def refresh_record_state(self) -> None:
self._refresh_record_state()
@@ -937,30 +937,6 @@ class StudioMainWindow(QMainWindow):
"""The charts' shared cursor scrubs the recording (through the controller)."""
self.replay_ctl.on_cursor_moved(t_rel_s)
# -------------------------------------------------------- replay (forwards)
# The replay policy lives in ``ReplayController``. These forward so the call sites
# still in this file — and the tests / ``tools/ui_check`` / ``tools/ui_snapshot``,
# which drive the window — keep working while the rest of the split lands. Each is
# a one-line delegation; they are deleted as the call sites move to
# ``self.replay_ctl.*``.
def _load_recording(self, directory: Path) -> bool:
return self.replay_ctl.load_recording(directory)
def _load_replay_video(self, video: Path) -> bool:
return self.replay_ctl.load_video(video)
def _render_replay_frame(self) -> None:
self.replay_ctl.render_frame()
def _start_playback(self) -> None:
self.replay_ctl.start_playback()
def _on_play_shortcut(self) -> None:
self.replay_ctl.on_play_shortcut()
def _replay_owns_image(self) -> bool:
return self.replay_ctl.owns_image()
def _drop_render_marker(self) -> None:
"""A recorded frame was on the panel: the next live packet must repaint it."""
if self.acquire_presenter is not None:
@@ -1118,15 +1094,6 @@ class StudioMainWindow(QMainWindow):
)
self.view_menu.addAction(self.act_show_diagnostics)
self.act_raw_auto_zero = QAction(self)
self.act_raw_auto_zero.setCheckable(True)
self.act_raw_auto_zero.setChecked(bool(self.prefs.get("raw_auto_zero", True)))
self.act_raw_auto_zero.toggled.connect(self.display_ctl.on_raw_auto_zero_toggled)
self.view_menu.addAction(self.act_raw_auto_zero)
self.act_raw_reestimate = QAction(self)
self.act_raw_reestimate.triggered.connect(self.display_ctl.request_raw_reestimate)
self.view_menu.addAction(self.act_raw_reestimate)
self.act_settings = QAction(self)
self.act_settings.setCheckable(True)
@@ -1174,7 +1141,7 @@ class StudioMainWindow(QMainWindow):
self.view_menu.addAction(self.act_reset)
# The graphics half of the old 重置: curves, window, cursor and the 3D camera.
# Separate because wanting a fresh device baseline is not the same intention as
# Separate because resetting the device tracker is not the same intention as
# wanting the views back, and only the first one belongs to the device.
self.act_reset_view = QAction(self)
self.act_reset_view.setShortcut("F8")
@@ -1415,10 +1382,6 @@ class StudioMainWindow(QMainWindow):
self.act_scene3d_settings.setText(t("menu.scene3d_settings"))
self.act_show_fps.setText(t("menu.show_fps"))
self.act_show_diagnostics.setText(t("menu.show_diagnostics"))
self.act_raw_auto_zero.setText(t("menu.raw_auto_zero"))
self.act_raw_auto_zero.setToolTip(t("menu.raw_auto_zero_tip"))
self.act_raw_reestimate.setText(t("menu.raw_reestimate"))
self.act_raw_reestimate.setToolTip(t("menu.raw_reestimate_tip"))
self.act_settings.setText(t("menu.settings"))
self.act_plot_settings.setText(t("menu.plot_settings"))
self.act_rotate_arrows.setText(t("menu.rotate_arrows"))
@@ -1819,6 +1782,8 @@ class StudioMainWindow(QMainWindow):
# Display-only zeroing floors; 0 means "use the estimated band".
deadband_n=settings["deadband_n"],
deadband_raw=settings["deadband_raw"],
newton_decimals=int(settings["newton_decimals"]),
raw_decimals=int(settings["raw_decimals"]),
units=settings["units"],
filters=self._filter_settings(),
# Asked *now*: the answer is a property of this moment, and the
@@ -2241,7 +2206,7 @@ class StudioMainWindow(QMainWindow):
self.statusBar().showMessage(t("status.preview_stopped"))
def _on_preview_tick(self) -> None:
if self._preview is None or self._replay_owns_image():
if self._preview is None or self.replay_ctl.owns_image():
return
frame = self._preview.read()
if frame is None:
@@ -2321,7 +2286,7 @@ class StudioMainWindow(QMainWindow):
# both units, which is the point of it (an offset reads as an offset in newtons
# too). This used to be a ``finger and raw`` conjunction, which silently disabled
# it for every other product.
enabled=bool(self.prefs.get("raw_auto_zero", True)),
enabled=True,
)
if self.acquire_presenter is not None:
self.acquire_presenter.clear_plot_marker()
@@ -2457,21 +2422,7 @@ class StudioMainWindow(QMainWindow):
self.statusBar().showMessage(t("status.run_stopped"))
def _on_reset(self) -> None:
"""Zero the *sensor tracker* — or, where there is no device, reset the display.
This used to call ``force_plots.reset()`` as well, so wanting a fresh baseline
also threw away the history, the window and the cursor on screen — two different
operator intentions on one button (and only one of them recorded, see
``note_reset_applied``). The display baseline is re-estimated instead of wiped:
a new device zero means the old idle estimate is stale, not that the operator
wants to lose what they were looking at.
Which of the two this button means is the *view's* answer, and the label says
it (``toolbar.reset_target``): a device session can be zeroed, a replay cannot
— its display can, and that is what the operator reached for. Neither branch
adds a second meaning: the live one is the tracker reset it always was, and the
replay one is the display reset ``_reset_display`` states in full.
"""
"""Reset the live sensor tracker or the replay display, as labeled."""
if self.toolbar.reset_target() == "display":
self._reset_display()
return
@@ -2481,10 +2432,6 @@ class StudioMainWindow(QMainWindow):
# A request, not the zeroing itself: the acquisition thread applies it
# between two frames and reports the frame and time it took effect.
self._active_worker.reset()
# The side effect the tooltip states: a new zero makes the old idle
# estimate stale, so the display baseline is re-learned from the samples
# that follow. The curve history itself is kept.
self.force_plots.request_presentation_reestimate()
if self.acquire_presenter is not None:
self.acquire_presenter.clear_plot_marker()
self.statusBar().showMessage(t("status.reset_pending"))
@@ -2794,7 +2741,7 @@ class StudioMainWindow(QMainWindow):
def _render_run_image(self, packet) -> None:
"""Live / recompute entry into the shared 2D→3D deliver path."""
if packet is None or self._replay_owns_image():
if packet is None or self.replay_ctl.owns_image():
return
self._deliver_measurement(packet, advance="tick")
apps/studio/src/orisys_studio/display_prefs.py
+30 -42
View File
@@ -29,6 +29,8 @@ from orisys_studio.shell_ports import DisplayShell
from orisys_studio.ui.plot_settings_dialog import (
DEFAULT_PLOT_DEADBAND_N,
DEFAULT_PLOT_DEADBAND_RAW,
DEFAULT_PLOT_NEWTON_DECIMALS,
DEFAULT_PLOT_RAW_DECIMALS,
DEFAULT_PLOT_FN_MAX,
DEFAULT_PLOT_FN_MIN,
DEFAULT_PLOT_FN_RAW_MAX,
@@ -51,12 +53,15 @@ PLOT_RANGE_DEFAULTS = {
"fn_raw_max": DEFAULT_PLOT_FN_RAW_MAX,
"shear_raw_max": DEFAULT_PLOT_SHEAR_RAW_MAX,
}
#: Display deadband floors, per unit. 0 = auto (the estimate from the quiet
#: startup window, with the presentation layer's own floor under it).
#: Absolute output thresholds per unit. Zero disables suppression.
PLOT_DEADBAND_DEFAULTS = {
"deadband_n": DEFAULT_PLOT_DEADBAND_N,
"deadband_raw": DEFAULT_PLOT_DEADBAND_RAW,
}
PLOT_PRECISION_DEFAULTS = {
"newton_decimals": DEFAULT_PLOT_NEWTON_DECIMALS,
"raw_decimals": DEFAULT_PLOT_RAW_DECIMALS,
}
# Kept as an alias so older imports and tests that name ``DisplayHost`` keep working.
DisplayHost = DisplayShell
@@ -99,10 +104,6 @@ class DisplayPrefsController(QObject):
#: Whether the panels explain the pipeline around the picture (see
#: ``on_show_diagnostics_toggled``). Off: each panel shows what the picture *is*.
self._show_diagnostics = bool(host.prefs.get("show_diagnostics", False))
#: The display baseline / deadband (``raw_auto_zero``): a display aid
#: that holds an idle channel at zero, for every sensor type and both
#: units. It never reaches the session store or forces.csv.
self._raw_auto_zero = bool(host.prefs.get("raw_auto_zero", True))
self._apply_diagnostics()
self.apply_presentation_prefs()
@@ -166,7 +167,9 @@ class DisplayPrefsController(QObject):
key: self._number(
f"plot_{key}", self._plot_defaults.get(key, fallback)
)
for key, fallback in {**PLOT_RANGE_DEFAULTS, **PLOT_DEADBAND_DEFAULTS}.items()
for key, fallback in {
**PLOT_RANGE_DEFAULTS, **PLOT_DEADBAND_DEFAULTS, **PLOT_PRECISION_DEFAULTS
}.items()
}
settings["units"] = self._toolbar.units()
return settings
@@ -191,7 +194,7 @@ class DisplayPrefsController(QObject):
A caller that knows only some of the keys leaves the rest as stored, so
this can never silently reset a range it was not told about.
"""
for key in (*PLOT_RANGE_DEFAULTS, *PLOT_DEADBAND_DEFAULTS):
for key in (*PLOT_RANGE_DEFAULTS, *PLOT_DEADBAND_DEFAULTS, *PLOT_PRECISION_DEFAULTS):
if values.get(key) is not None:
self._host.prefs[f"plot_{key}"] = values[key]
settings = self.plot_settings()
@@ -217,11 +220,28 @@ class DisplayPrefsController(QObject):
shear_raw_max=float(settings["shear_raw_max"]),
)
self._push_deadband_floors(settings["deadband_n"], settings["deadband_raw"])
decimals = getattr(self._charts, "set_value_decimals", None)
if callable(decimals):
decimals(
newton=int(settings["newton_decimals"]),
raw=int(settings["raw_decimals"]),
)
cal_panel = getattr(self._host, "cal_panel", None)
cal_settings = getattr(cal_panel, "set_output_settings", None)
if callable(cal_settings):
cal_settings(
raw_threshold=float(settings["deadband_raw"]),
raw_decimals=int(settings["raw_decimals"]),
)
def _push_deadband_floors(self, newton: float, raw: float) -> None:
floors = getattr(self._charts, "set_deadband_floors", None)
if callable(floors):
floors(newton=float(newton), raw=float(raw))
recording = getattr(self._host, "recording", None)
setter = getattr(recording, "set_force_thresholds", None)
if callable(setter):
setter(newton=float(newton), raw=float(raw))
def device_sample_rate(self) -> float:
"""The frame rate the low-pass coefficients would be designed for."""
@@ -301,32 +321,8 @@ class DisplayPrefsController(QObject):
if callable(setter):
setter(self._show_diagnostics)
# ------------------------------------------------------- raw display zeroing
@property
def raw_auto_zero(self) -> bool:
return bool(self._raw_auto_zero)
def set_raw_auto_zero(self, enabled: bool) -> None:
"""Persist whether the display baseline / deadband is on at all.
The preference keeps its name; it is no longer Finger-raw-only (see
:meth:`apply_presentation_prefs`).
"""
self._raw_auto_zero = bool(enabled)
self._host.prefs["raw_auto_zero"] = self._raw_auto_zero
self._host.persist_prefs()
self.apply_presentation_prefs()
def on_raw_auto_zero_toggled(self, enabled: bool) -> None:
self.set_raw_auto_zero(enabled)
def apply_presentation_prefs(self) -> None:
"""Push the display baseline policy onto the charts.
Every sensor type and both units. The aid holds an idle channel at zero:
a newton offset left in the numbers is the same display problem as a raw
count, so which device produced the reading is not what decides it.
"""
"""Push the deterministic per-unit threshold onto the charts."""
configure = getattr(self._charts, "configure_presentation", None)
sensor = (
self._picker.sensor_type()
@@ -338,16 +334,8 @@ class DisplayPrefsController(QObject):
configure(sensor_type=sensor, unit=unit)
enable = getattr(self._charts, "set_presentation_enabled", None)
if callable(enable):
enable(bool(self._raw_auto_zero))
enable(True)
self._push_deadband_floors(
self._number("plot_deadband_n", PLOT_DEADBAND_DEFAULTS["deadband_n"]),
self._number("plot_deadband_raw", PLOT_DEADBAND_DEFAULTS["deadband_raw"]),
)
def request_raw_reestimate(self) -> None:
"""Operator entry: restart the idle baseline window on the value cards."""
request = getattr(self._charts, "request_presentation_reestimate", None)
if callable(request):
request()
else:
self.apply_presentation_prefs()
apps/studio/src/orisys_studio/recording.py
+24 -3
View File
@@ -22,7 +22,7 @@ from typing import Any
from orisys_app_shared.record import ExperimentRecorder, RecordConfig, RecordReport
from orisys_app_shared.record.calibration_snapshot import store_calibration_snapshot
from orisys_app_shared.record.sample import to_record_sample
from orisys_app_shared.record.sample import process_record_sample, to_record_sample
logger = logging.getLogger(__name__)
@@ -51,6 +51,7 @@ class RecordingCoordinator:
self.last_report: RecordReport | None = None
self._active_calibration_id = ""
self._record_fps = 30.0
self._force_thresholds = (50.0, 0.0)
self.recorder.set_alarm_handler(self._on_alarm)
# ------------------------------------------------------------- acquisition
@@ -65,18 +66,32 @@ class RecordingCoordinator:
record = to_record_sample(
packet, calibration_version=self._active_calibration_id
)
raw_threshold, newton_threshold = self._force_thresholds
record = process_record_sample(
record, raw_threshold=raw_threshold, newton_threshold=newton_threshold
)
from orisys_app_shared.force.session import SessionSample
sample = SessionSample.from_record(record)
self.session_store.append(sample)
self.cal_recorder.feed(sample)
# Video + CSV + metadata share this one record path.
self.recorder.on_packet(packet)
self.recorder.on_packet(packet, sample=record)
fps = float(getattr(packet, "fps", 0.0) or 0.0)
if fps > 0.0:
self._record_fps = fps
return record
def set_force_thresholds(self, *, raw: float, newton: float) -> None:
values = (max(0.0, float(raw)), max(0.0, float(newton)))
if values == self._force_thresholds:
return
self._force_thresholds = values
if self.recorder.is_recording:
self.recorder.log_event(
"force_threshold_changed", f"raw={values[0]:g}, N={values[1]:g}"
)
def take_preroll(self, worker) -> list:
"""Frame history from the active worker, if it keeps one."""
take = getattr(worker, "take_preroll", None)
@@ -202,7 +217,13 @@ class RecordingCoordinator:
root=root,
config=config,
session_id=int(self.session_store.session_id),
metadata=metadata,
metadata={
**metadata,
"force_output_thresholds": {
"raw": self._force_thresholds[0],
"N": self._force_thresholds[1],
},
},
calibration_id=calibration_id,
calibration_version=calibration_version,
preroll=preroll,
apps/studio/src/orisys_studio/ui/cal_panel.py
+26 -7
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import numpy as np
from orisys_app_shared.force.presentation import threshold_force
from orisys_app_shared.force.calibration_io import (
ChannelKey,
)
@@ -77,6 +78,8 @@ class CalPanel(QFrame):
self._live_fn = 0.0
self._live_fx = 0.0
self._live_fy = 0.0
self._raw_threshold = 50.0
self._raw_decimals = 0
self._sensor_type = "box"
self._config_ref: str | None = None
self._mono = _mono_font()
@@ -305,6 +308,12 @@ class CalPanel(QFrame):
self._live_fy = float(fy)
self._render_live_label()
def set_output_settings(self, *, raw_threshold: float, raw_decimals: int) -> None:
self._raw_threshold = max(0.0, float(raw_threshold))
self._raw_decimals = max(0, int(raw_decimals))
self._render_live_label()
self._refresh_table()
def set_progress(self, fraction: float, elapsed_s: float, total_s: float) -> None:
self.progress.setVisible(True)
self.progress.setValue(max(0, min(1000, int(fraction * 1000))))
@@ -435,9 +444,10 @@ class CalPanel(QFrame):
("fx", self._live_fx),
("fy", self._live_fy),
):
rendered = f"{value:.{self._raw_decimals}f}" if np.isfinite(value) else "—"
parts.append(
f"<span style='color:{colors[key]}; font-weight:600'>{key.upper()}</span>"
f"&nbsp;{value:.3f}"
f"&nbsp;{rendered}"
)
self.live_label.setText(
f"<span style='color:{color(self._theme, 'text_secondary')}'>"
@@ -529,15 +539,21 @@ class CalPanel(QFrame):
self.empty_hint.setVisible(not points)
self._refresh_fit_label()
@staticmethod
def _axis_cell(point, channel: str) -> str:
value = point.axis_mean(channel)
def _axis_cell(self, point, channel: str) -> str:
column = {"fn": 2, "fx": 3, "fy": 4}[channel]
values = [
threshold_force(sample[column], self._raw_threshold)
for sample in point.samples if bool(sample[8]) and np.isfinite(sample[column])
]
value = float(np.mean(values)) if values else threshold_force(
point.axis_mean(channel), self._raw_threshold
)
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}"
return f"{value:.{self._raw_decimals}f} ({point.residual_n:+.4f})"
return f"{value:.{self._raw_decimals}f}"
@staticmethod
def _stability_cell(point) -> str:
@@ -598,7 +614,10 @@ class CalPanel(QFrame):
restore_diagnostics(point)
if not point.trace:
return None
return CalibrationPointDialog(point, theme=self._theme, parent=self)
return CalibrationPointDialog(
point, theme=self._theme, parent=self,
raw_threshold=self._raw_threshold, raw_decimals=self._raw_decimals,
)
def _on_row_activated(self) -> None:
"""Double click a row to inspect that point's three-axis curve."""
apps/studio/src/orisys_studio/ui/cal_point_dialog.py
+19 -3
View File
@@ -7,6 +7,7 @@ the three-panel layout, while a single point can still be examined in full.
from __future__ import annotations
import numpy as np
from orisys_app_shared.force.presentation import threshold_force, threshold_values
from orisys_app_shared.i18n import t
from orisys_app_shared.theme import color
from PySide6.QtGui import QFont, QFontDatabase
@@ -40,10 +41,15 @@ def _mono_font(point_size: int = 9) -> QFont:
class CalibrationPointDialog(QDialog):
"""One recorded point: all three raw traces plus its fit residual."""
def __init__(self, point, *, theme: str = "dark", parent=None) -> None:
def __init__(
self, point, *, theme: str = "dark", parent=None,
raw_threshold: float = 0.0, raw_decimals: int = 0,
) -> None:
super().__init__(parent)
self._point = point
self._theme = theme
self._raw_threshold = max(0.0, float(raw_threshold))
self._raw_decimals = max(0, int(raw_decimals))
self.setWindowTitle(
t(
"cal.point_title",
@@ -82,7 +88,15 @@ class CalibrationPointDialog(QDialog):
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}")
values = np.asarray(point.trace.get(key, ()), dtype=float)
if values.size:
values, _ = threshold_values(values, self._raw_threshold)
stats.append(
f"{key.upper()} {np.nanmean(values):.{self._raw_decimals}f}"
f" ± {np.nanstd(values):.{self._raw_decimals}f}"
)
else:
stats.append(f"{key.upper()} {threshold_force(axis.mean, self._raw_threshold):.{self._raw_decimals}f}")
value = QLabel(" ".join(stats))
value.setFont(_mono_font())
self._add_row(form, "cal.point_axes", "")
@@ -137,8 +151,10 @@ class CalibrationPointDialog(QDialog):
continue
step = self._point.trace_dt_s or 0.01
times = np.arange(len(values), dtype=float) * step
values = np.asarray(values, dtype=float)
values, _ = threshold_values(values, self._raw_threshold)
plot.plot(
times,
np.asarray(values, dtype=float),
values,
pen=pg.mkPen(color(self._theme, TRACE_TOKENS[row]), width=2),
)
apps/studio/src/orisys_studio/ui/force_plots.py
+50 -123
View File
@@ -10,8 +10,6 @@ from orisys_app_shared.acquisition.constants import FORCE_SAMPLE_RATE_HZ
from orisys_app_shared.force import MetricHistory
from orisys_app_shared.force.history import DEFAULT_MAX_FPS
from orisys_app_shared.force.presentation import (
DEFAULT_MIN_DEADBAND_N,
DEFAULT_MIN_DEADBAND_RAW,
ForcePresentation,
)
from orisys_app_shared.force.source import StoreCurveSource
@@ -108,8 +106,8 @@ def format_force(
if presentation is not None:
return presentation.format_force(value, unit=unit)
if unit == "N":
return f"{value:.3f}"
return f"{value:.1f}"
return f"{value:.2f}"
return f"{value:.0f}"
def unit_display(unit: str) -> str:
@@ -176,13 +174,10 @@ class ForcePlotsPanel(QFrame):
self._last_values: tuple[float, float, float] = (0.0, 0.0, 0.0)
#: Per axis: does ``_last_values`` hold a real value in the active unit?
self._available: tuple[bool, bool, bool] = (True, True, True)
#: Display-only baseline / deadband for the value cards *and* the curves.
self.presentation = ForcePresentation(enabled=False)
#: The operator's master switch for that baseline. It is owned by
#: ``display_prefs`` (the ``raw_auto_zero`` preference) and kept here so a
#: caller that does not know it cannot switch the aid off (see
#: ``configure_presentation``).
self._presentation_enabled = False
#: Shared absolute threshold and numeric precision for values and curves.
self.presentation = ForcePresentation(enabled=True)
#: Kept for callers that explicitly configure presentation state.
self._presentation_enabled = True
self._sensor_type = "box"
self._display_values: tuple[float, float, float] = (0.0, 0.0, 0.0)
self._display_suppressed: tuple[bool, bool, bool] = (False, False, False)
@@ -193,6 +188,7 @@ class ForcePlotsPanel(QFrame):
#: True once a source is attached (either kind). ``_store is None`` was
#: the old proxy for this and could not express a disk source.
self._source_ready = False
self._source_values_processed = False
#: Non-empty when there is no trace to draw *and that is a fact about the input*:
#: a bare video whose pass has not run, or a profile whose chain produced no force
#: for it. The mode line states it rather than leaving an empty plot under wording
@@ -557,31 +553,24 @@ class ForcePlotsPanel(QFrame):
self._apply_axis_mode()
def set_deadband_floors(self, *, newton: float | None = None, raw: float | None = None) -> None:
"""The operator's display deadband floors; 0 means "auto (estimated)".
A floor, not a replacement: the estimate from the quiet window still
applies underneath, so this is how an operator widens the zero zone (or
pins one when the calibration data is suspect) without re-fitting.
"""
"""Set absolute per-unit output thresholds; zero disables suppression."""
if newton is not None:
value = max(0.0, float(newton))
self.presentation.set_min_deadband(
"N", value if value > 0.0 else DEFAULT_MIN_DEADBAND_N
)
self.presentation.set_min_deadband("N", value)
if raw is not None:
value = max(0.0, float(raw))
self.presentation.set_min_deadband(
"raw", value if value > 0.0 else DEFAULT_MIN_DEADBAND_RAW
)
self.presentation.set_min_deadband("raw", value)
self._render_value_cards()
def set_value_decimals(self, *, newton: int, raw: int) -> None:
self.presentation.newton_decimals = max(0, int(newton))
self.presentation.raw_decimals = max(0, int(raw))
self._render_value_cards()
def set_presentation_enabled(self, enabled: bool) -> None:
"""The operator's master switch for the display baseline and deadband."""
"""Enable or disable thresholding for a standalone chart source."""
self._presentation_enabled = bool(enabled)
self.presentation.set_enabled(self._presentation_enabled, unit=self._unit)
# Switching the aid off must show the measurement again, and switching it
# on must show the policy — from the same values the curve is drawn with,
# or a card and its own curve would disagree for a frame.
self._refresh_display_values()
self._render_value_cards()
@@ -590,13 +579,7 @@ class ForcePlotsPanel(QFrame):
self._apply_axis_mode()
def reset(self) -> None:
"""Back to the view defaults: empty curves, follow mode, unit ``raw``.
The display baseline is deliberately *kept*: a view reset is not a
statement about the device, and the operator's zero estimate (or its
absence) survives it. A new session restarts the estimate through
``configure_presentation``.
"""
"""Restore view defaults while retaining the configured threshold."""
self.history.reset()
self._clear_curves()
self._last_values = (0.0, 0.0, 0.0)
@@ -626,57 +609,13 @@ class ForcePlotsPanel(QFrame):
unit: str,
enabled: bool | None = None,
) -> None:
"""Point the display baseline at the sensor and unit now being shown.
The aid is for every sensor type and both units — an idle channel that
reads a raw count or an offset in newtons is a display problem whichever
device produced it — so the sensor type only names what is drawn.
``enabled`` sets the master switch directly when given; the operator's
``raw_auto_zero`` preference is what the run-start call site passes.
Omitting it leaves the current switch alone.
"""
"""Configure the chart's unit and optional threshold switch."""
self._sensor_type = str(sensor_type or "box")
if enabled is not None:
self._presentation_enabled = bool(enabled)
self.presentation.enabled = bool(self._presentation_enabled)
self.presentation.reset_session(unit=unit)
def request_presentation_reestimate(self) -> None:
"""Restart the idle baseline window (display only: cards and curves)."""
self.presentation.request_reestimate()
# A restarted window starts from nothing, so the cards go back to the
# measurement for the couple of samples it takes to re-learn the offset
# — derived by the same call the curve uses, never left stale.
self._refresh_display_values()
self._render_value_cards()
def observe_for_baseline(
self,
values: tuple[float, float, float],
*,
t_rel_s: float,
valid: bool = True,
flow_p95: float | None = None,
contact_found: bool = False,
unit: str | None = None,
) -> None:
"""Feed the startup estimator in the unit these values are expressed in.
The curve and the record keep the raw measurement; the estimate is only
ever a display copy.
"""
if not self.presentation.enabled:
return
self.presentation.observe(
values,
t_rel_s=t_rel_s,
valid=valid,
flow_p95=flow_p95,
contact_found=contact_found,
unit=unit,
)
def push(
self,
t_rel_s: float,
@@ -712,16 +651,6 @@ class ForcePlotsPanel(QFrame):
self._last_values = (float(fn), float(fx), float(fy))
if available is not None:
self._available = tuple(bool(flag) for flag in available)
self.observe_for_baseline(
self._last_values,
t_rel_s=float(t_rel_s),
valid=bool(forces_valid),
flow_p95=flow_p95,
contact_found=contact_found,
# The values are in ``unit``, and the estimate is kept per unit, so
# the sample must be filed under the unit it was measured in.
unit=unit,
)
displayed = self.presentation.accept(
self._last_values, unit=unit, available=self._available
)
@@ -816,8 +745,6 @@ class ForcePlotsPanel(QFrame):
if unit == self._unit:
return
self._unit = unit
# The display policy keeps one estimate per unit, so it has to know which
# one is on screen: a stale "raw" is ready does not mean newtons are.
self.presentation.unit = unit
# ``clear`` keeps the buffer's time origin on purpose: the rebuilt history
# is in the same session-relative base, so the X axis must not restart.
@@ -852,6 +779,7 @@ class ForcePlotsPanel(QFrame):
self._live_store = store
self._source = StoreCurveSource(store)
self._source_ready = store is not None
self._source_values_processed = True
self._source_absent_reason = ""
self._on_source_changed()
@@ -861,6 +789,13 @@ class ForcePlotsPanel(QFrame):
self._source = RecordingCurveSource(reader)
self._source_ready = True
columns = getattr(reader.index(), "_columns", {})
self._source_values_processed = all(
name in columns for name in (
"FN_output_raw", "FX_output_raw", "FY_output_raw",
"FN_output_N", "FX_output_N", "FY_output_N",
)
)
self._source_absent_reason = ""
self._on_source_changed()
@@ -876,6 +811,7 @@ class ForcePlotsPanel(QFrame):
``source is None`` with a ``reason`` is a state, not an empty chart: the mode line
states that reason instead of leaving "following live" over a flat plot.
"""
self._source_values_processed = False
if source is None:
self._source = StoreCurveSource(self._live_store)
self._source_ready = False
@@ -1149,8 +1085,10 @@ class ForcePlotsPanel(QFrame):
# The cards describe this point, so they carry the display copy of it —
# including while browsing or replaying, where the values did not come
# from the live sample the last ``push`` presented.
presented, suppressed = self.presentation.present(
np.asarray([values], dtype=float), unit=self._unit
presented, suppressed = (
(np.asarray([values], dtype=float), np.zeros((1, 3), dtype=bool))
if self._source_values_processed
else self.presentation.present(np.asarray([values], dtype=float), unit=self._unit)
)
self._display_values = tuple(float(value) for value in presented[0])
self._display_suppressed = tuple(bool(flag) for flag in suppressed[0])
@@ -1248,9 +1186,7 @@ class ForcePlotsPanel(QFrame):
buckets=self._overview_points(), unit=self._unit
)
values = np.asarray(window.values, dtype=float)
# The envelope is drawn from the same store measurement as the window
# above, so it goes through the same presentation: the overview thumb
# must not show a jittering baseline the main chart has flattened.
# The overview uses the same processed values as the main curves.
values = self._curve_values(values)
column = values[:, 0]
finite = np.isfinite(column)
@@ -1397,7 +1333,7 @@ class ForcePlotsPanel(QFrame):
Raw mode used to auto-range, which collapsed the axis onto the idle
jitter and then rescaled on every contact. A fixed raw range keeps the
baseline visually flat: the raw maxima are sized so the idle jitter is
near-zero region visually flat: the raw maxima are sized so idle jitter is
well under one pixel.
"""
if self._unit == "N":
@@ -1460,30 +1396,23 @@ class ForcePlotsPanel(QFrame):
zero_line.setVisible(bool(ymin <= 0.0 <= ymax))
def _curve_values(self, values) -> np.ndarray:
"""The values the curves draw: the measurement through the display policy.
``history``, the session store and ``forces.csv`` keep the measurement;
the baseline and deadband are applied here, on the way to the plot only.
That is what makes a resting channel a flat line at exactly zero without
the record being edited. Browsing and a replay read the store, so they go
through this same presentation — the curves show the operator's display
policy, the store shows what was measured.
"""
displayed, _suppressed = self.presentation.present(values, unit=self._unit)
"""Use stored output values, or threshold a legacy unprocessed source."""
displayed, _suppressed = (
(np.asarray(values, dtype=float), None)
if self._source_values_processed
else self.presentation.present(values, unit=self._unit)
)
return displayed
def _refresh_display_values(self) -> None:
"""Re-derive the card values from the policy the curves are drawn with.
``push`` fills these from :meth:`ForcePresentation.accept`, whose
hysteresis is what keeps a *live* reading steady. Every other path — a
re-estimate, the master switch, a unit switch — goes through the same
stateless :meth:`ForcePresentation.present` the curve uses, so a card can
never state a value its own curve does not carry.
"""
displayed, suppressed = self.presentation.present(
"""Refresh cards from the same values that the curves draw."""
displayed, suppressed = (
(np.asarray([self._last_values], dtype=float), np.zeros((1, 3), dtype=bool))
if self._source_values_processed
else self.presentation.present(
np.asarray([self._last_values], dtype=float), unit=self._unit
)
)
self._display_values = tuple(float(value) for value in displayed[0])
self._display_suppressed = tuple(bool(flag) for flag in suppressed[0])
@@ -1499,7 +1428,11 @@ class ForcePlotsPanel(QFrame):
a single long line under the charts, which could not fit and pushed the
panel wider.
"""
fmt = "{value:.3f}" if self._unit == "N" else "{value:.1f}"
decimals = (
self.presentation.newton_decimals
if self._unit == "N" else self.presentation.raw_decimals
)
fmt = "{value:." + str(decimals) + "f}"
unavailable = t("chart.value_unavailable")
selecting = self._selection
for index, (value_label, unit_label, detail_label, value) in enumerate(
@@ -1516,12 +1449,6 @@ class ForcePlotsPanel(QFrame):
available = bool(np.isfinite(peak))
value = peak
elif self.presentation.enabled:
# The policy is *live from the first samples*: there is no
# separate "the estimate is ready" gate here. Such a gate is what
# let a card read FN 1118 raw while its own curve was already flat
# on zero, because the card fell back to the raw measurement and
# the curve did not. Both are ``_display_values`` / ``present()``
# of the same applied estimate.
value = self._display_values[index]
if not available:
# An axis with no curve in this unit: a dash, never a raw count
apps/studio/src/orisys_studio/ui/plot_settings_dialog.py
+28 -9
View File
@@ -48,12 +48,11 @@ DEFAULT_PLOT_SHEAR_MAX = 20.0
DEFAULT_PLOT_FN_RAW_MIN = 0.0
DEFAULT_PLOT_FN_RAW_MAX = 1000000.0
DEFAULT_PLOT_SHEAR_RAW_MAX = 500000.0
#: Display deadband floors ("归零阈值") in the unit each applies to. 0 means
#: "auto": the percentile the presentation layer estimates from the quiet
#: startup window. Non-zero forces a floor under that estimate, so an operator
#: can widen the zero zone (or pin one when calibration data is suspect).
#: Absolute output thresholds in each unit; zero disables suppression.
DEFAULT_PLOT_DEADBAND_N = 0.0
DEFAULT_PLOT_DEADBAND_RAW = 0.0
DEFAULT_PLOT_DEADBAND_RAW = 50.0
DEFAULT_PLOT_NEWTON_DECIMALS = 2
DEFAULT_PLOT_RAW_DECIMALS = 0
class PlotSettingsDialog(QDialog):
@@ -70,6 +69,8 @@ class PlotSettingsDialog(QDialog):
shear_raw_max: float = DEFAULT_PLOT_SHEAR_RAW_MAX,
deadband_n: float = DEFAULT_PLOT_DEADBAND_N,
deadband_raw: float = DEFAULT_PLOT_DEADBAND_RAW,
newton_decimals: int = DEFAULT_PLOT_NEWTON_DECIMALS,
raw_decimals: int = DEFAULT_PLOT_RAW_DECIMALS,
units: str = "newton",
filters: FilterSettings | None = None,
filters_locked: bool = False,
@@ -160,6 +161,17 @@ class PlotSettingsDialog(QDialog):
self.deadband_raw_spin.setSingleStep(5000.0)
self.deadband_raw_spin.setDecimals(0)
self.deadband_raw_spin.setValue(float(deadband_raw))
self.newton_decimals_spin = QSpinBox()
self.newton_decimals_spin.setRange(0, 6)
self.newton_decimals_spin.setValue(int(newton_decimals))
self.raw_decimals_spin = QSpinBox()
self.raw_decimals_spin.setRange(0, 6)
self.raw_decimals_spin.setValue(int(raw_decimals))
precision_row = QHBoxLayout()
precision_row.addWidget(QLabel("N"))
precision_row.addWidget(self.newton_decimals_spin)
precision_row.addWidget(QLabel("raw"))
precision_row.addWidget(self.raw_decimals_spin)
# How the finger's 2D canvas is turned for display. A *presentation* choice, so
# it lives with the other display ones (this dialog), not with the acquisition
@@ -186,6 +198,8 @@ class PlotSettingsDialog(QDialog):
form.addRow(self.shear_raw_max_label, self.shear_raw_max_spin)
form.addRow(self.deadband_n_label, self.deadband_n_spin)
form.addRow(self.deadband_raw_label, self.deadband_raw_spin)
self.precision_label = QLabel()
form.addRow(self.precision_label, precision_row)
root.addLayout(form)
self.orientation_group = QGroupBox()
@@ -438,7 +452,7 @@ class PlotSettingsDialog(QDialog):
}
def deadbands(self) -> dict[str, float]:
"""The per-unit display deadband floors. 0 means "auto" (estimated)."""
"""The per-unit absolute output thresholds."""
return {
"deadband_n": float(self.deadband_n_spin.value()),
"deadband_raw": float(self.deadband_raw_spin.value()),
@@ -446,7 +460,11 @@ class PlotSettingsDialog(QDialog):
def settings(self) -> dict[str, float]:
"""Everything the plot half of the dialog decides, in one mapping."""
return {**self.ranges(), **self.deadbands()}
return {
**self.ranges(), **self.deadbands(),
"newton_decimals": int(self.newton_decimals_spin.value()),
"raw_decimals": int(self.raw_decimals_spin.value()),
}
def orientation(self) -> str:
"""The 2D presentation the dialog is showing (see ``display_orientation``)."""
@@ -464,10 +482,10 @@ class PlotSettingsDialog(QDialog):
self.orientation_combo.setCurrentIndex(
max(self.orientation_combo.findData(DEFAULT_FINGER_ORIENTATION), 0)
)
# "Auto" is the shipped answer for the deadband: the estimate from the
# quiet startup window, with the layer's own floor under it.
self.deadband_n_spin.setValue(DEFAULT_PLOT_DEADBAND_N)
self.deadband_raw_spin.setValue(DEFAULT_PLOT_DEADBAND_RAW)
self.newton_decimals_spin.setValue(DEFAULT_PLOT_NEWTON_DECIMALS)
self.raw_decimals_spin.setValue(DEFAULT_PLOT_RAW_DECIMALS)
# "Reset" means the shipped defaults, and those filter nothing — the same
# answer the dialog gives for a preference file that says nothing.
self.flow_ema_spin.setValue(1.0)
@@ -498,6 +516,7 @@ class PlotSettingsDialog(QDialog):
self.deadband_n_spin.setToolTip(t("dialog.plot_deadband_n_tip"))
self.deadband_raw_label.setText(t("dialog.plot_deadband_raw"))
self.deadband_raw_spin.setToolTip(t("dialog.plot_deadband_raw_tip"))
self.precision_label.setText(t("dialog.plot_value_decimals"))
self.btn_reset.setText(t("dialog.reset_defaults"))
self.orientation_group.setTitle(t("dialog.plot_orientation_group"))
self.orientation_label.setText(t("dialog.plot_orientation"))
apps/studio/src/orisys_studio/ui/toolbar.py
+1 -2
View File
@@ -835,8 +835,7 @@ class ToolbarWidget(QWidget):
The operator's complaint was that 重置 read as "reset the charts" when it zeroes
the *sensor tracker* — and that in the replay view, where it is the display that
can be reset, it was unavailable. The label therefore names the target and the
tip states the side effect that is real but invisible: zeroing the tracker also
restarts the display baseline (the curve history is kept).
tip identifies the target so a tracker reset is not mistaken for a chart reset.
"""
if self.reset_target() == "display":
self.btn_reset.setText(t("toolbar.reset_display"))
apps/studio/tests/test_display_prefs.py
+3 -11
View File
@@ -286,28 +286,20 @@ def test_a_partial_plot_settings_mapping_leaves_the_rest_as_stored():
assert parts["charts"].deadband_floors[-1]["raw"] == 900.0
def test_the_display_baseline_is_not_a_finger_only_setting():
"""Every sensor type and both units: an idle newton offset is the same problem."""
def test_force_threshold_policy_is_available_for_each_sensor_and_unit():
for sensor, units in (("box", "newton"), ("box", "raw"),
("finger", "newton"), ("finger", "raw")):
controller, parts = _controller(prefs={"raw_auto_zero": True},
sensor_type=sensor, units=units)
controller, parts = _controller(sensor_type=sensor, units=units)
assert parts["charts"].presentation_enabled[-1] is True, (sensor, units)
assert parts["charts"].presentation[-1]["unit"] == (
"N" if units == "newton" else "raw"
)
controller, parts = _controller(prefs={"raw_auto_zero": False}, sensor_type="box",
units="newton")
assert parts["charts"].presentation_enabled[-1] is False
def test_the_deadband_floors_reach_the_charts_from_the_preferences():
controller, parts = _controller(prefs={"plot_deadband_n": 0.5, "plot_deadband_raw": 2000.0})
assert parts["charts"].deadband_floors[-1] == {"newton": 0.5, "raw": 2000.0}
controller.set_raw_auto_zero(True) # re-applies the whole policy
assert parts["charts"].deadband_floors[-1] == {"newton": 0.5, "raw": 2000.0}
assert parts["charts"].presentation_enabled[-1] is True
def test_the_sample_rate_prefers_the_measured_rate_over_the_stored_one():
apps/studio/tests/test_force_plots_display.py
+21 -199
View File
@@ -1,25 +1,12 @@
"""The force charts as the operator sees them: baseline, deadband and ranges.
The display policy is a *view* concern: an idle channel must read zero and draw a
flat line, while the session store and ``forces.csv`` keep the measurement. These
tests drive the panel on its own (no acquisition, no window) and read back what
the curves and the value cards were actually given.
"""
"""Force cards, curves, thresholds, precision and chart ranges."""
from __future__ import annotations
import numpy as np
from orisys_app_shared.force.session import SessionSample, SessionStore
from orisys_studio.ui.force_plots import ForcePlotsPanel, _nice_ticks, format_force
#: The two idle triplets from the customer's screenshots: a raw FN of ~9555 and
#: an FX whose bias sits ~13k counts off zero.
RAW_BIAS = np.array([9555.0, -13368.0, -353.0])
NEWTON_BIAS = np.array([0.4, -0.25, 0.1])
from orisys_studio.ui.force_plots import ForcePlotsPanel, _nice_ticks
def _panel(unit: str = "raw") -> ForcePlotsPanel:
"""A chart in one unit, with the display baseline switched on."""
panel = ForcePlotsPanel(theme="dark")
panel.set_units(unit)
panel.configure_presentation(sensor_type="box", unit=unit)
@@ -28,29 +15,6 @@ def _panel(unit: str = "raw") -> ForcePlotsPanel:
return panel
def _rest(
panel: ForcePlotsPanel,
bias: np.ndarray,
*,
unit: str = "raw",
samples: int = 80,
spread: float = 0.0,
sigma: float = 0.0,
seed: int = 0,
) -> None:
"""Feed a quiet startup window, the way a live session would.
``spread`` is a bounded wander (the device really does sit still) and
``sigma`` is the random jitter of a `noisier` product.
"""
rng = np.random.default_rng(seed)
for index in range(samples):
sample = bias + ((index % 5) - 2) * spread
if sigma:
sample = sample + rng.normal(0.0, sigma, size=3)
panel.push(index / 30.0, *sample, unit=unit)
def _ticks(panel: ForcePlotsPanel, channel: int) -> list[float]:
"""The Y tick values actually installed on one axis."""
levels = panel.plots[channel].getAxis("left")._tickLevels
@@ -63,174 +27,32 @@ def _curve(panel: ForcePlotsPanel, channel: int) -> np.ndarray:
return np.asarray(values, dtype=float)
# ------------------------------------------------------------ resting channel
def test_the_cards_read_zero_from_the_first_samples_of_a_session(qt_app):
"""The reported defect: FN 1118 raw, FX 34.16, FY -186.1 on the cards.
The screenshot was taken as the session started, so the aid has to be
effective from the first samples — not once the whole quiet window closed.
"""
# ------------------------------------------------------------- force output
def test_raw_threshold_and_precision_match_cards_and_curve(qt_app):
panel = _panel("raw")
for index in range(4):
panel.push(index / 30.0, *RAW_BIAS, unit="raw")
if index < 2:
# Nothing has been estimated from fewer than three samples yet; those
# first frames honestly show the measurement.
continue
assert [label.text() for label in panel.value_labels] == ["0"] * 3, index
assert not panel.presentation.ready, "the whole three second window was needed"
assert panel.presentation.display_quality == "provisional"
for channel in (0, 1, 2):
assert set(np.unique(_curve(panel, channel))) == {0.0}, channel
panel.set_deadband_floors(raw=50.0)
panel.push(0.0, 49.0, -50.0, 51.2, unit="raw")
assert [label.text() for label in panel.value_labels] == ["0", "0", "51"]
assert [_curve(panel, axis)[-1] for axis in range(3)] == [0.0, 0.0, 51.2]
panel.set_value_decimals(raw=1, newton=2)
assert panel.value_labels[2].text() == "51.2"
def test_the_newton_cards_read_zero_from_the_first_samples_too(qt_app):
"""Both units: a newton offset at rest is the same display problem."""
def test_newton_threshold_and_precision_match_cards_and_curve(qt_app):
panel = _panel("N")
for index in range(4):
panel.push(index / 30.0, *NEWTON_BIAS, unit="N")
if index < 2:
continue
assert [label.text() for label in panel.value_labels] == ["0.000"] * 3, index
assert not panel.presentation.ready
panel.set_deadband_floors(newton=0.5)
panel.push(0.0, 0.49, -0.5, 0.51, unit="N")
assert [label.text() for label in panel.value_labels] == ["0.00", "0.00", "0.51"]
assert [_curve(panel, axis)[-1] for axis in range(3)] == [0.0, 0.0, 0.51]
def test_a_card_and_its_own_curve_are_the_same_value(qt_app):
"""A card may never read a number its curve does not carry.
The extra "the estimate is ready" gate in ``_render_value_cards`` is what
used to let the card show the raw measurement while the curve was already
drawn through the applied estimate.
"""
def test_threshold_has_no_estimated_baseline(qt_app):
panel = _panel("raw")
for index in range(4):
panel.push(index / 30.0, *RAW_BIAS, unit="raw")
# A real contact, flagged as one so it does not enter the idle estimate.
panel.push(1.0, *(RAW_BIAS + [4000.0, 0.0, 0.0]), unit="raw", contact_found=True)
for channel in range(3):
drawn = float(_curve(panel, channel)[-1])
assert panel.value_labels[channel].text() == format_force(
drawn, "raw", presentation=panel.presentation
), channel
assert panel.value_labels[0].text() not in ("0", "1118"), "the contact was lost"
def test_a_resting_raw_channel_reads_zero_and_draws_a_flat_curve(qt_app):
"""FN 9555 / FX -13368 at rest must read 0 and draw a line on zero."""
panel = _panel("raw")
_rest(panel, RAW_BIAS, spread=20.0)
assert panel.presentation.ready, "the quiet window never closed"
for index, label in enumerate(panel.value_labels):
assert label.text() == "0", (index, label.text())
for channel in (0, 1, 2):
values = _curve(panel, channel)
assert values.size, channel
assert set(np.unique(values)) == {0.0}, (channel, np.unique(values))
def test_a_resting_newton_channel_is_baselined_too(qt_app):
"""The same aid in newtons: it is a display problem in either unit."""
panel = _panel("N")
_rest(panel, NEWTON_BIAS, unit="N")
assert panel.presentation.ready
assert [label.text() for label in panel.value_labels] == ["0.000"] * 3
assert set(np.unique(_curve(panel, 0))) == {0.0}
def test_noise_outside_the_estimated_band_is_still_rare(qt_app):
"""A p99 band, so a resting *noisy* channel is flat but not rigid.
The offset removal is what fixes a 13k-count bias; the percentile band holds
the remaining jitter at zero. A few idle samples may sit just outside it —
that is what the operator's deadband floor in the dialog is for.
"""
panel = _panel("raw")
_rest(panel, RAW_BIAS, sigma=30.0)
assert panel.presentation.ready
values = _curve(panel, 1)
zeroed = float(np.mean(values == 0.0))
assert zeroed >= 0.95, f"only {zeroed:.2%} of a resting channel was zeroed"
assert abs(float(np.median(values))) < 1.0
def test_the_record_keeps_the_measurement_while_the_display_is_zeroed(qt_app):
"""The deadband is a display copy: the buffer and the store see the truth."""
panel = _panel("raw")
store = SessionStore()
panel.set_store(store)
for index in range(6):
panel.push(index / 30.0, *RAW_BIAS, unit="raw")
store.append(
SessionSample(
index,
index / 30.0,
fn_raw=RAW_BIAS[0],
fx_raw=RAW_BIAS[1],
fy_raw=RAW_BIAS[2],
fn_n=NEWTON_BIAS[0],
fx_n=NEWTON_BIAS[1],
fy_n=NEWTON_BIAS[2],
valid=True,
)
)
# The very values the customer complained about are still measured, from the
# first sample on — in the live buffer...
state = panel.history.state()
assert state is not None
assert abs(float(np.median(state.values[:, 0])) - RAW_BIAS[0]) < 100.0
assert abs(float(np.median(state.values[:, 1])) - RAW_BIAS[1]) < 100.0
# ...and in the session store (which is what forces.csv is written from).
stored = store.data().raw
assert stored.shape[0] == 6
assert float(np.median(stored[:, 0])) == RAW_BIAS[0]
assert float(np.median(stored[:, 1])) == RAW_BIAS[1]
assert float(np.median(stored[:, 2])) == RAW_BIAS[2]
# ...while the cards and the curve drawn from it are already flat on zero.
assert [label.text() for label in panel.value_labels] == ["0"] * 3
assert set(np.unique(_curve(panel, 1))) == {0.0}
# ------------------------------------------------------------------ deadband
def test_a_small_excursion_is_suppressed_and_a_larger_one_is_not(qt_app):
panel = _panel("raw")
_rest(panel, RAW_BIAS, spread=20.0)
estimate = panel.presentation.last_estimate
assert estimate is not None and estimate.quality == "ok"
deadband = float(estimate.deadbands[0])
assert deadband > 0.0
panel.push(3.0, *(RAW_BIAS + [0.5 * deadband, 0.0, 0.0]), unit="raw")
assert panel.value_labels[0].text() == "0", "half the deadband was drawn"
panel.push(3.1, *(RAW_BIAS + [3.0 * deadband, 0.0, 0.0]), unit="raw")
assert panel.value_labels[0].text() != "0", (
"an excursion well outside the deadband was swallowed"
)
assert abs(_curve(panel, 0)[-1]) > 2.0 * deadband
# ...and the other axes, which stayed quiet, are still zero.
assert panel.value_labels[1].text() == "0"
def test_the_operator_floor_widens_the_deadband_at_once(qt_app):
panel = _panel("raw")
_rest(panel, RAW_BIAS, spread=20.0)
panel.set_deadband_floors(raw=250000.0)
panel.push(3.0, *(RAW_BIAS + [50000.0, 0.0, 0.0]), unit="raw")
assert panel.value_labels[0].text() == "0", (
"an excursion inside the operator's floor was drawn"
)
panel.set_deadband_floors(raw=0.0) # "auto": the layer's own floor returns
assert panel.presentation.min_deadband_for("raw") == 50.0
panel.set_deadband_floors(raw=50.0)
panel.push(0.0, 9555.0, -13368.0, 20.0, unit="raw")
assert [label.text() for label in panel.value_labels] == ["9555", "-13368", "0"]
panel.set_deadband_floors(raw=0.0)
assert panel.presentation.min_deadband_for("raw") == 0.0
# -------------------------------------------------------------------- ranges
def test_the_configured_ranges_reach_every_axis(qt_app):
apps/studio/tests/test_ui_smoke.py
+35 -41
View File
@@ -634,6 +634,8 @@ def test_unit_switch_rebuilds_from_the_session_store(make_window):
import math
window, stuck = _running_window(make_window)
window.recording.set_force_thresholds(raw=0.0, newton=0.0)
window.force_plots.set_deadband_floors(raw=0.0, newton=0.0)
window.toolbar.units_combo.setCurrentIndex(1) # raw
window.force_plots.reset()
for index in range(60):
@@ -1700,7 +1702,7 @@ def test_the_replay_view_loads_a_recording_and_scrubs_it(make_window, replay_rec
window.brand_bar.btn_playback.click()
QApplication.processEvents()
assert window._load_recording(replay_recording)
assert window.replay_ctl.load_recording(replay_recording)
QApplication.processEvents()
panel = window.force_plots
assert panel.is_recording_source(), "the charts did not switch to the recording"
@@ -1787,15 +1789,12 @@ def test_a_late_calibration_does_not_plot_raw_counts_on_the_newton_axis(make_win
# The sample that caused the switch still reaches the readouts; the early
# return used to leave the cards showing the previous sample. The number on
# the card is the *display* copy of it: this fake FN is a slow 2.0…2.4 N ramp
# with no contact flagged, which is exactly the resting drift the display
# baseline holds at zero. What matters here is that the card is not a raw
# count, not the previous sample and not a dash.
# the card uses the newton sample, with only the configured absolute threshold.
assert abs(plots._last_values[0] - (2.0 + 39 * 0.01)) < 1e-9
assert plots.presentation.display_quality != "none", (
assert plots.presentation.display_quality == "threshold", (
"the display policy was not applied to the freshly shown newton unit"
)
assert plots.value_labels[0].text() == "0.000", plots.value_labels[0].text()
assert plots.value_labels[0].text() == "2.39", plots.value_labels[0].text()
assert plots.value_labels[0].text() != t("chart.value_unavailable")
_finish(window, stuck)
@@ -1940,9 +1939,9 @@ def test_a_replay_keeps_the_2d_pane_from_the_live_camera(
window, stuck = _running_window(make_window)
window.brand_bar.btn_playback.click()
QApplication.processEvents()
assert window._load_recording(replay_recording), "the replay could not open it"
assert window.replay_ctl.load_recording(replay_recording), "the replay could not open it"
QApplication.processEvents()
assert window._replay_owns_image(), "the replay does not own the 2D pane"
assert window.replay_ctl.owns_image(), "the replay does not own the 2D pane"
label = window.arrow_panel.image_label
# Park the playhead at the end of the recording and remember what the pane and
@@ -1971,7 +1970,7 @@ def test_a_replay_keeps_the_2d_pane_from_the_live_camera(
# meaningful rather than a dead pane.
window.brand_bar.btn_live.click()
QApplication.processEvents()
assert not window._replay_owns_image()
assert not window.replay_ctl.owns_image()
stuck.emit(ui_check._fake_packet(9, mono_t=9 / 30.0, frame=live))
window._on_run_tick()
QApplication.processEvents()
@@ -1993,7 +1992,7 @@ def test_the_readout_describes_the_point_the_cursor_is_on(
window = _show(make_window(), (1440, 880))
window.brand_bar.btn_playback.click()
QApplication.processEvents()
assert window._load_recording(replay_recording)
assert window.replay_ctl.load_recording(replay_recording)
QApplication.processEvents()
plots = window.force_plots
assert plots.mode() == "browse", "opening a recording did not browse it"
@@ -2028,7 +2027,7 @@ def test_a_raw_only_recording_is_never_shown_in_newton(make_window, replay_recor
QApplication.processEvents()
window.brand_bar.btn_playback.click()
QApplication.processEvents()
assert window._load_recording(replay_recording)
assert window.replay_ctl.load_recording(replay_recording)
QApplication.processEvents()
panel = window.force_plots
@@ -2051,7 +2050,7 @@ def test_the_replay_transport_plays_and_pauses(make_window, replay_recording):
window = _show(make_window(), (1440, 880))
window.brand_bar.btn_playback.click()
QApplication.processEvents()
assert window._load_recording(replay_recording)
assert window.replay_ctl.load_recording(replay_recording)
QApplication.processEvents()
panel = window.force_plots
@@ -2091,7 +2090,7 @@ def test_the_recompute_switch_asks_for_analysis_not_an_acquire_chain(
window = _show(make_window(), (1440, 880))
window.brand_bar.btn_playback.click()
QApplication.processEvents()
assert window._load_recording(replay_recording)
assert window.replay_ctl.load_recording(replay_recording)
QApplication.processEvents()
panel = window.force_plots
assert panel.is_recording_source(), "the charts do not read the recording"
@@ -2140,7 +2139,7 @@ def test_the_replay_view_offers_a_stop_that_always_works(make_window, replay_rec
window = _show(make_window(), (1440, 880))
window.brand_bar.btn_playback.click()
QApplication.processEvents()
assert window._load_recording(replay_recording)
assert window.replay_ctl.load_recording(replay_recording)
QApplication.processEvents()
assert window.toolbar.btn_reset.isEnabled(), (
@@ -2197,7 +2196,7 @@ def test_escape_stops_the_replay_too(make_window, replay_recording):
window = _show(make_window(), (1440, 880))
window.brand_bar.btn_playback.click()
QApplication.processEvents()
assert window._load_recording(replay_recording)
assert window.replay_ctl.load_recording(replay_recording)
QApplication.processEvents()
window.toolbar.btn_run.setChecked(True)
QApplication.processEvents()
@@ -2252,7 +2251,7 @@ def test_the_replay_reset_is_the_display_reset_and_never_the_device(
window._finger_worker = stuck
window.brand_bar.btn_playback.click()
QApplication.processEvents()
assert window._load_recording(replay_recording)
assert window.replay_ctl.load_recording(replay_recording)
QApplication.processEvents()
assert window.toolbar.reset_target() == "display"
@@ -2298,9 +2297,8 @@ def test_the_replay_reset_is_the_display_reset_and_never_the_device(
window.close()
def test_the_live_reset_zeroes_the_device_and_restarts_the_display_baseline(make_window):
"""The live half of the same control: the device is asked, and the tip's side
effect (the display baseline restarting) really happens."""
def test_the_live_reset_zeroes_the_device_without_changing_threshold(make_window):
"""The device reset must not erase or re-estimate the configured threshold."""
from orisys_app_shared.i18n import t
from PySide6.QtWidgets import QApplication
@@ -2322,14 +2320,11 @@ def test_the_live_reset_zeroes_the_device_and_restarts_the_display_baseline(make
assert window.toolbar.btn_reset.isEnabled()
assert window.toolbar.btn_reset.text() == t("toolbar.reset_sensor")
assert window.toolbar.btn_reset.toolTip() == t("toolbar.reset_sensor_tip")
restarts: list = []
window.force_plots.request_presentation_reestimate = lambda: restarts.append(True)
threshold = window.force_plots.presentation.min_deadband_for("raw")
window.toolbar.btn_reset.click()
assert stuck.reset_requests == 1, "the live reset did not reach the device"
assert restarts == [True], (
"the display baseline was not restarted with the zero (the tooltip's side effect)"
)
assert window.force_plots.presentation.min_deadband_for("raw") == threshold
assert window.statusBar().currentMessage() == t("status.reset_pending"), (
window.statusBar().currentMessage()
)
@@ -2460,7 +2455,7 @@ def test_a_bare_video_plays_even_though_it_is_not_a_recording(
window.brand_bar.btn_playback.click()
QApplication.processEvents()
assert not window.replay.has_video()
assert window._load_replay_video(bare) is True, "a playable video was refused"
assert window.replay_ctl.load_video(bare) is True, "a playable video was refused"
QApplication.processEvents()
assert window.replay.has_video(), "nothing is playable"
assert window.replay.is_video_only(), "a bare video was taken for a recording"
@@ -2534,8 +2529,8 @@ def test_the_replay_and_the_live_paths_share_one_image_backend(
app_module.render_image = spy
try:
window.layout_ctl.apply_preset("replay")
assert window._load_replay_video(bare) is True
window._render_replay_frame()
assert window.replay_ctl.load_video(bare) is True
window.replay_ctl.render_frame()
QApplication.processEvents()
assert window.arrow_panel.plan_modes() == ("recorded",), (
"a recording must not offer a derived view"
@@ -2623,7 +2618,7 @@ def test_the_replay_entries_exist_only_in_the_replay_view(
assert not window.toolbar.btn_recompute.isEnabled()
# Programmatic use is refused too, and leaves nothing half-changed.
assert window._load_replay_video(bare) is False
assert window.replay_ctl.load_video(bare) is False
QApplication.processEvents()
assert not window.replay.has_video()
assert window.toolbar.source_mode() == "camera"
@@ -2637,7 +2632,7 @@ def test_the_replay_entries_exist_only_in_the_replay_view(
QApplication.processEvents()
for action in (window.act_open_video, window.act_playback, window.act_playback_latest):
assert action.isEnabled()
assert window._load_replay_video(bare) is True
assert window.replay_ctl.load_video(bare) is True
QApplication.processEvents()
assert window.arrow_panel.plan_modes() == ("recorded",)
assert window.toolbar.btn_recompute.isEnabled(), (
@@ -2670,7 +2665,7 @@ def test_a_file_that_is_not_a_video_at_all_is_refused_with_a_reason(
try:
window.brand_bar.btn_playback.click()
QApplication.processEvents()
assert window._load_replay_video(broken) is False, (
assert window.replay_ctl.load_video(broken) is False, (
"a file that cannot be decoded was accepted for playback"
)
QApplication.processEvents()
@@ -2700,7 +2695,7 @@ def test_a_video_inside_a_recording_opens_the_recording_and_plays(
QApplication.processEvents()
segments = sorted(Path(replay_recording).glob("raw_*.mp4"))
assert segments, "the fixture wrote no video segment"
assert window._load_replay_video(segments[0]) is True
assert window.replay_ctl.load_video(segments[0]) is True
QApplication.processEvents()
assert window.replay.is_loaded()
assert window.toolbar.btn_run.isEnabled(), "the transport is not offered"
@@ -2736,11 +2731,11 @@ def test_a_recording_without_a_video_says_it_cannot_be_played(make_window, recor
window = _show(make_window(), (1440, 880))
window.brand_bar.btn_playback.click()
QApplication.processEvents()
if not window._load_recording(directory):
if not window.replay_ctl.load_recording(directory):
pytest.skip("this recording layout was refused before it could be played")
QApplication.processEvents()
assert window.replay.duration_s() <= 0.0
window._start_playback()
window.replay_ctl.start_playback()
QApplication.processEvents()
assert not window._play_timer.isActive()
assert t("playback.empty_recording")[:8] in window.statusBar().currentMessage(), (
@@ -2784,7 +2779,7 @@ def test_a_selection_moves_its_statistics_onto_the_value_cards(make_window):
), summary
# The numbers a reader cannot fit in the strip are on the cards instead.
expected_peak = float(plots._selection.peaks[0])
expected = f"{expected_peak:.3f}"
expected = f"{expected_peak:.2f}"
assert plots.value_labels[0].text() == expected, (
f"the FN card is not the selection's peak: {plots.value_labels[0].text()!r}"
)
@@ -3074,9 +3069,9 @@ def test_the_plot_dialog_carries_the_ranges_and_the_zero_threshold(qt_app):
assert raw.settings()["fn_raw_min"] == 0.0
assert raw.settings()["fn_min"] == 0.0
assert raw.settings()["window_sec"] == 4.0
assert raw.deadbands() == {"deadband_n": 0.0, "deadband_raw": 0.0}, (
"the shipped deadband answer is 'auto'"
)
assert raw.deadbands() == {"deadband_n": 0.0, "deadband_raw": 50.0}
assert raw.settings()["newton_decimals"] == 2
assert raw.settings()["raw_decimals"] == 0
def test_the_plot_dialog_values_reach_the_chart(make_window):
@@ -3113,12 +3108,11 @@ def test_the_plot_dialog_values_reach_the_chart(make_window):
window.close()
def test_the_display_baseline_is_on_for_every_sensor_by_default(make_window):
"""It used to be Finger-raw only; an idle box sensor drifts just as much."""
def test_force_threshold_is_on_for_every_sensor_by_default(make_window):
window, stuck = _running_window(make_window) # sensor_type="box"
plots = window.force_plots
assert plots._presentation_enabled is True, "a box sensor lost its baseline"
assert plots._presentation_enabled is True
assert plots.presentation.enabled is True
# The per-unit floors came from the preferences, not from a sensor type.
assert plots.presentation.min_deadband_for("raw") == 50.0
shared/src/orisys_app_shared/force/__init__.py
+2 -2
View File
@@ -24,7 +24,7 @@ from orisys_app_shared.force.history import MetricHistory, PlotState, lowpass_fi
from orisys_app_shared.force.presentation import (
DisplayForce,
ForcePresentation,
ZeroEstimate,
threshold_force,
)
from orisys_app_shared.force.session import (
NA,
@@ -55,7 +55,7 @@ __all__ = [
"SessionSample",
"SessionSlice",
"SessionStore",
"ZeroEstimate",
"threshold_force",
"build_force_calibration_v2",
"describe_state",
"effective_filters",
shared/src/orisys_app_shared/force/presentation.py
+46 -525
View File
@@ -1,473 +1,76 @@
"""Display-only force presentation: baseline, deadband, and formatting.
Never mutates recorded samples. The session store, ``forces.csv`` and the metric
history keep the measurement; the value cards *and the curves* are drawn from
this layer, which removes the idle bias and holds a resting channel at exactly
zero.
The estimate is kept **per display unit**. A raw count and a newton reading of
the same axis are different scales, so one deadband in the wrong unit would
either hide a real contact or leave the noise outside it. Values arrive here
already expressed in the unit the operator is looking at (``force_plots.push``
passes the live triplet in its own unit), and the unit is part of the key.
The estimate is published in two stages, because waiting for the whole startup
window leaves the operator staring at the device's raw bias:
* **provisional** — as soon as a few quiet samples exist (see
:data:`DEFAULT_PROVISIONAL_SAMPLES`) an estimate is fitted from them and
*applied*; it is re-fitted on every further quiet sample. The cards and the
curves therefore read zero from the first tenths of a second of a session
instead of after ``baseline_seconds``.
* **frozen** — when the quiet window closes with quality ``ok`` the estimate is
frozen exactly as before and stops chasing the data.
A window that closes *without* a trustworthy estimate (contaminated, unstable,
insufficient samples) keeps the provisional estimate applied rather than
reverting to the raw bias: `quality` on :class:`ZeroEstimate` and
:attr:`ForcePresentation.display_quality` say where the numbers came from. A
window in which *no* quiet sample was ever seen has nothing to estimate from, so
nothing is invented — the measurement is displayed unchanged, which is how a
contact present at session start stays visible instead of being zeroed away.
"""
"""Deterministic per-axis force threshold and numeric presentation."""
from __future__ import annotations
from collections.abc import Sequence
from dataclasses import dataclass, field
from dataclasses import dataclass
import numpy as np
#: Floor under the estimated deadband, per display unit. 50 raw counts is a
#: meaningful noise floor for the uncalibrated device; 50 N would be a contact,
#: so newton mode defaults to no floor at all and relies on the estimated
#: percentile (see :func:`ForcePresentation.fit_zero`).
DEFAULT_MIN_DEADBAND_RAW = 50.0
DEFAULT_MIN_DEADBAND_N = 0.0
#: How many quiet samples are enough to publish a *provisional* estimate.
#: Three is the smallest window ``fit_zero`` can take a median and a residual
#: percentile from, and at the ~30 Hz force rate it is about a tenth of a
#: second: the operator sees the display policy working almost immediately
#: rather than watching the device's raw bias for the whole startup window.
#: Every further quiet sample re-fits the estimate, so three is a starting
#: point, not a decision — the frozen window still has the final say.
DEFAULT_PROVISIONAL_SAMPLES = 3
def normalise_unit(unit: str | None) -> str:
"""``"N"`` or ``"raw"`` — the two units a force reading can be shown in."""
return "N" if str(unit or "").upper().startswith("N") else "raw"
@dataclass(frozen=True)
class ZeroEstimate:
"""Outcome of an idle-window bias / noise estimate."""
def threshold_force(value: float, threshold: float) -> float:
"""Return a processed force without turning an unavailable axis into zero."""
value = float(value)
return 0.0 if np.isfinite(value) and abs(value) <= threshold else value
offsets: tuple[float, float, float]
deadbands: tuple[float, float, float]
sample_count: int
quality: str # "ok" | "insufficient" | "unstable" | "contaminated"
message: str = ""
def threshold_values(values, threshold: float) -> tuple[np.ndarray, np.ndarray]:
"""Threshold finite array elements and return values plus the changed mask."""
block = np.array(values, dtype=np.float64, copy=True)
suppressed = np.isfinite(block) & (0 < threshold) & (np.abs(block) <= threshold)
block[suppressed] = 0.0
return block, suppressed
@dataclass(frozen=True)
class DisplayForce:
"""One displayed triplet after offset + deadband (measurement unchanged)."""
values: tuple[float, float, float]
suppressed: tuple[bool, bool, bool]
unit: str
zero_zone: bool
@dataclass
class _UnitEstimate:
"""What was learned about one display unit's idle state."""
collecting: bool = False
samples: list[tuple[float, float, float]] = field(default_factory=list)
sample_times: list[float] = field(default_factory=list)
t0: float | None = None
#: Frozen estimate: set once the quiet window closes with quality ``ok``.
offsets: tuple[float, float, float] | None = None
deadbands: tuple[float, float, float] | None = None
estimate: ZeroEstimate | None = None
#: Provisional estimate, published from the first few quiet samples and
#: re-fitted on each further one. It is what the display uses while the
#: window is still open, and what it keeps if the window closes badly.
provisional_offsets: tuple[float, float, float] | None = None
provisional_deadbands: tuple[float, float, float] | None = None
provisional_samples: int = 0
#: A quiet sample was refused because a contact was present or the flow was
#: too high: this window is not a trustworthy idle window any more.
contaminated: bool = False
@property
def ready(self) -> bool:
"""A *frozen* estimate: the startup window closed with quality ``ok``."""
return self.offsets is not None and self.deadbands is not None
@property
def display_offsets(self) -> tuple[float, float, float] | None:
"""What the display is centred on: frozen if there is one, else provisional."""
return self.offsets if self.offsets is not None else self.provisional_offsets
@property
def display_deadbands(self) -> tuple[float, float, float] | None:
"""The deadband the display is using (frozen if there is one)."""
return (
self.deadbands if self.deadbands is not None else self.provisional_deadbands
)
@property
def display_ready(self) -> bool:
"""Whether *some* estimate — provisional or frozen — is being applied."""
return (
self.display_offsets is not None and self.display_deadbands is not None
)
@dataclass
class ForcePresentation:
"""Stateful display policy for the force numbers *and* the force curves.
Startup auto-baseline collects quiet samples per display unit, freezes
offsets and deadbands, then applies a symmetric hysteretic deadband to the
display copy. It is enabled for every sensor type and both units: an idle
channel that reads a raw count or a newton offset at rest is a display
problem whichever device produced it.
"""
"""One pure post-process for live values, curves, replay and export."""
enabled: bool = True
unit: str = "raw"
baseline_seconds: float = 3.0
min_samples: int = 30
#: Quiet samples needed before a provisional estimate is applied (see
#: :data:`DEFAULT_PROVISIONAL_SAMPLES`).
provisional_samples: int = DEFAULT_PROVISIONAL_SAMPLES
max_flow_p95: float = 0.4
require_no_contact: bool = True
#: Residual percentile used as the per-axis deadband after offset removal.
deadband_percentile: float = 99.0
#: Extra margin on top of the percentile residual.
deadband_margin: float = 1.05
#: Floor so tiny MAD windows do not produce a zero deadband (see above).
min_deadband_raw: float = DEFAULT_MIN_DEADBAND_RAW
min_deadband_n: float = DEFAULT_MIN_DEADBAND_N
hysteresis_ratio: float = 0.8
raw_significant_digits: int = 4
newton_decimals: int = 3
raw_decimals: int = 0
newton_decimals: int = 2
#: One idle estimate per display unit ("raw" / "N"), because the floors and
#: the values are in different scales.
_states: dict[str, _UnitEstimate] = field(
default_factory=dict, init=False, repr=False
)
_suppressed: tuple[bool, bool, bool] = field(
default=(False, False, False), init=False, repr=False
)
# ------------------------------------------------------------------ policy
def reset_session(self, *, unit: str = "raw") -> None:
"""Start a new idle estimate for one display unit."""
self.unit = normalise_unit(unit)
state = self._state(self.unit)
state.collecting = bool(self.enabled)
state.samples = []
state.sample_times = []
state.t0 = None
state.offsets = None
state.deadbands = None
state.estimate = None
state.provisional_offsets = None
state.provisional_deadbands = None
state.provisional_samples = 0
state.contaminated = False
self._suppressed = (False, False, False)
def request_reestimate(self) -> None:
"""Operator-triggered restart of the idle window."""
self.reset_session(unit=self.unit)
def set_enabled(self, enabled: bool, *, unit: str | None = None) -> None:
"""The master switch; switching it on restarts the quiet window."""
self.enabled = bool(enabled)
self.reset_session(unit=self.unit if unit is None else unit)
def min_deadband_for(self, unit: str | None = None) -> float:
"""The floor under the estimated deadband, for one display unit."""
chosen = self.unit if unit is None else unit
return (
float(self.min_deadband_n)
if normalise_unit(chosen) == "N"
else float(self.min_deadband_raw)
)
def set_min_deadband(self, unit: str, floor: float) -> None:
"""Move one unit's floor, and re-floor the estimate already applied.
The floor is a lower bound, never a replacement: the samples collected so
far are re-fitted, so setting it back to 0 ("auto") restores the
estimated band instead of leaving the raised one behind. Both the frozen
estimate and the provisional one in use are re-fitted, because either can
be what the cards and the curves are reading.
"""
floor = max(0.0, float(floor))
if normalise_unit(unit) == "N":
self.min_deadband_n = floor
else:
self.min_deadband_raw = floor
state = self._states.get(normalise_unit(unit))
if state is None or not state.samples:
return # a window still being collected will use the new floor
estimate = self.fit_zero(state.samples, unit=unit)
if estimate.quality != "ok":
return
if state.collecting or not state.ready:
# Nothing frozen yet: this is the provisional estimate on screen.
state.provisional_offsets = estimate.offsets
state.provisional_deadbands = estimate.deadbands
state.provisional_samples = estimate.sample_count
return
state.offsets = estimate.offsets
state.deadbands = estimate.deadbands
state.estimate = estimate
@property
def ready(self) -> bool:
"""A frozen estimate for the active unit (the startup window closed)."""
state = self._states.get(self.unit)
return bool(state is not None and state.ready)
@property
def display_ready(self) -> bool:
"""Whether the display policy is being applied to the active unit.
True from the first few quiet samples (provisional) and after the frozen
window as well — this is what the cards and the curves ask, because both
are drawn from the same applied estimate.
"""
state = self._states.get(self.unit)
return bool(state is not None and state.display_ready)
if unit is not None:
self.unit = normalise_unit(unit)
@property
def display_quality(self) -> str:
"""Where the numbers on screen come from, for diagnostics and reporting.
return "threshold" if self.enabled else "off"
``"off"`` — the aid is switched off; the measurement is shown.
``"provisional"`` — a provisional estimate from the first quiet samples
is applied; the startup window is still open.
``"ok"`` — the full quiet window closed and was frozen.
``"contaminated"`` / ``"unstable"`` / ``"insufficient"`` — the window
closed without a trustworthy estimate: the
provisional estimate stays applied (this is the
state that used to fall back to the raw bias).
``"none"`` — no quiet sample was ever seen, so nothing is
zeroed; a contact at session start is shown as it
is measured rather than hidden.
"""
if not self.enabled:
return "off"
state = self._states.get(self.unit)
if state is None or not state.display_ready:
return "none"
if state.ready:
return "ok"
if not state.collecting and state.estimate is not None:
return state.estimate.quality
return "provisional"
def min_deadband_for(self, unit: str | None = None) -> float:
chosen = normalise_unit(self.unit if unit is None else unit)
return self.min_deadband_n if chosen == "N" else self.min_deadband_raw
@property
def last_estimate(self) -> ZeroEstimate | None:
"""The window's outcome: the frozen estimate, the failed one, or the
provisional estimate while the window is still open."""
state = self._states.get(self.unit)
if state is None:
return None
if state.estimate is not None:
return state.estimate
if state.provisional_offsets is None or state.provisional_deadbands is None:
return None
return ZeroEstimate(
offsets=state.provisional_offsets,
deadbands=state.provisional_deadbands,
sample_count=state.provisional_samples,
quality="provisional",
message=f"provisional: {state.provisional_samples} quiet samples so far",
)
def set_min_deadband(self, unit: str, threshold: float) -> None:
value = max(0.0, float(threshold))
if normalise_unit(unit) == "N":
self.min_deadband_n = value
else:
self.min_deadband_raw = value
def _state(self, unit: str) -> _UnitEstimate:
"""The estimate for one unit, created on first use.
A unit that was never estimated — the operator switched mid-session —
starts its own quiet window instead of borrowing the other unit's, which
would be an offset in the wrong scale.
"""
key = normalise_unit(unit)
state = self._states.get(key)
if state is None:
state = _UnitEstimate(collecting=bool(self.enabled))
self._states[key] = state
return state
# --------------------------------------------------------------- estimator
def observe(
self,
values: Sequence[float],
*,
t_rel_s: float,
valid: bool = True,
flow_p95: float | None = None,
contact_found: bool = False,
unit: str | None = None,
) -> ZeroEstimate | None:
"""Feed one live sample into the startup estimator.
``values`` must already be expressed in ``unit`` (the displayed one).
Returns a :class:`ZeroEstimate` when the window closes (ok or failed).
"""
key = normalise_unit(self.unit if unit is None else unit)
state = self._state(key)
if not state.collecting:
return None
if state.t0 is None:
state.t0 = float(t_rel_s)
if not valid:
return None
if self.require_no_contact and contact_found:
# Contaminated window — the sample is not a quiet one, so it must not
# enter the estimate (that is how a press at session start stays
# visible). The window is marked so a close can report it honestly.
state.contaminated = True
return None
if flow_p95 is not None and float(flow_p95) > self.max_flow_p95:
state.contaminated = True
return None
sample = tuple(float(v) for v in values[:3])
if len(sample) < 3 or not all(np.isfinite(sample)):
return None
state.samples.append(sample) # type: ignore[arg-type]
state.sample_times.append(float(t_rel_s))
# Publish or refine the provisional estimate *before* the window is
# allowed to close, so the display policy is effective from the first
# samples and a later contamination cannot drop it back to the raw bias.
self._update_provisional(state, key)
elapsed = float(t_rel_s) - float(state.t0)
if elapsed < self.baseline_seconds and len(state.samples) < self.min_samples * 2:
return None
if len(state.samples) < self.min_samples:
# The window closed before it was ever a full quiet window. It is a
# *failure* of the startup estimate, and the honest thing to do is
# say so — but the display keeps whatever the provisional estimate
# already worked out, because falling back to the raw bias is the
# defect this class exists to remove.
quality = "contaminated" if state.contaminated else "insufficient"
detail = (
"window contaminated (contact or flow) after"
if state.contaminated
else "only"
)
estimate = ZeroEstimate(
offsets=(0.0, 0.0, 0.0),
deadbands=(0.0, 0.0, 0.0),
sample_count=len(state.samples),
quality=quality,
message=f"{detail} {len(state.samples)} quiet samples in {elapsed:.2f}s",
)
state.estimate = estimate
state.collecting = False
return estimate
estimate = self.fit_zero(state.samples, unit=key)
state.estimate = estimate
state.collecting = False
if estimate.quality == "ok":
state.offsets = estimate.offsets
state.deadbands = estimate.deadbands
return estimate
def _update_provisional(self, state: _UnitEstimate, unit: str) -> None:
"""Fit and apply the display estimate from the quiet samples so far.
Called on every accepted quiet sample. The estimate is only published
once there are enough samples to take a median (:data:`DEFAULT_PROVISIONAL_SAMPLES`);
from then on each sample sharpens it until the frozen window closes. A
sample set the fitter rejects (not enough finite rows, wildly unstable)
leaves the previous provisional estimate in place rather than dropping
the display back to the raw bias.
"""
if len(state.samples) < max(3, int(self.provisional_samples)):
return
estimate = self.fit_zero(state.samples, unit=unit)
if estimate.quality != "ok":
return
state.provisional_offsets = estimate.offsets
state.provisional_deadbands = estimate.deadbands
state.provisional_samples = estimate.sample_count
def fit_zero(
self, samples: Sequence[Sequence[float]], *, unit: str | None = None
) -> ZeroEstimate:
"""Estimate offsets and deadbands from quiet samples (does not apply them)."""
floor = self.min_deadband_for(unit)
block = np.asarray(samples, dtype=np.float64)
if block.ndim != 2 or block.shape[1] < 3 or block.shape[0] < 3:
return ZeroEstimate(
offsets=(0.0, 0.0, 0.0),
deadbands=(0.0, 0.0, 0.0),
sample_count=int(block.shape[0]) if block.ndim == 2 else 0,
quality="insufficient",
message="not enough samples to estimate a baseline",
)
offsets: list[float] = []
deadbands: list[float] = []
for axis in range(3):
col = block[:, axis]
finite = col[np.isfinite(col)]
if finite.size < 3:
return ZeroEstimate(
offsets=(0.0, 0.0, 0.0),
deadbands=(0.0, 0.0, 0.0),
sample_count=int(block.shape[0]),
quality="insufficient",
message=f"axis {axis} lacks finite samples",
)
med = float(np.median(finite))
residual = np.abs(finite - med)
p = float(np.percentile(residual, self.deadband_percentile))
db = max(floor, p * self.deadband_margin)
# Reject wildly unstable windows (p99 >> median magnitude scale).
mad = float(np.median(residual))
if mad > 0 and p > 20.0 * mad and p > 2000.0:
return ZeroEstimate(
offsets=(0.0, 0.0, 0.0),
deadbands=(0.0, 0.0, 0.0),
sample_count=int(block.shape[0]),
quality="unstable",
message=f"axis {axis} residual p{self.deadband_percentile:g}={p:.1f}",
)
offsets.append(med)
deadbands.append(db)
return ZeroEstimate(
offsets=(offsets[0], offsets[1], offsets[2]),
deadbands=(deadbands[0], deadbands[1], deadbands[2]),
sample_count=int(block.shape[0]),
quality="ok",
)
def apply_estimate(
self, estimate: ZeroEstimate, *, unit: str | None = None
) -> None:
"""Explicitly freeze an estimate (e.g. after user confirmation)."""
if estimate.quality != "ok":
return
state = self._state(self.unit if unit is None else unit)
state.offsets = estimate.offsets
state.deadbands = estimate.deadbands
state.collecting = False
state.estimate = estimate
# ------------------------------------------------------------------ display
def accept(
self,
values: Sequence[float],
@@ -475,116 +78,34 @@ class ForcePresentation:
unit: str | None = None,
available: tuple[bool, bool, bool] = (True, True, True),
) -> DisplayForce:
"""Return a display copy of one live sample; never mutates ``values``.
The deadband is hysteretic: an axis already held at zero stays there
until it leaves the full band, so a reading on the edge does not flicker
between a number and zero.
The offset it subtracts is the frozen estimate once the startup window
has closed cleanly, and the provisional one before (or instead of) that
— never "nothing", which is what used to leave the raw bias on the cards
for the first seconds of a session.
"""
unit = normalise_unit(self.unit if unit is None else unit)
raw = [float(v) for v in values[:3]]
while len(raw) < 3:
raw.append(float("nan"))
state = self._states.get(unit)
if not self.enabled or state is None or not state.display_ready:
return DisplayForce(
values=(raw[0], raw[1], raw[2]),
suppressed=(False, False, False),
unit=unit,
zero_zone=False,
)
out: list[float] = []
suppressed: list[bool] = []
prev = self._suppressed
for i in range(3):
if not available[i] or not np.isfinite(raw[i]):
out.append(raw[i])
suppressed.append(False)
continue
centered = raw[i] - state.display_offsets[i] # type: ignore[index]
on = state.display_deadbands[i] # type: ignore[index]
off = on * self.hysteresis_ratio
was = prev[i]
if was:
quiet = abs(centered) <= on
else:
quiet = abs(centered) <= off
if quiet:
out.append(0.0)
suppressed.append(True)
else:
out.append(centered)
suppressed.append(False)
self._suppressed = (suppressed[0], suppressed[1], suppressed[2])
return DisplayForce(
values=(out[0], out[1], out[2]),
suppressed=self._suppressed,
unit=unit,
zero_zone=any(self._suppressed),
chosen = normalise_unit(self.unit if unit is None else unit)
raw = tuple(float(values[i]) if i < len(values) else float("nan") for i in range(3))
processed, mask = self.present([raw], unit=chosen)
result = tuple(
float(processed[0, i]) if available[i] else raw[i] for i in range(3)
)
suppressed = tuple(bool(mask[0, i] and available[i]) for i in range(3))
return DisplayForce(result, suppressed, chosen, any(suppressed))
def present(
self, values, *, unit: str | None = None
) -> tuple[np.ndarray, np.ndarray]:
"""Present a whole block of readings: ``(display copy, suppressed mask)``.
One row per sample, for the curves — the live buffer draws a rolling
window, not one sample at a time. Vectorised and *stateless*: the
hysteresis latch in :meth:`accept` belongs to the live sample stream, and
a curve redrawn from a window must not depend on which frame drew it.
Non-finite readings stay non-finite — an axis with no value in this unit,
or a failed readout, is a gap and never a zero.
It reads the same applied estimate as :meth:`accept` (provisional while
the startup window is open), so a card and its own curve can never
disagree about what the same sample is worth.
"""
unit = normalise_unit(self.unit if unit is None else unit)
"""Apply the same absolute threshold independently to each finite axis."""
block = np.array(values, dtype=np.float64, copy=True, ndmin=2)[:, :3]
if block.shape[1] < 3:
pad = np.full((block.shape[0], 3 - block.shape[1]), np.nan)
block = np.hstack([block, pad])
block = np.pad(
block, ((0, 0), (0, 3 - block.shape[1])), constant_values=np.nan
)
suppressed = np.zeros(block.shape, dtype=bool)
state = self._states.get(unit)
if not self.enabled or state is None or not state.display_ready:
return block, suppressed
offsets = state.display_offsets
deadbands = state.display_deadbands
for axis in range(3):
column = block[:, axis]
finite = np.isfinite(column)
if not finite.any():
continue
centered = column[finite] - offsets[axis] # type: ignore[index]
quiet = np.abs(centered) <= deadbands[axis] # type: ignore[index]
column[finite] = np.where(quiet, 0.0, centered)
suppressed[finite, axis] = quiet
if self.enabled:
threshold = self.min_deadband_for(unit)
block, suppressed = threshold_values(block, threshold)
return block, suppressed
def format_force(self, value: float, unit: str | None = None) -> str:
"""Stable digit formatting for value cards."""
unit = normalise_unit(self.unit if unit is None else unit)
if not np.isfinite(value):
return "—"
# Avoid signed zero after deadband.
if value == 0.0 or value == -0.0:
value = 0.0
if unit == "N":
return f"{value:.{self.newton_decimals}f}"
digits = max(2, int(self.raw_significant_digits))
abs_v = abs(value)
if abs_v >= 1000:
return f"{value:.0f}"
if abs_v >= 100:
return f"{value:.1f}"
if abs_v >= 10:
return f"{value:.2f}"
# Significant figures for small residual readings.
return f"{value:.{digits}g}"
chosen = normalise_unit(self.unit if unit is None else unit)
decimals = self.newton_decimals if chosen == "N" else self.raw_decimals
decimals = max(0, int(decimals))
return f"{0.0 if round(value, decimals) == 0 else value:.{decimals}f}"
shared/src/orisys_app_shared/force/session.py
+54 -11
View File
@@ -35,7 +35,7 @@ from orisys_app_shared.record.sample import plottable_axis_mask
NA = float("nan")
DEFAULT_BLOCK_SIZE = 8192
#: ~4.5 h at 30 Hz of three raw + three calibrated float64 channels.
#: ~4.5 h at 30 Hz; each sample keeps original and processed force channels.
DEFAULT_MAX_SAMPLES = 500_000
# Column layout of the packed store.
@@ -48,7 +48,13 @@ _COL_FN_N = 5
_COL_FX_N = 6
_COL_FY_N = 7
_COL_VALID = 8
_COLUMNS = 9
_COL_FN_OUTPUT_RAW = 9
_COL_FX_OUTPUT_RAW = 10
_COL_FY_OUTPUT_RAW = 11
_COL_FN_OUTPUT_N = 12
_COL_FX_OUTPUT_N = 13
_COL_FY_OUTPUT_N = 14
_COLUMNS = 15
@dataclass(frozen=True)
@@ -69,6 +75,12 @@ class SessionSample:
#: Axes that actually have a calibration curve. An axis outside this set has
#: no newton value, which the UI shows as a dash rather than as a number.
calibrated_axes: tuple[str, ...] = ()
fn_raw_processed: float = NA
fx_raw_processed: float = NA
fy_raw_processed: float = NA
fn_n_processed: float = NA
fx_n_processed: float = NA
fy_n_processed: float = NA
@classmethod
def from_record(cls, record) -> SessionSample:
@@ -87,6 +99,12 @@ class SessionSample:
fn_n=float(record.fn_n),
fx_n=float(record.fx_n),
fy_n=float(record.fy_n),
fn_raw_processed=float(record.fn_raw_processed),
fx_raw_processed=float(record.fx_raw_processed),
fy_raw_processed=float(record.fy_raw_processed),
fn_n_processed=float(record.fn_n_processed),
fx_n_processed=float(record.fx_n_processed),
fy_n_processed=float(record.fy_n_processed),
valid=bool(record.valid),
calibration_id=str(record.calibration_version or ""),
error=str(record.invalid_reason or ""),
@@ -116,6 +134,8 @@ class SessionSlice:
valid: np.ndarray
frame_ids: np.ndarray
truncated: bool = False
output_raw: np.ndarray | None = None
output_newtons: np.ndarray | None = None
def __len__(self) -> int:
return int(self.times.size)
@@ -126,6 +146,11 @@ class SessionSlice:
return self.newtons
return self.raw
def output_for_unit(self, unit: str) -> np.ndarray:
if str(unit).upper().startswith("N"):
return self.output_newtons if self.output_newtons is not None else self.newtons
return self.output_raw if self.output_raw is not None else self.raw
def plottable(self, unit: str) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
"""``(times, values, mask)`` for drawing, under one shared rule.
@@ -156,7 +181,9 @@ class SelectionStats:
stds: tuple[float, float, float] = (NA, NA, NA)
def _decimate_preserving_peaks(rows: np.ndarray, max_points: int, unit: str = "raw") -> np.ndarray:
def _decimate_preserving_peaks(
rows: np.ndarray, max_points: int, unit: str = "raw", *, processed: bool = False
) -> np.ndarray:
"""Reduce ``rows`` to about ``max_points`` while keeping every spike.
Each bucket contributes the extreme sample of **every plotted axis** in the
@@ -167,10 +194,15 @@ def _decimate_preserving_peaks(rows: np.ndarray, max_points: int, unit: str = "r
total = rows.shape[0]
if max_points < 4 or total <= max_points:
return rows
if str(unit).upper().startswith("N"):
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)
(_COL_FN_OUTPUT_N, _COL_FX_OUTPUT_N, _COL_FY_OUTPUT_N)
if processed else (_COL_FN_N, _COL_FX_N, _COL_FY_N)
)
else:
columns = (
(_COL_FN_OUTPUT_RAW, _COL_FX_OUTPUT_RAW, _COL_FY_OUTPUT_RAW)
if processed else (_COL_FN_RAW, _COL_FX_RAW, _COL_FY_RAW)
)
bucket = max(2, total // max(2, max_points // (2 * len(columns))))
usable = (total // bucket) * bucket
@@ -261,6 +293,12 @@ class SessionStore:
float(sample.fx_n),
float(sample.fy_n),
valid,
float(sample.fn_raw_processed) if np.isfinite(sample.fn_raw_processed) else float(sample.fn_raw),
float(sample.fx_raw_processed) if np.isfinite(sample.fx_raw_processed) else float(sample.fx_raw),
float(sample.fy_raw_processed) if np.isfinite(sample.fy_raw_processed) else float(sample.fy_raw),
float(sample.fn_n_processed) if np.isfinite(sample.fn_n_processed) else float(sample.fn_n),
float(sample.fx_n_processed) if np.isfinite(sample.fx_n_processed) else float(sample.fx_n),
float(sample.fy_n_processed) if np.isfinite(sample.fy_n_processed) else float(sample.fy_n),
)
self._write_row(row)
self._last_t = float(sample.t_rel_s)
@@ -378,7 +416,8 @@ class SessionStore:
return (float(block[0, _COL_T]), float(block[rows - 1, _COL_T]))
def visible(
self, t0: float, t1: float, max_points: int = 2000, *, unit: str = "raw"
self, t0: float, t1: float, max_points: int = 2000, *,
unit: str = "raw", processed: bool = False,
) -> SessionSlice:
"""Decimated slice for drawing, keeping peaks on every plotted axis.
@@ -395,7 +434,7 @@ class SessionStore:
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, unit)
rows = _decimate_preserving_peaks(rows, max_points, unit, processed=processed)
return replace(self._slice_from(rows), truncated=truncated)
def _packed_window(self, t0: float, t1: float) -> np.ndarray:
@@ -421,7 +460,9 @@ class SessionStore:
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:
def statistics(
self, t0: float, t1: float, unit: str = "raw", *, processed: bool = False
) -> SelectionStats:
"""Duration, peak, mean and std over a selection, per channel."""
t0 = float(t0)
t1 = float(t1)
@@ -438,9 +479,9 @@ class SessionStore:
if valid_rows.shape[0] == 0:
return SelectionStats(duration_s=duration, samples=0)
columns = (
(_COL_FN_N, _COL_FX_N, _COL_FY_N)
((_COL_FN_OUTPUT_N, _COL_FX_OUTPUT_N, _COL_FY_OUTPUT_N) if processed else (_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)
else ((_COL_FN_OUTPUT_RAW, _COL_FX_OUTPUT_RAW, _COL_FY_OUTPUT_RAW) if processed else (_COL_FN_RAW, _COL_FX_RAW, _COL_FY_RAW))
)
peaks: list[float] = []
means: list[float] = []
@@ -478,6 +519,8 @@ class SessionStore:
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(),
output_raw=packed[:, [_COL_FN_OUTPUT_RAW, _COL_FX_OUTPUT_RAW, _COL_FY_OUTPUT_RAW]].copy(),
output_newtons=packed[:, [_COL_FN_OUTPUT_N, _COL_FX_OUTPUT_N, _COL_FY_OUTPUT_N]].copy(),
valid=packed[:, _COL_VALID] > 0.5,
frame_ids=packed[:, _COL_FRAME_ID].astype(np.int64, copy=True),
truncated=self._dropped > 0,
shared/src/orisys_app_shared/force/source.py
+3 -3
View File
@@ -123,7 +123,7 @@ class StoreCurveSource:
frame_ids=np.empty(0, dtype=np.int64),
)
return self.store.visible(
t0, t1, max_points=max_points or 0, unit=unit
t0, t1, max_points=max_points or 0, unit=unit, processed=True
)
def window(
@@ -137,7 +137,7 @@ class StoreCurveSource:
slice_ = self._slice(t0, t1, max_points, unit)
return SourceWindow(
times=slice_.times,
values=slice_.value_for_unit(unit),
values=slice_.output_for_unit(unit),
valid=slice_.valid,
frame_ids=slice_.frame_ids,
unit=unit,
@@ -154,7 +154,7 @@ class StoreCurveSource:
def statistics(self, t0: float, t1: float, *, unit: str = "raw") -> SourceStats:
if self.store is None:
return SourceStats()
stats = self.store.statistics(t0, t1, unit=unit)
stats = self.store.statistics(t0, t1, unit=unit, processed=True)
return SourceStats(
duration_s=stats.duration_s,
samples=stats.samples,
shared/src/orisys_app_shared/force/units.py
+3 -2
View File
@@ -63,8 +63,9 @@ def select_display_forces(*, units_pref: str, sample) -> DisplayForces:
show: there is no second interpretation of what a value means.
"""
want_newton = str(units_pref or "newton").lower().startswith("n")
raw = tuple(sample.raw_for(axis) for axis in AXES)
newton = tuple(sample.newton_for(axis) for axis in AXES)
processed = getattr(sample, "processed_for", None)
raw = tuple(processed(axis, "raw") if callable(processed) else sample.raw_for(axis) for axis in AXES)
newton = tuple(processed(axis, "N") if callable(processed) else sample.newton_for(axis) for axis in AXES)
requested = "N" if want_newton else "raw"
if not want_newton:
shared/src/orisys_app_shared/prefs/store.py
+4 -5
View File
@@ -56,11 +56,11 @@ _DEFAULTS: dict[str, Any] = {
"plot_fn_raw_min": 0.0,
"plot_fn_raw_max": 1000000.0,
"plot_shear_raw_max": 500000.0,
# Display deadband floors, per display unit. 0 = auto: the noise estimated
# from the quiet startup window (raw keeps its 50-count floor under it).
# Display only — the session store and forces.csv keep the measurement.
# Absolute force-output thresholds. Original measurements are retained.
"plot_deadband_n": 0.0,
"plot_deadband_raw": 0.0,
"plot_deadband_raw": 50.0,
"plot_newton_decimals": 2,
"plot_raw_decimals": 0,
"image_view_mode": "arrows_pure",
"arrow_rotation_deg": 90,
#: How the finger's 2D canvas is presented. A display convention (the unrolled
@@ -69,7 +69,6 @@ _DEFAULTS: dict[str, Any] = {
#: the stored flow and mask are never reshaped.
"finger_2d_orientation": "rotate180",
"show_fps": False,
"raw_auto_zero": True,
#: The GL profile the 3D view is started with. One per process: the surface
#: format is chosen before ``QApplication`` exists and cannot be switched later.
"gl_profile": "desktop33",
shared/src/orisys_app_shared/record/experiment.py
+15 -2
View File
@@ -46,7 +46,7 @@ from orisys_app_shared.record.video_writer import VideoSegments
logger = logging.getLogger(__name__)
RECORD_FORMAT_VERSION = "1.1"
RECORD_FORMAT_VERSION = "1.2"
# Lifecycle states. ``failed`` and ``stopped`` are terminal; a recorder that has
# not reached a terminal state must not be reused.
@@ -71,6 +71,12 @@ FORCES_HEADER = (
"FY_N",
"valid",
"calibration_id",
"FN_output_raw",
"FX_output_raw",
"FY_output_raw",
"FN_output_N",
"FX_output_N",
"FY_output_N",
)
FRAMES_HEADER = (
"frame_id",
@@ -513,7 +519,7 @@ class ExperimentRecorder:
self._enqueue(("event", self._next_seq(), event))
# ----------------------------------------------------------- sink interface
def on_packet(self, packet) -> None:
def on_packet(self, packet, *, sample: RecordSample | None = None) -> None:
"""Called on the acquisition thread: enqueue, never do I/O here.
A data row is small and a frame is large, so video is what gives way —
@@ -532,6 +538,7 @@ class ExperimentRecorder:
if self._state == STATE_FAILED:
self._stats.rejected_after_failure += 1
return
if sample is None:
sample = to_record_sample(
packet, calibration_version=self._calibration_version
)
@@ -1107,6 +1114,12 @@ class _SessionWriter:
format_value(sample.fy_n),
1 if sample.valid else 0,
sample.calibration_version or self.calibration_id,
format_value(sample.fn_raw_processed),
format_value(sample.fx_raw_processed),
format_value(sample.fy_raw_processed),
format_value(sample.fn_n_processed),
format_value(sample.fx_n_processed),
format_value(sample.fy_n_processed),
)
)
shared/src/orisys_app_shared/record/index.py
+10 -1
View File
@@ -35,6 +35,10 @@ CHECKPOINT_ROWS = 4096
DEFAULT_BUCKETS = 1200
_FORCE_COLUMNS = ("FN_raw", "FX_raw", "FY_raw", "FN_N", "FX_N", "FY_N")
_OUTPUT_COLUMNS = (
"FN_output_raw", "FX_output_raw", "FY_output_raw",
"FN_output_N", "FX_output_N", "FY_output_N",
)
def _parse(text: str) -> float:
@@ -235,7 +239,12 @@ class RecordingIndex:
timestamp = float(cell("t_rel_s"))
if not np.isfinite(timestamp) or not -(2**63) <= frame_id < 2**63:
return None
values = tuple(_parse(cell(name)) for name in _FORCE_COLUMNS)
columns = (
_OUTPUT_COLUMNS
if all(name in self._columns for name in _OUTPUT_COLUMNS)
else _FORCE_COLUMNS
)
values = tuple(_parse(cell(name)) for name in columns)
return frame_id, timestamp, values[:3], values[3:], cell("valid").strip() == "1"
except (KeyError, IndexError, ValueError, StopIteration, csv.Error):
return None
shared/src/orisys_app_shared/record/sample.py
+40 -1
View File
@@ -26,7 +26,7 @@ from __future__ import annotations
import math
from collections.abc import Iterable
from dataclasses import dataclass
from dataclasses import dataclass, replace
from typing import TYPE_CHECKING
if TYPE_CHECKING: # numpy is imported lazily, by the one function that needs it
@@ -68,6 +68,12 @@ class RecordSample:
fy_n: float = float("nan")
calibrated_axes: tuple[str, ...] = ()
calibration_version: str = ""
fn_raw_processed: float = float("nan")
fx_raw_processed: float = float("nan")
fy_raw_processed: float = float("nan")
fn_n_processed: float = float("nan")
fx_n_processed: float = float("nan")
fy_n_processed: float = float("nan")
# ------------------------------------------------------------- accessors
def raw_triplet(self) -> tuple[float, float, float]:
@@ -76,6 +82,14 @@ class RecordSample:
def newton_triplet(self) -> tuple[float, float, float]:
return (self.fn_n, self.fx_n, self.fy_n)
def processed_triplet(self, unit: str) -> tuple[float, float, float]:
if str(unit).upper().startswith("N"):
return (self.fn_n_processed, self.fx_n_processed, self.fy_n_processed)
return (self.fn_raw_processed, self.fx_raw_processed, self.fy_raw_processed)
def processed_for(self, axis: str, unit: str) -> float:
return dict(zip(AXES, self.processed_triplet(unit))).get(axis, float("nan"))
def raw_for(self, axis: str) -> float:
return {"fn": self.fn_raw, "fx": self.fx_raw, "fy": self.fy_raw}.get(
axis, float("nan")
@@ -203,11 +217,36 @@ def to_record_sample(packet, *, calibration_version: str = "") -> RecordSample:
fn_n=newton[0],
fx_n=newton[1],
fy_n=newton[2],
fn_raw_processed=raw[0],
fx_raw_processed=raw[1],
fy_raw_processed=raw[2],
fn_n_processed=newton[0],
fx_n_processed=newton[1],
fy_n_processed=newton[2],
calibrated_axes=axes,
calibration_version=version,
)
def process_record_sample(
sample: RecordSample, *, raw_threshold: float, newton_threshold: float
) -> RecordSample:
"""Keep acquired values and attach the same per-axis output used by Studio."""
from orisys_app_shared.force.presentation import threshold_force
raw = tuple(threshold_force(v, raw_threshold) for v in sample.raw_triplet())
newton = tuple(threshold_force(v, newton_threshold) for v in sample.newton_triplet())
return replace(
sample,
fn_raw_processed=raw[0],
fx_raw_processed=raw[1],
fy_raw_processed=raw[2],
fn_n_processed=newton[0],
fx_n_processed=newton[1],
fy_n_processed=newton[2],
)
def axes_without_newton(sample: RecordSample) -> tuple[str, ...]:
"""Axes that produced no newton value — used by the UI to explain blanks."""
return tuple(axis for axis in AXES if not sample.has_newton(axis))
shared/tests/test_force_presentation.py
+26 -385
View File
@@ -1,394 +1,35 @@
"""Display-only force baseline / deadband (value cards and curves)."""
from __future__ import annotations
"""Absolute force thresholds preserve unavailable axes and original values."""
import numpy as np
from orisys_app_shared.force.presentation import ForcePresentation, ZeroEstimate
#: The idle triplet from the customer's screenshot at run start: the numbers the
#: cards were showing because the estimate was not applied yet.
REPORTED_RAW_BIAS = (1118.0, 34.16, -186.1)
REPORTED_NEWTON_BIAS = (0.4, -0.25, 0.1)
from orisys_app_shared.force.presentation import ForcePresentation, threshold_force
def test_fit_zero_rejects_insufficient_samples() -> None:
presentation = ForcePresentation(enabled=True, min_samples=30)
estimate = presentation.fit_zero([(1.0, 2.0, 3.0), (1.1, 2.1, 3.1)])
assert estimate.quality == "insufficient"
def test_threshold_is_absolute_and_independent_per_axis():
policy = ForcePresentation(min_deadband_raw=50.0, min_deadband_n=0.5)
displayed = policy.accept((49.0, -50.0, 51.0), unit="raw")
assert displayed.values == (0.0, 0.0, 51.0)
assert displayed.suppressed == (True, True, False)
newtons = policy.accept((0.4, -0.6, float("nan")), unit="N")
assert newtons.values[:2] == (0.0, -0.6)
assert np.isnan(newtons.values[2])
def test_fit_zero_and_accept_deadband() -> None:
rng = np.random.default_rng(0)
bias = np.array([1262.0, -11.0, -353.0])
noise = rng.normal(0.0, 20.0, size=(80, 3))
samples = (bias + noise).tolist()
presentation = ForcePresentation(
enabled=True,
min_samples=30,
baseline_seconds=0.0,
min_deadband_raw=50.0,
deadband_percentile=99.0,
)
estimate = presentation.fit_zero(samples)
assert estimate.quality == "ok"
presentation.apply_estimate(estimate)
idle = presentation.accept(tuple(bias), unit="raw")
assert idle.zero_zone
assert idle.values == (0.0, 0.0, 0.0)
# Curves / CSV keep the measurement: accept never mutates the input tuple.
measured = (1262.0, -11.0, -353.0)
presentation.accept(measured, unit="raw")
assert measured == (1262.0, -11.0, -353.0)
def test_scalar_and_array_paths_agree_without_mutating_input():
source = np.array([[49.0, -51.0, np.nan]])
policy = ForcePresentation(min_deadband_raw=50.0)
output, mask = policy.present(source, unit="raw")
assert output[0, :2].tolist() == [0.0, -51.0]
assert mask.tolist() == [[True, False, False]]
assert source[0, 0] == 49.0
assert threshold_force(49.0, 50.0) == output[0, 0]
def test_observe_skips_contact_contamination() -> None:
presentation = ForcePresentation(
enabled=True,
min_samples=5,
baseline_seconds=0.1,
require_no_contact=True,
)
for i in range(10):
out = presentation.observe(
(1000.0, 0.0, 0.0),
t_rel_s=i * 0.05,
contact_found=True,
flow_p95=0.1,
)
assert out is None
assert not presentation.ready
def test_observe_freezes_after_quiet_window() -> None:
presentation = ForcePresentation(
enabled=True,
min_samples=10,
baseline_seconds=0.5,
min_deadband_raw=10.0,
)
estimate: ZeroEstimate | None = None
for i in range(40):
estimate = presentation.observe(
(100.0 + (i % 3), -5.0, 2.0),
t_rel_s=i * 0.02,
flow_p95=0.1,
contact_found=False,
)
if estimate is not None:
break
assert estimate is not None
assert estimate.quality == "ok"
assert presentation.ready
# Further observes do not chase a new zero once frozen.
later = presentation.observe((5000.0, 0.0, 0.0), t_rel_s=10.0, flow_p95=0.0)
assert later is None
def test_the_estimate_is_kept_per_display_unit() -> None:
"""A newton baseline and a raw baseline are different scales, not one number."""
presentation = ForcePresentation(enabled=True, min_samples=5, baseline_seconds=1.0)
for i in range(12):
presentation.observe(
(1000.0 + i * 0.1, 0.0, 0.0), t_rel_s=i * 0.02, unit="raw"
)
assert presentation.unit == "raw"
assert presentation.ready
assert presentation.accept((1000.55, 0.0, 0.0), unit="raw").values[0] == 0.0
# The other unit has no estimate of its own and must show what was measured.
assert presentation.accept((1000.55, 0.0, 0.0), unit="N").values == (
1000.55,
0.0,
0.0,
)
presentation.unit = "N"
assert not presentation.ready, "the raw estimate made newtons ready"
presentation.unit = "raw"
# ...and it collects its own window, in its own scale.
for i in range(12):
presentation.observe((0.5 + i * 0.01, 0.0, 0.0), t_rel_s=i * 0.02, unit="N")
presentation.unit = "N"
assert presentation.ready
assert presentation.accept((0.555, 0.0, 0.0), unit="N").values[0] == 0.0
def test_the_newton_floor_is_not_the_raw_floor() -> None:
"""50 raw counts is a noise floor; 50 N would be a contact."""
presentation = ForcePresentation(enabled=True)
assert presentation.min_deadband_for("raw") == 50.0
assert presentation.min_deadband_for("N") == 0.0
rng = np.random.default_rng(3)
quiet = (0.5 + rng.normal(0.0, 0.002, size=(60, 3))).tolist()
estimate = presentation.fit_zero(quiet, unit="N")
assert estimate.quality == "ok"
assert max(estimate.deadbands) < 0.01, estimate.deadbands
def test_present_holds_a_resting_block_at_zero_and_keeps_an_excursion() -> None:
rng = np.random.default_rng(7)
bias = np.array([9555.0, -13368.0, -353.0])
quiet = (bias + rng.normal(0.0, 20.0, size=(80, 3))).tolist()
presentation = ForcePresentation(
enabled=True, min_samples=30, baseline_seconds=0.0, deadband_percentile=99.0
)
estimate = presentation.fit_zero(quiet)
assert estimate.quality == "ok"
presentation.apply_estimate(estimate)
block = np.asarray(
[bias, bias + 5.0, bias + 4000.0, [np.nan, np.nan, np.nan]], dtype=float
)
displayed, suppressed = presentation.present(block, unit="raw")
assert displayed[0].tolist() == [0.0, 0.0, 0.0], "a resting sample drifted"
assert suppressed[0].all()
assert displayed[1].tolist() == [0.0, 0.0, 0.0], "noise outside the deadband?"
# A real excursion survives, centred on the estimated offset.
assert 3900.0 < displayed[2][0] < 4100.0, displayed[2]
assert not suppressed[2].any()
# A gap is a gap: an unavailable axis must never become a zero.
assert np.isnan(displayed[3]).all()
assert not suppressed[3].any()
def test_present_is_a_display_copy_and_never_touches_the_input() -> None:
presentation = ForcePresentation(enabled=True, min_samples=5, baseline_seconds=1.0)
for i in range(12):
presentation.observe((100.0 + i * 0.01, 0.0, 0.0), t_rel_s=i * 0.02)
source = np.array([[100.055, 0.0, 0.0]])
displayed, _suppressed = presentation.present(source, unit="raw")
assert source.tolist() == [[100.055, 0.0, 0.0]], "present mutated the measurement"
assert displayed[0][0] == 0.0
def test_the_operator_floor_moves_the_frozen_deadband() -> None:
presentation = ForcePresentation(enabled=True, min_samples=5, baseline_seconds=1.0)
for i in range(12):
presentation.observe((100.0 + i * 0.01, 0.0, 0.0), t_rel_s=i * 0.02, unit="raw")
assert presentation.ready
presentation.set_min_deadband("raw", 500.0)
assert presentation.accept((300.0, 0.0, 0.0), unit="raw").values[0] == 0.0, (
"200 raw counts above the device's own band but inside the operator's floor"
)
# Back to auto: the floor is a floor, so the estimated band returns.
presentation.set_min_deadband("raw", 0.0)
assert presentation.min_deadband_for("raw") == 0.0
assert presentation.accept((300.0, 0.0, 0.0), unit="raw").values[0] != 0.0
def test_format_force_stable_digits() -> None:
presentation = ForcePresentation(enabled=True)
assert presentation.format_force(0.0, unit="raw") == "0"
assert presentation.format_force(1262.4, unit="raw") == "1262"
assert presentation.format_force(1.2345, unit="N") == "1.234"
# ------------------------------------------------------- the reported defect
def _quiet_until_zeroed(
presentation: ForcePresentation,
bias: tuple[float, float, float],
*,
unit: str,
limit: int = 29,
) -> int:
"""Feed quiet samples until ``accept`` reads zero; return the sample count.
Returns ``limit + 1`` (i.e. "never within the limit") if it did not zero.
"""
for index in range(limit):
presentation.observe(bias, t_rel_s=index / 30.0, flow_p95=0.1, unit=unit)
if presentation.accept(bias, unit=unit).values == (0.0, 0.0, 0.0):
return index + 1
return limit + 1
def test_a_resting_raw_channel_reads_zero_from_the_first_samples() -> None:
"""FN 1118 / FX 34.16 / FY -186.1 must be zero long before 30 samples / 3 s.
This is the reported defect: the cards kept the device's raw bias for the
whole startup window, so a session that started at rest looked like it was
under load.
"""
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
zeroed_at = _quiet_until_zeroed(presentation, REPORTED_RAW_BIAS, unit="raw")
assert zeroed_at <= 3, f"the raw bias was still on the cards after {zeroed_at} samples"
assert not presentation.ready, "zeroing should not have needed the whole quiet window"
assert presentation.display_ready, "the display policy was not applied while collecting"
assert presentation.display_quality == "provisional"
# ...and no invented zero: the offset is the one the quiet samples measured.
offsets = presentation.last_estimate.offsets
assert offsets[0] == REPORTED_RAW_BIAS[0]
assert offsets[1] == REPORTED_RAW_BIAS[1]
assert offsets[2] == REPORTED_RAW_BIAS[2]
def test_a_resting_newton_channel_reads_zero_from_the_first_samples() -> None:
"""The same aid in newtons — the estimate is per display unit."""
presentation = ForcePresentation(
enabled=True, unit="N", min_samples=30, baseline_seconds=3.0
)
zeroed_at = _quiet_until_zeroed(presentation, REPORTED_NEWTON_BIAS, unit="N")
assert zeroed_at <= 3, f"the newton bias survived {zeroed_at} samples"
assert not presentation.ready
assert presentation.display_quality == "provisional"
def test_a_noisy_channel_is_zeroed_from_the_first_samples_too() -> None:
"""The provisional band covers the samples it was fitted from.
With a handful of samples the p99 residual is essentially the largest one, so
the newest sample of the window is inside the band: the first tenths of a
second are zeroed, and the *first* sample is not claimed to be zero because
nothing has been estimated from it yet.
"""
rng = np.random.default_rng(11)
bias = np.array(REPORTED_RAW_BIAS)
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
for index in range(8):
sample = bias + rng.normal(0.0, 30.0, size=3)
presentation.observe(tuple(sample), t_rel_s=index / 30.0, flow_p95=0.1)
if index < 2:
# Nothing has been estimated from two samples: those frames honestly
# show the measurement (and are asserted to do so elsewhere).
continue
displayed = presentation.accept(tuple(sample), unit="raw")
assert displayed.values == (0.0, 0.0, 0.0), (
f"sample {index} of a resting noisy channel was drawn"
)
def test_a_contact_present_at_session_start_is_never_zeroed() -> None:
"""A press at run start must stay visible: no estimate may be invented for it.
The contact guard decides what feeds the estimate, so a window that has only
ever seen a contact has nothing to subtract and displays the measurement.
"""
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
pressed = (6000.0, 40.0, -180.0)
for index in range(20):
assert (
presentation.observe(
pressed, t_rel_s=index / 30.0, contact_found=True, flow_p95=0.1
)
is None
)
displayed = presentation.accept(pressed, unit="raw")
assert displayed.values[0] == pressed[0], "a contact at startup was zeroed away"
assert not displayed.suppressed[0]
assert not presentation.display_ready
assert presentation.display_quality == "none"
def test_a_contaminated_window_never_falls_back_to_the_raw_bias() -> None:
"""A window that closes badly keeps the estimate it already worked out.
Falling back to the raw offset is the reported defect. Both ways a startup
window can be spoiled — a contact, or too much flow — must leave the display
zeroed and say so through ``display_quality``.
"""
bias = REPORTED_RAW_BIAS
pressed = (6000.0, 40.0, -180.0)
for spoiled_by in ("contact", "flow"):
presentation = ForcePresentation(
enabled=True, min_samples=30, baseline_seconds=3.0
)
for index in range(3):
presentation.observe(bias, t_rel_s=index / 30.0, flow_p95=0.1)
assert presentation.display_ready and not presentation.ready
for index in range(90): # the window is spoiled before it can close
presentation.observe(
pressed,
t_rel_s=0.5 + index / 30.0,
contact_found=(spoiled_by == "contact"),
flow_p95=0.9 if spoiled_by == "flow" else 0.1,
)
estimate = presentation.observe(bias, t_rel_s=4.0, flow_p95=0.1)
assert estimate is not None, spoiled_by
assert estimate.quality == "contaminated", (spoiled_by, estimate)
assert not presentation.ready, spoiled_by
assert presentation.display_quality == "contaminated", spoiled_by
assert presentation.accept(bias, unit="raw").values == (0.0, 0.0, 0.0), spoiled_by
def test_a_window_that_closed_short_still_applies_what_it_learned() -> None:
"""A session shorter than the quiet window is not a reason to show the bias."""
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=0.5)
for index in range(3):
presentation.observe(REPORTED_RAW_BIAS, t_rel_s=index / 30.0, flow_p95=0.1)
estimate = presentation.observe(REPORTED_RAW_BIAS, t_rel_s=0.6, flow_p95=0.1)
assert estimate is not None and estimate.quality == "insufficient"
assert presentation.display_quality == "insufficient"
assert presentation.accept(REPORTED_RAW_BIAS, unit="raw").values == (0.0, 0.0, 0.0)
def test_a_frozen_window_stops_chasing_the_data() -> None:
"""The provisional phase is a start, not a licence to re-zero a press."""
presentation = ForcePresentation(
enabled=True, min_samples=5, baseline_seconds=3.0, min_deadband_raw=50.0
)
for index in range(10):
presentation.observe(REPORTED_RAW_BIAS, t_rel_s=index / 30.0, flow_p95=0.1)
assert presentation.ready
assert presentation.display_quality == "ok"
# A held press after the window is a reading, not a new zero.
assert presentation.accept((5000.0, 34.16, -186.1), unit="raw").values[0] > 3000.0
def test_the_cards_and_the_curves_are_the_same_presentation() -> None:
"""What ``accept`` puts on a card is the row ``present`` draws on its curve.
This is what the extra "the estimate is ready" gate in the value cards broke:
the card kept the raw measurement while the curve was already drawn from the
applied estimate, so the two disagreed about the same sample.
"""
bias = REPORTED_RAW_BIAS
#: The samples the acquisition would flag as a contact: they must not feed
#: the idle estimate, or the display would zero the press away.
presses = {
3: (bias[0] + 4000.0, bias[1] + 10.0, bias[2] - 10.0),
5: (bias[0] - 4000.0, bias[1], bias[2]),
}
stream = [
bias,
bias,
bias,
presses[3],
bias,
presses[5],
bias,
]
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
for index, sample in enumerate(stream):
presentation.observe(
sample,
t_rel_s=index / 30.0,
flow_p95=0.1,
contact_found=index in presses,
)
card = presentation.accept(sample, unit="raw")
curve, suppressed = presentation.present([sample], unit="raw")
assert card.values == tuple(float(value) for value in curve[0]), index
assert card.suppressed == tuple(bool(flag) for flag in suppressed[0]), index
# The assertion is not vacuous: the excursion really is on both.
assert presentation.accept(stream[3], unit="raw").values[0] > 3000.0
def test_present_uses_the_provisional_estimate_before_the_window_closes() -> None:
"""The curves get the same early zero as the cards, not only after 3 s."""
presentation = ForcePresentation(enabled=True, min_samples=30, baseline_seconds=3.0)
block = np.asarray([REPORTED_RAW_BIAS] * 4, dtype=float)
raw, _suppressed = presentation.present(block, unit="raw")
assert raw[0].tolist() == list(REPORTED_RAW_BIAS), "nothing was learned yet"
for index in range(3):
presentation.observe(REPORTED_RAW_BIAS, t_rel_s=index / 30.0, flow_p95=0.1)
displayed, suppressed = presentation.present(block, unit="raw")
assert displayed.tolist() == [[0.0, 0.0, 0.0]] * 4
assert suppressed.all()
assert not presentation.ready, "the whole quiet window was not needed"
def test_precision_is_per_unit_and_does_not_change_threshold():
policy = ForcePresentation(min_deadband_raw=50.0, min_deadband_n=0.5)
assert policy.format_force(50.6, "raw") == "51"
assert policy.format_force(0.506, "N") == "0.51"
assert policy.format_force(-0.004, "N") == "0.00"
assert policy.format_force(float("nan"), "raw") == "—"
policy.raw_decimals = 1
assert policy.format_force(50.6, "raw") == "50.6"
shared/tests/test_record_semantics.py
+24 -1
View File
@@ -16,7 +16,7 @@ import numpy as np
import pytest
from orisys_app_shared.acquisition.sdk_worker import SdkPacket
from orisys_app_shared.record import ExperimentRecorder, RecordConfig
from orisys_app_shared.record.sample import STALE_REASON, to_record_sample
from orisys_app_shared.record.sample import STALE_REASON, process_record_sample, to_record_sample
SCRATCH = Path(__file__).resolve().parent / "_scratch"
@@ -73,6 +73,29 @@ def test_uncalibrated_sensor_writes_no_newton_values():
assert rows[0]["FY_N"] == ""
def test_export_preserves_original_and_processed_columns(tmp_path):
packet = _packet(
0, force_calibrated=True, calibrated_axes=("fn", "fx", "fy"),
normal_force_raw=49.0, shear_x_raw=-51.0, shear_y_raw=20.0,
normal_force=0.4, shear_x=-0.6, shear_y=0.1,
)
sample = process_record_sample(
to_record_sample(packet), raw_threshold=50.0, newton_threshold=0.5
)
recorder = ExperimentRecorder(RecordConfig(video=False))
directory = recorder.start(root=tmp_path, session_id=1, metadata={})
recorder.on_packet(packet, sample=sample)
recorder.stop()
with (directory / "forces.csv").open(encoding="utf-8", newline="") as handle:
row = next(csv.DictReader(handle))
assert row["FN_raw"] == "49.0"
assert row["FN_output_raw"] == "0.0"
assert row["FX_raw"] == "-51.0"
assert row["FX_output_raw"] == "-51.0"
assert row["FN_N"] == "0.4"
assert row["FN_output_N"] == "0.0"
def test_partially_calibrated_sensor_fills_only_the_calibrated_axes():
rows = _record(
[