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

add session browsing, calibration diagnostics and the 2D layout preset

Browse Source 
Phase 2 of the review: browse the whole session instead of only the rolling
window, diagnose a calibration from the recorded points, and provide the
reviewed two-panel layout.

What
- SessionStore gains extent(), visible(t0, t1, max_points) and
  statistics(t0, t1, unit). visible() walks only the blocks that overlap the
  window and decimates by taking each bucket's minimum and maximum, so a spike
  survives decimation and browsing a long session does not copy all of it.
- ForcePlotsPanel gains follow/browse modes, a whole-session overview strip
  with the current window, pan and wheel zoom, a rubber-band zoom toggle,
  a shared time cursor across all three channels, and a selection readout
  (duration, valid samples, per-axis peak/mean/std).
- CalibrationPoint records a bounded per-axis trace, an excluded flag, a
  validation flag and its fit residual. CalibrationRecorder gains
  set_excluded/set_validation/fit; the fit is a least-squares line over the
  non-excluded, non-validation points with per-point residuals and an
  independent validation error.
- The calibration table becomes the reviewed diagnostic view
  (reference load | FN | FX | FY | stability | state) with the residual shown
  next to the loaded axis, plus exclude/restore and mark-validation actions and
  a fit summary line. Double clicking a row opens the point's three-axis trace.
- Add a layout preset: free layout or 2D mode (depth map on the left, force
  charts on the right, calibration folded away), persisted in preferences.

Why
- Browsing is a display pause, not a capture pause: the record channel keeps
  running while the charts show an older window.
- Excluding a point removes it from the fit and from the exported calibration
  but keeps it on record, so the evidence for the decision is not destroyed.
- The non-loaded axes are recorded but never given a reference load: their true
  load is unknown during a single-axis calibration, so it stays empty.

Verification
- shared/tests/test_record_channel.py grows to 18 cases, covering extent,
  peak-preserving decimation, windowed reads, inverted/empty windows, unit-aware
  statistics and the time origin.
- tools/ui_check.py gains "session browsing" and "layout preset" sections
  (223 checks, 0 failures overall).
- apps/studio/tests/test_ui_smoke.py covers the same invariants.

