refactor(studio): 分拆 app_window 第一步 —— 会话事实(记录元数据/滤波器/采集协商)搬进 session_metadata

- What: 新增 `apps/studio/src/orisys_studio/session_metadata.py`(273 行,纯函数 + 一个 `SessionFacts`
  数据类,无 Qt、无状态),把 `StudioMainWindow` 里"记录必须自述"的那部分逻辑搬过去:
  `config_sha256` / `sdk_version` / `device_serial` / `capture_metadata` / `filter_metadata` /
  `filter_settings_from_prefs` / `video_fps` / `device_sample_rate` / `build_session_metadata`。
  `app_window` 侧只留一行转发(保留 `_config_hash`/`_sdk_version`/`_device_serial`/`_capture_metadata`/
  `_filter_metadata`/`_filter_settings`/`_video_fps`/`_device_sample_rate` 这些名字,因为测试与
  `ui_check` 会直接调它们),`_build_session_metadata` 缩减为"把只有窗口能看见的事实装进
  `SessionFacts`,然后交给模块"。`app_window.py` 3045 → 2901 行(-196/+51)。
- Why: 用户已同意分拆 `app_window.py`(当前 155 个方法、fan-out 36、唯一 fan-in=1 的中心),并选定四个
  新模块的命名。A 步(会话/采集/记录)本身约 1100 行、40 处调用点,一次搬完风险高,所以按"两种变化原因
  分开命名"的原则切成两步:**A1(本提交)= 无状态的事实与规则**,A2 = 带状态的 `SessionController`
  (worker 生命周期、phase、定时器、packet 扇出),后者将调用本模块组装记录块。
  顺带把这几条规则从"只在记录流程里被间接覆盖"变成可单测:录制视频的 fps 对相机取**配置值**、对文件取
  **实测值**、且永不写 0 fps;滤波器从**真正运行它们的 worker** 读回(旧 SDK 可能少应用),并把
  `unsupported_by_sdk` 明确记下;任何元数据读取失败都只是空块,绝不让一次记录失败。
- Verification: 新增 `apps/studio/tests/test_session_metadata.py` **15 项**:配置内容哈希(含缺失/空路径)、
  SDK 版本恒为字符串、设备序列号三种字段与嵌套 `info.SN`、capture 状态缺失/抛异常都返回空块、
  滤波器回读与 `unsupported_by_sdk`、空/损坏 preferences 等于"什么都不滤"、
  相机 vs 文件的 fps 规则与 0 fps 兜底、Nyquist 用的采样率来源与垃圾输入、完整 session 块(设备/版本/采集/
  标定/深度/约定/滤波/录像模式/video_config/3D 块逐项断言)、worker 消失且从未收到包时仍产出完整块、
  3D 描述原样透传、三种录像模式如实记录。全量:shared+apps 同一次运行 **603 项 0 失败 15 跳过**
  (较上一提交 +15)、`ui_check` 587 项 0 失败、`locale_check` 0 缺失。
- Not in this change: 行为零改动(同一批函数、同一批默认值),也没有搬任何**状态**——`_latest_fps`/
  `_latest_process_ms`/`_latest_capture_*`/`_record_error_count`/`_capture_checked`/`_dropped_reported`
  仍在窗口里,A2 连同 worker 生命周期、phase、定时器与 packet 扇出一起搬;
  回放(B)与视图/显示(C)两个控制器未开始。四个模块的名字已定:
  `SessionController` / `ReplayController`(引擎已改名 `ReplayEngine`)/ `ViewLayoutController` /
  `DisplayPrefsController`。
