diff --git a/apps/studio/config/compound_eye.png b/apps/studio/config/compound_eye.png index b3f37b3..d3977c6 100644 Binary files a/apps/studio/config/compound_eye.png and b/apps/studio/config/compound_eye.png differ diff --git a/apps/studio/src/orisys_studio/recording.py b/apps/studio/src/orisys_studio/recording.py index 17c8254..5f0ef2a 100644 --- a/apps/studio/src/orisys_studio/recording.py +++ b/apps/studio/src/orisys_studio/recording.py @@ -199,6 +199,7 @@ class RecordingCoordinator: """Start a recording. Raises when the directory cannot be created.""" directory = self.recorder.start( root=root, + config=config, session_id=int(self.session_store.session_id), metadata=metadata, calibration_id=calibration_id, diff --git a/shared/src/orisys_app_shared/force/source.py b/shared/src/orisys_app_shared/force/source.py index 8b1192b..1d7d94c 100644 --- a/shared/src/orisys_app_shared/force/source.py +++ b/shared/src/orisys_app_shared/force/source.py @@ -63,7 +63,7 @@ class SourceWindow: means.append(NA) stds.append(NA) continue - peaks.append(float(np.nanmax(column))) + peaks.append(float(column[np.argmax(np.abs(column))])) means.append(float(np.nanmean(column))) stds.append(float(np.nanstd(column))) span = float(self.times[-1] - self.times[0]) if self.times.size > 1 else 0.0 diff --git a/shared/src/orisys_app_shared/record/depth.py b/shared/src/orisys_app_shared/record/depth.py index 6ec93e0..12f43b9 100644 --- a/shared/src/orisys_app_shared/record/depth.py +++ b/shared/src/orisys_app_shared/record/depth.py @@ -186,9 +186,8 @@ class DepthRecomputer: still produced, and the caller is told it may differ from live). Finding a reset event is **not** by itself a recovered state: the reset - frame has to exist in the index *and* in the video, and it has to be in - the same segment, because the SDK cannot carry its state across a segment - boundary. + frame has to exist in the index *and* in the video. Every intervening + segment must be read with the same processor to preserve its state. """ warnings: list[str] = [] segment_path = self.reader.segment_path(frame.segment) @@ -205,6 +204,12 @@ class DepthRecomputer: # The boundary frame is not in the video, so the state at that # point cannot be rebuilt: replay from here is an approximation. warnings.append("reset_frame_not_in_video") + elif any( + entry.dropped or not entry.in_video + for entry in self.reader.info().frames + if origin.frame_id <= entry.frame_id <= frame.frame_id + ): + warnings.append("state_missing_frames") elif origin.segment == frame.segment: plan = ReplayPlan( segment=segment_path, @@ -228,21 +233,24 @@ class DepthRecomputer: # The reset is in an earlier segment. The processor survives a # source swap, so the state *can* be carried: read the tail of # the reset's segment, then continue in the target's. - reset_segment = self.reader.segment_path(origin.segment) - last_of_reset_segment = self._last_segment_frame(origin.segment) - if reset_segment is None or last_of_reset_segment is None: + steps = [] + for segment in range(int(origin.segment), int(frame.segment) + 1): + source = self.reader.segment_path(segment) + last = stop_at if segment == frame.segment else self._last_segment_frame(segment) + if source is None or not source.is_file() or last is None: + steps = [] + break + steps.append(ReplayStep( + source=source, + to_frame=int(last), + reset_after=int(origin.segment_frame) if segment == origin.segment else None, + )) + if not steps: warnings.append("state_across_segment") else: plan = ReplayPlan( segment=segment_path, - steps=( - ReplayStep( - source=reset_segment, - to_frame=int(last_of_reset_segment), - reset_after=int(origin.segment_frame), - ), - ReplayStep(source=segment_path, to_frame=stop_at), - ), + steps=tuple(steps), stop_at=stop_at, reset_frame_id=int(reset_frame), anchor="reset", @@ -405,7 +413,7 @@ class DepthRecomputer: while self._position < target: depth = self._read_one() if depth is None: - break + raise RuntimeError(f"replay source ended before frame {target}: {step.source}") last = depth if step.reset_after is not None and self._position == int(step.reset_after): # The operator's zeroing: the live run cleared the tracker here. diff --git a/shared/src/orisys_app_shared/record/experiment.py b/shared/src/orisys_app_shared/record/experiment.py index bb503ee..36c9aee 100644 --- a/shared/src/orisys_app_shared/record/experiment.py +++ b/shared/src/orisys_app_shared/record/experiment.py @@ -32,7 +32,8 @@ import queue import threading import time from collections import deque -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace +from uuid import uuid4 from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional @@ -308,6 +309,7 @@ class ExperimentRecorder: root: Path, session_id: int, metadata: dict[str, Any], + config: Optional[RecordConfig] = None, calibration_id: str = "", calibration_version: str = "", preroll=(), @@ -325,10 +327,13 @@ class ExperimentRecorder: f"cannot start recording from state {self._state!r}" ) self._state = STATE_STARTING + # A running writer owns its configuration. Copy only after the + # lifecycle checks, so caller mutations cannot change this run. + self.config = replace(config if config is not None else self.config) stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") - directory = Path(root) / f"{stamp}_session{session_id:04d}" + directory = Path(root) / f"{stamp}_session{session_id:04d}_{uuid4().hex}" try: - directory.mkdir(parents=True, exist_ok=True) + directory.mkdir(parents=True, exist_ok=False) except Exception as exc: # noqa: BLE001 - reported to the caller # A directory that cannot be created is a *terminal* failure, not a # stuck ``starting``: the operator can fix the path and press record diff --git a/shared/src/orisys_app_shared/record/index.py b/shared/src/orisys_app_shared/record/index.py index 83d3cf5..e42a226 100644 --- a/shared/src/orisys_app_shared/record/index.py +++ b/shared/src/orisys_app_shared/record/index.py @@ -11,15 +11,15 @@ the interval being drawn. It also builds a multi-level overview (per-bucket min/max/count) while streaming the file once, which is what a whole-session strip needs and what it cannot get by sampling a few thousand rows. -Nothing here touches Qt, and nothing here trusts the file to be in order: a -legacy recording is sorted at load time (see ``playback.RecordingReader``). +Nothing here touches Qt. Unordered recordings keep a stable time-to-byte-offset +mapping, shared by all interval reads and statistics. """ from __future__ import annotations import csv import logging -from bisect import bisect_left, bisect_right +from array import array from dataclasses import dataclass, field from pathlib import Path from typing import Iterable, Iterator, Optional @@ -95,7 +95,7 @@ class ForceWindow: means.append(float("nan")) stds.append(float("nan")) continue - peaks.append(float(np.nanmax(column))) + peaks.append(float(column[np.argmax(np.abs(column))])) means.append(float(np.nanmean(column))) stds.append(float(np.nanstd(column))) span = 0.0 @@ -153,6 +153,10 @@ class RecordingIndex: self._rows = 0 self._byte_size = 0 self._built = False + self._offsets = None # Only unordered files need one byte offset per row. + self._columns = {} + self.ordered = True + self.dropped_rows = 0 # ------------------------------------------------------------------ building @property @@ -175,35 +179,43 @@ class RecordingIndex: valid: list[bool] = [] frame_ids: list[int] = [] checkpoints: list[_Checkpoint] = [] + offsets = array("Q") + self.dropped_rows = 0 if path.is_file(): with path.open("rb") as handle: header = handle.readline().decode("utf-8", "replace").strip() columns = {name: index for index, name in enumerate(header.split(","))} - time_column = columns.get("t_rel_s", 2) - valid_column = columns.get("valid", 9) - frame_column = columns.get("frame_id", 1) + self._columns = columns offset = handle.tell() for row, line in enumerate(handle): next_offset = handle.tell() - text = line.decode("utf-8", "replace").rstrip("\r\n") - if row % self.checkpoint_rows == 0: + record = self._parse_row(line) + if record is None: + self.dropped_rows += 1 + offset = next_offset + continue + frame_id, timestamp, _raw, _newtons, is_valid = record + logical_row = len(times) + if logical_row % self.checkpoint_rows == 0: checkpoints.append( - _Checkpoint(row=row, time=_time_of(text), offset=offset) + _Checkpoint(row=logical_row, time=timestamp, offset=offset) ) - fields = next(csv.reader([text])) if text else [] - if len(fields) > max(time_column, valid_column, frame_column): - try: - times.append(float(fields[time_column])) - valid.append(fields[valid_column].strip() == "1") - frame_ids.append(int(fields[frame_column])) - except (TypeError, ValueError): - times.append(float("nan")) - valid.append(False) - frame_ids.append(-1) + times.append(timestamp) + valid.append(is_valid) + frame_ids.append(frame_id) + offsets.append(offset) offset = next_offset self.times = np.asarray(times, dtype=np.float64) self.valid = np.asarray(valid, dtype=bool) self.frame_ids = np.asarray(frame_ids, dtype=np.int64) + self.ordered = bool(np.all(self.times[1:] >= self.times[:-1])) + self._offsets = None + if not self.ordered: + order = np.argsort(self.times, kind="stable") + self.times = self.times[order] + self.valid = self.valid[order] + self.frame_ids = self.frame_ids[order] + self._offsets = np.asarray(offsets, dtype=np.uint64)[order] self._checkpoints = checkpoints self._rows = int(self.times.size) try: @@ -213,6 +225,21 @@ class RecordingIndex: self._built = True return self + def _parse_row(self, line: bytes): + """Parse atomically; malformed rows never shift one column alone.""" + try: + fields = next(csv.reader([line.decode("utf-8", "replace")])) + def cell(name): + return fields[self._columns[name]] + frame_id = int(cell("frame_id")) + 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) + return frame_id, timestamp, values[:3], values[3:], cell("valid").strip() == "1" + except (KeyError, IndexError, ValueError, StopIteration, csv.Error): + return None + def ensure_built(self) -> "RecordingIndex": if not self._built: self.build() @@ -220,6 +247,7 @@ class RecordingIndex: # ------------------------------------------------------------------- queries def extent(self) -> tuple[float, float]: + self.ensure_built() if self._rows == 0: return (0.0, 0.0) return (float(self.times[0]), float(self.times[-1])) @@ -235,8 +263,8 @@ class RecordingIndex: return (0, 0) if float(t1) < float(t0): t0, t1 = t1, t0 - start = max(0, bisect_left(self.times.tolist(), float(t0)) - 1) - stop = min(self._rows, bisect_right(self.times.tolist(), float(t1))) + start = int(self.times.searchsorted(float(t0), side="left")) + stop = int(self.times.searchsorted(float(t1), side="right")) return (start, max(start, stop)) def _checkpoint_before(self, row: int) -> _Checkpoint: @@ -260,9 +288,9 @@ class RecordingIndex: With ``max_points`` the rows are decimated **while streaming**: each bucket contributes the extreme row of every axis in ``unit``, so a long - interval never has to exist in memory in full. The extremes are taken in - both the raw and the requested unit's axis, which is why a shear spike - survives a flat normal channel. + interval never has to exist in memory in full. Extrema are taken independently on all three axes in the requested unit, + so a shear spike survives a flat normal channel. A bucket keeps at most + nine rows; budgets smaller than nine may therefore return nine points. """ self.ensure_built() start = max(0, min(int(start), self._rows)) @@ -276,72 +304,54 @@ class RecordingIndex: newton_unit = str(unit).upper().startswith("N") budget = int(max_points) if max_points else 0 total = stop - start - bucket = ( - max(1, total // max(1, budget // 6)) - if budget and total > budget * 2 - else 1 - ) - - checkpoint = self._checkpoint_before(start) - pending: list[tuple] = [] + # At most six extrema, two endpoints and a missing-data marker per + # bucket. No bucket holds a list proportional to the recording length. + bucket = max(1, (total + max(1, budget // 9) - 1) // max(1, budget // 9)) if budget and total > budget else 1 picked: list[tuple] = [] + extrema = {} + first = last = missing = None + count = 0 - def flush() -> None: - if not pending: - return + def flush(): + nonlocal first, last, missing, count + chosen = {item[0]: item for item in extrema.values()} + for item in (first, last, missing): + if item is not None: + chosen[item[0]] = item + picked.extend(item[1:] for item in sorted(chosen.values(), key=lambda item: item[0])) + extrema.clear() + first = last = missing = None + count = 0 + + for item in self.iter_rows(start, stop): + # item: logical row, frame id, time, raw, newtons, validity. if bucket == 1: - picked.extend(pending) - else: - values = np.asarray( - [row[3] if newton_unit else row[2] for row in pending], - dtype=np.float64, - ) - for axis in range(3): - column = values[:, axis] - finite = np.isfinite(column) - if not finite.any(): - continue - ordered = np.where(finite, column, -np.inf) - for index in {int(np.argmin(ordered)), int(np.argmax(ordered))}: - picked.append(pending[index]) - pending.clear() - - with path.open("rb") as handle: - # Every checkpoint offset points at a *data* row: the build records - # ``tell()`` after the header, so there is nothing to skip here. - handle.seek(checkpoint.offset) - for row, line in enumerate(handle, start=checkpoint.row): - if row >= stop: - break - if row < start: + picked.append(item[1:]) + continue + if first is None: + first = item + last = item + values = item[4] if newton_unit else item[3] + for axis, value in enumerate(values): + if not item[5] or not np.isfinite(value): + if missing is None: + missing = item continue - fields = next(csv.reader([line.decode("utf-8", "replace")])) - if len(fields) < 11: - continue - try: - t_rel = float(fields[2]) - frame_id = int(fields[1]) - except (TypeError, ValueError): - continue - values = [_parse(fields[index]) for index in (3, 4, 5, 6, 7, 8)] - pending.append( - ( - frame_id, - t_rel, - tuple(values[:3]), - tuple(values[3:]), - fields[9].strip() == "1", - ) - ) - if len(pending) >= bucket: - flush() + for side in ('min', 'max'): + key = (axis, side) + previous = extrema.get(key) + old = previous[4 if newton_unit else 3][axis] if previous is not None else None + if old is None or (value < old if side == 'min' else value > old): + extrema[key] = item + count += 1 + if count >= bucket: + flush() flush() if not picked: return _empty_window() # A row can be the extreme of several axes; keep one copy, in time order. - unique = {row[0]: row for row in picked} - rows = sorted(unique.values(), key=lambda row: row[1]) + rows = picked # Already in stable time order; duplicate frame IDs are not deduplicated. return ForceWindow( times=np.asarray([row[1] for row in rows], dtype=np.float64), raw=np.asarray([row[2] for row in rows], dtype=np.float64).reshape(-1, 3), @@ -352,36 +362,68 @@ class RecordingIndex: ) def iter_rows(self, start: int = 0, stop: Optional[int] = None) -> Iterator[tuple]: - """Yield ``(row, frame_id, t_rel_s, raw3, newtons3, valid)`` in file order.""" + """Yield logical rows in stable time order, seeking only the selection.""" self.ensure_built() - path = self.path - if not path.is_file(): - return + start = max(0, int(start)) stop = self._rows if stop is None else min(int(stop), self._rows) - with path.open("rb") as handle: - handle.readline() # header - for row, line in enumerate(handle): + if start >= stop or not self.path.is_file(): + return + with self.path.open("rb") as handle: + if self._offsets is not None: + for row in range(start, stop): + handle.seek(int(self._offsets[row])) + record = self._parse_row(handle.readline()) + if record is not None: + yield (row, *record) + return + checkpoint = self._checkpoint_before(start) + handle.seek(checkpoint.offset) + row = checkpoint.row + for line in handle: + record = self._parse_row(line) + if record is None: + continue + if row >= start: + yield (row, *record) + row += 1 if row >= stop: return - if row < start: - continue - fields = next(csv.reader([line.decode("utf-8", "replace")])) - if len(fields) < 11: - continue - try: - frame_id = int(fields[1]) - t_rel = float(fields[2]) - except (TypeError, ValueError): - continue - values = [_parse(fields[index]) for index in (3, 4, 5, 6, 7, 8)] - yield ( - row, - frame_id, - t_rel, - tuple(values[:3]), - tuple(values[3:]), - fields[9].strip() == "1", - ) + + def statistics(self, t0: float, t1: float, *, unit: str = "raw") -> WindowStats: + """Exact selected-row statistics with constant working memory.""" + start, stop = self.row_range(t0, t1) + counts = np.zeros(3, dtype=np.int64) + means = np.zeros(3) + m2 = np.zeros(3) + peaks = np.full(3, np.nan) + samples = 0 + first = last = None + newton = str(unit).upper().startswith("N") + for _, _, timestamp, raw, newtons, valid in self.iter_rows(start, stop): + if first is None: + first = timestamp + last = timestamp + if not valid: + continue + values = np.asarray(newtons if newton else raw) + mask = np.isfinite(values) + samples += int(mask.any()) + for axis in np.flatnonzero(mask): + value = values[axis] + counts[axis] += 1 + delta = value - means[axis] + means[axis] += delta / counts[axis] + m2[axis] += delta * (value - means[axis]) + if not np.isfinite(peaks[axis]) or abs(value) > abs(peaks[axis]): + peaks[axis] = value + available = counts > 0 + stds = np.full(3, np.nan) + stds[available] = np.sqrt(np.maximum(0, m2[available] / counts[available])) + means[~available] = np.nan + return WindowStats( + duration_s=0.0 if first is None else last - first, + samples=samples, peaks=tuple(peaks), means=tuple(means), stds=tuple(stds), + ) # ------------------------------------------------------------------ overview def overview(self, *, unit: str = "raw", buckets: int = DEFAULT_BUCKETS) -> Overview: diff --git a/shared/src/orisys_app_shared/record/playback.py b/shared/src/orisys_app_shared/record/playback.py index 5801f3e..8b9a565 100644 --- a/shared/src/orisys_app_shared/record/playback.py +++ b/shared/src/orisys_app_shared/record/playback.py @@ -179,8 +179,8 @@ class RecordingReader: newtons = window.newtons valid = window.valid frame_ids = window.frame_ids - forces_ordered = True - dropped = 0 + forces_ordered = index.ordered + dropped = index.dropped_rows else: (times, raw, newtons, valid, frame_ids, dropped, forces_ordered) = ( self._load_forces() @@ -244,71 +244,10 @@ class RecordingReader: return rows, ordered def _load_forces(self): - path = self.directory / "forces.csv" - if not path.is_file(): - # A directory with no data (an empty or half-written recording) is - # readable: it simply has no samples, and nothing is out of order. - empty = np.empty(0, dtype=np.float64) - return ( - empty, - np.empty((0, 3)), - np.empty((0, 3)), - np.empty(0, bool), - np.empty(0, int), - 0, - True, - ) - - def parse(text: str) -> float: - text = (text or "").strip() - if not text: - # An empty cell means "not available"; it must never become 0. - return float("nan") - try: - return float(text) - except ValueError: - return float("nan") - - times: list[float] = [] - raw: list[tuple[float, float, float]] = [] - newtons: list[tuple[float, float, float]] = [] - valid: list[bool] = [] - frame_ids: list[int] = [] - dropped = 0 - with path.open(encoding="utf-8", newline="") as handle: - for row in csv.DictReader(handle): - try: - times.append(float(row["t_rel_s"])) - frame_ids.append(int(row["frame_id"])) - except (KeyError, TypeError, ValueError): - dropped += 1 - continue - raw.append( - (parse(row.get("FN_raw")), parse(row.get("FX_raw")), parse(row.get("FY_raw"))) - ) - newtons.append( - (parse(row.get("FN_N")), parse(row.get("FX_N")), parse(row.get("FY_N"))) - ) - valid.append((row.get("valid") or "0").strip() == "1") - times_arr = np.asarray(times, dtype=np.float64) - raw_arr = np.asarray(raw, dtype=np.float64).reshape(-1, 3) - newtons_arr = np.asarray(newtons, dtype=np.float64).reshape(-1, 3) - valid_arr = np.asarray(valid, dtype=bool) - ids_arr = np.asarray(frame_ids, dtype=np.int64) - - # Older recordings can have rows out of time order (the writer used to - # drain a backlog before the queue). Every window query below assumes - # sorted time, so sort once here and report it instead of answering - # ``searchsorted`` on unsorted input. - ordered = _is_time_ordered(times) - if not ordered and times_arr.size: - order = np.argsort(times_arr, kind="stable") - times_arr = times_arr[order] - raw_arr = raw_arr[order] - newtons_arr = newtons_arr[order] - valid_arr = valid_arr[order] - ids_arr = ids_arr[order] - return times_arr, raw_arr, newtons_arr, valid_arr, ids_arr, dropped, ordered + index = self.index() + window = index.read_rows(0, index.rows) + return (window.times, window.raw, window.newtons, window.valid, + window.frame_ids, index.dropped_rows, index.ordered) # ------------------------------------------------------------------ queries def frame_at(self, t_rel_s: float) -> Optional[RecordingFrame]: @@ -475,7 +414,7 @@ class RecordingReader: def statistics(self, t0: float, t1: float, *, unit: str = "raw"): """Statistics of an interval, computed from the interval's own rows.""" - return self.window(t0, t1, unit=unit).statistics(unit) + return self.index().statistics(t0, t1, unit=unit) # ------------------------------------------------------------------ policy def depth_policy(self) -> dict: