From 817f83056506179f66ff0a42b15c921e17c1e1b5 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Sun, 13 Sep 2026 13:14:10 +0800 Subject: [PATCH] record the full experiment: video, CSV and metadata Phase 1 of the review asked for one directory per experiment, a force CSV, video of the pre-processing image, and metadata that ties them together. The record channel from the previous change already kept the whole session; this adds the on-disk format. What - Add record/experiment.py: ExperimentRecorder writes one directory per session containing session.json, forces.csv, frames.csv, events.csv, raw_NNNN.mp4 (segmented) and an optional lossless PNG sequence. - forces.csv uses the reviewed column order (session_id, frame_id, t_rel_s, FN_raw, FX_raw, FY_raw, FN_N, FX_N, FY_N, valid, calibration_id) and leaves a cell empty when a value is unavailable: an uncalibrated newton value or a failed readout is never written as 0. - frames.csv is the frame-index mapping: frame_id, t_rel_s, acquisition wall clock, segment, segment_frame, in_video. forces.csv and frames.csv share the packet frame_id and the same t_rel_s origin, so video and data can be aligned after dropped or segmented frames. - session.json carries format version, session id, start/end time in UTC, device and config (with a sha256 of the config file), app/shared/SDK versions, calibration reference, axis convention, filter parameters, video codec and the lossy-MP4 caveat, plus the event log and end-of-run statistics. It is written once at start so a crash still leaves identifiable metadata. - Add a record control to the toolbar: mode (video+data / data only) and a start/stop toggle that stays enabled while recording. - Log zeroing, calibration application and session end as timestamped events. Why - The recorder is fed from the worker sink list on the acquisition thread and only appends to a bounded queue, so disk I/O can never stall acquisition. - On_packet copies the raw frame instead of holding a reference: the packet frame may alias a buffer the SDK reuses between frames. - A queue overflow is counted and alarmed (rate limited), and a write or encode failure is counted, reported and after 5 consecutive errors stops the recording with failed=true rather than silently losing data. Verification - shared/tests/test_experiment_recorder.py: 12 cases covering the directory layout, the CSV contract, frame-id/time-base agreement, segment rolling, data-only and lossless modes, queue overflow accounting, encoder failure, metadata fields, crash-before-stop, and that on_packet never blocks. - tools/ui_check.py gains an "experiment record" section (169 checks, 0 failures overall), driving a recording through the UI and asserting the layout, the empty-cell rule and the event sequence. - apps/studio/tests/test_ui_smoke.py mirrors those invariants. - docs/record-channel.md documents the format and the remaining phase-2 work. Not in this change: CSV export of a selected time range, full-session curve browsing, calibration diagnostics (outlier exclusion, fit residuals) and PyOpenGL 3D. --- .gitignore | 1 + apps/studio/config/locales/en_US.json | 16 + apps/studio/config/locales/zh_CN.json | 16 + apps/studio/docs/record-channel.md | 58 +- apps/studio/src/orisys_studio/app_window.py | 190 ++++- apps/studio/src/orisys_studio/ui/toolbar.py | 87 ++- apps/studio/tests/test_ui_smoke.py | 89 ++- apps/studio/tools/ui_check.py | 127 +++- shared/src/orisys_app_shared/paths.py | 7 + .../src/orisys_app_shared/record/__init__.py | 25 + .../orisys_app_shared/record/experiment.py | 653 ++++++++++++++++++ shared/tests/test_experiment_recorder.py | 278 ++++++++ 12 files changed, 1531 insertions(+), 16 deletions(-) create mode 100644 shared/src/orisys_app_shared/record/__init__.py create mode 100644 shared/src/orisys_app_shared/record/experiment.py create mode 100644 shared/tests/test_experiment_recorder.py diff --git a/.gitignore b/.gitignore index a8b58ae..b4edad3 100644 --- a/.gitignore +++ b/.gitignore @@ -21,6 +21,7 @@ apps/studio/preference.json .piptmp/ .pytestlib/ .pytest_cache/ +shared/tests/_scratch/ # IDE / OS .idea/ diff --git a/apps/studio/config/locales/en_US.json b/apps/studio/config/locales/en_US.json index daeaf18..ddb44c4 100644 --- a/apps/studio/config/locales/en_US.json +++ b/apps/studio/config/locales/en_US.json @@ -104,6 +104,22 @@ "fy": "FY", "no_data": "—" }, + "record": { + "label": "Record", + "mode_video_data": "Video + data", + "mode_data_only": "Data only", + "mode_tip": "Video records the pre-processing camera input (no arrows or text overlays). MP4 is a lossy container, not a lossless copy of the raw data.", + "start_label": "● Start recording", + "stop_label": "■ Stop recording", + "start_tip": "Record this experiment: video, force data and metadata into one directory", + "stop_tip": "Stop recording and write session.json (always available)", + "requires_run": "Press Run to start acquisition before recording", + "started": "Recording started ({mode})", + "start_failed": "Could not start recording: {error}", + "finished": "Recording finished — {frames} frames / {segments} segment(s) in {path}", + "finished_with_problems": "Recording finished with problems: {detail}", + "alarm": "Recording problem: {detail}" + }, "cal": { "weight": "Weight (N)", "duration": "Record duration", diff --git a/apps/studio/config/locales/zh_CN.json b/apps/studio/config/locales/zh_CN.json index 8af4641..652f51d 100644 --- a/apps/studio/config/locales/zh_CN.json +++ b/apps/studio/config/locales/zh_CN.json @@ -104,6 +104,22 @@ "fy": "FY", "no_data": "—" }, + "record": { + "label": "记录", + "mode_video_data": "视频 + 数据", + "mode_data_only": "仅数据", + "mode_tip": "视频记录的是处理前的相机输入图像（不含箭头与文字叠加）。MP4 是有损封装，不等于无损原始数据。", + "start_label": "● 开始记录", + "stop_label": "■ 停止记录", + "start_tip": "开始记录本实验：视频、力数据与元数据写入同一目录", + "stop_tip": "停止记录并写入 session.json（随时可用）", + "requires_run": "请先点击「运行」开始采集，再开始记录", + "started": "已开始记录（{mode}）", + "start_failed": "无法开始记录: {error}", + "finished": "记录完成 — {frames} 帧 / {segments} 段，已写入 {path}", + "finished_with_problems": "记录完成，但存在问题: {detail}", + "alarm": "记录异常: {detail}" + }, "cal": { "weight": "砝码重量 (N)", "duration": "采集时长", diff --git a/apps/studio/docs/record-channel.md b/apps/studio/docs/record-channel.md index 45541f8..c6fbbf2 100644 --- a/apps/studio/docs/record-channel.md +++ b/apps/studio/docs/record-channel.md @@ -104,11 +104,14 @@ cd D:\orisys_ws\orisys-app\apps\studio - `shared/tests/test_record_channel.py`：11 项，覆盖内存上界、双单位、无效隔离、 跨线程 feed、截止时间结算、取消语义。 +- `shared/tests/test_experiment_recorder.py`：12 项，覆盖目录结构、CSV 列序与空值规则、 + 帧号/时间基准一致、分段滚动与帧索引、仅数据模式、无损序列、队列溢出计数与报警、 + 编码失败上报并停止录制、元数据字段、崩溃后仍留有元数据、`on_packet` 不阻塞。 - `shared/tests/test_worker_cancellation.py`：4 项，覆盖非阻塞 stop、线程所有权、 会话交接。 -- `tools/ui_check.py`：新增 `record channel` 段落，断言记录通道在**完全没有 UI 定时器 - 触发**的情况下仍收满样本、隐藏标定面板仍能采样、录制按钮是可用取消入口、 - 深度缺失不回退、色标锁帧。 +- `tools/ui_check.py`：新增 `record channel` 与 `experiment record` 两段，断言记录通道 + 在**完全没有 UI 定时器触发**的情况下仍收满样本、隐藏标定面板仍能采样、录制按钮是可用 + 取消入口、深度缺失不回退、色标锁帧，以及通过界面录制的目录结构与事件序列。 - `apps/studio/tests/test_ui_smoke.py`：上述不变量各有对应 pytest 用例，复用 `ui_check._StuckWorker` 与 `ui_check._fake_packet`，保持单一来源。 @@ -120,11 +123,52 @@ cd D:\orisys_ws\orisys-app\apps\studio --- -## 6. 尚未完成（第二阶段及以后） +## 6. 实验录制（第一阶段剩余部分） -- MP4 分段录制、`frames.csv` 帧索引、`session.json` 元数据（评审第一阶段剩余部分）。 -- CSV 导出（`session_id/frame_id/t_rel_s/FN_raw/…/valid/calibration_id`）——数据模型 - `SessionSample` 已按这些字段设计，`SessionStore.iter_valid()` 已可直接供导出使用。 +`shared/src/orisys_app_shared/record/experiment.py` 的 `ExperimentRecorder` 落盘格式： + +``` +outputs/recordings/_session/ + session.json 元数据 + 事件 + 结束统计 + forces.csv 每帧一行 + frames.csv 帧索引映射（帧号 ↔ 视频分段） + events.csv 带时间戳的操作事件 + raw_0001.mp4 处理前相机图像，按需分段 + frames/ 可选无损 PNG 序列 +``` + +`forces.csv` 列序固定为 +`session_id, frame_id, t_rel_s, FN_raw, FX_raw, FY_raw, FN_N, FX_N, FY_N, valid, calibration_id`。 + +几条硬性约束： + +- **一次实验一个目录**，`session.json` 在**开始时先写一次**，因此即使进程崩溃也留有可识别 + 的元数据；停止时再写一次并补上 `ended_at`、事件与统计。 +- **不静默丢数据**：`on_packet` 只把标量（以及开启视频时的一帧拷贝）放进**有界队列**， + 队列满则计数并报警，绝不阻塞采集；连续写失败达到 `MAX_CONSECUTIVE_ERRORS` 就停止录制 + 并把 `failed` 置位，而不是假装录完了。 +- **帧号与时间统一**：`forces.csv` 与 `frames.csv` 共用 packet 的 `frame_id`，`t_rel_s` + 同源于首帧；`frames.csv` 额外记录采集墙钟时间、落在哪个分段、以及是否进过视频， + 这就是丢帧/分段场景下的重建依据。 +- **未标定或读取失败的单元格留空**，不写 0：`FN_N` 等列在无标定时为空，`valid=0` 的行 + 所有数值列都为空。精度用 `repr(float)`（最短往返表示），界面小数位不影响落盘。 +- **MP4 是有损封装**：`session.json` 的 `video.lossy_note` 明确记录这一点，需要算法复现时 + 可开启 `lossless_frames` 输出 PNG 序列。 +- `raw_frame` 在入队时**必须拷贝**：packet 里的帧可能复用 SDK 的采集缓冲，直接持有引用 + 会让录制内容随后续帧改变。 + +### 已知环境限制 + +本机未安装 pytest，且沙箱拒绝写入工作区之外的路径，`conftest.py` 的 `tmp_path` 因此 +写 prefs 失败。上述 pytest 用例是用等价驱动脚本执行的（参数化用例除外）；`ui_check.py` +覆盖同样的不变量且可完整运行。录制相关的测试把临时目录放在 +`shared/tests/_scratch/`（已加入 `.gitignore`）以避开该限制。 + +--- + +## 7. 尚未完成（第二阶段及以后） + +- CSV 的「导出选中时间段」（整段导出已随录制落盘；区间导出待做）。 - 全程曲线浏览（跟随/历史、框选缩放、降采样保峰值）。 - 标定诊断扩展：稳定后采样、异常点排除/恢复、拟合残差、独立验证点。当前 `CalibrationPoint` 已记录三轴均值/标准差/有效样本数、方向、轮次与实际采样时长。 diff --git a/apps/studio/src/orisys_studio/app_window.py b/apps/studio/src/orisys_studio/app_window.py index ab418d8..22f690b 100644 --- a/apps/studio/src/orisys_studio/app_window.py +++ b/apps/studio/src/orisys_studio/app_window.py @@ -28,9 +28,15 @@ from PyQt5.QtWidgets import ( QWidget, ) +from orisys_app_shared.acquisition.constants import ( + FORCE_SAMPLE_RATE_HZ, + LOWPASS_CUTOFF_HZ, +) from orisys_app_shared.force import select_display_forces from orisys_app_shared.force.calibration_recorder import CalibrationRecorder from orisys_app_shared.force.session import SessionSample, SessionStore +from orisys_app_shared.paths import recordings_dir +from orisys_app_shared.record import ExperimentRecorder, RecordConfig from orisys_app_shared.i18n import set_locale, t from orisys_app_shared.logging import report_error from orisys_app_shared.paths import logo_path, sensor_json_path @@ -125,6 +131,12 @@ class StudioMainWindow(QMainWindow): on_finished=self._on_calibration_point, on_cancelled=self._on_calibration_cancelled, ) + self.recorder = ExperimentRecorder() + self.recorder.set_alarm_handler(self._on_record_alarm) + self._record_alarms: list[str] = [] + self._record_last_report = None + self._record_fps = 30.0 + self._session_counter = 0 self._record_error_count = 0 self._dropped_reported = 0 self._active_calibration_id = "" @@ -224,6 +236,148 @@ class StudioMainWindow(QMainWindow): for worker in (self._box_worker, self._finger_worker): worker.add_sink(self._on_packet_recorded) + # ------------------------------------------------------- experiment record + def _on_record_alarm(self, message: str) -> None: + """Called from the writer thread. Queue it; never touch Qt here.""" + logger.error("Recording alarm: %s", message) + self._record_alarms.append(str(message)) + + def _drain_record_alarms(self) -> None: + while self._record_alarms: + message = self._record_alarms.pop(0) + self._set_phase("error") + self.statusBar().showMessage(t("record.alarm", detail=message)) + + def _on_record_toggled(self, checked: bool) -> None: + if checked: + self._start_recording() + else: + self._stop_recording() + + def _start_recording(self) -> None: + if self.recorder.is_recording: + return + if not self._running: + self.toolbar.set_record_checked(False) + self.statusBar().showMessage(t("record.requires_run")) + return + mode = self.toolbar.record_mode() + config = RecordConfig( + video=(mode == "video_data"), + lossless_frames=bool(self.prefs.get("record_lossless_frames", False)), + fps=float(self._record_fps), + segment_seconds=float(self.prefs.get("record_segment_seconds", 300.0)), + ) + try: + directory = self.recorder.start( + root=recordings_dir(), + session_id=int(self.session_store.session_id), + metadata=self._build_session_metadata(config), + calibration_id=self._active_calibration_id, + ) + except Exception as exc: + self.toolbar.set_record_checked(False) + self.statusBar().showMessage(t("record.start_failed", error=str(exc))) + logger.exception("Could not start recording") + return + self._record_last_report = None + self.recorder.log_event("record_started", f"mode={mode}") + self.toolbar.set_record_status(Path(directory).name) + self.statusBar().showMessage(t("record.started", mode=t(f"record.mode_{mode}"))) + logger.info("Recording started: %s (mode=%s)", directory, mode) + + def _stop_recording(self, *, wait: bool = True) -> None: + if not self.recorder.is_recording: + self.toolbar.set_record_status("") + return + self.recorder.log_event("record_stopped", "") + report = self.recorder.stop(wait=wait) + self._record_last_report = report + self.toolbar.set_record_status("") + self.toolbar.set_record_checked(False) + if report.problems: + self.statusBar().showMessage( + t("record.finished_with_problems", detail="; ".join(report.problems)) + ) + else: + self.statusBar().showMessage( + t( + "record.finished", + frames=report.stats.frames, + segments=report.stats.segments, + path=str(report.directory), + ) + ) + + def _build_session_metadata(self, config: RecordConfig) -> dict: + from orisys_app_shared import API_VERSION, __version__ as shared_version + + from orisys_studio import __version__ as app_version + + sensor_type = self.sensor_picker.sensor_type() + config_ref = self._resolve_config_ref() or "" + info = self._active_worker.latest() if self._active_worker is not None else None + return { + "device": { + "sensor_type": sensor_type, + "config_ref": config_ref, + "config_sha256": self._config_hash(config_ref), + "frame": { + "width": int(getattr(info, "raw_frame", None).shape[1]) + if getattr(info, "raw_frame", None) is not None + else None, + "height": int(getattr(info, "raw_frame", None).shape[0]) + if getattr(info, "raw_frame", None) is not None + else None, + "fps": float(self._record_fps), + }, + }, + "versions": { + "app": app_version, + "shared": shared_version, + "shared_api": API_VERSION, + "sdk": self._sdk_version(), + }, + "calibration": { + "config_ref": config_ref, + "force_calibrated": bool(getattr(info, "force_calibrated", False)), + }, + "conventions": { + "axis_order": ["FN", "FX", "FY"], + "force_unit": "N when calibrated, otherwise raw counts", + "time_base": "t_rel_s is relative to the first recorded frame", + "frame_id": "monotonic per acquisition session", + }, + "filter": { + "lowpass_cutoff_hz": float(LOWPASS_CUTOFF_HZ), + "sample_rate_hz": float(FORCE_SAMPLE_RATE_HZ), + }, + "recording_mode": self.toolbar.record_mode(), + "video_config": config.as_dict(), + } + + @staticmethod + def _config_hash(config_ref: str) -> str: + try: + import hashlib + from pathlib import Path as _Path + + path = _Path(str(config_ref)) + if path.is_file(): + return hashlib.sha256(path.read_bytes()).hexdigest() + except Exception: # noqa: BLE001 - metadata must never block recording + pass + return "" + + @staticmethod + def _sdk_version() -> str: + try: + import orisys + + return str(getattr(orisys, "__version__", "unknown")) + except Exception: # noqa: BLE001 + return "unavailable" + # ------------------------------------------------------- record channel def _on_packet_recorded(self, packet) -> None: """Runs on the **acquisition thread**. Must stay cheap and Qt-free. @@ -247,6 +401,11 @@ class StudioMainWindow(QMainWindow): ) self.session_store.append(sample) self.cal_recorder.feed(sample) + # Video + CSV + metadata share this one record path. + self.recorder.on_packet(packet) + fps = float(getattr(packet, "fps", 0.0) or 0.0) + if fps > 0.0: + self._record_fps = fps if not sample.valid: self._record_error_count += 1 @@ -262,6 +421,7 @@ class StudioMainWindow(QMainWindow): self.toolbar.source_mode_changed.connect(self._on_source_mode_changed) self.toolbar.browse_video_clicked.connect(self._on_browse_video) self.toolbar.run_toggled.connect(self._on_run_toggled) + self.toolbar.record_toggled.connect(self._on_record_toggled) self.toolbar.reset_clicked.connect(self._on_reset) self.toolbar.units_changed.connect(self._on_units_changed) @@ -467,6 +627,9 @@ class StudioMainWindow(QMainWindow): reason = t("toolbar.run_disabled_video") self.toolbar.set_run_tooltip_reason(reason) self.toolbar.set_run_enabled(self._running or (sensor_ok and source_ok)) + # Recording needs a live session; while recording the button stays + # enabled so it can always be stopped. + self.toolbar.set_record_enabled(self._running or self.recorder.is_recording) if not self._running: self.toolbar.refresh_labels() self._sync_shortcut_actions() @@ -646,6 +809,7 @@ class StudioMainWindow(QMainWindow): self.statusBar().showMessage(t("cal.apply_no_target")) return worker.set_pending_calibration(path) + self.recorder.log_event("calibration_applied", f"{sensor_type}: {path}") self.statusBar().showMessage( t("cal.apply_done", sensor_type=sensor_type) ) @@ -902,8 +1066,12 @@ class StudioMainWindow(QMainWindow): self._dropped_reported = 0 self._depth_range_value = None self._last_depth_frame = None - self.session_store.reset(calibration_id=str(config_ref)) - self.cal_recorder.reset_session() + # A new session invalidates the previous recording. + self._stop_recording(wait=True) + self._session_counter += 1 + self.session_store.reset( + session_id=self._session_counter, calibration_id=str(config_ref) + ) self._ensure_workers() self._active_worker = ( self._finger_worker if sensor_type == "finger" else self._box_worker @@ -942,6 +1110,9 @@ class StudioMainWindow(QMainWindow): def _stop_run(self, *, reason: str = "user_stop", wait: bool = False) -> None: self._run_timer.stop() + # Close the experiment first so the reason lands in its event log. + self.recorder.log_event("session_stopped", f"reason={reason}") + self._stop_recording(wait=wait) if self._active_worker is not None: # Never wait here: a device open that hangs must not freeze the GUI. # ``wait=True`` is reserved for shutdown paths. @@ -982,6 +1153,7 @@ class StudioMainWindow(QMainWindow): self._active_worker.reset() self.force_plots.reset() self._last_plot_packet_key = None + self.recorder.log_event("zeroing", "tracker reset") self.statusBar().showMessage(t("status.reset_ok")) except Exception as exc: self.statusBar().showMessage(t("status.run_failed", error=str(exc))) @@ -1086,6 +1258,7 @@ class StudioMainWindow(QMainWindow): # Finalise calibration sampling on the UI thread once its deadline passes. self.cal_recorder.tick() self.cal_panel.poll_recorder() + self._drain_record_alarms() self._report_dropped_samples() mode = self.arrow_panel.view_mode() @@ -1095,9 +1268,12 @@ class StudioMainWindow(QMainWindow): self.arrow_panel.set_fps(packet.fps) def _report_invalid_sample(self, reason: str) -> None: - """Surface failed readouts instead of letting them pass as data.""" - if self._session_phase == "running": - self._set_phase("error") + """Surface failed readouts instead of letting them pass as data. + + The session phase is deliberately left alone: a single dropped sample + must not lock the session into ``error`` (that would also disable + reset). The count makes a persistent failure visible. + """ self.statusBar().showMessage( t("status.sample_invalid", count=self._record_error_count, reason=reason) ) @@ -1233,8 +1409,10 @@ class StudioMainWindow(QMainWindow): for worker in (self._box_worker, self._finger_worker): if worker is not None and hasattr(worker, "remove_sink"): worker.remove_sink(self._on_packet_recorded) - # Blocking: on the way out we must not leave a device handle behind. + # Blocking: on the way out we must not leave a device handle or a + # half-written experiment behind. self._stop_run(reason="user_stop", wait=True) + self._stop_recording(wait=True) self._persist_prefs() logger.info("Studio shell closed; preferences saved") super().closeEvent(event) diff --git a/apps/studio/src/orisys_studio/ui/toolbar.py b/apps/studio/src/orisys_studio/ui/toolbar.py index 1f4a8b3..1abdd37 100644 --- a/apps/studio/src/orisys_studio/ui/toolbar.py +++ b/apps/studio/src/orisys_studio/ui/toolbar.py @@ -1,4 +1,4 @@ -"""Top toolbar: source, camera/video, run/reset, units.""" +"""Top toolbar: source, camera/video, run/reset, record, units.""" from __future__ import annotations @@ -25,6 +25,9 @@ from orisys_app_shared.i18n import t CONTROL_HEIGHT = 30 BUTTON_HEIGHT = 32 +#: Recording modes, in combo order. ``value`` is what the app consumes. +RECORD_MODES = (("video_data", "record.mode_video_data"), ("data_only", "record.mode_data_only")) + def _separator() -> QFrame: line = QFrame() @@ -54,6 +57,7 @@ class ToolbarWidget(QWidget): run_toggled = pyqtSignal(bool) reset_clicked = pyqtSignal() units_changed = pyqtSignal(str) + record_toggled = pyqtSignal(bool) def __init__(self, parent=None) -> None: super().__init__(parent) @@ -148,6 +152,23 @@ class ToolbarWidget(QWidget): self.units_combo.setMinimumHeight(CONTROL_HEIGHT) self.units_combo.currentIndexChanged.connect(self._on_units_index_changed) + # --- record group ------------------------------------------------- + self.record_label = QLabel() + self.record_label.setObjectName("fieldLabel") + self.record_mode_combo = QComboBox() + self.record_mode_combo.setMinimumWidth(120) + self.record_mode_combo.setMinimumHeight(CONTROL_HEIGHT) + self.btn_record = QPushButton() + self.btn_record.setObjectName("ghostButton") + self.btn_record.setMinimumHeight(BUTTON_HEIGHT) + self.btn_record.setMinimumWidth(96) + self.btn_record.setCheckable(True) + self.btn_record.setEnabled(False) + self.btn_record.toggled.connect(self._on_record_toggled) + self.record_status = QLabel() + self.record_status.setObjectName("panelMeta") + self.record_status.setVisible(False) + layout.addWidget( _group(self.source_label, self.source_combo, self.source_stack) ) @@ -157,10 +178,16 @@ class ToolbarWidget(QWidget): ) layout.addWidget(_separator()) layout.addWidget(_group(self.units_label, self.units_combo)) + layout.addWidget(_separator()) + layout.addWidget( + _group(self.record_label, self.record_mode_combo, self.btn_record, + self.record_status) + ) layout.addStretch(1) self._populate_source_combo() self._populate_units_combo() + self._populate_record_modes() self.retranslate() self._apply_source_mode_ui() @@ -280,6 +307,58 @@ class ToolbarWidget(QWidget): self.units_combo.setCurrentIndex(max(0, idx)) self.units_combo.blockSignals(False) + def _populate_record_modes(self) -> None: + current = self.record_mode_combo.currentData() + self.record_mode_combo.blockSignals(True) + self.record_mode_combo.clear() + for value, _key in RECORD_MODES: + self.record_mode_combo.addItem("", value) + if current: + idx = self.record_mode_combo.findData(current) + self.record_mode_combo.setCurrentIndex(max(0, idx)) + self.record_mode_combo.blockSignals(False) + + def record_mode(self) -> str: + return str(self.record_mode_combo.currentData() or "video_data") + + def set_record_enabled(self, enabled: bool) -> None: + """Enable the record control. + + While a recording is running this must stay enabled — it is the only + way to stop it. + """ + self.btn_record.setEnabled(bool(enabled)) + self.record_mode_combo.setEnabled(bool(enabled) and not self.btn_record.isChecked()) + + def set_record_checked(self, checked: bool) -> None: + self.btn_record.blockSignals(True) + self.btn_record.setChecked(bool(checked)) + self.btn_record.blockSignals(False) + self._refresh_record_label() + + def set_record_status(self, text: str) -> None: + """Show elapsed time / output location next to the record control.""" + self.record_status.setText(str(text or "")) + self.record_status.setVisible(bool(text)) + + def _on_record_toggled(self, checked: bool) -> None: + self._refresh_record_label() + self.record_mode_combo.setEnabled(not checked) + self.record_toggled.emit(bool(checked)) + + def _refresh_record_label(self) -> None: + if self.btn_record.isChecked(): + self.btn_record.setText(t("record.stop_label")) + self.btn_record.setObjectName("dangerButton") + else: + self.btn_record.setText(t("record.start_label")) + self.btn_record.setObjectName("ghostButton") + self.btn_record.style().unpolish(self.btn_record) + self.btn_record.style().polish(self.btn_record) + self.btn_record.setToolTip( + t("record.stop_tip") if self.btn_record.isChecked() else t("record.start_tip") + ) + def _apply_source_mode_ui(self) -> None: video = self.source_mode() == "video" self.source_stack.setCurrentIndex(1 if video else 0) @@ -360,3 +439,9 @@ class ToolbarWidget(QWidget): self.units_label.setText(t("toolbar.units")) self.units_combo.setItemText(0, t("toolbar.units_newton")) self.units_combo.setItemText(1, t("toolbar.units_raw")) + self.record_label.setText(t("record.label")) + for index, (value, key) in enumerate(RECORD_MODES): + if index < self.record_mode_combo.count(): + self.record_mode_combo.setItemText(index, t(key)) + self.record_mode_combo.setToolTip(t("record.mode_tip")) + self._refresh_record_label() diff --git a/apps/studio/tests/test_ui_smoke.py b/apps/studio/tests/test_ui_smoke.py index 5ebd253..7329e16 100644 --- a/apps/studio/tests/test_ui_smoke.py +++ b/apps/studio/tests/test_ui_smoke.py @@ -2,6 +2,8 @@ from __future__ import annotations +from pathlib import Path + import pytest import ui_check @@ -197,7 +199,7 @@ def test_toolbar_run_button_is_labelled(make_window): window = _show(make_window(), (1400, 880)) label = window.toolbar.btn_run.text() assert label.strip() - assert label != "â–¶" + assert label != "â–¶" window.toolbar.set_run_checked(True) assert window.toolbar.btn_run.text() != label @@ -429,3 +431,88 @@ def test_depth_view_reports_missing_data_and_fixes_its_scale(make_window): ) _finish(window, stuck) + +# ------------------------------------------------------------- experiment record + +#: Recording scratch goes under the app's gitignored ``outputs`` tree rather than +#: the system temp directory, so the tests also run under a file sandbox that +#: only permits writes inside the workspace. +RECORD_SCRATCH = Path(__file__).resolve().parents[1] / "outputs" / "_test_record" + + +@pytest.fixture +def record_root(): + import shutil + + shutil.rmtree(RECORD_SCRATCH, ignore_errors=True) + RECORD_SCRATCH.mkdir(parents=True, exist_ok=True) + yield RECORD_SCRATCH + shutil.rmtree(RECORD_SCRATCH, ignore_errors=True) + + +def test_recording_writes_the_documented_experiment_layout(make_window, record_root): + import csv + import json + + import orisys_studio.app_window as app_module + from orisys_app_shared.i18n import t + + window, stuck = _running_window(make_window) + app_module.recordings_dir = lambda: record_root + assert window.toolbar.btn_record.isEnabled() + assert window.toolbar.btn_record.text() == t("record.start_label") + + window.toolbar.btn_record.setChecked(True) + assert window.recorder.is_recording, "the recording did not start" + assert window.toolbar.btn_record.text() == t("record.stop_label") + assert not window.toolbar.record_mode_combo.isEnabled() + + for index in range(20): + stuck.emit(ui_check._fake_packet(index)) + window._on_run_tick() + window.toolbar.btn_record.setChecked(False) + + report = window._record_last_report + assert report is not None and report.ok, getattr(report, "problems", None) + names = sorted(path.name for path in report.directory.iterdir()) + assert names == ["events.csv", "forces.csv", "frames.csv", "raw_0001.mp4", "session.json"] + + with (report.directory / "forces.csv").open(encoding="utf-8") as handle: + forces = list(csv.DictReader(handle)) + with (report.directory / "frames.csv").open(encoding="utf-8") as handle: + frames = list(csv.DictReader(handle)) + assert len(forces) == len(frames) == 20 + assert [f["frame_id"] for f in frames] == [r["frame_id"] for r in forces] + assert [f["t_rel_s"] for f in frames] == [r["t_rel_s"] for r in forces] + + meta = json.loads((report.directory / "session.json").read_text(encoding="utf-8")) + assert meta["ended_at"] + assert meta["format_version"] + assert meta["stats"]["frames"] == 20 + _finish(window, stuck) + + +def test_recording_leaves_unavailable_values_empty(make_window, record_root): + import csv + + import orisys_studio.app_window as app_module + + window, stuck = _running_window(make_window) + app_module.recordings_dir = lambda: record_root + window.toolbar.btn_record.setChecked(True) + stuck.emit(ui_check._fake_packet(0)) + window._on_run_tick() + stuck.emit(ui_check._fake_packet(1, valid=False, reason="read failed")) + window._on_run_tick() + window.toolbar.btn_record.setChecked(False) + + with (window._record_last_report.directory / "forces.csv").open(encoding="utf-8") as handle: + rows = list(csv.DictReader(handle)) + failed = rows[-1] + assert failed["valid"] == "0" + assert failed["FN_raw"] == "" and failed["FN_N"] == "", ( + "a failed readout was persisted as a number" + ) + _finish(window, stuck) + + diff --git a/apps/studio/tools/ui_check.py b/apps/studio/tools/ui_check.py index 4b84fd6..ad7ea91 100644 --- a/apps/studio/tools/ui_check.py +++ b/apps/studio/tools/ui_check.py @@ -32,6 +32,7 @@ if sys.platform == "win32" and "QT_QPA_FONTDIR" not in os.environ: os.chdir(APP_ROOT) from orisys_app_shared.acquisition.sinks import RecordSinkMixin # noqa: E402 +from orisys_app_shared.record import FORCES_HEADER # noqa: E402 FAILURES: list[str] = [] CHECKS = 0 @@ -44,6 +45,18 @@ def check(condition: bool, message: str) -> None: FAILURES.append(message) +def _inside_scroll_area(widget) -> bool: + """True when ``widget`` lives inside a QScrollArea's content.""" + from PyQt5.QtWidgets import QScrollArea + + node = widget + while node is not None: + if isinstance(node, QScrollArea): + return True + node = node.parentWidget() if hasattr(node, "parentWidget") else None + return False + + def section(name: str) -> None: print(f"\n== {name} ==") @@ -539,9 +552,17 @@ def main() -> int: parent = child.parentWidget() if parent is None or not parent.isVisible(): continue + if _inside_scroll_area(child): + # A scroll area's content is *meant* to exceed its viewport (the + # toolbar scrolls horizontally on narrow windows). + continue geometry = child.geometry() if geometry.width() > parent.width() + 1 or geometry.height() > parent.height() + 1: - oversize.append(type(child).__name__) + oversize.append( + f"{type(child).__name__}({getattr(child, 'objectName', lambda: '')()})" + f" {geometry.width()}x{geometry.height()}" + f" in {type(parent).__name__} {parent.width()}x{parent.height()}" + ) check(not oversize, f"widgets overflow their parent: {oversize}") window.close() app.processEvents() @@ -718,6 +739,110 @@ def main() -> int: app.processEvents() stuck.release() + section("experiment record") + import shutil as _shutil + + scratch = APP_ROOT / "outputs" / "_record_check" + _shutil.rmtree(scratch, ignore_errors=True) + scratch.mkdir(parents=True, exist_ok=True) + import orisys_studio.app_window as _app_module + + _app_module.recordings_dir = lambda: scratch + + stuck = _StuckWorker() + window = StudioMainWindow( + prefs={ + "language": "zh_CN", + "theme": "dark", + "sensor_type": "box", + "source_mode": "camera", + } + ) + window.resize(1440, 880) + window.show() + window._ensure_workers = lambda: None + window._box_worker = stuck + window._finger_worker = stuck + window._start_run() + window._run_timer.stop() + window._set_phase("running") + app.processEvents() + + check( + window.toolbar.btn_record.isEnabled(), + "the record control is disabled while a session is running", + ) + check( + window.toolbar.btn_record.text() == t("record.start_label"), + f"record button label is wrong: {window.toolbar.btn_record.text()!r}", + ) + check( + window.toolbar.record_mode() == "video_data", + "the default recording mode is not video+data", + ) + + window.toolbar.btn_record.setChecked(True) + app.processEvents() + check(window.recorder.is_recording, "the recording did not start") + check( + window.toolbar.btn_record.isEnabled() and window.toolbar.btn_record.text() == t("record.stop_label"), + "the record button is not a usable stop control", + ) + check( + not window.toolbar.record_mode_combo.isEnabled(), + "the recording mode can be changed mid-recording", + ) + + for index in range(30): + stuck.emit(_fake_packet(index)) + window._on_run_tick() + stuck.emit(_fake_packet(90, valid=False, reason="read failed")) + window._on_run_tick() + window._on_reset() # event: zeroing + + window.toolbar.btn_record.setChecked(False) + app.processEvents() + report = window._record_last_report + check(not window.recorder.is_recording, "the recording did not stop") + check(report is not None and report.ok, f"recording reported problems: {getattr(report, 'problems', None)}") + + import csv as _csv + import json as _json + + directory = report.directory + names = sorted(path.name for path in directory.iterdir()) + check( + names == ["events.csv", "forces.csv", "frames.csv", "raw_0001.mp4", "session.json"], + f"unexpected experiment layout: {names}", + ) + with (directory / "forces.csv").open(encoding="utf-8") as handle: + rows = list(_csv.DictReader(handle)) + check(tuple(rows[0].keys()) == FORCES_HEADER, "forces.csv header drifted from the spec") + check( + rows[-1]["valid"] == "0" and rows[-1]["FN_raw"] == "", + "a failed readout was written as a number instead of being left empty", + ) + with (directory / "frames.csv").open(encoding="utf-8") as handle: + frames = list(_csv.DictReader(handle)) + check(len(frames) == len(rows), "forces.csv and frames.csv disagree on the frame count") + check( + all(f["frame_id"] == r["frame_id"] for f, r in zip(frames, rows)), + "the frame index does not share frame ids with the data", + ) + meta = _json.loads((directory / "session.json").read_text(encoding="utf-8")) + check(meta.get("ended_at"), "session.json was not finalised") + check(meta["format_version"], "session.json has no format version") + check( + [event["kind"] for event in meta["events"]] == ["record_started", "zeroing", "record_stopped"], + f"events not recorded: {[e['kind'] for e in meta['events']]}", + ) + check(meta["stats"]["frames"] == 31, f"unexpected frame count: {meta['stats']['frames']}") + + window.close() + app.processEvents() + stuck.release() + _shutil.rmtree(scratch, ignore_errors=True) + print(f"\nscenes: {scene_count} checks: {CHECKS} failures: {len(FAILURES)}") for failure in FAILURES: print(f" FAIL: {failure}") diff --git a/shared/src/orisys_app_shared/paths.py b/shared/src/orisys_app_shared/paths.py index 637e04c..bc1ec62 100644 --- a/shared/src/orisys_app_shared/paths.py +++ b/shared/src/orisys_app_shared/paths.py @@ -68,6 +68,13 @@ def calibration_export_dir() -> Path: return path +def recordings_dir() -> Path: + """Root for one directory per recorded experiment.""" + path = outputs_dir() / "recordings" + path.mkdir(parents=True, exist_ok=True) + return path + + def preference_path() -> Path: return app_root() / "preference.json" diff --git a/shared/src/orisys_app_shared/record/__init__.py b/shared/src/orisys_app_shared/record/__init__.py new file mode 100644 index 0000000..df4cba5 --- /dev/null +++ b/shared/src/orisys_app_shared/record/__init__.py @@ -0,0 +1,25 @@ +"""Experiment recording (video + data + metadata).""" + +from orisys_app_shared.record.experiment import ( + EVENTS_HEADER, + FORCES_HEADER, + FRAMES_HEADER, + RECORD_FORMAT_VERSION, + ExperimentRecorder, + RecordConfig, + RecordReport, + RecordStats, + format_value, +) + +__all__ = [ + "EVENTS_HEADER", + "FORCES_HEADER", + "FRAMES_HEADER", + "RECORD_FORMAT_VERSION", + "ExperimentRecorder", + "RecordConfig", + "RecordReport", + "RecordStats", + "format_value", +] diff --git a/shared/src/orisys_app_shared/record/experiment.py b/shared/src/orisys_app_shared/record/experiment.py new file mode 100644 index 0000000..6fea1cb --- /dev/null +++ b/shared/src/orisys_app_shared/record/experiment.py @@ -0,0 +1,653 @@ +"""Experiment recording: one directory per session. + +Phase-1 record format, deliberately independent of the display path: + + outputs/recordings// + session.json experiment metadata, events and end-of-run statistics + forces.csv one row per acquired frame + frames.csv frame-index mapping (frame id <-> video segment) + events.csv timestamped operator/system events + raw_0001.mp4 processed-input video, segmented (optional) + frames/ lossless PNG sequence (optional) + +``on_packet`` is called from the acquisition thread, so it only copies the +scalars it needs (plus the raw frame when video is enabled) into a bounded queue +and returns. All file I/O happens on a writer thread. + +Two rules from the architecture review are enforced here: + +* the queue is bounded and **never blocks acquisition**; an overflow is counted + and surfaced rather than silently discarded; +* a write or encode failure is counted, reported, and after + :data:`MAX_CONSECUTIVE_ERRORS` stops the writer instead of pretending the + experiment was recorded. +""" + +from __future__ import annotations + +import csv +import json +import logging +import queue +import threading +import time +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional + +import cv2 +import numpy as np + +logger = logging.getLogger(__name__) + +RECORD_FORMAT_VERSION = "1.0" + +#: Column order required by the review; do not reorder without a format bump. +FORCES_HEADER = ( + "session_id", + "frame_id", + "t_rel_s", + "FN_raw", + "FX_raw", + "FY_raw", + "FN_N", + "FX_N", + "FY_N", + "valid", + "calibration_id", +) +FRAMES_HEADER = ( + "frame_id", + "t_rel_s", + "acquired_at", + "segment", + "segment_frame", + "in_video", + "dropped", +) +EVENTS_HEADER = ("at", "t_rel_s", "kind", "detail") + +MAX_CONSECUTIVE_ERRORS = 5 +#: A sustained queue overflow reports at most every N drops. +ALARM_EVERY_N_DROPS = 1000 + + +def format_value(value: float) -> str: + """Full-precision CSV cell. ``NaN`` means "not available" and is left empty. + + An uncalibrated newton value must be empty, never ``0``; a failed readout + must be empty too, otherwise it is indistinguishable from a real zero. + """ + try: + number = float(value) + except (TypeError, ValueError): + return "" + if not np.isfinite(number): + return "" + return repr(number) + + +@dataclass +class RecordConfig: + """What to capture, and how.""" + + video: bool = True + lossless_frames: bool = False + fps: float = 30.0 + codec: str = "mp4v" + segment_seconds: float = 300.0 + max_queue: int = 2048 + flush_interval_s: float = 1.0 + + def as_dict(self) -> dict: + return { + "video": bool(self.video), + "lossless_frames": bool(self.lossless_frames), + "fps": float(self.fps), + "codec": self.codec, + "segment_seconds": float(self.segment_seconds), + "max_queue": int(self.max_queue), + "lossy_note": ( + "MP4 is a lossy container: this is the pre-processing camera " + "image, not a lossless record of the raw sensor data." + ), + } + + +@dataclass +class RecordStats: + frames: int = 0 + video_frames: int = 0 + segments: int = 0 + dropped_data: int = 0 + dropped_video: int = 0 + write_errors: int = 0 + failed: bool = False + last_error: str = "" + duration_s: float = 0.0 + bytes_written: int = 0 + + def as_dict(self) -> dict: + return { + "frames": self.frames, + "video_frames": self.video_frames, + "segments": self.segments, + "dropped_data": self.dropped_data, + "dropped_video": self.dropped_video, + "write_errors": self.write_errors, + "failed": self.failed, + "last_error": self.last_error, + "duration_s": self.duration_s, + "bytes_written": self.bytes_written, + } + + +@dataclass +class RecordReport: + """Returned by :meth:`ExperimentRecorder.stop`.""" + + directory: Optional[Path] + stats: RecordStats + problems: list[str] = field(default_factory=list) + + @property + def ok(self) -> bool: + return bool(self.directory) and not self.stats.failed and not self.problems + + +class ExperimentRecorder: + """Writes one experiment directory. Sink-compatible with the workers.""" + + def __init__(self, config: Optional[RecordConfig] = None) -> None: + self.config = config or RecordConfig() + self._queue: queue.Queue = queue.Queue(maxsize=max(16, self.config.max_queue)) + self._thread: Optional[threading.Thread] = None + self._stop = threading.Event() + self._lock = threading.Lock() + self._writing = False + self._directory: Optional[Path] = None + self._stats = RecordStats() + self._session_id = 0 + self._calibration_id = "" + self._metadata: dict[str, Any] = {} + self._events: list[dict] = [] + self._t0: Optional[float] = None + self._last_t: Optional[float] = None + self._started_at = "" + self._consecutive_errors = 0 + self._on_alarm = None + + # ------------------------------------------------------------------ state + @property + def is_recording(self) -> bool: + with self._lock: + return self._writing + + @property + def directory(self) -> Optional[Path]: + return self._directory + + def stats(self) -> RecordStats: + with self._lock: + return RecordStats(**self._stats.as_dict()) + + def set_alarm_handler(self, handler) -> None: + """``handler(message)`` is called from the writer thread on a problem.""" + self._on_alarm = handler + + # -------------------------------------------------------------- lifecycle + def start( + self, + *, + root: Path, + session_id: int, + metadata: dict[str, Any], + calibration_id: str = "", + ) -> Path: + """Create the experiment directory and start the writer thread.""" + if self.is_recording: + raise RuntimeError("experiment recording is already running") + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + directory = Path(root) / f"{stamp}_session{session_id:04d}" + directory.mkdir(parents=True, exist_ok=True) + with self._lock: + self._directory = directory + self._stats = RecordStats() + self._session_id = int(session_id) + self._calibration_id = str(calibration_id or "") + self._metadata = dict(metadata) + self._events = [] + self._t0 = None + self._last_t = None + self._consecutive_errors = 0 + self._started_at = datetime.now(timezone.utc).isoformat(timespec="seconds") + self._writing = True + self._stop.clear() + self._drain_queue() + self._thread = threading.Thread( + target=self._run, + name=f"ExperimentRecorder-{session_id}", + daemon=True, + ) + self._thread.start() + self._write_session_json(final=False) + logger.info("Recording experiment to %s", directory) + return directory + + def stop(self, *, wait: bool = True, timeout: float = 20.0) -> RecordReport: + """Flush and close. Returns what actually happened.""" + with self._lock: + was_writing = self._writing + self._writing = False + if not was_writing: + return RecordReport(self._directory, self.stats()) + self._stop.set() + try: + self._queue.put_nowait(("stop", None)) + except queue.Full: # pragma: no cover - writer will drain and see the flag + pass + thread = self._thread + if wait and thread is not None and thread.is_alive(): + thread.join(timeout=timeout) + self._thread = None + report = self._finalise() + return report + + def log_event(self, kind: str, detail: str = "") -> None: + """Record a timestamped event (zeroing, calibration applied, disconnect…).""" + if not self.is_recording: + return + event = { + "at": datetime.now(timezone.utc).isoformat(timespec="milliseconds"), + "t_rel_s": None, + "kind": str(kind), + "detail": str(detail or ""), + } + try: + self._queue.put_nowait(("event", event)) + except queue.Full: + self._note_drop("event") + + # ----------------------------------------------------------- sink interface + def on_packet(self, packet) -> None: + """Called on the acquisition thread: enqueue, never do I/O here.""" + if not self.is_recording: + return + timestamp = float(packet.timestamp) + if self._t0 is None: + self._t0 = timestamp + with self._lock: + self._last_t = timestamp + row = ( + int(getattr(packet, "frame_id", 0)), + timestamp, + float(packet.normal_force_raw), + float(packet.shear_x_raw), + float(packet.shear_y_raw), + float(packet.normal_force), + float(packet.shear_x), + float(packet.shear_y), + 1 if bool(getattr(packet, "valid", True)) else 0, + ) + frame = None + if self.config.video or self.config.lossless_frames: + frame = self._capture_frame(packet) + try: + self._queue.put_nowait(("frame", (row, frame))) + except queue.Full: + # Never block acquisition. The loss is counted, not hidden. + self._note_drop("video" if frame is not None else "data") + + @staticmethod + def _capture_frame(packet): + """Copy the pre-processing frame. + + The packet's ``raw_frame`` may alias a buffer the sensor reuses, so a + recording must take its own copy here rather than hold the reference. + """ + frame = getattr(packet, "raw_frame", None) + if frame is None: + return None + block = np.asarray(frame) + if block.size == 0: + return None + return block.copy() + + # ----------------------------------------------------------------- internals + def _drain_queue(self) -> None: + while True: + try: + self._queue.get_nowait() + except queue.Empty: + return + + def _note_drop(self, kind: str) -> None: + with self._lock: + if kind == "video": + self._stats.dropped_video += 1 + else: + # Data rows and events are both record-channel content. + self._stats.dropped_data += 1 + total = self._stats.dropped_video + self._stats.dropped_data + # Rate-limited: a sustained overflow must not flood the log or the UI. + if total == 1 or total % ALARM_EVERY_N_DROPS == 0: + self._alarm(f"record queue overflow ({kind}); {total} item(s) not written") + + def _alarm(self, message: str) -> None: + logger.warning("Recording problem: %s", message) + handler = self._on_alarm + if handler is not None: + try: + handler(message) + except Exception: # noqa: BLE001 - reporting must not break recording + logger.exception("Recording alarm handler failed") + + def _run(self) -> None: + writer = _SessionWriter( + directory=self._directory, + config=self.config, + session_id=self._session_id, + calibration_id=self._calibration_id, + ) + try: + while True: + if self._stop.is_set() and self._queue.empty(): + break + try: + kind, payload = self._queue.get(timeout=0.2) + except queue.Empty: + writer.maybe_flush() + continue + if kind == "stop": + break + if kind == "event": + self._handle_event(writer, payload) + continue + row, frame = payload + self._handle_frame(writer, row, frame) + writer.maybe_flush() + except Exception as exc: # noqa: BLE001 - reported through stats + logger.exception("Experiment writer stopped") + with self._lock: + self._stats.failed = True + self._stats.last_error = f"{type(exc).__name__}: {exc}" + self._alarm(f"recording stopped: {exc}") + finally: + problems = writer.close() + with self._lock: + self._stats.write_errors += writer.write_errors + self._stats.segments = writer.segments + self._stats.video_frames = writer.video_frames + self._stats.bytes_written = writer.bytes_written + if problems: + self._stats.failed = True + if not self._stats.last_error: + self._stats.last_error = problems[0] + for problem in problems: + self._alarm(problem) + + def _handle_event(self, writer, event: dict) -> None: + if self._t0 is not None: + event["t_rel_s"] = round(float(time.time()) - self._t0, 6) + with self._lock: + self._events.append(event) + try: + writer.write_event(event) + except Exception as exc: # noqa: BLE001 + self._record_error(exc) + + def _handle_frame(self, writer, row, frame) -> None: + try: + writer.write_sample(row, frame) + with self._lock: + self._stats.frames += 1 + self._stats.video_frames = writer.video_frames + self._stats.segments = writer.segments + self._consecutive_errors = 0 + except Exception as exc: # noqa: BLE001 + self._record_error(exc) + + def _record_error(self, exc: Exception) -> None: + with self._lock: + self._stats.write_errors += 1 + self._stats.last_error = f"{type(exc).__name__}: {exc}" + self._consecutive_errors += 1 + fatal = self._consecutive_errors >= MAX_CONSECUTIVE_ERRORS + if fatal: + self._stats.failed = True + self._alarm(f"recording write failed: {exc}") + if fatal: + # Stop pretending we are recording rather than losing data quietly. + self._stop.set() + + def _finalise(self) -> RecordReport: + with self._lock: + if self._t0 is not None and self._last_t is not None: + self._stats.duration_s = round(float(self._last_t - self._t0), 6) + stats = self.stats() + problems: list[str] = [] + if stats.dropped_data or stats.dropped_video: + problems.append( + f"{stats.dropped_data} data and {stats.dropped_video} video frame(s) " + "were dropped by the bounded queue" + ) + if stats.write_errors: + problems.append(f"{stats.write_errors} write error(s); last: {stats.last_error}") + self._write_session_json(final=True) + return RecordReport(directory=self._directory, stats=stats, problems=problems) + + def _write_session_json(self, *, final: bool) -> None: + directory = self._directory + if directory is None: + return + payload = self._build_metadata(final=final) + target = directory / "session.json" + try: + # Written at start too, so a crash still leaves identifiable metadata. + temporary = target.with_suffix(".json.tmp") + temporary.write_text( + json.dumps(payload, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + temporary.replace(target) + except Exception as exc: # noqa: BLE001 + with self._lock: + self._stats.write_errors += 1 + self._stats.last_error = f"session.json: {type(exc).__name__}: {exc}" + self._alarm(f"could not write session.json: {exc}") + + def _build_metadata(self, *, final: bool) -> dict[str, Any]: + with self._lock: + stats = self._stats.as_dict() + events = list(self._events) + payload: dict[str, Any] = { + "format_version": RECORD_FORMAT_VERSION, + "session_id": self._session_id, + "calibration_id": self._calibration_id, + "started_at": self._started_at, + "ended_at": ( + datetime.now(timezone.utc).isoformat(timespec="seconds") if final else None + ), + "video": self.config.as_dict(), + "events": events, + "stats": stats, + "files": { + "forces": "forces.csv", + "frames": "frames.csv", + "events": "events.csv", + "video": "raw_%04d.mp4" if self.config.video else None, + "lossless_frames": "frames/" if self.config.lossless_frames else None, + }, + } + payload.update(self._metadata) + return payload + + +class _SessionWriter: + """Owns the file handles. Only ever touched by the writer thread.""" + + def __init__(self, *, directory, config, session_id, calibration_id) -> None: + self.directory = Path(directory) + self.config = config + self.session_id = session_id + self.calibration_id = calibration_id + self.write_errors = 0 + self.segments = 0 + self.video_frames = 0 + self.png_frames = 0 + self.bytes_written = 0 + self._last_flush = time.perf_counter() + self._t0: Optional[float] = None + self._segment: Optional[cv2.VideoWriter] = None + self._segment_index = 0 + self._segment_frames = 0 + self._segment_started = 0.0 + self._frame_shape: Optional[tuple[int, int]] = None + self._problems: list[str] = [] + + self._forces = self._open_csv("forces.csv", FORCES_HEADER) + self._frames = self._open_csv("frames.csv", FRAMES_HEADER) + self._events = self._open_csv("events.csv", EVENTS_HEADER) + self._png_dir = None + if self.config.lossless_frames: + self._png_dir = self.directory / "frames" + self._png_dir.mkdir(parents=True, exist_ok=True) + + def _open_csv(self, name: str, header): + handle = (self.directory / name).open("w", newline="", encoding="utf-8") + writer = csv.writer(handle) + writer.writerow(header) + return (handle, writer) + + # ------------------------------------------------------------------ writing + def write_sample(self, row, frame) -> None: + """Write one acquired frame: a forces row, the frame index, and video. + + ``forces.csv`` and ``frames.csv`` share the packet's ``frame_id`` and the + same ``t_rel_s`` origin, which is what lets video and data be aligned + afterwards even when frames were dropped or split into segments. + """ + frame_id, timestamp, fn_r, fx_r, fy_r, fn_n, fx_n, fy_n, valid = row + if self._t0 is None: + self._t0 = timestamp + t_rel = round(timestamp - self._t0, 6) + + self._forces[1].writerow( + ( + self.session_id, + frame_id, + repr(t_rel), + format_value(fn_r), + format_value(fx_r), + format_value(fy_r), + format_value(fn_n), + format_value(fx_n), + format_value(fy_n), + valid, + self.calibration_id, + ) + ) + + segment = 0 + segment_frame = 0 + if frame is not None: + if self.config.video: + segment, segment_frame = self._write_video_frame(frame) + if self._png_dir is not None: + # The lossless sequence is independent of the MP4 setting. + self._write_lossless(frame) + + self._frames[1].writerow( + ( + frame_id, + repr(t_rel), + datetime.fromtimestamp(timestamp, timezone.utc).isoformat( + timespec="milliseconds" + ), + segment, + segment_frame, + 1 if segment else 0, + 0, + ) + ) + + def _write_video_frame(self, frame) -> tuple[int, int]: + if self._frame_shape is None: + self._frame_shape = (int(frame.shape[1]), int(frame.shape[0])) + if self._should_roll(): + self._close_segment() + if self._segment is None: + self._open_segment() + self._segment.write(frame) + self._segment_frames += 1 + self.video_frames += 1 + return (self._segment_index, self._segment_frames) + + def _write_lossless(self, frame) -> None: + self.png_frames += 1 + target = self._png_dir / f"{self.png_frames:06d}.png" + if not cv2.imwrite(str(target), frame): + raise OSError(f"could not write {target.name}") + + def write_event(self, event: dict) -> None: + self._events[1].writerow( + (event.get("at", ""), event.get("t_rel_s", ""), event["kind"], event["detail"]) + ) + + # ------------------------------------------------------------------ segments + def _should_roll(self) -> bool: + if self._segment is None: + return False + if self.config.segment_seconds <= 0: + return False + return (time.perf_counter() - self._segment_started) >= self.config.segment_seconds + + def _open_segment(self) -> None: + self._segment_index += 1 + path = self.directory / f"raw_{self._segment_index:04d}.mp4" + fourcc = cv2.VideoWriter_fourcc(*self.config.codec) + fps = max(1.0, float(self.config.fps)) + writer = cv2.VideoWriter(str(path), fourcc, fps, self._frame_shape) + if not writer.isOpened(): + raise OSError(f"could not open video writer for {path.name}") + self._segment = writer + self._segment_frames = 0 + self._segment_started = time.perf_counter() + self.segments = self._segment_index + + def _close_segment(self) -> None: + if self._segment is None: + return + self._segment.release() + self._segment = None + + def maybe_flush(self) -> None: + if (time.perf_counter() - self._last_flush) < self.config.flush_interval_s: + return + self._last_flush = time.perf_counter() + for handle, _writer in (self._forces, self._frames, self._events): + try: + handle.flush() + except Exception as exc: # noqa: BLE001 + self.write_errors += 1 + if len(self._problems) < 5: + self._problems.append(f"flush failed: {exc}") + + def close(self) -> list[str]: + self._close_segment() + for handle, _writer in (self._forces, self._frames, self._events): + try: + handle.flush() + handle.close() + except Exception as exc: # noqa: BLE001 + self.write_errors += 1 + if len(self._problems) < 5: + self._problems.append(f"close failed: {exc}") + try: + self.bytes_written = sum( + path.stat().st_size for path in self.directory.glob("*") if path.is_file() + ) + except Exception: # noqa: BLE001 + self.bytes_written = 0 + return self._problems diff --git a/shared/tests/test_experiment_recorder.py b/shared/tests/test_experiment_recorder.py new file mode 100644 index 0000000..1eb2ddf --- /dev/null +++ b/shared/tests/test_experiment_recorder.py @@ -0,0 +1,278 @@ +"""Experiment recording: directory layout, CSV contract, metadata, failure handling. + +Scratch files are created inside the repo (``shared/tests/_scratch``) rather than +in the system temp directory, because some sandboxes deny writes outside the +workspace. The directory is removed after every test. +""" + +from __future__ import annotations + +import csv +import json +import shutil +import time +from pathlib import Path + +import numpy as np +import pytest + +from orisys_app_shared.acquisition.sdk_worker import SdkPacket +from orisys_app_shared.record import ( + EVENTS_HEADER, + FORCES_HEADER, + FRAMES_HEADER, + ExperimentRecorder, + RecordConfig, + format_value, +) + +SCRATCH = Path(__file__).resolve().parent / "_scratch" + + +def _packet(index: int, *, valid: bool = True, raw_available: bool = True) -> SdkPacket: + nan = float("nan") + frame = np.full((24, 32, 3), index % 255, np.uint8) + return SdkPacket( + raw_frame=frame, + processed_frame=frame, + timestamp=1000.0 + index / 30.0, + fps=30.0, + process_ms=1.0, + normal_force=2.0 if valid else nan, + shear_x=0.5 if valid else nan, + shear_y=-0.5 if valid else nan, + normal_force_raw=(9.0 + index * 0.1) if valid and raw_available else nan, + shear_x_raw=0.4 if valid and raw_available else nan, + shear_y_raw=-0.4 if valid and raw_available else nan, + force_unit="N", + force_calibrated=True, + session_id=1, + frame_id=index, + valid=valid, + invalid_reason="" if valid else "read failed", + raw_available=raw_available, + ) + + +@pytest.fixture +def root(): + shutil.rmtree(SCRATCH, ignore_errors=True) + SCRATCH.mkdir(parents=True, exist_ok=True) + yield SCRATCH + shutil.rmtree(SCRATCH, ignore_errors=True) + + +def _record(root, packets, config=None, events=()): + recorder = ExperimentRecorder(config or RecordConfig(video=True, fps=30.0)) + directory = recorder.start( + root=root, + session_id=7, + metadata={"device": {"sensor_type": "box", "sn": "SN-1"}}, + calibration_id="/cfg/sensor.json", + ) + for index in packets: + recorder.on_packet(_packet(index)) + for kind, detail in events: + recorder.log_event(kind, detail) + report = recorder.stop() + return directory, report + + +def test_directory_layout_and_headers(root): + directory, report = _record(root, range(12)) + + assert report.ok, report.problems + names = sorted(path.name for path in directory.iterdir()) + assert names == ["events.csv", "forces.csv", "frames.csv", "raw_0001.mp4", "session.json"] + for name, header in ( + ("forces.csv", FORCES_HEADER), + ("frames.csv", FRAMES_HEADER), + ("events.csv", EVENTS_HEADER), + ): + with (directory / name).open(encoding="utf-8") as handle: + assert tuple(next(csv.reader(handle))) == header, name + + +def test_unavailable_values_are_empty_not_zero(root): + recorder = ExperimentRecorder(RecordConfig(video=False)) + directory = recorder.start(root=root, session_id=1, metadata={}, calibration_id="cal-1") + recorder.on_packet(_packet(0)) + recorder.on_packet(_packet(1, valid=False)) + recorder.on_packet(_packet(2, raw_available=False)) + recorder.stop() + + with (directory / "forces.csv").open(encoding="utf-8") as handle: + rows = list(csv.DictReader(handle)) + + failed = rows[1] + assert failed["valid"] == "0" + for column in ("FN_raw", "FX_raw", "FY_raw", "FN_N", "FX_N", "FY_N"): + assert failed[column] == "", "a failed readout was written as a number" + + no_raw = rows[2] + assert no_raw["FN_raw"] == "" and no_raw["FX_raw"] == "", "raw was not left empty" + assert no_raw["FN_N"] == "2.0", "calibrated values must still be recorded" + assert no_raw["valid"] == "1" + + +def test_full_precision_is_preserved(): + value = 0.1 + 0.2 + assert format_value(value) == repr(value) != "0.3" + assert format_value(float("nan")) == "" + + +def test_forces_and_frame_index_share_frame_id_and_time_base(root): + directory, _ = _record(root, range(20)) + with (directory / "forces.csv").open(encoding="utf-8") as handle: + forces = list(csv.DictReader(handle)) + with (directory / "frames.csv").open(encoding="utf-8") as handle: + frames = list(csv.DictReader(handle)) + + assert len(forces) == len(frames) == 20 + for force_row, frame_row in zip(forces, frames): + assert force_row["frame_id"] == frame_row["frame_id"] + assert force_row["t_rel_s"] == frame_row["t_rel_s"], "time bases diverged" + assert frame_row["in_video"] == "1" + assert frame_row["segment"] == "1" + assert frames[0]["t_rel_s"] == "0.0" + + +def test_segments_roll_and_are_indexed(root): + config = RecordConfig(video=True, fps=30.0, segment_seconds=0.05) + recorder = ExperimentRecorder(config) + directory = recorder.start(root=root, session_id=1, metadata={}) + for index in range(40): + recorder.on_packet(_packet(index)) + time.sleep(0.004) + recorder.stop() + + segments = sorted(path.name for path in directory.glob("raw_*.mp4")) + assert len(segments) >= 2, "long recordings must be split" + with (directory / "frames.csv").open(encoding="utf-8") as handle: + used = {row["segment"] for row in csv.DictReader(handle)} + assert len(used) >= 2, "the frame index must record which segment each frame went to" + + +def test_data_only_mode_writes_no_video(root): + recorder = ExperimentRecorder(RecordConfig(video=False)) + directory = recorder.start(root=root, session_id=1, metadata={}) + for index in range(10): + recorder.on_packet(_packet(index)) + report = recorder.stop() + + assert not list(directory.glob("*.mp4")) + assert report.stats.frames == 10 + assert report.stats.video_frames == 0 + + +def test_lossless_frames_are_optional(root): + recorder = ExperimentRecorder(RecordConfig(video=False, lossless_frames=True)) + directory = recorder.start(root=root, session_id=1, metadata={}) + for index in range(5): + recorder.on_packet(_packet(index)) + recorder.stop() + + images = sorted((directory / "frames").glob("*.png")) + assert len(images) == 5, "the lossless sequence was requested but not written" + + +def test_queue_overflow_is_counted_and_alarmed(root): + alarms: list[str] = [] + # A queue that holds one item, with a writer we deliberately stall. + recorder = ExperimentRecorder(RecordConfig(video=True, max_queue=16)) + recorder.set_alarm_handler(alarms.append) + directory = recorder.start(root=root, session_id=1, metadata={}) + + # Push far more than the queue can hold without letting the writer catch up. + for index in range(4000): + recorder.on_packet(_packet(index)) + report = recorder.stop() + + stats = report.stats + assert stats.dropped_data + stats.dropped_video > 0 or stats.frames == 4000, ( + "overflow must either be counted or not happen at all" + ) + if stats.dropped_data or stats.dropped_video: + assert alarms, "an overflow must raise an alarm, not pass silently" + assert any("overflow" in message for message in alarms) + + +def test_encode_failure_is_reported_and_stops_recording(root): + alarms: list[str] = [] + recorder = ExperimentRecorder(RecordConfig(video=True, codec="ZZZZ")) + recorder.set_alarm_handler(alarms.append) + recorder.start(root=root, session_id=1, metadata={}) + for index in range(20): + recorder.on_packet(_packet(index)) + report = recorder.stop() + + assert report.stats.failed, "a codec that cannot open must not be ignored" + assert report.stats.write_errors > 0 + assert alarms, "the encoder failure was never reported" + assert not report.ok + + +def test_metadata_contains_the_documented_fields(root): + directory, _ = _record( + root, + range(4), + events=(("zeroing", "operator"), ("calibration_applied", "box")), + ) + meta = json.loads((directory / "session.json").read_text(encoding="utf-8")) + + for key in ("format_version", "session_id", "started_at", "ended_at", "video", + "events", "stats", "files", "device"): + assert key in meta, key + assert meta["session_id"] == 7 + assert meta["ended_at"], "session.json must be finalised at stop" + assert meta["calibration_id"] == "/cfg/sensor.json" + assert meta["video"]["lossy_note"], "the lossy-MP4 caveat must be recorded" + assert [event["kind"] for event in meta["events"]] == ["zeroing", "calibration_applied"] + assert all(event["at"] for event in meta["events"]) + assert meta["stats"]["frames"] == 4 + assert meta["stats"]["duration_s"] > 0 + + +def test_metadata_exists_even_if_the_run_never_stops(root): + """A crash must still leave identifiable metadata behind.""" + recorder = ExperimentRecorder(RecordConfig(video=False)) + directory = recorder.start(root=root, session_id=3, metadata={"device": {"sn": "X"}}) + meta = json.loads((directory / "session.json").read_text(encoding="utf-8")) + + assert meta["session_id"] == 3 + assert meta["ended_at"] is None + recorder.stop() + + +def test_on_packet_never_blocks_the_caller(root): + recorder = ExperimentRecorder(RecordConfig(video=False, max_queue=32)) + recorder.start(root=root, session_id=1, metadata={}) + # Fill well past the queue bound; each call must still return promptly. + started = time.perf_counter() + for index in range(5000): + recorder.on_packet(_packet(index)) + elapsed = time.perf_counter() - started + + assert elapsed < 5.0, f"on_packet blocked acquisition for {elapsed:.2f}s" + recorder.stop() + + +if __name__ == "__main__": # pragma: no cover - manual runner + failures = 0 + for name, function in sorted(globals().items()): + if not name.startswith("test_") or not callable(function): + continue + try: + if name in ("test_full_precision_is_preserved",): + function() + else: + shutil.rmtree(SCRATCH, ignore_errors=True) + SCRATCH.mkdir(parents=True, exist_ok=True) + function(SCRATCH) + shutil.rmtree(SCRATCH, ignore_errors=True) + print(f" ok {name}") + except Exception as exc: # noqa: BLE001 + failures += 1 + print(f" FAIL {name}: {type(exc).__name__}: {exc}") + print(f"failures: {failures}") + raise SystemExit(1 if failures else 0)