Not in this change: synchronised playback of video, depth and curves. It needs
a decision on how depth is stored for replay (per-frame PNG, reduced-rate PNG,
or offline recomputation); the cursor plumbing (cursor_moved / set_cursor) is
already in place for it.
This commit is contained in:
ykcheungab
2026-09-13 13:34:26 +08:00
parent 817f830565
commit 92da51217c
13 changed files with 1576 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shared/src/orisys_app_shared/force/__init__.py
+2
View File
@@ -16,6 +16,7 @@ from orisys_app_shared.force.calibration_recorder import (
from orisys_app_shared.force.history import MetricHistory, PlotState, lowpass_filter
from orisys_app_shared.force.session import (
NA,
SelectionStats,
SessionSample,
SessionSlice,
SessionStore,
@@ -34,6 +35,7 @@ __all__ = [
"DisplayForces",
"MetricHistory",
"PlotState",
"SelectionStats",
"SessionSample",
"SessionSlice",
"SessionStore",
shared/src/orisys_app_shared/force/calibration_recorder.py
+173
View File
@@ -69,10 +69,30 @@ class CalibrationPoint:
raw: dict[str, AxisStats] = field(default_factory=dict)
newtons: dict[str, AxisStats] = field(default_factory=dict)
status: str = STATUS_OK
#: Downsampled per-axis raw trace, so a row can be inspected after the fact.
trace: dict[str, list[float]] = field(default_factory=dict)
trace_dt_s: float = 0.0
#: Diagnostics, all operator-controlled. Excluded points leave the fit;
#: validation points are held out of it and used to test it.
excluded: bool = False
is_validation: bool = False
#: Filled in by :meth:`CalibrationRecorder.fit`.
residual_n: float = NA
def loaded_stats(self) -> AxisStats:
return self.raw.get(self.channel, AxisStats())
def axis_mean(self, channel: str) -> float:
stats = self.raw.get(channel)
return stats.mean if stats is not None else NA
def state_key(self) -> str:
if self.excluded:
return "cal.state_excluded"
if self.is_validation:
return "cal.state_validation"
return "cal.state_ok"
def as_dict(self) -> dict:
return {
"channel": self.channel,
@@ -89,11 +109,51 @@ class CalibrationPoint:
"n_total": self.n_total,
"n_valid": self.n_valid,
"status": self.status,
"excluded": self.excluded,
"is_validation": self.is_validation,
"residual_n": self.residual_n,
"raw": {key: stats.as_dict() for key, stats in self.raw.items()},
"newtons": {key: stats.as_dict() for key, stats in self.newtons.items()},
}
@dataclass(frozen=True)
class FitResult:
"""Least-squares result for one channel."""
channel: ChannelKey
gain: float = NA
offset: float = NA
n_points: int = 0
max_residual_n: float = NA
validation_error_n: float = NA
validation_points: int = 0
@property
def usable(self) -> bool:
return self.n_points >= 2 and np.isfinite(self.gain) and np.isfinite(self.offset)
def predict(self, raw_value: float) -> float:
if not self.usable:
return NA
return float(self.gain) * float(raw_value) + float(self.offset)
def as_dict(self) -> dict:
return {
"channel": self.channel,
"gain_n_per_raw": self.gain,
"offset_n": self.offset,
"points_in_fit": self.n_points,
"max_residual_n": self.max_residual_n,
"validation_error_n": self.validation_error_n,
"validation_points": self.validation_points,
}
#: Cap on stored trace samples per axis; peaks are preserved when reducing.
MAX_TRACE_POINTS = 400
class CalibrationRecorder:
"""Owns the calibration sampling lifecycle."""
@@ -270,6 +330,8 @@ class CalibrationRecorder:
raw={"fn": axis_stats(1), "fx": axis_stats(2), "fy": axis_stats(3)},
newtons={"fn": axis_stats(4), "fx": axis_stats(5), "fy": axis_stats(6)},
status=STATUS_OK if valid_block.shape[0] >= 2 else STATUS_FEW_SAMPLES,
trace=_trace_from(valid_block),
trace_dt_s=_trace_dt(valid_block),
)
with self._lock:
self.points.append(point)
@@ -277,8 +339,119 @@ class CalibrationRecorder:
self._on_finished(point)
return point
# ------------------------------------------------------------ diagnostics
def set_excluded(self, index: int, excluded: bool) -> None:
"""Remove (or restore) a point from the fit and from the export."""
with self._lock:
if not 0 <= index < len(self.points):
return
self.points[index].excluded = bool(excluded)
if excluded:
self.points[index].is_validation = False
def set_validation(self, index: int, validation: bool) -> None:
"""Hold a point out of the fit and use it to test the fit instead."""
with self._lock:
if not 0 <= index < len(self.points):
return
self.points[index].is_validation = bool(validation)
if validation:
self.points[index].excluded = False
def points_for(self, channel: ChannelKey) -> list[CalibrationPoint]:
with self._lock:
return [point for point in self.points if point.channel == channel]
def fit(self, channel: ChannelKey) -> FitResult:
"""Least-squares gain/offset from the raw mean to the reference load.
Excluded points are dropped, validation points are held out, and every
point gets its residual filled in so the table can show the misfit.
"""
points = self.points_for(channel)
usable = [
point
for point in points
if not point.excluded and not point.is_validation
and np.isfinite(point.reference_load_n)
and np.isfinite(point.loaded_stats().mean)
]
for point in points:
point.residual_n = NA
if len(usable) < 2:
return FitResult(channel=channel, n_points=len(usable))
raws = np.asarray([point.loaded_stats().mean for point in usable], dtype=float)
loads = np.asarray([point.reference_load_n for point in usable], dtype=float)
if np.allclose(raws, raws[0]):
return FitResult(channel=channel, n_points=len(usable))
gain, offset = np.polyfit(raws, loads, 1)
residuals = (gain * raws + offset) - loads
for point, residual in zip(usable, residuals):
point.residual_n = float(residual)
validation = [
point
for point in points
if not point.excluded and point.is_validation
and np.isfinite(point.reference_load_n)
and np.isfinite(point.loaded_stats().mean)
]
validation_error = NA
if validation:
errors = [
abs(gain * point.loaded_stats().mean + offset - point.reference_load_n)
for point in validation
]
validation_error = float(max(errors))
return FitResult(
channel=channel,
gain=float(gain),
offset=float(offset),
n_points=len(usable),
max_residual_n=float(np.max(np.abs(residuals))),
validation_error_n=validation_error,
validation_points=len(validation),
)
def _next_round(self, channel: ChannelKey) -> int:
with self._lock:
index = self._rounds.get(channel, 0)
self._rounds[channel] = index + 1
return index
def _trace_from(valid_block: np.ndarray) -> dict[str, list[float]]:
"""Downsampled raw trace per axis, bounded by ``MAX_TRACE_POINTS``."""
if valid_block.size == 0:
return {}
traces: dict[str, list[float]] = {}
for key, column in (("fn", 1), ("fx", 2), ("fy", 3)):
values = valid_block[:, column]
values = values[np.isfinite(values)]
if values.size == 0:
continue
if values.size > MAX_TRACE_POINTS:
# Keep the shape of the trace, including its extremes.
edges = np.linspace(0, values.size, MAX_TRACE_POINTS + 1, dtype=int)
reduced = [
values[start:max(start + 1, stop)]
for start, stop in zip(edges[:-1], edges[1:])
]
values = np.asarray(
[
bucket[int(np.argmax(np.abs(bucket)))]
for bucket in reduced
if bucket.size
]
)
traces[key] = [float(value) for value in values]
return traces
def _trace_dt(valid_block: np.ndarray) -> float:
if valid_block.shape[0] < 2:
return 0.0
span = float(valid_block[-1, 0] - valid_block[0, 0])
return span / max(1, valid_block.shape[0] - 1)
shared/src/orisys_app_shared/force/session.py
+151 -1
View File
@@ -24,7 +24,7 @@ Design notes
from __future__ import annotations
import threading
from dataclasses import dataclass
from dataclasses import dataclass, replace
from typing import Iterable, Iterator, Optional
import numpy as np
@@ -99,6 +99,55 @@ class SessionSlice:
return self.raw
@dataclass(frozen=True)
class SelectionStats:
"""What the browser shows for the current selection."""
duration_s: float = 0.0
samples: int = 0
peaks: tuple[float, float, float] = (NA, NA, NA)
means: tuple[float, float, float] = (NA, NA, NA)
stds: tuple[float, float, float] = (NA, NA, NA)
def _decimate_preserving_peaks(rows: np.ndarray, max_points: int) -> np.ndarray:
"""Reduce ``rows`` to about ``max_points`` while keeping every spike.
Each bucket contributes its minimum and its maximum sample (in time order),
so the envelope of the signal survives decimation instead of being averaged
flat.
"""
total = rows.shape[0]
if max_points < 4 or total <= max_points:
return rows
bucket = max(2, total // max(2, max_points // 2))
usable = (total // bucket) * bucket
head = rows[:usable].reshape(-1, bucket, rows.shape[1])
tail = rows[usable:]
# Index of the extreme value on the first channel; ties go to the earlier one.
values = head[:, :, _COL_FN_RAW]
finite = np.isfinite(values)
values = np.where(finite, values, -np.inf)
low = np.argmin(values, axis=1)
high = np.argmax(values, axis=1)
first = np.minimum(low, high)
second = np.maximum(low, high)
index = np.arange(head.shape[0])
picked = np.concatenate(
[
head[index, first],
head[index, second],
],
axis=0,
)
# ``concatenate`` above groups all lows then all highs; restore time order.
order = np.argsort(picked[:, _COL_FRAME_ID], kind="stable")
picked = picked[order]
if tail.shape[0]:
picked = np.vstack([picked, tail])
return picked
class SessionStore:
"""Thread-safe, memory-bounded, whole-session sample store."""
@@ -219,6 +268,11 @@ class SessionStore:
return 0.0
return float(self._last_t - self._t0)
def origin_time(self) -> float:
"""Absolute acquisition timestamp that ``t_rel_s == 0`` refers to."""
with self._lock:
return float(self._t0) if self._t0 is not None else 0.0
def _packed(self) -> np.ndarray:
if self._count == 0:
return np.empty((0, _COLUMNS), dtype=np.float64)
@@ -255,6 +309,102 @@ class SessionStore:
mask = (packed[:, _COL_T] >= float(t0)) & (packed[:, _COL_T] <= float(t1))
return self._slice_from(packed[mask])
# -------------------------------------------------------------- browsing
def extent(self) -> tuple[float, float]:
"""``(first_t, last_t)`` of the stored session; ``(0, 0)`` when empty."""
with self._lock:
if self._count == 0 or self._t0 is None or self._last_t is None:
return (0.0, 0.0)
first = self._blocks[0][0, _COL_T] if self._blocks else 0.0
return (float(first), float(self._last_t - self._t0))
def _block_span(self, block: np.ndarray, rows: int) -> tuple[float, float]:
return (float(block[0, _COL_T]), float(block[rows - 1, _COL_T]))
def visible(self, t0: float, t1: float, max_points: int = 2000) -> SessionSlice:
"""Decimated slice for drawing, keeping peaks.
Only the blocks that overlap ``[t0, t1]`` are touched, so browsing a
long session does not copy the whole thing, and the decimation emits the
min and the max of each bucket so a spike can never be smoothed away.
"""
t0 = float(t0)
t1 = float(t1)
if t1 < t0:
t0, t1 = t1, t0
with self._lock:
rows = self._packed_window(t0, t1)
truncated = self._dropped > 0
if max_points and rows.shape[0] > max_points:
rows = _decimate_preserving_peaks(rows, max_points)
return replace(self._slice_from(rows), truncated=truncated)
def _packed_window(self, t0: float, t1: float) -> np.ndarray:
"""Concatenate only the rows inside the window. Caller holds the lock."""
if self._count == 0:
return np.empty((0, _COLUMNS), dtype=np.float64)
selected: list[np.ndarray] = []
remaining = self._count
for block in self._blocks:
rows = min(remaining, block.shape[0])
remaining -= rows
if rows <= 0:
break
span = self._block_span(block, rows)
if span[1] < t0 or span[0] > t1:
continue # blocks are time-ordered: skip whole blocks
times = block[:rows, _COL_T]
start = int(np.searchsorted(times, t0, side="left"))
stop = int(np.searchsorted(times, t1, side="right"))
if stop > start:
selected.append(block[start:stop])
if not selected:
return np.empty((0, _COLUMNS), dtype=np.float64)
return np.vstack(selected) if len(selected) > 1 else selected[0]
def statistics(self, t0: float, t1: float, unit: str = "raw") -> "SelectionStats":
"""Duration, peak, mean and std over a selection, per channel."""
t0 = float(t0)
t1 = float(t1)
if t1 < t0:
t0, t1 = t1, t0
with self._lock:
rows = self._packed_window(t0, t1)
if rows.shape[0] == 0:
return SelectionStats(duration_s=0.0, samples=0)
mask = rows[:, _COL_VALID] > 0.5
valid_rows = rows[mask]
times = rows[:, _COL_T]
duration = float(times[-1] - times[0]) if times.size else 0.0
if valid_rows.shape[0] == 0:
return SelectionStats(duration_s=duration, samples=0)
columns = (
(_COL_FN_N, _COL_FX_N, _COL_FY_N)
if str(unit).upper().startswith("N")
else (_COL_FN_RAW, _COL_FX_RAW, _COL_FY_RAW)
)
peaks: list[float] = []
means: list[float] = []
stds: list[float] = []
for column in columns:
values = valid_rows[:, column]
values = values[np.isfinite(values)]
if values.size == 0:
peaks.append(NA)
means.append(NA)
stds.append(NA)
continue
peaks.append(float(values[np.argmax(np.abs(values))]))
means.append(float(values.mean()))
stds.append(float(values.std()))
return SelectionStats(
duration_s=duration,
samples=int(valid_rows.shape[0]),
peaks=tuple(peaks),
means=tuple(means),
stds=tuple(stds),
)
def _slice_from(self, packed: np.ndarray) -> SessionSlice:
if packed.shape[0] == 0:
empty = np.empty(0, dtype=np.float64)
shared/tests/test_record_channel.py
+97
View File
@@ -16,6 +16,7 @@ import math
import threading
import time
import numpy as np
import pytest
from orisys_app_shared.force.calibration_recorder import (
@@ -100,6 +101,102 @@ def test_store_window_queries_are_relative_to_the_first_sample():
assert float(first_second.times[0]) == 0.0
# ------------------------------------------------------------- session browsing
def _ramp_store(count: int = 3000, *, spike_at: int = 1500) -> SessionStore:
store = SessionStore(max_samples=20_000, block_size=64)
store.reset(session_id=1)
for index in range(count):
fn = index * 0.01
if index == spike_at:
fn = 99.0
store.append(
SessionSample(
frame_id=index,
t_rel_s=index / 30.0,
fn_raw=fn,
fx_raw=1.0,
fy_raw=-1.0,
fn_n=fn / 10.0,
fx_n=0.1,
fy_n=-0.1,
valid=(index % 10) != 3,
)
)
return store
def test_extent_covers_the_stored_session():
store = _ramp_store(600)
first, last = store.extent()
assert first == 0.0
assert last == pytest.approx(599 / 30.0, abs=1e-6)
assert SessionStore().extent() == (0.0, 0.0)
def test_visible_decimates_but_keeps_the_peak():
store = _ramp_store()
full = store.data()
window = store.visible(0.0, 100.0, max_points=200)
assert len(window) <= 400, "decimation did not bound the point count"
assert len(window) < len(full), "a long window should be decimated"
assert float(window.value_for_unit("raw")[:, 0].max()) == 99.0, (
"decimation averaged away the spike"
)
assert bool((np.diff(window.times) >= 0).all()), "decimation broke time order"
def test_visible_reads_only_the_requested_window():
store = _ramp_store()
window = store.visible(10.0, 20.0, max_points=100_000)
assert float(window.times.min()) == pytest.approx(10.0, abs=1 / 30.0)
assert float(window.times.max()) == pytest.approx(20.0, abs=1 / 30.0)
assert len(window) == pytest.approx(301, abs=1)
def test_visible_survives_an_empty_or_inverted_window():
store = _ramp_store(120)
assert len(store.visible(500.0, 600.0)) == 0
# An inverted window is normalised rather than returning nothing.
assert len(store.visible(20.0, 10.0)) == len(store.visible(10.0, 20.0))
def test_statistics_report_duration_peak_mean_and_std():
store = _ramp_store(600)
stats = store.statistics(0.0, 20.0, unit="raw")
assert stats.samples == 540, "invalid samples must not be counted"
assert stats.duration_s == pytest.approx(20.0, abs=0.05)
# Last sample inside the window is index 599, so the peak is index * 0.01.
assert stats.peaks[0] == pytest.approx(5.99, abs=0.02)
assert stats.means[1] == pytest.approx(1.0)
assert stats.stds[1] == pytest.approx(0.0, abs=1e-9)
def test_statistics_follow_the_display_unit():
store = _ramp_store(600)
raw = store.statistics(0.0, 20.0, unit="raw")
newtons = store.statistics(0.0, 20.0, unit="N")
assert newtons.peaks[0] == pytest.approx(raw.peaks[0] / 10.0, abs=0.01)
def test_statistics_of_an_empty_selection_are_blank_not_zero():
store = _ramp_store(120)
stats = store.statistics(900.0, 1000.0)
assert stats.samples == 0
assert all(math.isnan(value) for value in stats.means)
def test_origin_time_matches_the_first_sample():
store = _ramp_store(10)
assert store.origin_time() == pytest.approx(0.0) # _sample starts at t=0
store.reset()
store.append(SessionSample(frame_id=0, t_rel_s=5000.0, fn_raw=1.0))
assert store.origin_time() == pytest.approx(5000.0)
# ------------------------------------------------------- CalibrationRecorder