This commit is contained in:
ykcheungab
2026-09-15 08:51:17 +08:00
parent 8d374e5850
commit c2671e5c2f
3 changed files with 531 additions and 196 deletions
+50 -195
View File
@@ -49,6 +49,7 @@ from orisys_app_shared.record import RecordConfig
from orisys_app_shared.record.sample import to_record_sample from orisys_app_shared.record.sample import to_record_sample
from orisys_studio.recording import RecordingCoordinator, record_config_from_prefs from orisys_studio.recording import RecordingCoordinator, record_config_from_prefs
from orisys_studio.replay import ReplayEngine from orisys_studio.replay import ReplayEngine
from orisys_studio import session_metadata
from orisys_studio.scene3d.prefs import settings_to_prefs from orisys_studio.scene3d.prefs import settings_to_prefs
from orisys_app_shared.i18n import set_locale, t from orisys_app_shared.i18n import set_locale, t
from orisys_app_shared.logging import report_error from orisys_app_shared.logging import report_error
@@ -372,28 +373,17 @@ class StudioMainWindow(QMainWindow):
return self.recording.record_fps return self.recording.record_fps
def _video_fps(self) -> float: def _video_fps(self) -> float:
"""The rate the recorded video is written at. """The rate the recorded video is written at (see ``session_metadata``).
For a live camera that is the sensor configuration's own frame rate: the Delegated rather than deleted: tests and the dependency-free UI checker call
container's fps must describe the spacing of the pictures, and using the this on the window, and the rule itself now lives with the other record
measured counter made a 30 fps sensor produce a video that declared (and metadata (where it is unit-tested on its own).
played at) 48 fps. A file source has no configured rate to honour, so its
measured rate is the honest one.
""" """
measured = float(self._record_fps) return session_metadata.video_fps(
if not isinstance(self.toolbar.active_source(), str): self._record_fps,
try: source_is_file=isinstance(self.toolbar.active_source(), str),
from orisys_app_shared.acquisition.models import load_fixed_mode config_ref=self._resolve_config_ref(),
)
mode = load_fixed_mode(self._resolve_config_ref() or "")
if mode.fps > 0:
return float(mode.fps)
except Exception: # noqa: BLE001 - the measured rate is the fallback
logger.debug("Could not read the configured frame rate", exc_info=True)
# Neither a stream that has not yet delivered a second of frames nor a
# config that could not be read may reach the recorder as a rate: a
# container of "0 fps" would thin the video to one picture per second.
return measured if measured > 0.0 else float(RecordConfig().fps)
@property @property
def _active_calibration_id(self) -> str: def _active_calibration_id(self) -> str:
@@ -544,91 +534,30 @@ class StudioMainWindow(QMainWindow):
@staticmethod @staticmethod
def _device_serial(info) -> str: def _device_serial(info) -> str:
"""The sensor's own serial number, when the SDK exposes one.""" """The sensor's own serial number, when the SDK exposes one."""
for attribute in ("sn", "serial", "serial_number"): return session_metadata.device_serial(info)
value = getattr(info, attribute, None)
if value:
return str(value)
serial = getattr(info, "info", None)
return str(getattr(serial, "SN", "") or "")
def _build_session_metadata(self, config: RecordConfig) -> dict: def _build_session_metadata(self, config: RecordConfig) -> dict:
from orisys_app_shared import API_VERSION, __version__ as shared_version """Gather the facts only the window can see, then let the record describe them.
from orisys_studio import __version__ as app_version The block itself (and the rules about which frame rate a container may declare,
how filters are read back, what a failure looks like) lives in
sensor_type = self.sensor_picker.sensor_type() ``session_metadata``; this only says *what* is on screen right now.
config_ref = self._resolve_config_ref() or "" """
info = self._active_worker.latest() if self._active_worker is not None else None facts = session_metadata.SessionFacts(
return { sensor_type=self.sensor_picker.sensor_type(),
"device": { config_ref=self._resolve_config_ref() or "",
"sensor_type": sensor_type, worker=self._active_worker,
"config_ref": config_ref, info=(
"config_sha256": self._config_hash(config_ref), self._active_worker.latest() if self._active_worker is not None else None
"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._video_fps()),
# The device's own measured rate, kept so a mismatch between
# the two can be diagnosed after the fact.
"measured_fps": float(self._record_fps),
},
},
"versions": {
"app": app_version,
"shared": shared_version,
"shared_api": API_VERSION,
"sdk": self._sdk_version(),
},
# What the camera was asked for and what the driver gave. A recording
# that runs at 16.7 fps instead of 30 has to say why by itself.
"capture": self._capture_metadata(),
"calibration": {
"config_ref": config_ref,
"force_calibrated": bool(getattr(info, "force_calibrated", False)),
# Content hash of the coefficients the device confirmed. It is
# filled in by the acquisition thread's confirmation, so it can
# be empty here; the CSV's per-row version is authoritative.
"version": self._active_calibration_id,
# The operator-entered serial of the *calibrated* device, as the
# calibration panel has it (the device SN when the config has one).
"sn": self.cal_panel.sn_edit.text().strip(),
"device_sn": self._device_serial(info),
},
# Depth is deliberately NOT stored (decided with the team): it is a
# function of the frame and the sensor configuration, and the replay
# view shows the recorded frames only. Re-running the pipeline over
# raw_*.mp4 is the "recompute" mode's job; the hashes below are what
# such a pass (and any other consumer) checks before trusting it.
"depth": {
"stored": False,
"replay": "none",
"quantity": "relative deformation / algorithm response, not a calibrated length",
"method": "re-run the SDK over raw_*.mp4 with this config_ref",
"config_sha256": self._config_hash(config_ref),
"sdk_version": self._sdk_version(),
},
"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; an event "
"that predates it has no relative base and is left empty"
), ),
"frame_id": "monotonic per acquisition session", video_fps=float(self._video_fps()),
}, measured_fps=float(self._record_fps),
"filter": self._filter_metadata(), active_calibration_id=self._active_calibration_id,
"recording_mode": self.toolbar.record_mode(), operator_sn=self.cal_panel.sn_edit.text(),
"video_config": config.as_dict(), recording_mode=self.toolbar.record_mode(),
# What the 3D pane was, for a reader who has to decide whether a picture visualization3d=self._visualization3d_metadata(),
# was part of the observation. Display-only facts: no measurement, and no )
# counters (those are only known when the recording stops and travel as an return session_metadata.build_session_metadata(config, facts)
# event).
"visualization3d": self._visualization3d_metadata(),
}
def _visualization3d_metadata(self) -> dict: def _visualization3d_metadata(self) -> dict:
"""The pane's state for ``session.json``. Must never block a recording. """The pane's state for ``session.json``. Must never block a recording.
@@ -653,68 +582,21 @@ class StudioMainWindow(QMainWindow):
@staticmethod @staticmethod
def _config_hash(config_ref: str) -> str: def _config_hash(config_ref: str) -> str:
try: """Content hash of the sensor config (see ``session_metadata``)."""
import hashlib return session_metadata.config_sha256(config_ref)
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 @staticmethod
def _sdk_version() -> str: def _sdk_version() -> str:
try: """The SDK version in use, or ``"unavailable"`` (see ``session_metadata``)."""
import orisys return session_metadata.sdk_version()
return str(getattr(orisys, "__version__", "unknown"))
except Exception: # noqa: BLE001
return "unavailable"
def _filter_settings(self) -> FilterSettings: def _filter_settings(self) -> FilterSettings:
"""The filters the operator asked for, from the preferences file. """The filters the operator asked for, from the preferences file."""
return session_metadata.filter_settings_from_prefs(self.prefs)
One place reads them, so the dialog that writes them and the session that
runs them cannot drift apart. An empty or unreadable preference is
"nothing filters" — the same default the shipped sensor configs state.
"""
keys = (
"flow_ema_alpha",
"force_median_window",
"force_lowpass_cutoff_hz",
"force_lowpass_order",
"force_lowpass_sample_rate_hz",
"show_unfiltered",
"preset",
)
stored = {}
for key in keys:
try:
value = self.prefs.get(f"plot_{key}")
except Exception: # noqa: BLE001 - a broken preference is not fatal
value = None
if value is not None:
stored[key] = value
return settings_from_mapping(stored)
def _capture_metadata(self) -> dict: def _capture_metadata(self) -> dict:
"""Requested vs negotiated capture parameters, for ``session.json``. """Requested vs negotiated capture parameters, for ``session.json``."""
return session_metadata.capture_metadata(self._active_worker)
Empty while nothing is open. A recording that runs at 16.7 fps on a 30 fps
config has to be able to explain itself; from the frames alone nobody can
tell whether the camera or the pipeline was the ceiling.
"""
worker = getattr(self, "_active_worker", None)
reader = getattr(worker, "capture_status", None) if worker is not None else None
if not callable(reader):
return {}
try:
return dict(reader() or {})
except Exception: # noqa: BLE001 - metadata must never block a record
logger.exception("Could not read the capture status")
return {}
def _check_capture_negotiation(self) -> None: def _check_capture_negotiation(self) -> None:
"""Say it out loud when the camera delivers less than the config asks. """Say it out loud when the camera delivers less than the config asks.
@@ -727,15 +609,8 @@ class StudioMainWindow(QMainWindow):
""" """
if self._capture_checked: if self._capture_checked:
return return
worker = getattr(self, "_active_worker", None) status = session_metadata.capture_metadata(self._active_worker)
reader = getattr(worker, "capture_status", None) if worker is not None else None if not status:
if not callable(reader):
return
try:
status = dict(reader() or {})
except Exception: # noqa: BLE001
logger.exception("Could not read the capture status")
self._capture_checked = True
return return
measured = float(self._latest_capture_fps or 0.0) measured = float(self._latest_capture_fps or 0.0)
if measured <= 0.0: if measured <= 0.0:
@@ -748,10 +623,9 @@ class StudioMainWindow(QMainWindow):
def _filter_metadata(self) -> dict: def _filter_metadata(self) -> dict:
"""The filters this session actually ran, for ``session.json``. """The filters this session actually ran, for ``session.json``.
Read from the worker that is running the device, not from the settings The block (and the rule that the filters are read back from the worker that ran
the operator chose: those are two different things whenever the SDK is them, not from what was requested) lives in ``session_metadata``; this only
older than the request, and the record has to describe the data it holds. resolves *which* worker to ask.
Never raises — a record that cannot name its filters is still a record.
""" """
try: try:
worker = self._active_worker worker = self._active_worker
@@ -760,19 +634,7 @@ class StudioMainWindow(QMainWindow):
except Exception: # noqa: BLE001 except Exception: # noqa: BLE001
logger.exception("Could not find the worker to ask about the filters") logger.exception("Could not find the worker to ask about the filters")
return FilterSettings().metadata() return FilterSettings().metadata()
try: return session_metadata.filter_metadata(worker)
state = worker.filter_state() if worker is not None else {}
settings = worker.filter_settings() if worker is not None else None
unsupported = tuple(worker.filters_unsupported()) if worker is not None else ()
except Exception: # noqa: BLE001
logger.exception("Could not read the filter state for the record")
state, settings, unsupported = {}, None, ()
block = (settings or FilterSettings()).metadata(state or None)
if unsupported:
# Named, not hidden: the record says which settings the installed SDK
# could not apply, so the numbers can be read for what they are.
block["unsupported_by_sdk"] = list(unsupported)
return block
# ------------------------------------------------------- record channel # ------------------------------------------------------- record channel
def _on_packet_recorded(self, packet) -> None: def _on_packet_recorded(self, packet) -> None:
@@ -1959,19 +1821,12 @@ class StudioMainWindow(QMainWindow):
) )
def _device_sample_rate(self) -> float: def _device_sample_rate(self) -> float:
"""The frame rate the low-pass coefficients would be designed for. """The frame rate the low-pass coefficients would be designed for."""
return session_metadata.device_sample_rate(
The device's measured rate once it has one, else the configured 30 fps — self._latest_fps,
the dialog needs a number to check a cutoff against Nyquist. self.prefs.get("measured_fps"),
""" default=float(FORCE_SAMPLE_RATE_HZ),
for source in (self._latest_fps, self.prefs.get("measured_fps")): )
try:
rate = float(source or 0.0)
except (TypeError, ValueError):
rate = 0.0
if rate > 0:
return rate
return float(FORCE_SAMPLE_RATE_HZ)
def _apply_plot_prefs_to_panel(self) -> None: def _apply_plot_prefs_to_panel(self) -> None:
self.force_plots.set_window_sec( self.force_plots.set_window_sec(
@@ -0,0 +1,272 @@
"""The facts a recording has to state about the session that produced it.
``session.json`` is the only witness a reader has when they ask "what was this?" months
later, so its rules are the reason this module exists rather than living in the window:
* the recorded video's frame rate is the **configured** rate for a camera and the
**measured** one for a file — using the measured counter made a 30 fps sensor write a
container that declared (and played at) 48 fps, and an unreadable config must not
become a "0 fps" container either;
* the filters are read back from the worker that actually ran them, not from the
settings the operator asked for: an older SDK may have applied less than requested,
and the record has to describe the data it holds;
* the device's measured capture rate is kept next to the requested one, because a
recording that runs at 16.7 fps on a 30 fps config has to be able to explain itself;
* none of this may block a recording: every read is guarded and a failure becomes an
empty block rather than an exception.
Everything here is a **pure function of the facts the caller passes in**: no widgets, no
Qt, no session state. That is what makes it testable on its own, and it is the stateless
half of the session concern — the stateful half (workers, phase, timers, the packet
fan-out) is ``SessionController``.
"""
from __future__ import annotations
import hashlib
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Mapping
from orisys_app_shared.force.filters import FilterSettings, settings_from_mapping
from orisys_app_shared.record import RecordConfig
logger = logging.getLogger(__name__)
#: The preference keys that describe the filter chain. One list, so the dialog that
#: writes them and the session that runs them cannot drift apart.
FILTER_PREF_KEYS = (
"flow_ema_alpha",
"force_median_window",
"force_lowpass_cutoff_hz",
"force_lowpass_order",
"force_lowpass_sample_rate_hz",
"show_unfiltered",
"preset",
)
def config_sha256(config_ref: str | Path) -> str:
"""Content hash of the sensor config, or "" when it cannot be read."""
try:
path = Path(str(config_ref))
if path.is_file():
return hashlib.sha256(path.read_bytes()).hexdigest()
except Exception: # noqa: BLE001 - metadata must never block a recording
logger.debug("Could not hash the config %s", config_ref, exc_info=True)
return ""
def sdk_version() -> str:
"""The SDK version in use, or ``"unavailable"``."""
try:
import orisys
return str(getattr(orisys, "__version__", "unknown"))
except Exception: # noqa: BLE001 - a missing SDK is a fact about the record
return "unavailable"
def device_serial(info: Any) -> str:
"""The sensor's own serial number, when the SDK exposes one."""
for attribute in ("sn", "serial", "serial_number"):
value = getattr(info, attribute, None)
if value:
return str(value)
serial = getattr(info, "info", None)
return str(getattr(serial, "SN", "") or "")
def capture_metadata(worker: Any) -> dict:
"""Requested vs negotiated capture parameters, for ``session.json``.
Empty while nothing is open. From the frames alone nobody can tell whether the
camera or the pipeline was the ceiling, so the worker's own status is copied.
"""
reader = getattr(worker, "capture_status", None) if worker is not None else None
if not callable(reader):
return {}
try:
return dict(reader() or {})
except Exception: # noqa: BLE001 - metadata must never block a record
logger.exception("Could not read the capture status")
return {}
def filter_metadata(worker: Any) -> dict:
"""The filters this session actually ran, for ``session.json``.
Never raises: a record that cannot name its filters is still a record.
"""
try:
state = worker.filter_state() if worker is not None else {}
settings = worker.filter_settings() if worker is not None else None
unsupported = tuple(worker.filters_unsupported()) if worker is not None else ()
except Exception: # noqa: BLE001
logger.exception("Could not read the filter state for the record")
state, settings, unsupported = {}, None, ()
block = (settings or FilterSettings()).metadata(state or None)
if unsupported:
# Named, not hidden: the record says which settings the installed SDK could
# not apply, so the numbers can be read for what they are.
block["unsupported_by_sdk"] = list(unsupported)
return block
def filter_settings_from_prefs(prefs: Mapping[str, Any]) -> FilterSettings:
"""The filters the operator asked for, from the preferences file.
An empty or unreadable preference is "nothing filters" — the same default the
shipped sensor configs state.
"""
stored: dict[str, Any] = {}
for key in FILTER_PREF_KEYS:
try:
value = dict(prefs or {}).get(f"plot_{key}")
except Exception: # noqa: BLE001 - a broken preference is not fatal
value = None
if value is not None:
stored[key] = value
return settings_from_mapping(stored)
def video_fps(
measured_fps: float,
*,
source_is_file: bool,
config_ref: str | Path | None,
) -> float:
"""The rate the recorded video is written at.
A camera has a configured rate and the container must honour it; a file source has
none, so its measured rate is the honest one. Never returns 0: a container declared
as "0 fps" would thin the video to one picture per second.
"""
measured = float(measured_fps or 0.0)
if not source_is_file:
try:
from orisys_app_shared.acquisition.models import load_fixed_mode
mode = load_fixed_mode(str(config_ref or ""))
if mode.fps > 0:
return float(mode.fps)
except Exception: # noqa: BLE001 - the measured rate is the fallback
logger.debug("Could not read the configured frame rate", exc_info=True)
return measured if measured > 0.0 else float(RecordConfig().fps)
def device_sample_rate(
measured_fps: Any, stored_measured_fps: Any, *, default: float
) -> float:
"""The rate the low-pass coefficients would be designed for.
The device's measured rate once it has one, else whatever the preferences last
recorded, else the caller's default — the filter dialog needs a number to check a
cutoff against Nyquist.
"""
for source in (measured_fps, stored_measured_fps):
try:
rate = float(source or 0.0)
except (TypeError, ValueError):
rate = 0.0
if rate > 0:
return rate
return float(default)
@dataclass(frozen=True)
class SessionFacts:
"""The handful of facts only the window can see, for the metadata block."""
sensor_type: str
config_ref: str
#: The worker that is running the device (its capture status and filters are read
#: back from *it*, not from what was requested).
worker: Any = None
#: The last packet: frame size and whether the force channels are calibrated.
info: Any = None
video_fps: float = 30.0
measured_fps: float = 0.0
active_calibration_id: str = ""
#: The serial the operator entered in the calibration panel.
operator_sn: str = ""
recording_mode: str = "video"
#: The display block, already assembled by the pane that owns it.
visualization3d: Mapping[str, Any] = field(default_factory=dict)
def build_session_metadata(config: RecordConfig, facts: SessionFacts) -> dict:
"""The ``session.json`` metadata written when a recording starts."""
from orisys_app_shared import API_VERSION, __version__ as shared_version
from orisys_studio import __version__ as app_version
info = facts.info
raw_frame = getattr(info, "raw_frame", None)
config_ref = str(facts.config_ref or "")
config_hash = config_sha256(config_ref)
return {
"device": {
"sensor_type": str(facts.sensor_type),
"config_ref": config_ref,
"config_sha256": config_hash,
"frame": {
"width": int(raw_frame.shape[1]) if raw_frame is not None else None,
"height": int(raw_frame.shape[0]) if raw_frame is not None else None,
"fps": float(facts.video_fps),
# The device's own measured rate, kept so a mismatch between the two
# can be diagnosed after the fact.
"measured_fps": float(facts.measured_fps),
},
},
"versions": {
"app": app_version,
"shared": shared_version,
"shared_api": API_VERSION,
"sdk": sdk_version(),
},
# What the camera was asked for and what the driver gave. A recording that runs
# at 16.7 fps instead of 30 has to say why by itself.
"capture": capture_metadata(facts.worker),
"calibration": {
"config_ref": config_ref,
"force_calibrated": bool(getattr(info, "force_calibrated", False)),
# Content hash of the coefficients the device confirmed. It is filled in by
# the acquisition thread's confirmation, so it can be empty here; the CSV's
# per-row version is authoritative.
"version": str(facts.active_calibration_id or ""),
# The operator-entered serial of the *calibrated* device, as the
# calibration panel has it (the device SN when the config has one).
"sn": str(facts.operator_sn or "").strip(),
"device_sn": device_serial(info),
},
# Depth is deliberately NOT stored (decided with the team): it is a function of
# the frame and the sensor configuration, and the replay view shows the recorded
# frames only. Re-running the pipeline over raw_*.mp4 is the "recompute" mode's
# job; the hashes below are what such a pass checks before trusting it.
"depth": {
"stored": False,
"replay": "none",
"quantity": "relative deformation / algorithm response, not a calibrated length",
"method": "re-run the SDK over raw_*.mp4 with this config_ref",
"config_sha256": config_hash,
"sdk_version": sdk_version(),
},
"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; an event "
"that predates it has no relative base and is left empty"
),
"frame_id": "monotonic per acquisition session",
},
"filter": filter_metadata(facts.worker),
"recording_mode": str(facts.recording_mode),
"video_config": config.as_dict(),
# What the 3D pane was, for a reader who has to decide whether a picture was
# part of the observation. Display-only facts: no measurement, and no counters
# (those are only known when the recording stops and travel as an event).
"visualization3d": dict(facts.visualization3d or {}),
}
+208
View File
@@ -0,0 +1,208 @@
"""Acceptance for the facts a recording states about its own session.
``session.json`` is the only witness a reader has later, so these are the rules that
keep it honest: which frame rate a container may declare, that the filters are read back
from the worker that ran them, that no metadata read can fail a recording, and that the
3D pane's own description is passed through untouched.
"""
from __future__ import annotations
import hashlib
from pathlib import Path
from types import SimpleNamespace
import numpy as np
import pytest
from orisys_app_shared.force.filters import FilterSettings
from orisys_app_shared.record import RecordConfig
from orisys_studio import session_metadata as sm
CONFIG = (
Path(__file__).resolve().parents[1]
/ "config"
/ "sensors"
/ "finger"
/ "sensor.json"
)
# --------------------------------------------------------------------- config hash
def test_the_config_hash_is_the_file_content_or_nothing(tmp_path):
path = tmp_path / "sensor.json"
path.write_text('{"camera": {}}', encoding="utf-8")
assert sm.config_sha256(path) == hashlib.sha256(path.read_bytes()).hexdigest()
assert sm.config_sha256(tmp_path / "gone.json") == ""
assert sm.config_sha256("") == ""
def test_the_sdk_version_is_a_string_even_with_no_sdk():
value = sm.sdk_version()
assert isinstance(value, str) and value
# ------------------------------------------------------------------- device serial
def test_the_device_serial_comes_from_whichever_field_the_sdk_uses():
assert sm.device_serial(SimpleNamespace(sn="A1")) == "A1"
assert sm.device_serial(SimpleNamespace(serial="B2")) == "B2"
assert sm.device_serial(SimpleNamespace(serial_number="C3")) == "C3"
assert sm.device_serial(SimpleNamespace(info=SimpleNamespace(SN="D4"))) == "D4"
assert sm.device_serial(SimpleNamespace()) == ""
assert sm.device_serial(None) == ""
# -------------------------------------------------------------- capture / filters
def test_capture_metadata_copies_what_the_worker_says_and_survives_failure():
assert sm.capture_metadata(None) == {}
assert sm.capture_metadata(SimpleNamespace()) == {}
worker = SimpleNamespace(capture_status=lambda: {"requested_fps": 30.0})
assert sm.capture_metadata(worker) == {"requested_fps": 30.0}
def explode():
raise RuntimeError("driver gone")
# A recording must not fail because the camera cannot describe itself.
assert sm.capture_metadata(SimpleNamespace(capture_status=explode)) == {}
def test_filter_metadata_names_what_the_sdk_could_not_apply():
settings = FilterSettings()
worker = SimpleNamespace(
filter_state=lambda: {"flow_ema_alpha": 1.0},
filter_settings=lambda: settings,
filters_unsupported=lambda: ("force_lowpass_order",),
)
block = sm.filter_metadata(worker)
assert block["unsupported_by_sdk"] == ["force_lowpass_order"]
assert block # the applied settings are stated too
# No worker, and a worker that raises, both give a block rather than an exception.
assert sm.filter_metadata(None)
def explode():
raise RuntimeError("worker is gone")
assert sm.filter_metadata(SimpleNamespace(filter_state=explode))
def test_filter_settings_default_to_nothing_filters():
"""The shipped configs state "no filters"; an empty preference must agree."""
empty = sm.filter_settings_from_prefs({})
assert empty == FilterSettings()
stored = sm.filter_settings_from_prefs(
{"plot_force_median_window": 3, "plot_flow_ema_alpha": 0.5, "unrelated": 1}
)
assert stored != empty
assert stored.median_window == 3
assert stored.flow_ema_alpha == pytest.approx(0.5)
def test_a_broken_preferences_object_is_not_fatal():
class Broken(dict):
def get(self, key, default=None): # noqa: D102 - deliberately broken
raise RuntimeError("corrupt preferences")
assert sm.filter_settings_from_prefs(Broken()) == FilterSettings()
# ------------------------------------------------------------------- frame rates
def test_a_camera_declares_its_configured_rate_and_a_file_its_measured_one():
"""The container's fps must describe the spacing of the pictures it holds."""
if CONFIG.is_file():
assert sm.video_fps(48.0, source_is_file=False, config_ref=CONFIG) == 30.0
# A file source has no configured rate to honour.
assert sm.video_fps(17.5, source_is_file=True, config_ref=CONFIG) == 17.5
# An unreadable config falls back to the measured rate, never to zero.
assert sm.video_fps(12.0, source_is_file=False, config_ref="missing.json") == 12.0
assert sm.video_fps(0.0, source_is_file=True, config_ref=None) == float(
RecordConfig().fps
)
def test_the_filter_dialog_gets_a_rate_it_can_check_nyquist_against():
assert sm.device_sample_rate(25.0, None, default=30.0) == 25.0
assert sm.device_sample_rate(None, "15", default=30.0) == 15.0
assert sm.device_sample_rate("garbage", "also garbage", default=30.0) == 30.0
assert sm.device_sample_rate(None, 0, default=30.0) == 30.0
# ------------------------------------------------------------- the whole metadata
def test_the_session_block_states_the_device_the_filters_and_the_picture():
info = SimpleNamespace(
raw_frame=np.zeros((48, 64, 3), dtype=np.uint8), force_calibrated=True, sn="SN-9"
)
worker = SimpleNamespace(
capture_status=lambda: {"requested_fps": 30.0, "measured_fps": 21.0},
filter_state=lambda: {},
filter_settings=lambda: FilterSettings(),
filters_unsupported=lambda: (),
)
facts = sm.SessionFacts(
sensor_type="finger",
config_ref=str(CONFIG) if CONFIG.is_file() else "missing.json",
worker=worker,
info=info,
video_fps=30.0,
measured_fps=21.0,
active_calibration_id="cal-7",
operator_sn=" OPERATOR-1 ",
recording_mode="video",
visualization3d={"block_version": 1, "enabled": True},
)
config = RecordConfig()
block = sm.build_session_metadata(config, facts)
assert block["device"]["sensor_type"] == "finger"
assert block["device"]["frame"] == {
"width": 64,
"height": 48,
"fps": 30.0,
"measured_fps": 21.0,
}
assert block["versions"]["sdk"]
assert block["capture"] == {"requested_fps": 30.0, "measured_fps": 21.0}
assert block["calibration"]["force_calibrated"] is True
assert block["calibration"]["version"] == "cal-7"
assert block["calibration"]["sn"] == "OPERATOR-1"
assert block["calibration"]["device_sn"] == "SN-9"
assert block["depth"]["stored"] is False
assert block["depth"]["config_sha256"] == block["device"]["config_sha256"]
assert block["filter"]
assert block["recording_mode"] == "video"
assert block["video_config"] == config.as_dict()
assert block["visualization3d"] == {"block_version": 1, "enabled": True}
assert block["conventions"]["axis_order"] == ["FN", "FX", "FY"]
def test_the_block_survives_a_worker_that_is_gone_and_a_packet_that_never_came():
"""Metadata that cannot be read is empty, never an exception (a record is running)."""
block = sm.build_session_metadata(
RecordConfig(),
sm.SessionFacts(sensor_type="box", config_ref="", worker=None, info=None),
)
assert block["device"]["frame"]["width"] is None
assert block["capture"] == {}
assert block["filter"]
assert block["device"]["config_sha256"] == ""
def test_the_3d_pane_description_is_passed_through_untouched():
described = {"scene": {"asset_id": "x"}, "gl": {}, "display": {}, "stream": {}}
block = sm.build_session_metadata(
RecordConfig(),
sm.SessionFacts(
sensor_type="finger",
config_ref="",
visualization3d={"visualization3d": described, "block_version": 1},
),
)
assert block["visualization3d"]["visualization3d"] == described
@pytest.mark.parametrize("mode", ("video", "raw", "lossless"))
def test_every_recording_mode_is_recorded_as_stated(mode):
block = sm.build_session_metadata(
RecordConfig(),
sm.SessionFacts(sensor_type="finger", config_ref="", recording_mode=mode),
)
assert block["recording_mode"] == mode