feat: 新增单路流类型合同 stream-contract/1

冻结逐帧身份、采集时钟、测量头和事件信封:FrameIdentity、FrameClock、FrameStamp、MeasurementFrame、StreamEvent,并从包根导出。采用严格读取:未知字段和未列出的版本一律拒绝,任何字段变更都升版本,读取方在 SUPPORTED_CONTRACT_VERSIONS 中列出可读版本;未知事件 kind 仍兼容。事件可哈希,kind 为 lower_snake。合同写入 docs/stream-contract.zh.md;不接入现有运行时。窄套件 139 项、lint 通过。
This commit is contained in:
ykcheungab
2026-10-21 11:03:53 +08:00
parent cbccbd14b5
commit f6f3c45285
4 changed files with 1218 additions and 0 deletions
+176
View File
@@ -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 格式和非法负载拒绝。类型本身不启动传感器、流水线或会话。
+23
View File
@@ -19,6 +19,18 @@ from .pipeline import (
StreamPipeline, StreamPipeline,
) )
from .sensor import Sensor 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 #: The explicit force-readout session (``studio-force/1``) used by the Studio
#: workflow. Imported after ``Sensor`` because it composes the sensor, model and #: workflow. Imported after ``Sensor`` because it composes the sensor, model and
#: output blocks; the historical ``Sensor.read_info``/``ForceReadout.apply`` #: output blocks; the historical ``Sensor.read_info``/``ForceReadout.apply``
@@ -70,6 +82,17 @@ __all__ = [
'ForceSession', 'ForceSession',
'WORKFLOW_VERSION', '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 # configs utilities functions
'list_config', 'list_config',
'export_config', 'export_config',
+578
View File
@@ -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),
)
+441
View File
@@ -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