From f6f3c45285d138ee99231b163cfd454ab09c6a28 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Wed, 21 Oct 2026 11:03:53 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E5=8D=95=E8=B7=AF?= =?UTF-8?q?=E6=B5=81=E7=B1=BB=E5=9E=8B=E5=90=88=E5=90=8C=20stream-contract?= =?UTF-8?q?/1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 冻结逐帧身份、采集时钟、测量头和事件信封：FrameIdentity、FrameClock、FrameStamp、MeasurementFrame、StreamEvent，并从包根导出。采用严格读取：未知字段和未列出的版本一律拒绝，任何字段变更都升版本，读取方在 SUPPORTED_CONTRACT_VERSIONS 中列出可读版本；未知事件 kind 仍兼容。事件可哈希，kind 为 lower_snake。合同写入 docs/stream-contract.zh.md；不接入现有运行时。窄套件 139 项、lint 通过。 --- docs/stream-contract.zh.md | 176 +++++++++++ src/orisys/__init__.py | 23 ++ src/orisys/stream_contract.py | 578 ++++++++++++++++++++++++++++++++++ tests/test_stream_contract.py | 441 ++++++++++++++++++++++++++ 4 files changed, 1218 insertions(+) create mode 100644 docs/stream-contract.zh.md create mode 100644 src/orisys/stream_contract.py create mode 100644 tests/test_stream_contract.py diff --git a/docs/stream-contract.zh.md b/docs/stream-contract.zh.md new file mode 100644 index 0000000..b8b9472 --- /dev/null +++ b/docs/stream-contract.zh.md @@ -0,0 +1,176 @@ +# 单路流类型合同（`stream-contract/1`） + +用途：reference。范围：`orisys.stream_contract` 冻结的逐帧身份、采集时钟、测量头和事件信封。 + +状态：T03 已验收（done），2026-10-21。只冻结类型、校验和 JSON 往返，不接入现有 Sensor、StreamPipeline 或 ForceSession；T05、T06、T08、T11、T13–T15 复用这些头，不再各自另立字段。 + +## 为什么先冻结共享头 + +阶段 0 修复的是参考版本传递和复位线程归属；阶段 1 要让图像块、力会话、单路入口以及 Studio 的记录与分析使用同一份逐帧身份和时标。若每层各写一套字段，就会重新出现“同一 PTS 的两个源帧被当成重复”“文件名代替标定版本”“处理耗时冒充源时间”这类旧缺陷。本任务把这些共享字段冻结成可 JSON 往返的 frozen 数据类；后续任务只增加各自负载，不改已有字段含义。 + +## 稳定性规则 + +- `STREAM_CONTRACT_VERSION = "stream-contract/1"` 写入每个可持久化对象的 `to_dict()`。采用严格读取（政策 A）：**任何新增、改名、删除或改类型的字段都升版本**，不存在同版本新增字段。 +- 读取方只读 `SUPPORTED_CONTRACT_VERSIONS` 中显式列出的版本，并拒绝未列出的版本和未知字段；写方要加字段时必须升版本，并把新版本加入读取方列表。 +- 增加新事件 `kind` 属于兼容变化：消费者忽略不认识的 `kind`，不因未知事件失败；但 `kind` 本身的名称格式受信封合同约束（见下）。 +- 构造即校验；非法数据立即失败，不生成“看起来合理”的默认值。缺失组身份不伪造同步，缺失时钟不补假时间。 +- 本任务不改变已发布接口：`Sensor.reset()`、`StreamPipeline` 快照和当前 `ForceSession` 行为不变；T06 重做会话时才改为直接消费这里定义的测量头。 + +## 帧身份 + +`FrameIdentity` 表达“哪一路的哪一帧”，字段如下。 + +| 字段 | 类型 | 说明 | +|---|---|---| +| `sensor_id` | `str` | 逻辑 Sensor 身份，非空 | +| `session_id` | `str` | 一次运行/录制的会话身份，非空 | +| `frame_id` | `int` | 该路连续输出序号，`>= 0` | +| `source_frame_id` | `int \| None` | 源自己的帧号；源没有编号时为 `None`，有编号时必须保留原值 | +| `acquisition_id` | `str \| None` | 可选组身份之一，与 `sync_group_id` 同时出现或同时缺失 | +| `sync_group_id` | `str \| None` | 可选组身份之一，不得只给一半 | + +规则： + +- 去重、缓存和迟到判断以源身份为准；时间戳不是身份。两个源帧可以有相同的 PTS，它们仍是两帧。 +- 相同时间戳不触发去重、清零、复位或缓存替换；入口如何忽略重复输入属于 T08，本任务只保证类型层面可区分。 +- 组身份缺失表示本路独立，不表示已同步；不伪造 acquisition/sync 值。 +- 组身份即使存在，也只说明采集组归属；跨时钟映射和同步容差仍需后续计划实测，不能由字段存在推断。 + +## 采集时钟 + +`FrameClock` 保存一帧的采集时间和它的时间来源。 + +| 字段 | 类型 | 说明 | +|---|---|---| +| `capture_time_ns` | `int` | 采集时间，整数纳秒，`>= 0` | +| `clock_id` | `str` | 时钟域标识，例如 `host_monotonic`、`video:recording-4` | +| `capture_time_basis` | `str` 枚举 | 见下表 | +| `source_pts` | `int \| float \| None` | 源时间戳原值（单位由源声明），与处理时间无关 | + +`capture_time_basis` 的取值： + +| 取值 | 含义 | +|---|---| +| `capture_monotonic` | 实时源的主机单调采集/接收时间 | +| `recorded_signal` | 录像保存的信号时间（回放/debug） | +| `video_pts` | 容器真实 PTS 换算成纳秒 | +| `estimated_source` | 无可靠 PTS，按源帧号和名义帧率推算；消费方必须当作估算值 | + +规则： + +- 不能用“处理计数 × 名义周期”冒充 `video_pts`，不用离线处理耗时作为信号时间，也不伪造递增 PTS。没有可信 PTS 时明确标为 `estimated_source`。 +- `source_pts` 保留原值，不因换算成 `capture_time_ns` 而丢弃；发布日期、写入时间由结果信封另记，不写进采集时间。 +- 时钟域之间不隐式换算；需要换算时由上层显式声明来源。 + +## 帧头 FrameStamp + +`FrameStamp` 是身份加时钟的扁平帧头，可持久化形式如下；`identity` 和 `clock` 属性返回对应的两个组件，`from_parts()` 从组件构造。 + +```json +{ + "schema": "stream-contract/1", + "sensor_id": "FINGER-001", + "session_id": "run-7", + "frame_id": 12, + "source_frame_id": 41, + "capture_time_ns": 1600000000000000000, + "clock_id": "host_monotonic", + "capture_time_basis": "capture_monotonic", + "source_pts": 41.5, + "acquisition_id": null, + "sync_group_id": null +} +``` + +## 测量头 MeasurementFrame + +| 字段 | 类型 | 说明 | +|---|---|---| +| `stamp` | `FrameStamp` | 本帧身份与采集时钟 | +| `source_dt_s` | `float \| None` | 与同一 Sensor 上一条实际处理帧之间的源时间差，单位秒；第一帧为 `None`，其余 `>= 0` | +| `reference_generation` | `int` | 计算该帧光流时生效的参考版本，`>= 0` | + +`source_dt_s` 的定义和边界： + +- 用声明的采集时钟计算源时间差，不是处理速度，也不是相邻输出序号的固定周期。 +- 相同源时间（`0.0`）是合法值，表示两帧时间相等；不能因此把两帧合并。 +- 长间隔只产生警告和质量标记，不自动复位、清零或改滤波设计帧率；阈值由后续实测确定（K4.5、K9）。 +- `reference_generation` 由图像线程在该帧计算光流时写入，随后沿用到结果；不能在 HHD 输出或 Studio 消费时读取全局最新计数补标签（K8.1）。迟到旧版本的样本不能进入新零点基线。 + +测量负载（flow、HHD 场、接触、场标定特征）以及力结果、批次和记录格式不在本任务冻结；T05、T06 定义负载并在本头上扩展，T11、T13–T15 复用同一身份和时标。 + +## 事件信封 StreamEvent + +`StreamEvent` 是状态变化事件的公共信封：参考复位、归零进度、滤波切换、长间隔、跳帧等都由后续任务登记具体 `kind` 与 `payload` 字段。 + +| 字段 | 类型 | 说明 | +|---|---|---| +| `kind` | `str` | 稳定的 lower_snake 事件名，必须匹配 `^[a-z][a-z0-9_]*$`；未知 kind 必须被消费者忽略而不是报错 | +| `stamp` | `FrameStamp \| None` | 事件所属帧；无帧事件为 `None` | +| `payload` | `Mapping[str, Any]` | 只允许 JSON 类型，构造时校验并冻结为只读深拷贝 | +| `event_version` | `int` | 信封版本，`>= 1`；信封结构不兼容变化时递增 | + +`payload` 拒绝 `NaN`、无穷、`bytes`、集合和非字符串键，因为这些值无法无损通过 `json.dumps`/`json.loads` 往返。事件序列化后必须能原样读回，记录才能真正复现在线状态。`StreamEvent` 可哈希：`payload` 冻结但不参与哈希，哈希只由 `kind`、`stamp` 和 `event_version` 决定，因此相等事件哈希相同；不同 payload 的事件允许落在同一哈希桶。 + +```json +{ + "schema": "stream-contract/1", + "kind": "source_gap", + "event_version": 1, + "stamp": { + "schema": "stream-contract/1", + "sensor_id": "BOX-3", + "session_id": "run-7", + "frame_id": 3, + "source_frame_id": 12, + "capture_time_ns": 1600000000000000000, + "clock_id": "host_monotonic", + "capture_time_basis": "capture_monotonic", + "source_pts": null, + "acquisition_id": null, + "sync_group_id": null + }, + "payload": {"source_dt_s": 0.5, "dropped_frames": 3} +} +``` + +## JSON 往返与严格校验 + +每个可持久化类型提供 `to_dict()` 和 `from_dict()`。`to_dict()` 带 `schema`；`from_dict()` 只接受 `SUPPORTED_CONTRACT_VERSIONS` 中显式列出的版本，并校验必填字段、未知字段和字段类型，损坏的记录立即失败而不是被部分接受。下面的例子可以直接运行： + +```python +import json + +from orisys import FrameClock, FrameIdentity, FrameStamp, MeasurementFrame + +stamp = FrameStamp.from_parts( + FrameIdentity( + sensor_id="FINGER-001", + session_id="run-7", + frame_id=12, + source_frame_id=41, + ), + FrameClock( + capture_time_ns=1_600_000_000_000_000_000, + clock_id="host_monotonic", + capture_time_basis="capture_monotonic", + ), +) +text = json.dumps(MeasurementFrame(stamp, source_dt_s=0.033).to_dict()) +measurement = MeasurementFrame.from_dict(json.loads(text)) +assert measurement.stamp == stamp +``` + +未知字段被拒绝，是为了让不同版本的记录不会静默错配。采用政策 A：字段变更（包括新增字段）一律升 `STREAM_CONTRACT_VERSION`，同时把新版本加入读取方的 `SUPPORTED_CONTRACT_VERSIONS`；同版本内不新增字段。 + +## 不包含什么 + +- 不实现 SensorStream 入口、实时最新选帧、完整回放和背压（T08）。 +- 不实现图像块拆分、测量负载、ForceSession 重做、零点与输出滤波（T04–T07）。 +- 不实现批次、配组、多流调度（后续多流计划）。 +- 不实现录制/回放格式、Studio 迁移和离线重新滤波（T11、T13–T15）。 +- 不修改现有 `Sensor`、`StreamPipeline` 和当前 `ForceSession`；现有 `ForceSession.FrameStamp` 在 T06 重做时被本合同的帧头替换。 + +## 验证 + +`tests/test_stream_contract.py` 覆盖 T03 的四项验收：逐路身份与时钟、组字段可缺但不半组、同 PTS 帧可区分、`source_dt_s` 与事件 JSON 往返；同时覆盖 frozen 校验、严格版本/未知字段拒绝（政策 A）、`SUPPORTED_CONTRACT_VERSIONS`、事件可哈希、`kind` 的 lower_snake 格式和非法负载拒绝。类型本身不启动传感器、流水线或会话。 diff --git a/src/orisys/__init__.py b/src/orisys/__init__.py index c414946..a68b19c 100644 --- a/src/orisys/__init__.py +++ b/src/orisys/__init__.py @@ -19,6 +19,18 @@ from .pipeline import ( StreamPipeline, ) from .sensor import Sensor +from .stream_contract import ( + STREAM_CONTRACT_VERSION, + SUPPORTED_CONTRACT_VERSIONS, + TIME_BASES, + FrameClock, + TimeBasis, + FrameIdentity, + FrameStamp, + MeasurementFrame, + StreamEvent, +) + #: The explicit force-readout session (``studio-force/1``) used by the Studio #: workflow. Imported after ``Sensor`` because it composes the sensor, model and #: output blocks; the historical ``Sensor.read_info``/``ForceReadout.apply`` @@ -70,6 +82,17 @@ __all__ = [ 'ForceSession', 'WORKFLOW_VERSION', + # frozen single-stream contract (stream-contract/1) + 'STREAM_CONTRACT_VERSION', + 'SUPPORTED_CONTRACT_VERSIONS', + 'TIME_BASES', + 'TimeBasis', + 'FrameIdentity', + 'FrameClock', + 'FrameStamp', + 'MeasurementFrame', + 'StreamEvent', + # configs utilities functions 'list_config', 'export_config', diff --git a/src/orisys/stream_contract.py b/src/orisys/stream_contract.py new file mode 100644 index 0000000..4436a0d --- /dev/null +++ b/src/orisys/stream_contract.py @@ -0,0 +1,578 @@ +"""Frozen public types for the single-stream SensorStream contract. + +This module is the stage-1A **type contract** (task T03): the frame identity, +capture clock, measurement header and event envelope that the per-sensor stream +(T08), the rebuilt force session (T06) and the recording/analysis consumers +(T11, T13-T15) share. It deliberately does not change any existing class: +``Sensor``, ``StreamPipeline`` and the current ``ForceSession`` keep their +published behavior, and the later tasks adopt these headers instead of growing +per-layer variants. + +The contract rules are documented in ``docs/stream-contract.zh.md``. In short: + +* identity is per stream - ``sensor_id`` + ``session_id`` + source/original + frame number and a separate contiguous output number; +* a group identity (``acquisition_id`` + ``sync_group_id``) is optional and is + never fabricated: both parts are present or neither is; +* a capture time is a number of nanoseconds plus the clock domain and the basis + it was derived from; a raw source PTS is carried separately and is never + replaced by processing time; +* two frames with the same timestamp are still different frames (identity, not + the timestamp, is the deduplication key); +* ``source_dt_s`` is the source-time interval to the previous *actually + processed* frame of the same stream (``None`` for the first) and never a + processing-speed measurement; +* events round-trip through JSON with a JSON-safe payload. + +Versioning follows one strict rule (policy A): **any new, renamed, removed or +re-typed field bumps ``STREAM_CONTRACT_VERSION``**. A reader rejects unknown +fields and only reads the versions it lists in +``SUPPORTED_CONTRACT_VERSIONS``; a writer that needs a field must bump the +version and both sides must list it. Event ``kind`` values are not schema +fields: consumers must ignore kinds they do not know, so new kinds stay +additive within the same version. +""" + +from __future__ import annotations + +import math +import re +from collections.abc import Mapping +from dataclasses import dataclass, field +from types import MappingProxyType +from typing import Any, Literal + +__all__ = [ + "STREAM_CONTRACT_VERSION", + "SUPPORTED_CONTRACT_VERSIONS", + "TIME_BASES", + "FrameClock", + "FrameIdentity", + "FrameStamp", + "MeasurementFrame", + "StreamEvent", + "TimeBasis", +] + +#: Version of the serialized contract (``to_dict()``/``from_dict()``). +#: Policy A: bump it on every field addition, removal, rename or meaning +#: change; there are no same-version field additions. +STREAM_CONTRACT_VERSION = "stream-contract/1" + +#: Serialized schemas this reader accepts, and the only ones it may read. +#: A writer that bumps ``STREAM_CONTRACT_VERSION`` must add the new version +#: here (and in every reader that must read it); old versions stay listed as +#: long as the loader supports them. +SUPPORTED_CONTRACT_VERSIONS: tuple[str, ...] = (STREAM_CONTRACT_VERSION,) + +#: How ``FrameClock.capture_time_ns`` was derived. +#: +#: ``capture_monotonic`` - host monotonic capture/receive time of a live source. +#: ``recorded_signal`` - signal time stored by the recording (playback/debug). +#: ``video_pts`` - a real container PTS converted to nanoseconds. +#: ``estimated_source`` - derived from the source frame number and a nominal +#: rate because no trustworthy PTS exists; consumers must treat it as estimated. +TimeBasis = Literal[ + "capture_monotonic", + "recorded_signal", + "video_pts", + "estimated_source", +] + +TIME_BASES: tuple[TimeBasis, ...] = ( + "capture_monotonic", + "recorded_signal", + "video_pts", + "estimated_source", +) + +#: Event kinds are a stable lower_snake identifier; this grammar is part of the +#: frozen envelope contract (unknown *values* stay compatible, malformed ones do +#: not). +_EVENT_KIND_RE = re.compile(r"^[a-z][a-z0-9_]*$") + + +# --------------------------------------------------------------- validation +def _require_str(value: Any, name: str) -> str: + if not isinstance(value, str): + raise TypeError(f"{name} must be a string") + if not value: + raise ValueError(f"{name} must not be empty") + return value + + +def _require_int(value: Any, name: str, *, minimum: int | None = None) -> int: + # ``type`` rather than ``isinstance``: a bool is an int in Python but never a + # frame number or generation. + if type(value) is not int: + raise TypeError(f"{name} must be an integer") + if minimum is not None and value < minimum: + raise ValueError(f"{name} must be >= {minimum}") + return value + + +def _require_optional_int( + value: Any, name: str, *, minimum: int | None = None +) -> int | None: + if value is None: + return None + return _require_int(value, name, minimum=minimum) + + +def _require_finite_number(value: Any, name: str) -> int | float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise TypeError(f"{name} must be a real number") + if not math.isfinite(float(value)): + raise ValueError(f"{name} must be finite") + return value + + +def _require_time_basis(value: Any) -> TimeBasis: + if value not in TIME_BASES: + raise ValueError( + f"capture_time_basis must be one of {list(TIME_BASES)}, got {value!r}" + ) + return value + + +def _require_source_dt(value: Any) -> float | None: + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise TypeError("source_dt_s must be a real number or None") + number = float(value) + if not math.isfinite(number): + raise ValueError("source_dt_s must be finite") + if number < 0.0: + raise ValueError("source_dt_s must be >= 0") + return number + + +# ----------------------------------------------------------- JSON payloads +def _freeze_json(value: Any, path: str) -> Any: + """Validate a JSON-serializable value and return an immutable deep copy. + + Only JSON types are accepted (``None``/bool/int/finite float/str/mapping + with string keys/list/tuple). ``NaN`` and infinities are rejected because + they do not survive ``json.dumps``/``json.loads`` unchanged. + """ + if value is None or isinstance(value, (bool, str)): + return value + if isinstance(value, int): + return value + if isinstance(value, float): + if not math.isfinite(value): + raise ValueError(f"{path} contains a non-finite float") + return value + if isinstance(value, Mapping): + frozen: dict[str, Any] = {} + for key, item in value.items(): + if not isinstance(key, str): + raise TypeError( + f"{path} keys must be strings, got {type(key).__name__}" + ) + frozen[key] = _freeze_json(item, f"{path}.{key}") + return MappingProxyType(frozen) + if isinstance(value, (list, tuple)): + return tuple( + _freeze_json(item, f"{path}[{index}]") for index, item in enumerate(value) + ) + raise TypeError(f"{path} contains non-JSON value {type(value).__name__}") + + +def _thaw_json(value: Any) -> Any: + if isinstance(value, Mapping): + return {key: _thaw_json(item) for key, item in value.items()} + if isinstance(value, tuple): + return [_thaw_json(item) for item in value] + return value + + +def _mapping( + data: Any, name: str, allowed: frozenset[str], required: frozenset[str] +) -> Mapping: + """Read one serialized object under policy A (strict field set). + + Unknown fields are rejected, not ignored: a same-version record may only + contain the fields of that frozen version. A new field therefore requires a + new ``STREAM_CONTRACT_VERSION`` and an entry in + ``SUPPORTED_CONTRACT_VERSIONS``; components without a schema (identity and + clock) follow the same allow-list rule through their owners. + """ + if not isinstance(data, Mapping): + raise TypeError(f"{name} must be a mapping") + unknown = set(data) - allowed + if unknown: + raise ValueError( + f"{name} has unknown field(s) {sorted(unknown)}; adding a field " + "requires a new STREAM_CONTRACT_VERSION" + ) + missing = required - set(data) + if missing: + raise ValueError(f"{name} is missing field(s) {sorted(missing)}") + return data + + +def _require_schema(value: Any, name: str) -> str: + """Accept only the serialized versions this reader explicitly lists.""" + if not isinstance(value, str): + raise TypeError(f"{name}.schema must be a string") + if value not in SUPPORTED_CONTRACT_VERSIONS: + raise ValueError( + f"{name}.schema must be one of {list(SUPPORTED_CONTRACT_VERSIONS)}, " + f"got {value!r}" + ) + return value + + +# ------------------------------------------------------------- frame types +@dataclass(frozen=True) +class FrameIdentity: + """Identity of one frame in one stream. + + ``source_frame_id`` keeps the source's own frame number (``None`` when the + source does not provide one); ``frame_id`` is the contiguous output number + of this stream. Both are needed: the same timestamp does not identify a + frame, and two different source frames may share a PTS. + + ``acquisition_id``/``sync_group_id`` describe an optional capture group + (for example one hardware trigger spanning several sensors). They are + both present or both ``None``; a missing group is not replaced with a + fabricated value and does not imply synchronization. + """ + + sensor_id: str + session_id: str + frame_id: int + source_frame_id: int | None + acquisition_id: str | None = None + sync_group_id: str | None = None + + def __post_init__(self) -> None: + _require_str(self.sensor_id, "sensor_id") + _require_str(self.session_id, "session_id") + _require_int(self.frame_id, "frame_id", minimum=0) + _require_optional_int(self.source_frame_id, "source_frame_id", minimum=0) + if (self.acquisition_id is None) != (self.sync_group_id is None): + raise ValueError( + "group identity is optional but must be complete: set both " + "acquisition_id and sync_group_id, or leave both None" + ) + if self.acquisition_id is not None: + _require_str(self.acquisition_id, "acquisition_id") + _require_str(self.sync_group_id, "sync_group_id") + + @property + def has_group(self) -> bool: + """Whether this frame carries a capture-group identity.""" + return self.acquisition_id is not None + + @property + def group_identity(self) -> tuple[str, str] | None: + """``(acquisition_id, sync_group_id)``, or ``None`` when absent.""" + if self.acquisition_id is None: + return None + return (str(self.acquisition_id), str(self.sync_group_id)) + + def to_dict(self) -> dict[str, Any]: + return { + "sensor_id": self.sensor_id, + "session_id": self.session_id, + "frame_id": self.frame_id, + "source_frame_id": self.source_frame_id, + "acquisition_id": self.acquisition_id, + "sync_group_id": self.sync_group_id, + } + + @classmethod + def from_dict(cls, data: Mapping) -> FrameIdentity: + allowed = frozenset( + { + "sensor_id", + "session_id", + "frame_id", + "source_frame_id", + "acquisition_id", + "sync_group_id", + } + ) + required = frozenset({"sensor_id", "session_id", "frame_id", "source_frame_id"}) + fields = _mapping(data, "FrameIdentity", allowed, required) + return cls( + sensor_id=fields["sensor_id"], + session_id=fields["session_id"], + frame_id=fields["frame_id"], + source_frame_id=fields["source_frame_id"], + acquisition_id=fields.get("acquisition_id"), + sync_group_id=fields.get("sync_group_id"), + ) + + +@dataclass(frozen=True) +class FrameClock: + """Capture time of one frame, with its clock domain and time basis. + + ``capture_time_ns`` is always nanoseconds, but what it means is declared by + ``capture_time_basis``; ``clock_id`` names the clock domain (for example + ``host_monotonic`` or a source/video clock identity). ``source_pts`` keeps + the source's own presentation timestamp untouched (its unit is the source's + declared timebase); it is never replaced by processing time or by an + estimated value. A result's *published* time is recorded separately by the + result envelope, not here. + """ + + capture_time_ns: int + clock_id: str + capture_time_basis: TimeBasis + source_pts: int | float | None = None + + def __post_init__(self) -> None: + _require_int(self.capture_time_ns, "capture_time_ns", minimum=0) + _require_str(self.clock_id, "clock_id") + _require_time_basis(self.capture_time_basis) + if self.source_pts is not None: + _require_finite_number(self.source_pts, "source_pts") + + def to_dict(self) -> dict[str, Any]: + return { + "capture_time_ns": self.capture_time_ns, + "clock_id": self.clock_id, + "capture_time_basis": self.capture_time_basis, + "source_pts": self.source_pts, + } + + @classmethod + def from_dict(cls, data: Mapping) -> FrameClock: + allowed = frozenset( + {"capture_time_ns", "clock_id", "capture_time_basis", "source_pts"} + ) + required = frozenset({"capture_time_ns", "clock_id", "capture_time_basis"}) + fields = _mapping(data, "FrameClock", allowed, required) + return cls( + capture_time_ns=fields["capture_time_ns"], + clock_id=fields["clock_id"], + capture_time_basis=fields["capture_time_basis"], + source_pts=fields.get("source_pts"), + ) + + +@dataclass(frozen=True) +class FrameStamp: + """Per-stream frame identity plus capture clock; the shared frame header. + + The flat fields are the serialized contract. ``identity`` and ``clock`` + expose the two components for code that wants to carry them independently. + """ + + sensor_id: str + session_id: str + frame_id: int + source_frame_id: int | None + capture_time_ns: int + clock_id: str + capture_time_basis: TimeBasis + source_pts: int | float | None = None + acquisition_id: str | None = None + sync_group_id: str | None = None + + def __post_init__(self) -> None: + # Validate through the components so every frame header obeys the same + # rules as a separately supplied identity/clock pair. + self.identity + self.clock + + @classmethod + def from_parts(cls, identity: FrameIdentity, clock: FrameClock) -> FrameStamp: + """Build a stamp from its two component types.""" + if not isinstance(identity, FrameIdentity): + raise TypeError("identity must be a FrameIdentity") + if not isinstance(clock, FrameClock): + raise TypeError("clock must be a FrameClock") + return cls( + sensor_id=identity.sensor_id, + session_id=identity.session_id, + frame_id=identity.frame_id, + source_frame_id=identity.source_frame_id, + capture_time_ns=clock.capture_time_ns, + clock_id=clock.clock_id, + capture_time_basis=clock.capture_time_basis, + source_pts=clock.source_pts, + acquisition_id=identity.acquisition_id, + sync_group_id=identity.sync_group_id, + ) + + @property + def identity(self) -> FrameIdentity: + return FrameIdentity( + sensor_id=self.sensor_id, + session_id=self.session_id, + frame_id=self.frame_id, + source_frame_id=self.source_frame_id, + acquisition_id=self.acquisition_id, + sync_group_id=self.sync_group_id, + ) + + @property + def clock(self) -> FrameClock: + return FrameClock( + capture_time_ns=self.capture_time_ns, + clock_id=self.clock_id, + capture_time_basis=self.capture_time_basis, + source_pts=self.source_pts, + ) + + def to_dict(self) -> dict[str, Any]: + return { + "schema": STREAM_CONTRACT_VERSION, + **self.identity.to_dict(), + **self.clock.to_dict(), + } + + @classmethod + def from_dict(cls, data: Mapping) -> FrameStamp: + allowed = frozenset( + { + "schema", + "sensor_id", + "session_id", + "frame_id", + "source_frame_id", + "capture_time_ns", + "clock_id", + "capture_time_basis", + "source_pts", + "acquisition_id", + "sync_group_id", + } + ) + required = frozenset( + { + "schema", + "sensor_id", + "session_id", + "frame_id", + "source_frame_id", + "capture_time_ns", + "clock_id", + "capture_time_basis", + } + ) + fields = _mapping(data, "FrameStamp", allowed, required) + _require_schema(fields["schema"], "FrameStamp") + return cls( + sensor_id=fields["sensor_id"], + session_id=fields["session_id"], + frame_id=fields["frame_id"], + source_frame_id=fields["source_frame_id"], + capture_time_ns=fields["capture_time_ns"], + clock_id=fields["clock_id"], + capture_time_basis=fields["capture_time_basis"], + source_pts=fields.get("source_pts"), + acquisition_id=fields.get("acquisition_id"), + sync_group_id=fields.get("sync_group_id"), + ) + + +@dataclass(frozen=True) +class MeasurementFrame: + """Minimal measurement header handed from the image block to force session. + + ``reference_generation`` is the optical-flow reference version that was in + effect when this frame's flow was computed (the image thread writes it; it + is the ``FrameSnapshot.tracker_reset_seq``/``request_reset`` target concept + from stage 0). ``source_dt_s`` is the source-time distance to the previous + *actually processed* frame of this stream, or ``None`` for the first one. + A long interval is data for a quality marker and a warning; it never resets + a reference, filter or zero by itself. + + The measurement payload (flow/HHD/field/contact) and the force result are + defined by T05/T06 and the result envelope; this header is the part frozen + now. + """ + + stamp: FrameStamp + source_dt_s: float | None + reference_generation: int = 0 + + def __post_init__(self) -> None: + if not isinstance(self.stamp, FrameStamp): + raise TypeError("stamp must be a FrameStamp") + _require_source_dt(self.source_dt_s) + _require_int(self.reference_generation, "reference_generation", minimum=0) + + def to_dict(self) -> dict[str, Any]: + return { + "schema": STREAM_CONTRACT_VERSION, + "stamp": self.stamp.to_dict(), + "source_dt_s": None + if self.source_dt_s is None + else float(self.source_dt_s), + "reference_generation": self.reference_generation, + } + + @classmethod + def from_dict(cls, data: Mapping) -> MeasurementFrame: + allowed = frozenset({"schema", "stamp", "source_dt_s", "reference_generation"}) + required = frozenset({"schema", "stamp", "source_dt_s"}) + fields = _mapping(data, "MeasurementFrame", allowed, required) + _require_schema(fields["schema"], "MeasurementFrame") + return cls( + stamp=FrameStamp.from_dict(fields["stamp"]), + source_dt_s=fields["source_dt_s"], + reference_generation=fields.get("reference_generation", 0), + ) + + +@dataclass(frozen=True) +class StreamEvent: + """One serializable event of a stream's lifecycle. + + ``kind`` is a stable lower_snake identifier matching + ``^[a-z][a-z0-9_]*$`` (for example ``reference_reset``, + ``zero_requested``, ``source_gap``); consumers must ignore kinds they do not + know instead of failing, so new kinds are additive. ``stamp`` is the frame + the event belongs to when there is one. ``payload`` must be JSON-safe and is + frozen/validated on construction; the frozen payload is excluded from the + hash so equal events hash equally while ``hash(event)`` stays usable. + ``event_version`` versions the envelope itself (``STREAM_CONTRACT_VERSION`` + versions the field contract). + """ + + kind: str + stamp: FrameStamp | None = None + payload: Mapping[str, Any] = field(default_factory=dict, hash=False) + event_version: int = 1 + + def __post_init__(self) -> None: + _require_str(self.kind, "kind") + if _EVENT_KIND_RE.fullmatch(self.kind) is None: + raise ValueError(f"kind must match ^[a-z][a-z0-9_]*$, got {self.kind!r}") + if self.stamp is not None and not isinstance(self.stamp, FrameStamp): + raise TypeError("stamp must be a FrameStamp or None") + _require_int(self.event_version, "event_version", minimum=1) + if not isinstance(self.payload, Mapping): + raise TypeError("payload must be a mapping") + object.__setattr__(self, "payload", _freeze_json(self.payload, "payload")) + + def to_dict(self) -> dict[str, Any]: + return { + "schema": STREAM_CONTRACT_VERSION, + "kind": self.kind, + "event_version": self.event_version, + "stamp": None if self.stamp is None else self.stamp.to_dict(), + "payload": _thaw_json(self.payload), + } + + @classmethod + def from_dict(cls, data: Mapping) -> StreamEvent: + allowed = frozenset({"schema", "kind", "event_version", "stamp", "payload"}) + required = frozenset({"schema", "kind"}) + fields = _mapping(data, "StreamEvent", allowed, required) + _require_schema(fields["schema"], "StreamEvent") + stamp = fields.get("stamp") + return cls( + kind=fields["kind"], + stamp=None if stamp is None else FrameStamp.from_dict(stamp), + payload=fields.get("payload") or {}, + event_version=fields.get("event_version", 1), + ) diff --git a/tests/test_stream_contract.py b/tests/test_stream_contract.py new file mode 100644 index 0000000..4a6b0f2 --- /dev/null +++ b/tests/test_stream_contract.py @@ -0,0 +1,441 @@ +"""The frozen single-stream type contract: identity, clock, source_dt_s, events. + +The contract is defined in ``orisys.stream_contract`` and documented in +``docs/stream-contract.zh.md``. These tests pin the four acceptance rules of +T03: per-stream identity and clock, an optional-but-never-partial group +identity, distinguishability of frames that share a timestamp, and JSON +round-trips for ``source_dt_s`` and events. They are pure type tests: nothing +here starts a sensor, a pipeline or a session. +""" + +from __future__ import annotations + +import json + +import pytest + +from orisys import ( + STREAM_CONTRACT_VERSION, + SUPPORTED_CONTRACT_VERSIONS, + FrameClock, + FrameIdentity, + FrameStamp, + MeasurementFrame, + StreamEvent, +) + + +def _round_trip(value): + """JSON round-trip exactly as a recording would read the contract back.""" + payload = json.loads(json.dumps(value.to_dict())) + return type(value).from_dict(payload) + + +def _stamp(**overrides) -> FrameStamp: + fields = { + "sensor_id": "FINGER-001", + "session_id": "run-7", + "frame_id": 12, + "source_frame_id": 41, + "capture_time_ns": 1_600_000_000_000_000_000, + "clock_id": "host_monotonic", + "capture_time_basis": "capture_monotonic", + } + fields.update(overrides) + return FrameStamp(**fields) + + +def test_the_contract_version_is_exported_and_explicit() -> None: + assert STREAM_CONTRACT_VERSION == "stream-contract/1" + # Every frozen type and the version are importable from the package root, + # which is the public boundary downstream repositories import. + import orisys + + for name in ( + "FrameIdentity", + "FrameClock", + "FrameStamp", + "MeasurementFrame", + "StreamEvent", + "TIME_BASES", + "STREAM_CONTRACT_VERSION", + ): + assert hasattr(orisys, name), name + + +# --------------------------------------------------------------- identity +def test_identity_keeps_a_per_stream_output_number_and_the_source_number() -> None: + identity = FrameIdentity( + sensor_id="BOX-3", + session_id="run-7", + frame_id=12, + source_frame_id=41, + ) + assert (identity.sensor_id, identity.session_id) == ("BOX-3", "run-7") + assert identity.frame_id == 12 + assert identity.source_frame_id == 41 + # A source without its own numbering may say so; the output number still exists. + identity = FrameIdentity( + sensor_id="BOX-3", session_id="run-7", frame_id=0, source_frame_id=None + ) + assert identity.frame_id == 0 and identity.source_frame_id is None + with pytest.raises(ValueError): + FrameIdentity( + sensor_id="", session_id="run-7", frame_id=0, source_frame_id=None + ) + with pytest.raises(ValueError): + FrameIdentity( + sensor_id="BOX-3", session_id="run-7", frame_id=-1, source_frame_id=None + ) + with pytest.raises(TypeError): + FrameIdentity( + sensor_id="BOX-3", session_id="run-7", frame_id=True, source_frame_id=None + ) + + +def test_group_identity_is_optional_but_never_partial() -> None: + without_group = FrameIdentity( + sensor_id="BOX-3", session_id="run-7", frame_id=0, source_frame_id=0 + ) + assert without_group.has_group is False + assert without_group.group_identity is None + assert without_group.to_dict()["acquisition_id"] is None + assert without_group.to_dict()["sync_group_id"] is None + + grouped = FrameIdentity( + sensor_id="BOX-3", + session_id="run-7", + frame_id=0, + source_frame_id=0, + acquisition_id="acq-1", + sync_group_id="group-1", + ) + assert grouped.has_group is True + assert grouped.group_identity == ("acq-1", "group-1") + assert _round_trip(grouped) == grouped + + # A half group would look like synchronization without an owner. + with pytest.raises(ValueError, match="both acquisition_id and sync_group_id"): + FrameIdentity( + sensor_id="BOX-3", + session_id="run-7", + frame_id=0, + source_frame_id=0, + acquisition_id="acq-1", + ) + with pytest.raises(ValueError, match="both acquisition_id and sync_group_id"): + FrameIdentity( + sensor_id="BOX-3", + session_id="run-7", + frame_id=0, + source_frame_id=0, + sync_group_id="group-1", + ) + with pytest.raises(ValueError): + FrameIdentity( + sensor_id="BOX-3", + session_id="run-7", + frame_id=0, + source_frame_id=0, + acquisition_id="", + sync_group_id="group-1", + ) + + +# ------------------------------------------------------------------ clock +def test_clock_keeps_the_time_domain_basis_and_the_raw_source_pts() -> None: + clock = FrameClock( + capture_time_ns=1_600_000_000_000_000_000, + clock_id="video:recording-4", + capture_time_basis="video_pts", + source_pts=90000, + ) + assert clock.capture_time_basis == "video_pts" + assert clock.source_pts == 90000 + # A derived file timestamp is explicitly labelled as estimated. + assert FrameClock( + 0, "source:frame-numbers", "estimated_source" + ).capture_time_basis == ("estimated_source") + with pytest.raises(ValueError): + FrameClock(0, "host", "wall_clock") + with pytest.raises(ValueError): + FrameClock(-1, "host", "capture_monotonic") + with pytest.raises(ValueError): + FrameClock(0, "", "capture_monotonic") + with pytest.raises(ValueError): + FrameClock(0, "host", "capture_monotonic", source_pts=float("nan")) + with pytest.raises(TypeError): + FrameClock(0, "host", "capture_monotonic", source_pts=True) + + +# ------------------------------------------------------------------ stamp +def test_stamp_composes_identity_and_clock_and_round_trips() -> None: + identity = FrameIdentity( + sensor_id="FINGER-001", + session_id="run-7", + frame_id=12, + source_frame_id=41, + acquisition_id="acq-1", + sync_group_id="group-1", + ) + clock = FrameClock( + capture_time_ns=1_600_000_000_000_000_000, + clock_id="host_monotonic", + capture_time_basis="capture_monotonic", + source_pts=41.5, + ) + stamp = FrameStamp.from_parts(identity, clock) + assert stamp.identity == identity + assert stamp.clock == clock + + payload = stamp.to_dict() + assert payload["schema"] == STREAM_CONTRACT_VERSION + assert payload["sensor_id"] == "FINGER-001" + assert payload["source_frame_id"] == 41 + assert payload["capture_time_ns"] == clock.capture_time_ns + assert _round_trip(stamp) == stamp + + # The flattened fields are the same objects the components report. + assert stamp.sensor_id == identity.sensor_id + assert stamp.session_id == identity.session_id + assert stamp.frame_id == identity.frame_id + assert stamp.source_frame_id == identity.source_frame_id + assert stamp.capture_time_ns == clock.capture_time_ns + assert stamp.clock_id == clock.clock_id + assert stamp.capture_time_basis == clock.capture_time_basis + assert stamp.source_pts == clock.source_pts + assert (stamp.acquisition_id, stamp.sync_group_id) == ("acq-1", "group-1") + + +def test_frames_that_share_a_pts_are_still_distinct_frames() -> None: + """Timestamps are not identity: a recording must keep both frames.""" + first = _stamp(source_frame_id=41, frame_id=10) + second = _stamp(source_frame_id=42, frame_id=11) + assert first.capture_time_ns == second.capture_time_ns + assert first.source_pts == second.source_pts + assert first != second + assert first.to_dict() != second.to_dict() + assert _round_trip(first) != _round_trip(second) + + # A source without its own numbering can still emit repeated timestamps; + # the contiguous output number keeps the frames apart. + first = _stamp(source_frame_id=None, frame_id=10) + second = _stamp(source_frame_id=None, frame_id=11) + assert first != second + assert (first.capture_time_ns, first.source_pts) == ( + second.capture_time_ns, + second.source_pts, + ) + + # Different sessions are different streams even with identical numbers. + other_session = _stamp(session_id="run-8") + assert other_session != first and other_session != second + + +# -------------------------------------------------------- measurement frame +def test_measurement_frame_carries_source_dt_s_and_round_trips() -> None: + stamp = _stamp() + first = MeasurementFrame(stamp=stamp, source_dt_s=None) + assert first.source_dt_s is None + assert first.reference_generation == 0 + assert _round_trip(first) == first + + following = MeasurementFrame( + stamp=_stamp(frame_id=13, source_frame_id=42), + source_dt_s=0.033, + reference_generation=3, + ) + payload = following.to_dict() + assert payload["schema"] == STREAM_CONTRACT_VERSION + assert payload["source_dt_s"] == pytest.approx(0.033) + assert payload["reference_generation"] == 3 + assert _round_trip(following) == following + + # Equal source timestamps are valid data; they do not collapse frames and do + # not produce a negative/None interval. + same_time = MeasurementFrame(stamp=_stamp(), source_dt_s=0.0) + assert same_time.source_dt_s == 0.0 + + +def test_measurement_frame_rejects_invalid_source_dt_and_generation() -> None: + stamp = _stamp() + with pytest.raises(ValueError): + MeasurementFrame(stamp=stamp, source_dt_s=-0.001) + with pytest.raises(ValueError): + MeasurementFrame(stamp=stamp, source_dt_s=float("nan")) + with pytest.raises(ValueError): + MeasurementFrame(stamp=stamp, source_dt_s=float("inf")) + with pytest.raises(TypeError): + MeasurementFrame(stamp=stamp, source_dt_s=True) + with pytest.raises(ValueError): + MeasurementFrame(stamp=stamp, source_dt_s=None, reference_generation=-1) + with pytest.raises(TypeError): + MeasurementFrame(stamp=stamp, source_dt_s=None, reference_generation=1.5) + with pytest.raises(TypeError): + MeasurementFrame(stamp={}, source_dt_s=None) + + +# ------------------------------------------------------------------ events +def test_events_round_trip_through_json_with_a_frozen_payload() -> None: + stamp = _stamp() + event = StreamEvent( + kind="source_gap", + stamp=stamp, + payload={ + "dropped_frames": 3, + "source_dt_s": 0.5, + "proportion": 0.25, + "flags": [True, False, None], + "context": {"reason": "overload", "counts": [1, 2]}, + }, + event_version=1, + ) + assert event.payload["dropped_frames"] == 3 + # The payload is read-only after construction. + with pytest.raises(TypeError): + event.payload["dropped_frames"] = 4 # type: ignore[index] + + payload = event.to_dict() + assert payload["schema"] == STREAM_CONTRACT_VERSION + assert payload["event_version"] == 1 + assert _round_trip(event) == event + + # A lifecycle event may have no frame, and unknown kinds stay additive. + event = StreamEvent(kind="stream_started", payload={}) + assert event.stamp is None + assert _round_trip(event) == event + assert _round_trip(StreamEvent(kind="some_future_kind", payload={"x": 1})).kind == ( + "some_future_kind" + ) + + +def test_events_reject_payloads_that_cannot_round_trip() -> None: + with pytest.raises(ValueError): + StreamEvent(kind="bad", payload={"nan": float("nan")}) + with pytest.raises(ValueError): + StreamEvent(kind="bad", payload={"inf": float("inf")}) + with pytest.raises(TypeError): + StreamEvent(kind="bad", payload={1: "non-string key"}) + with pytest.raises(TypeError): + StreamEvent(kind="bad", payload={"bytes": b"raw"}) + with pytest.raises(TypeError): + StreamEvent(kind="bad", payload={"set": {1, 2}}) + with pytest.raises(ValueError): + StreamEvent(kind="", payload={}) + with pytest.raises(ValueError): + StreamEvent(kind="bad", payload={}, event_version=0) + + +# -------------------------------------------------------------- read-back +@pytest.mark.parametrize( + "factory", + [ + lambda: _stamp(), + lambda: _stamp(acquisition_id="acq-1", sync_group_id="group-1"), + lambda: MeasurementFrame(_stamp(), None), + lambda: MeasurementFrame(_stamp(), 0.033, 2), + lambda: StreamEvent("reference_reset", _stamp(), {"from": 1, "to": 2}), + ], +) +def test_every_frozen_type_round_trips_through_its_dictionary(factory) -> None: + value = factory() + assert _round_trip(value) == value + + +def test_unknown_fields_and_wrong_schema_are_rejected() -> None: + stamp_payload = _stamp().to_dict() + # Policy A: no same-version field additions - an unknown field names the + # versioning rule instead of being silently dropped. + with pytest.raises(ValueError, match="new STREAM_CONTRACT_VERSION"): + FrameStamp.from_dict({**stamp_payload, "unexpected": 1}) + with pytest.raises(ValueError, match="schema"): + FrameStamp.from_dict({**stamp_payload, "schema": "stream-contract/2"}) + with pytest.raises(ValueError, match="missing field"): + FrameStamp.from_dict( + {key: value for key, value in stamp_payload.items() if key != "sensor_id"} + ) + with pytest.raises(ValueError, match="unknown field"): + StreamEvent.from_dict( + {"schema": STREAM_CONTRACT_VERSION, "kind": "x", "extra": 1} + ) + with pytest.raises(ValueError, match="unknown field"): + # A nested frame header follows the same policy. + MeasurementFrame.from_dict( + { + "schema": STREAM_CONTRACT_VERSION, + "stamp": {**stamp_payload, "unexpected": 1}, + "source_dt_s": None, + } + ) + with pytest.raises(ValueError, match="schema"): + MeasurementFrame.from_dict( + { + "schema": "stream-contract/0", + "stamp": stamp_payload, + "source_dt_s": None, + } + ) + + +def test_the_reader_lists_the_versions_it_accepts() -> None: + """Policy A: strict reading, and only explicitly listed versions are read.""" + assert SUPPORTED_CONTRACT_VERSIONS == (STREAM_CONTRACT_VERSION,) + payload = _stamp().to_dict() + # The listed version is the one this reader was built for and it reads. + assert FrameStamp.from_dict(payload) == _stamp() + + # A different version is a different field contract: it is rejected rather + # than guessed at. Bumping STREAM_CONTRACT_VERSION and adding the new value + # to SUPPORTED_CONTRACT_VERSIONS is what makes a new version readable. + for schema in ("stream-contract/0", "stream-contract/2", "other/1"): + with pytest.raises(ValueError, match="must be one of"): + FrameStamp.from_dict({**payload, "schema": schema}) + with pytest.raises(ValueError, match="must be one of"): + MeasurementFrame.from_dict( + {"schema": schema, "stamp": payload, "source_dt_s": None} + ) + with pytest.raises(ValueError, match="must be one of"): + StreamEvent.from_dict({"schema": schema, "kind": "x"}) + + +# --------------------------------------------------------------- event extras +def test_events_are_hashable_and_equal_events_share_a_hash() -> None: + """The frozen event is usable as a dict/set key; equality and hash agree.""" + first = StreamEvent( + kind="reference_reset", stamp=_stamp(), payload={"from": 1, "to": 2} + ) + second = StreamEvent( + kind="reference_reset", + stamp=_stamp(), + payload={"to": 2, "from": 1}, + ) + assert first == second + assert hash(first) == hash(second) + # A JSON read-back is still the same value, so it must hash the same too. + assert _round_trip(first) == first + assert hash(_round_trip(first)) == hash(first) + # The frozen payload is excluded from the hash, so it cannot make hashing + # fail; events differing only in payload may still share a hash bucket. + assert hash(StreamEvent(kind="source_gap", stamp=_stamp())) == hash( + StreamEvent(kind="source_gap", stamp=_stamp(), payload={"dropped": 1}) + ) + + +@pytest.mark.parametrize( + "kind", + ["NotSnake", "bad kind", "1leading", "with-dash", "_leading", ""], +) +def test_event_kind_rejects_names_outside_lower_snake(kind) -> None: + with pytest.raises(ValueError): + StreamEvent(kind=kind) + with pytest.raises(ValueError): + StreamEvent.from_dict({"schema": STREAM_CONTRACT_VERSION, "kind": kind}) + + +@pytest.mark.parametrize( + "kind", ["a", "source_gap", "reference_reset", "zero2", "a_b_c_1"] +) +def test_event_kind_accepts_lower_snake_and_round_trips(kind) -> None: + event = StreamEvent(kind=kind, payload={"x": 1}) + assert event.kind == kind + assert _round_trip(event) == event