subject: store the whole session and drive calibration from the acquisition stream

shared/src/orisys_app_shared/force/__init__.py

@@ -1,4 +1,4 @@
-"""Force history and unit helpers."""
+"""Force history, session storage, calibration sampling and unit helpers."""
 
 from orisys_app_shared.force.calibration_io import (
     CHANNEL_KEYS,
@@ -8,19 +8,39 @@ from orisys_app_shared.force.calibration_io import (
     export_force_calibration_v2,
     meta_from_sensor_json,
 )
+from orisys_app_shared.force.calibration_recorder import (
+    AxisStats,
+    CalibrationPoint,
+    CalibrationRecorder,
+)
 from orisys_app_shared.force.history import MetricHistory, PlotState, lowpass_filter
+from orisys_app_shared.force.session import (
+    NA,
+    SessionSample,
+    SessionSlice,
+    SessionStore,
+    summarise,
+)
 from orisys_app_shared.force.units import DisplayForces, select_display_forces
 
 __all__ = [
     "CHANNEL_KEYS",
+    "NA",
+    "AxisStats",
     "CalSample",
     "CalSessionStore",
+    "CalibrationPoint",
+    "CalibrationRecorder",
     "DisplayForces",
     "MetricHistory",
     "PlotState",
+    "SessionSample",
+    "SessionSlice",
+    "SessionStore",
     "build_force_calibration_v2",
     "export_force_calibration_v2",
     "lowpass_filter",
     "meta_from_sensor_json",
     "select_display_forces",
+    "summarise",
 ]

shared/src/orisys_app_shared/force/calibration_recorder.py

@@ -0,0 +1,284 @@
+"""Calibration sampling state machine, independent of any Qt widget.
+
+Two responsibilities are deliberately split across two threads:
+
+``feed(sample)``
+    Called from the acquisition thread (via the worker's sink list) for **every**
+    acquired sample. It only appends to a thread-safe buffer, so recording keeps
+    running when the calibration panel is hidden, when the plots are redrawing
+    slowly, or when the UI is busy. It never touches Qt.
+
+``tick()``
+    Called from the UI thread. It evaluates the sampling deadline and turns the
+    buffered samples into a finished :class:`CalibrationPoint`, invoking the
+    callbacks on the UI thread where it is safe to touch widgets.
+
+Each point stores all three axes rather than only the loaded one, so crosstalk
+between axes is observable from the recorded data. The non-loaded axes carry raw
+statistics but **no** reference load: their true load is unknown during a
+single-axis calibration, so it is left empty rather than filled with zero.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Callable, Literal, Optional
+
+import numpy as np
+
+from orisys_app_shared.force.session import NA, SessionSample, summarise
+
+ChannelKey = Literal["fn", "fx", "fy"]
+CHANNEL_KEYS: tuple[ChannelKey, ...] = ("fn", "fx", "fy")
+
+STATUS_OK = "ok"
+STATUS_FEW_SAMPLES = "few_samples"
+STATUS_CANCELLED = "cancelled"
+
+
+@dataclass(frozen=True)
+class AxisStats:
+    """Mean / standard deviation / valid count for one axis."""
+
+    mean: float = NA
+    std: float = NA
+    n: int = 0
+
+    def as_dict(self) -> dict:
+        return {"mean": self.mean, "std": self.std, "n": self.n}
+
+
+@dataclass
+class CalibrationPoint:
+    """One recorded calibration point, with all three axes captured."""
+
+    channel: ChannelKey
+    reference_load_n: float
+    direction: str = "load"
+    round_index: int = 0
+    note: str = ""
+    started_at: str = ""
+    ended_at: str = ""
+    duration_s: float = 0.0
+    requested_duration_s: float = 0.0
+    n_total: int = 0
+    n_valid: int = 0
+    raw: dict[str, AxisStats] = field(default_factory=dict)
+    newtons: dict[str, AxisStats] = field(default_factory=dict)
+    status: str = STATUS_OK
+
+    def loaded_stats(self) -> AxisStats:
+        return self.raw.get(self.channel, AxisStats())
+
+    def as_dict(self) -> dict:
+        return {
+            "channel": self.channel,
+            "reference_load_n": self.reference_load_n,
+            # Only the loaded axis has a known reference load; the other two are
+            # intentionally absent rather than zero.
+            "direction": self.direction,
+            "round_index": self.round_index,
+            "note": self.note,
+            "started_at": self.started_at,
+            "ended_at": self.ended_at,
+            "duration_s": self.duration_s,
+            "requested_duration_s": self.requested_duration_s,
+            "n_total": self.n_total,
+            "n_valid": self.n_valid,
+            "status": self.status,
+            "raw": {key: stats.as_dict() for key, stats in self.raw.items()},
+            "newtons": {key: stats.as_dict() for key, stats in self.newtons.items()},
+        }
+
+
+class CalibrationRecorder:
+    """Owns the calibration sampling lifecycle."""
+
+    def __init__(
+        self,
+        *,
+        on_finished: Optional[Callable[[CalibrationPoint], None]] = None,
+        on_cancelled: Optional[Callable[[], None]] = None,
+        clock: Callable[[], float] = time.perf_counter,
+        wall_clock: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
+    ) -> None:
+        self._on_finished = on_finished
+        self._on_cancelled = on_cancelled
+        self._clock = clock
+        self._wall_clock = wall_clock
+        self._lock = threading.Lock()
+        self._recording = False
+        self._channel: ChannelKey = "fn"
+        self._reference_load_n = 0.0
+        self._direction = "load"
+        self._note = ""
+        self._started_perf = 0.0
+        self._duration_s = 0.0
+        self._started_at = ""
+        self._samples: list[tuple[float, float, float, float, float, float, float, bool]] = []
+        self._rounds: dict[ChannelKey, int] = {}
+        self.points: list[CalibrationPoint] = []
+
+    # ------------------------------------------------------------------ state
+    @property
+    def is_recording(self) -> bool:
+        with self._lock:
+            return self._recording
+
+    @property
+    def channel(self) -> ChannelKey:
+        with self._lock:
+            return self._channel
+
+    @property
+    def requested_duration_s(self) -> float:
+        with self._lock:
+            return self._duration_s
+
+    def elapsed_s(self) -> float:
+        with self._lock:
+            if not self._recording:
+                return 0.0
+            return max(0.0, self._clock() - self._started_perf)
+
+    def progress(self) -> float:
+        elapsed = self.elapsed_s()
+        duration = self.requested_duration_s
+        if duration <= 0:
+            return 0.0
+        return min(1.0, elapsed / duration)
+
+    def note(self) -> str:
+        with self._lock:
+            return self._note
+
+    # ------------------------------------------------------------- lifecycle
+    def reset_session(self) -> None:
+        """Clear per-session state; called when a new acquisition starts."""
+        with self._lock:
+            self._recording = False
+            self._samples = []
+            self._rounds = {}
+            self.points = []
+
+    def start(
+        self,
+        *,
+        channel: ChannelKey,
+        reference_load_n: float,
+        duration_s: float,
+        direction: str = "load",
+        note: str = "",
+    ) -> None:
+        with self._lock:
+            self._channel = channel if channel in CHANNEL_KEYS else "fn"
+            self._reference_load_n = float(reference_load_n)
+            self._duration_s = max(0.05, float(duration_s))
+            self._direction = str(direction or "load")
+            self._note = str(note or "")
+            self._started_perf = self._clock()
+            self._started_at = self._wall_clock().isoformat(timespec="seconds")
+            self._samples = []
+            self._recording = True
+
+    def cancel(self) -> None:
+        with self._lock:
+            if not self._recording:
+                return
+            self._recording = False
+            self._samples = []
+        if self._on_cancelled is not None:
+            self._on_cancelled()
+
+    def feed(self, sample: SessionSample) -> None:
+        """Buffer one sample. Safe to call from the acquisition thread."""
+        with self._lock:
+            if not self._recording:
+                return
+            self._samples.append(
+                (
+                    float(sample.t_rel_s),
+                    float(sample.fn_raw),
+                    float(sample.fx_raw),
+                    float(sample.fy_raw),
+                    float(sample.fn_n),
+                    float(sample.fx_n),
+                    float(sample.fy_n),
+                    bool(sample.valid),
+                )
+            )
+
+    def tick(self) -> Optional[CalibrationPoint]:
+        """Finish on the UI thread once the deadline has passed."""
+        with self._lock:
+            if not self._recording:
+                return None
+            elapsed = self._clock() - self._started_perf
+            if elapsed < self._duration_s:
+                return None
+        return self._finish()
+
+    def finish_now(self) -> Optional[CalibrationPoint]:
+        """Stop early and keep whatever has been captured so far."""
+        with self._lock:
+            if not self._recording:
+                return None
+        return self._finish()
+
+    # --------------------------------------------------------------- internals
+    def _finish(self) -> Optional[CalibrationPoint]:
+        with self._lock:
+            samples = self._samples
+            self._samples = []
+            self._recording = False
+            channel = self._channel
+            reference = self._reference_load_n
+            direction = self._direction
+            note = self._note
+            started_at = self._started_at
+            duration = self._duration_s
+        ended_at = self._wall_clock().isoformat(timespec="seconds")
+        actual_duration = 0.0
+        if samples:
+            actual_duration = max(0.0, samples[-1][0] - samples[0][0])
+
+        block = np.asarray(samples, dtype=np.float64) if samples else np.zeros((0, 8))
+        valid_mask = block[:, 7] > 0.5 if block.size else np.zeros(0, dtype=bool)
+        valid_block = block[valid_mask] if block.size else block
+
+        def axis_stats(column: int) -> AxisStats:
+            if valid_block.size == 0:
+                return AxisStats()
+            mean, std, count = summarise(valid_block[:, column])
+            return AxisStats(mean=mean, std=std, n=count)
+
+        point = CalibrationPoint(
+            channel=channel,
+            reference_load_n=reference,
+            direction=direction,
+            round_index=self._next_round(channel),
+            note=note,
+            started_at=started_at,
+            ended_at=ended_at,
+            duration_s=actual_duration,
+            requested_duration_s=duration,
+            n_total=int(block.shape[0]),
+            n_valid=int(valid_block.shape[0]),
+            raw={"fn": axis_stats(1), "fx": axis_stats(2), "fy": axis_stats(3)},
+            newtons={"fn": axis_stats(4), "fx": axis_stats(5), "fy": axis_stats(6)},
+            status=STATUS_OK if valid_block.shape[0] >= 2 else STATUS_FEW_SAMPLES,
+        )
+        with self._lock:
+            self.points.append(point)
+        if self._on_finished is not None:
+            self._on_finished(point)
+        return point
+
+    def _next_round(self, channel: ChannelKey) -> int:
+        with self._lock:
+            index = self._rounds.get(channel, 0)
+            self._rounds[channel] = index + 1
+            return index

shared/src/orisys_app_shared/force/history.py

@@ -102,6 +102,16 @@ class MetricHistory:
             None for _ in range(self.channels)
         ]
 
+    def clear(self, *, keep_origin: bool = True) -> None:
+        """Drop every sample.
+
+        Unlike :meth:`reset` this keeps the time origin by default, so a buffer
+        rebuilt after a display-unit switch does not restart the X axis at zero.
+        """
+        origin = self._t0 if keep_origin else None
+        self.reset()
+        self._t0 = origin
+
     def _ensure_capacity(self, needed: int) -> None:
         needed = max(needed, self.max_points, 2)
         if needed <= self._capacity:

shared/src/orisys_app_shared/force/session.py

@@ -0,0 +1,304 @@
+"""Full-session force sample store.
+
+The visible plot buffer (:class:`~orisys_app_shared.force.history.MetricHistory`)
+deliberately drops every sample outside its time window, so it can never be the
+source of truth for an export, a calibration point or an audit trail.
+
+This store keeps the whole session. It is fed from the acquisition thread via
+the worker's sink list, so what gets recorded does not depend on the UI refresh
+rate, on which panel happens to be visible, or on how fast the plots redraw.
+
+Design notes
+------------
+* Samples are appended into fixed-size numpy blocks, so ``append`` is O(1) and
+  memory stays bounded. When ``max_samples`` is reached the oldest block is
+  released and ``dropped`` is incremented — callers must surface that counter
+  rather than let data vanish silently.
+* Both the raw and the calibrated triplet are stored per sample, so switching
+  the display unit never has to reinterpret old data.
+* Invalid samples (a failed readout) are stored with ``valid=False`` and NaN
+  values. They are never silently rewritten to zero, and consumers that build
+  calibration points or fits must filter on ``valid``.
+"""
+
+from __future__ import annotations
+
+import threading
+from dataclasses import dataclass
+from typing import Iterable, Iterator, Optional
+
+import numpy as np
+
+#: Stands in for "this value is not available / not valid".
+NA = float("nan")
+
+DEFAULT_BLOCK_SIZE = 8192
+#: ~4.5 h at 30 Hz of three raw + three calibrated float64 channels.
+DEFAULT_MAX_SAMPLES = 500_000
+
+# Column layout of the packed store.
+_COL_FRAME_ID = 0
+_COL_T = 1
+_COL_FN_RAW = 2
+_COL_FX_RAW = 3
+_COL_FY_RAW = 4
+_COL_FN_N = 5
+_COL_FX_N = 6
+_COL_FY_N = 7
+_COL_VALID = 8
+_COLUMNS = 9
+
+
+@dataclass(frozen=True)
+class SessionSample:
+    """One acquisition step, as handed to the record channel."""
+
+    frame_id: int
+    t_rel_s: float
+    fn_raw: float = NA
+    fx_raw: float = NA
+    fy_raw: float = NA
+    fn_n: float = NA
+    fx_n: float = NA
+    fy_n: float = NA
+    valid: bool = True
+    calibration_id: str = ""
+    error: str = ""
+
+    def raw_triplet(self) -> tuple[float, float, float]:
+        return (self.fn_raw, self.fx_raw, self.fy_raw)
+
+    def newton_triplet(self) -> tuple[float, float, float]:
+        return (self.fn_n, self.fx_n, self.fy_n)
+
+    def has_newton(self) -> bool:
+        return all(np.isfinite(v) for v in self.newton_triplet())
+
+    def has_raw(self) -> bool:
+        return all(np.isfinite(v) for v in self.raw_triplet())
+
+
+@dataclass(frozen=True)
+class SessionSlice:
+    """A copy of a time range, ready for plotting or analysis."""
+
+    times: np.ndarray
+    raw: np.ndarray
+    newtons: np.ndarray
+    valid: np.ndarray
+    frame_ids: np.ndarray
+    truncated: bool = False
+
+    def __len__(self) -> int:
+        return int(self.times.size)
+
+    def value_for_unit(self, unit: str) -> np.ndarray:
+        """Return the ``(n, 3)`` block matching a display unit."""
+        if str(unit).upper().startswith("N"):
+            return self.newtons
+        return self.raw
+
+
+class SessionStore:
+    """Thread-safe, memory-bounded, whole-session sample store."""
+
+    def __init__(
+        self,
+        *,
+        max_samples: int = DEFAULT_MAX_SAMPLES,
+        block_size: int = DEFAULT_BLOCK_SIZE,
+    ) -> None:
+        self._lock = threading.RLock()
+        self.max_samples = max(2, int(max_samples))
+        self.block_size = max(16, int(block_size))
+        self._blocks: list[np.ndarray] = []
+        self._count = 0
+        self._dropped = 0
+        self._session_id = 0
+        self._calibration_id = ""
+        self._t0: Optional[float] = None
+        self._last_t: Optional[float] = None
+        self._invalid = 0
+
+    # ------------------------------------------------------------- lifecycle
+    def reset(self, *, session_id: int = 0, calibration_id: str = "") -> None:
+        with self._lock:
+            self._blocks = []
+            self._count = 0
+            self._dropped = 0
+            self._invalid = 0
+            self._session_id = int(session_id)
+            self._calibration_id = str(calibration_id or "")
+            self._t0 = None
+            self._last_t = None
+
+    def set_calibration_id(self, calibration_id: str) -> None:
+        with self._lock:
+            self._calibration_id = str(calibration_id or "")
+
+    # ---------------------------------------------------------------- writing
+    def append(self, sample: SessionSample) -> None:
+        """Record one sample. Never raises on bad values — it marks them invalid."""
+        with self._lock:
+            if self._t0 is None:
+                self._t0 = float(sample.t_rel_s)
+            t = float(sample.t_rel_s) - self._t0
+            valid = 1.0 if sample.valid else 0.0
+            if not sample.valid:
+                self._invalid += 1
+            row = (
+                int(sample.frame_id),
+                t,
+                float(sample.fn_raw),
+                float(sample.fx_raw),
+                float(sample.fy_raw),
+                float(sample.fn_n),
+                float(sample.fx_n),
+                float(sample.fy_n),
+                valid,
+            )
+            self._write_row(row)
+            self._last_t = float(sample.t_rel_s)
+
+    def _write_row(self, row: tuple) -> None:
+        if self._count % self.block_size == 0:
+            self._blocks.append(
+                np.full((self.block_size, _COLUMNS), np.nan, dtype=np.float64)
+            )
+        block = self._blocks[-1]
+        offset = self._count % self.block_size
+        block[offset, :] = row
+        self._count += 1
+        if self._count > self.max_samples:
+            self._release_oldest()
+
+    def _release_oldest(self) -> None:
+        """Drop whole blocks from the front so memory stays bounded.
+
+        The block currently being written is never released, so the most recent
+        ``block_size`` samples always survive.
+        """
+        while self._count > self.max_samples and len(self._blocks) > 1:
+            self._blocks.pop(0)
+            self._count -= self.block_size
+            self._dropped += self.block_size
+        if self._count < 0:
+            self._count = 0
+
+    # ---------------------------------------------------------------- reading
+    @property
+    def count(self) -> int:
+        with self._lock:
+            return self._count
+
+    @property
+    def dropped(self) -> int:
+        """Samples evicted by the memory bound — must be surfaced, never hidden."""
+        with self._lock:
+            return self._dropped
+
+    @property
+    def invalid(self) -> int:
+        with self._lock:
+            return self._invalid
+
+    @property
+    def session_id(self) -> int:
+        with self._lock:
+            return self._session_id
+
+    @property
+    def calibration_id(self) -> str:
+        with self._lock:
+            return self._calibration_id
+
+    @property
+    def duration_s(self) -> float:
+        with self._lock:
+            if self._last_t is None or self._t0 is None:
+                return 0.0
+            return float(self._last_t - self._t0)
+
+    def _packed(self) -> np.ndarray:
+        if self._count == 0:
+            return np.empty((0, _COLUMNS), dtype=np.float64)
+        blocks: list[np.ndarray] = []
+        remaining = self._count
+        for block in self._blocks:
+            take = min(remaining, block.shape[0])
+            blocks.append(block[:take])
+            remaining -= take
+            if remaining <= 0:
+                break
+        return np.vstack(blocks) if len(blocks) > 1 else blocks[0]
+
+    def data(self) -> SessionSlice:
+        """The whole stored session (used by export)."""
+        with self._lock:
+            return self._slice_from(self._packed())
+
+    def tail(self, seconds: float) -> SessionSlice:
+        """The last ``seconds`` of the session."""
+        with self._lock:
+            packed = self._packed()
+        if packed.shape[0] == 0:
+            return self._slice_from(packed)
+        t_end = float(packed[-1, _COL_T])
+        mask = packed[:, _COL_T] >= (t_end - max(0.0, float(seconds)))
+        return self._slice_from(packed[mask])
+
+    def window(self, t0: float, t1: float) -> SessionSlice:
+        with self._lock:
+            packed = self._packed()
+        if packed.shape[0] == 0:
+            return self._slice_from(packed)
+        mask = (packed[:, _COL_T] >= float(t0)) & (packed[:, _COL_T] <= float(t1))
+        return self._slice_from(packed[mask])
+
+    def _slice_from(self, packed: np.ndarray) -> SessionSlice:
+        if packed.shape[0] == 0:
+            empty = np.empty(0, dtype=np.float64)
+            return SessionSlice(
+                times=empty,
+                raw=np.empty((0, 3)),
+                newtons=np.empty((0, 3)),
+                valid=np.empty(0, dtype=bool),
+                frame_ids=np.empty(0, dtype=np.int64),
+            )
+        return SessionSlice(
+            times=packed[:, _COL_T].copy(),
+            raw=packed[:, [_COL_FN_RAW, _COL_FX_RAW, _COL_FY_RAW]].copy(),
+            newtons=packed[:, [_COL_FN_N, _COL_FX_N, _COL_FY_N]].copy(),
+            valid=packed[:, _COL_VALID] > 0.5,
+            frame_ids=packed[:, _COL_FRAME_ID].astype(np.int64, copy=True),
+            truncated=self._dropped > 0,
+        )
+
+    def iter_valid(self, unit: str = "raw") -> Iterator[tuple[float, float, float, float]]:
+        """Yield ``(t_rel_s, v0, v1, v2)`` for valid samples only.
+
+        Calibration fitting uses this so a failed readout can never be averaged
+        in as a zero.
+        """
+        slice_ = self.data()
+        values = slice_.value_for_unit(unit)
+        for index in range(len(slice_)):
+            if not bool(slice_.valid[index]):
+                continue
+            row = values[index]
+            if not all(np.isfinite(v) for v in row):
+                continue
+            yield (
+                float(slice_.times[index]),
+                float(row[0]),
+                float(row[1]),
+                float(row[2]),
+            )
+
+
+def summarise(values: Iterable[float]) -> tuple[float, float, int]:
+    """Return ``(mean, std, n)`` over the finite values, NaN when empty."""
+    block = np.asarray([v for v in values if np.isfinite(v)], dtype=np.float64)
+    if block.size == 0:
+        return (NA, NA, 0)
+    return (float(block.mean()), float(block.std()), int(block.size))