From e3b60f3d6d342c44a88f733af092f9fd673bdc82 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Wed, 21 Oct 2026 13:35:02 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=B0=86=E5=9B=BE=E5=83=8F=E5=9D=97?= =?UTF-8?q?=E6=8B=86=E5=88=86=E4=B8=BA=E5=8F=AF=E7=8B=AC=E7=AB=8B=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E7=9A=84=20ImagePipeline?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 光流、HHD 和场从 Sensor 中抽到 ImageSettings/ImagePipeline，Sensor 与 FingerRuntime 改为委托，公开接口与数值不变（222 个基线数组逐位一致，含运行时 flow_ema_alpha 场景）。参考版本只有一个来源：StreamPipeline 在 flow 线程领取的版本逐帧交给图像块。对外数组为只读副本；输出带 stream-contract 的 MeasurementFrame 头。回归 357 项、lint 通过。 --- docs/stream-contract.zh.md | 47 ++- src/orisys/finger/runtime.py | 12 + src/orisys/image_pipeline.py | 592 +++++++++++++++++++++++++++++++++++ src/orisys/pipeline.py | 9 + src/orisys/sensor.py | 183 +++++++---- tests/test_image_pipeline.py | 521 ++++++++++++++++++++++++++++++ 6 files changed, 1306 insertions(+), 58 deletions(-) create mode 100644 src/orisys/image_pipeline.py create mode 100644 tests/test_image_pipeline.py diff --git a/docs/stream-contract.zh.md b/docs/stream-contract.zh.md index 5fa3f7f..54aa839 100644 --- a/docs/stream-contract.zh.md +++ b/docs/stream-contract.zh.md @@ -182,7 +182,7 @@ assert measurement.stamp == stamp - 每个 Sensor 实例独立绑定，没有模块级共享缓存；创建或使用另一个 Sensor 不会改变本实例的上下文。 - 显式 `Sensor.set_calibration()` 换标定时原子替换上下文：SN/配置/资产绑定保持不变，标定路径与指纹更新；`set_calibration(None)` 后 `calibration_path` 立即为 `None`、指纹为 `"none"`，不保留构造时请求过的旧路径。 - 绑定异常只写入 `diagnostics`，不把原本可构造的 Sensor 变成构造失败。 -- `assets`/`manifest_digest` 是 **manifest 声明的绑定**，不是运行时实际加载的路径：Finger 的导出目录由 `finger/exports.py` 在 cwd、打包目录、可编辑安装目录等候选中另行查找，实际加载路径由 T05 的图像块绑定；需要复现实验时以 T05/记录的身份为准。 +- `assets`/`manifest_digest` 是 **manifest 声明的绑定**，不是运行时实际加载的路径：Finger 的导出目录由 `finger/exports.py` 在 cwd、打包目录、可编辑安装目录等候选中另行查找；T05 的 ImagePipeline 消费已准备帧，不重新选择资产，需要复现实验时以记录的身份为准。 - 资产取应用包内相对路径，不复制、不上传、不按顺序推断；配置指纹取实际加载后的配置内容，而不是文件名。 - 本任务（T04）不改变 Sensor 的数值与既有资产包解析行为，也不把上下文接入流入口；T05/T08 消费该上下文。 @@ -203,14 +203,53 @@ identity: FrameIdentity = context.frame_identity( assert identity.sensor_id == "BOX-001" ``` +## ImagePipeline（T05） + +图像块（ImageSettings → 光流 / HHD / 场）已从 `Sensor` 抽成可独立运行的对象；`Sensor`、`FingerRuntime` 与 `StreamPipeline` 委托同一实现（Finger 的 UV 展开与采集仍留在运行时侧，`runtime.image_pipeline` 暴露该对象）。 + +| 组成 | 说明 | +|---|---| +| `ImageSettings` | 一次图像参数的声明式身份：tracker 配置、HHD 下采样/后端与 masked-Poisson 参数、光流 EMA 初值、cuda；`from_config(sensor.json)` 读取与 Sensor 相同的键。 | +| `ImagePipeline` | 拥有 tracker、nHHD、EMA 与 scratch 状态；`compute_flow()`、`decompose()`、`current_field()`；光流 EMA 系数与历史由传入的 `ForceReadout` 单一持有（独立使用时自建一个）。 | +| `ImageMeasurement` | T03 `MeasurementFrame` 头 + 光流/HHD/场/force 的只读副本；`frame.reference_generation` 由计算该帧的图像线程写入。 | + +规则： + +- 输入是**已准备**的帧（采集层完成 stitch/unroll/resize/mirror）；独立实验直接喂灰度帧即可，不必构造 Sensor：`SensorContext` 给 SN/身份，`ImageSettings` 给图像参数。 +- `include=` 选择要复制的数组（image/flow/vnormal/vshear/field/valid_mask）；返回数组是独立副本且 `writeable=False`，发布后不会被后续帧覆写（K11.3）。 +- 参考版本只有一个来源：驱动它的 `StreamPipeline` 领取版本后经 `set_stream_reference_generation()` 交给图像块（逐帧、计算前）；独立实验把版本直接传给 `measure(reference_generation=...)`。`ImagePipeline` 不自己计数，`reset_image_state()`/`reset_tracker()` 只重置状态。 +- 光流 EMA 的系数与历史属于 `ForceReadout`，因此已发布的 `Sensor.configure(flow_ema_alpha=...)` 从下一帧起生效（与拆分前一致）；运行时修改后 `ImageSettings.flow_ema_alpha` 仍只表示构造时声明。 +- 帧身份复用 T03 `FrameIdentity`/`FrameClock`，`measure()` 构造 `MeasurementFrame`；`Sensor` 的公开方法与数值不变（流入口 T08 才接 MeasurementFrame）。 + +```python +import numpy as np +from orisys import configs +from orisys.image_pipeline import ImagePipeline, ImageSettings, sensor_clock +from orisys.sensors import build_sensor_context + +context = build_sensor_context(config_name="box", config=configs.load_config("box"), + device_id="BOX-001", bundle_candidates=["box"]) +pipeline = ImagePipeline(ImageSettings.from_config(configs.load_config("box")), context=context) +prepared_grey_frame = np.full((365, 365), 120, dtype=np.uint8) +measurement = pipeline.measure( + prepared_grey_frame, + identity=context.frame_identity(session_id="run-1", frame_id=0, source_frame_id=0), + clock=sensor_clock(1_000_000, clock_id="host"), + include=("flow", "field"), +) +assert measurement.frame.reference_generation == 0 +``` + ## 不包含什么 - 不实现 SensorStream 入口、实时最新选帧、完整回放和背压（T08）。 -- 不实现图像块拆分、测量负载、ForceSession 重做、零点与输出滤波（T05–T07）。 +- 不实现测量负载、ForceSession 重做、零点与输出滤波（T06–T07）。 - 不实现批次、配组、多流调度（后续多流计划）。 - 不实现录制/回放格式、Studio 迁移和离线重新滤波（T11、T13–T15）。 -- 不修改现有 `Sensor`、`StreamPipeline` 和当前 `ForceSession`；现有 `ForceSession.FrameStamp` 在 T06 重做时被本合同的帧头替换。 +- T05 只把 flow/HHD/场移到 `ImagePipeline`，不改变 `Sensor` 的公开接口与数值；当前 `ForceSession` 在 T06 重做时改用本合同的帧头。 ## 验证 -`tests/test_stream_contract.py` 覆盖 T03 的四项验收：逐路身份与时钟、组字段可缺但不半组、同 PTS 帧可区分、`source_dt_s` 与事件 JSON 往返；同时覆盖 frozen 校验、严格版本/未知字段拒绝（政策 A）、`SUPPORTED_CONTRACT_VERSIONS`、事件可哈希、`kind` 的 lower_snake 格式和非法负载拒绝。`tests/test_sensor_context.py` 覆盖 T04 的只读 SensorContext：显式 SN/配置/资产/标定路径绑定、不猜 SN、实例隔离、换标定原子替换、绑定失败只报 diagnostics，以及 config source/资产包绑定。类型与上下文本身不启动传感器、流水线或会话。 +`tests/test_stream_contract.py` 覆盖 T03 的四项验收：逐路身份与时钟、组字段可缺但不半组、同 PTS 帧可区分、`source_dt_s` 与事件 JSON 往返；同时覆盖 frozen 校验、严格版本/未知字段拒绝（政策 A）、`SUPPORTED_CONTRACT_VERSIONS`、事件可哈希、`kind` 的 lower_snake 格式和非法负载拒绝。`tests/test_sensor_context.py` 覆盖 T04 的只读 SensorContext：显式 SN/配置/资产/标定路径绑定、不猜 SN、实例隔离、换标定原子替换、绑定失败只报 diagnostics，以及 config source/资产包绑定。 + +`tests/test_image_pipeline.py` 覆盖 T05：独立流水线与 Sensor 委托路径逐帧逐位一致、只读副本不被后续帧覆写、`include` 选择、`Sensor.configure(flow_ema_alpha=...)` 的运行时生效与 EMA 单属主、由 StreamPipeline 驱动时图像块版本逐帧等于快照 `tracker_reset_seq`（含复用 Sensor、直接 `reset()` 后），以及独立 `measure(reference_generation=...)` 的在途版本保留。 diff --git a/src/orisys/finger/runtime.py b/src/orisys/finger/runtime.py index 3949999..e637aa7 100644 --- a/src/orisys/finger/runtime.py +++ b/src/orisys/finger/runtime.py @@ -546,6 +546,18 @@ class FingerRuntime: return StreamPipeline.for_finger_runtime(self, **kwargs) + @property + def image_pipeline(self): + """The ImagePipeline the tactile sensor delegates its image block to. + + Capture and UV unrolling stay in this runtime; flow/HHD/field come from + the tactile sensor's single ``ImagePipeline``, so an offline experiment + can inspect or reuse the same ``ImageSettings``/``SensorContext`` without + a running stream. + """ + raw = getattr(self.tactile, "raw_sensor", None) + return None if raw is None else getattr(raw, "image_pipeline", None) + def close(self): if self.capture_source is not None: self.capture_source.close() diff --git a/src/orisys/image_pipeline.py b/src/orisys/image_pipeline.py new file mode 100644 index 0000000..0362cfd --- /dev/null +++ b/src/orisys/image_pipeline.py @@ -0,0 +1,592 @@ +"""Image block: ImageSettings and an independently runnable ImagePipeline. + +Stage-1 architecture (K8.1, K11.1, K11.3, task T05) separates the image side of +measurement from :class:`~orisys.sensor.Sensor`: + +* :class:`ImageSettings` is the immutable description of one image + configuration (optical-flow tracker settings, HHD grid/backend, flow EMA). +* :class:`ImagePipeline` owns the tracker, the HHD solver and their scratch + state. It turns a prepared grey frame into flow/HHD/field results, writes the + optical-flow reference generation at the moment it computes a frame, and + publishes measurement arrays as read-only copies. + +``Sensor`` delegates to an ``ImagePipeline``; ``FingerRuntime`` reaches the same +object through its tactile sensor, so all three published paths keep one image +implementation. The class can also be used without a Sensor at all:: + + from orisys import configs + from orisys.image_pipeline import ImagePipeline, ImageSettings, sensor_clock + + context = sensor.context # or build_sensor_context(...) + settings = ImageSettings.from_config(configs.load_config("box")) + pipeline = ImagePipeline(settings, context=context) + measurement = pipeline.measure( + prepared_grey_frame, + identity=context.frame_identity(session_id="run-1", frame_id=0, source_frame_id=0), + clock=sensor_clock(capture_time_ns, clock_id="host", basis="capture_monotonic"), + ) + print(measurement.force_raw, measurement.flow.shape, measurement.frame.reference_generation) + +Image preparation (stitch/unroll/resize/mirror) remains in the acquisition layer +for now; the pipeline starts at the prepared grey frame, which is exactly what +``Sensor.compute_flow``/``compute_hhd`` have always consumed. +""" + +from __future__ import annotations + +import time +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from dataclasses import field as dataclass_field +from typing import Any + +import cv2 +import numpy as np + +from .hhd import nHHD +from .stream_contract import ( + FrameClock, + FrameIdentity, + FrameStamp, + MeasurementFrame, + TimeBasis, +) +from .tracker import Tracker + +__all__ = [ + "ImageDecomposition", + "ImageMeasurement", + "ImagePipeline", + "ImageSettings", + "sensor_clock", +] + +#: Arrays an independent measurement may request with ``include``. +_INCLUDE_FIELDS = frozenset( + {"image", "flow", "vnormal", "vshear", "field", "valid_mask"} +) + + +def sensor_clock( + capture_time_ns: int, + *, + clock_id: str, + basis: TimeBasis = "capture_monotonic", +) -> FrameClock: + """Convenience constructor for the capture clock of a prepared frame. + + ``basis`` is one of :data:`~orisys.stream_contract.TIME_BASES`; the value is + validated by :class:`~orisys.stream_contract.FrameClock`. + """ + return FrameClock( + capture_time_ns=int(capture_time_ns), + clock_id=str(clock_id), + capture_time_basis=basis, + ) + + +@dataclass(frozen=True) +class ImageSettings: + """One image configuration: tracker, HHD and flow-EMA settings. + + ``from_config`` reads the same config keys ``Sensor`` has always read. The + flow EMA alpha is resolved through :class:`~orisys.force_readout.ForceReadout` + so the existing filter-preset interpretation has exactly one implementation. + """ + + tracker_config: Mapping[str, Any] + cuda: bool + verbose: bool + hhd_downsample: int + poisson_backend: str + masked_poisson_solver: str + masked_poisson_max_boundary: int + masked_poisson_boundary_stride: int + masked_poisson_max_iters: int + masked_poisson_tol: float + masked_poisson_reg: float + flow_ema_alpha: float = 1.0 + + @classmethod + def from_config( + cls, + config: Mapping[str, Any] | None, + *, + cuda: bool = True, + backend: str = "auto", + verbose: bool = False, + flow_ema_alpha: float | None = None, + ) -> ImageSettings: + """Build settings from ``sensor.json``-shaped configuration.""" + cfg: Mapping[str, Any] = config if isinstance(config, Mapping) else {} + tracker = dict(cfg.get("tracker") or {}) + raw_ds = tracker.get("hhd_downsample", 1) + try: + hhd_downsample = max(1, int(raw_ds)) if raw_ds is not None else 1 + except (TypeError, ValueError): + hhd_downsample = 1 + if flow_ema_alpha is None: + # Reuse the existing filter-preset resolution instead of a second + # interpretation of the filters block (that block owns the EMA). + from .force_readout import ForceReadout + + flow_ema_alpha = ForceReadout(cfg.get("filters")).flow_ema_alpha + backend_from_config = tracker.get("backend", None) + poisson_backend = str( + backend_from_config if backend_from_config is not None else backend + ) + return cls( + tracker_config=tracker, + cuda=bool(cuda), + verbose=bool(verbose), + hhd_downsample=hhd_downsample, + poisson_backend=poisson_backend, + masked_poisson_solver=str(tracker.get("masked_poisson_solver", "sparse")), + masked_poisson_max_boundary=int( + tracker.get("masked_poisson_max_boundary", 96) + ), + masked_poisson_boundary_stride=int( + tracker.get("masked_poisson_boundary_stride", 1) + ), + masked_poisson_max_iters=int(tracker.get("masked_poisson_max_iters", 3)), + masked_poisson_tol=float(tracker.get("masked_poisson_tol", 1e-4)), + masked_poisson_reg=float(tracker.get("masked_poisson_reg", 1e-8)), + flow_ema_alpha=float(flow_ema_alpha), + ) + + def as_dict(self) -> dict[str, Any]: + """Stable identity of the image settings, for records and comparisons.""" + return { + "tracker": dict(self.tracker_config), + "cuda": self.cuda, + "hhd_downsample": self.hhd_downsample, + "poisson_backend": self.poisson_backend, + "masked_poisson_solver": self.masked_poisson_solver, + "masked_poisson_max_boundary": self.masked_poisson_max_boundary, + "masked_poisson_boundary_stride": self.masked_poisson_boundary_stride, + "masked_poisson_max_iters": self.masked_poisson_max_iters, + "masked_poisson_tol": self.masked_poisson_tol, + "masked_poisson_reg": self.masked_poisson_reg, + "flow_ema_alpha": self.flow_ema_alpha, + } + + +@dataclass(frozen=True) +class ImageDecomposition: + """Flow/HHD result for one prepared frame (internal state objects, not copies).""" + + flow: np.ndarray + vnormal: np.ndarray + vshear: np.ndarray + force_raw: tuple[float, float, float] + torque: float + valid_mask_hhd: np.ndarray | None + + +def _readonly_copy(value: Any) -> Any: + """Copy arrays (recursively for the field payload) and freeze them.""" + if isinstance(value, np.ndarray): + out = np.array(value, copy=True) + out.setflags(write=False) + return out + if isinstance(value, dict): + return {key: _readonly_copy(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return type(value)(_readonly_copy(item) for item in value) + return value + + +@dataclass(frozen=True) +class ImageMeasurement: + """One image measurement: T03 header plus read-only flow/HHD/field payload.""" + + frame: MeasurementFrame + force_raw: tuple[float, float, float] + torque: float + flow: np.ndarray | None = None + vnormal: np.ndarray | None = None + vshear: np.ndarray | None = None + valid_mask: np.ndarray | None = None + field: dict | None = None + image: np.ndarray | None = None + elapsed_s: float = 0.0 + diagnostics: tuple[tuple[str, str | float], ...] = dataclass_field( + default_factory=tuple + ) + + +class ImagePipeline: + """Own the image half of measurement: tracker, HHD, field and reference version. + + Thread ownership mirrors stage 0: one instance is used by one image thread. + The reference version is **not counted here**: the stream that owns it (the + stage-0 ``StreamPipeline`` reset protocol) adopts it through + :meth:`set_reference_generation` or passes it to :meth:`measure`; the image + block only stamps what it was told. :meth:`reset_image_state` rebases the + tracker/flow EMA and never invents a version. The force/contact/model half + stays in ``Sensor``. + """ + + def __init__( + self, + settings: ImageSettings, + *, + context: Any = None, + verbose: bool = False, + flow_ema: Any = None, + ) -> None: + from .force_readout import ForceReadout + + self.settings = settings + self.context = context + self.verbose = bool(verbose) + # One owner for the flow-EMA coefficient *and* its history: the published + # ForceReadout when a Sensor binds this pipeline, otherwise a private one. + # Sensor.configure(flow_ema_alpha=...) therefore keeps working on the next + # frame, exactly as it did before the image block was split out. + self._flow_ema = ( + flow_ema + if flow_ema is not None + else ForceReadout({"flow_ema_alpha": settings.flow_ema_alpha}) + ) + self._tracker = Tracker( + adaptive=True, + cuda=settings.cuda, + verbose=settings.verbose, + config_tracker=dict(settings.tracker_config), + ) + self._hhd: nHHD | None = None + self._hhd_grid_shape: tuple[int, int] | None = None + self._hhd_ref_shape: tuple[int, int] | None = None + self._valid_mask_hhd: np.ndarray | None = None + self._valid_mask: np.ndarray | None = None + self._flow_ds_buf: np.ndarray | None = None + self._vn_buf: np.ndarray | None = None + self._vs_buf: np.ndarray | None = None + self._mask_f32: np.ndarray | None = None + self._flow: np.ndarray | None = None + self._vnormal: np.ndarray | None = None + self._vshear: np.ndarray | None = None + self._force_raw: tuple[float, float, float] = (0.0, 0.0, 0.0) + self._torque = 0.0 + self._reference_generation = 0 + + # ------------------------------------------------------------- read access + @property + def flow(self) -> np.ndarray | None: + return self._flow + + @property + def vnormal(self) -> np.ndarray | None: + return self._vnormal + + @property + def vshear(self) -> np.ndarray | None: + return self._vshear + + @property + def valid_mask_hhd(self) -> np.ndarray | None: + return self._valid_mask_hhd + + @property + def force_raw(self) -> tuple[float, float, float]: + return self._force_raw + + @property + def torque(self) -> float: + return self._torque + + @property + def reference_generation(self) -> int: + """Reference version most recently adopted from the owning stream.""" + return int(self._reference_generation) + + def set_reference_generation(self, generation: int) -> None: + """Adopt the stream's reference version (called on the image thread). + + The stage-0 ``StreamPipeline`` claims the version and passes it here + before computing the frame's flow; ``measure`` may stamp the same value + explicitly for an independent experiment. This class never increments it. + """ + if isinstance(generation, bool) or not isinstance(generation, int): + raise TypeError("reference_generation must be an integer") + if generation < 0: + raise ValueError("reference_generation must be >= 0") + self._reference_generation = int(generation) + + @property + def flow_ema_alpha(self) -> float: + """Runtime flow-EMA coefficient (owned by the readout, not this pipeline).""" + return float(self._flow_ema.flow_ema_alpha) + + # ----------------------------------------------------------- reset seams + def reset_image_state(self) -> None: + """Rebase the tracker and drop the flow-EMA history (image thread only). + + A reference version is not claimed here: the owning stream sets it with + :meth:`set_reference_generation`; this method only resets image state. + """ + self._tracker.reset() + self._flow_ema.reset_flow_ema() + + def reset_tracker(self) -> None: + """Drop tracker/EMA history without touching the reference identity. + + This is the frame-boundary zeroing half used by a force session before a + baseline window: the reference frame itself is unchanged. + """ + self._tracker.reset() + self._flow_ema.reset_flow_ema() + + # ------------------------------------------------------------- flow / HHD + def _apply_flow_ema(self, flow: np.ndarray | None) -> np.ndarray | None: + """Temporal flow smoothing, delegated to its single owner (ForceReadout).""" + return self._flow_ema.apply_flow_ema(flow) + + def _compute_flow_state( + self, image: np.ndarray, *, valid_mask: np.ndarray | None = None + ) -> np.ndarray: + """Flow for one frame (the caller has already adopted the version).""" + flow = self._tracker.track(image, valid_mask=valid_mask) + if hasattr(flow, "download"): + flow = flow.download() + flow = self._apply_flow_ema(flow) + self._flow = flow + return flow + + def compute_flow( + self, + image: np.ndarray, + *, + valid_mask: np.ndarray | None = None, + reference_generation: int | None = None, + ) -> np.ndarray: + """Compute and return this frame's flow (one call per frame).""" + if image is None: + raise Exception("Warning: Get image first") + if reference_generation is not None: + self.set_reference_generation(reference_generation) + return self._compute_flow_state(image, valid_mask=valid_mask) + + def _resize_mask( + self, mask: np.ndarray | None, size: tuple[int, int] + ) -> np.ndarray | None: + if mask is None: + return None + mask_u8 = np.asarray(mask, dtype=np.uint8) + resized = cv2.resize(mask_u8, size, interpolation=cv2.INTER_NEAREST) + return resized.astype(bool) + + def _hhd_grid_size(self) -> tuple[int, int] | None: + mask = self._valid_mask + if mask is None: + return None + h, w = mask.shape[:2] + k = max(1, int(self.settings.hhd_downsample)) + if k <= 1: + return (w, h) + return (max(1, w // k), max(1, h // k)) + + def decompose( + self, flow: np.ndarray | None = None, *, valid_mask: np.ndarray | None = None + ) -> ImageDecomposition: + """Run HHD on one flow field and publish normal/shear/force results.""" + flow = self._flow if flow is None else flow + if flow is None: + raise Exception("Warning: Compute flow first") + self._valid_mask = valid_mask + + orig_h, orig_w = flow.shape[:2] + try: + k = max(1, int(self.settings.hhd_downsample)) + except (TypeError, ValueError): + k = 1 + + mask_ds = None + if valid_mask is not None: + ds_size = self._hhd_grid_size() + if ( + self._valid_mask_hhd is not None + and ds_size is not None + and self._valid_mask_hhd.shape[1] == ds_size[0] + and self._valid_mask_hhd.shape[0] == ds_size[1] + ): + mask_ds = self._valid_mask_hhd + else: + mask_ds = np.asarray(valid_mask, dtype=bool) + + if k > 1: + ds_w = max(1, orig_w // k) + ds_h = max(1, orig_h // k) + flow_ds = self._flow_ds_buf + if ( + flow_ds is None + or flow_ds.shape != (ds_h, ds_w, 2) + or flow_ds.dtype != np.float32 + ): + flow_ds = np.empty((ds_h, ds_w, 2), dtype=np.float32) + self._flow_ds_buf = flow_ds + cv2.resize(flow, (ds_w, ds_h), dst=flow_ds, interpolation=cv2.INTER_AREA) + if mask_ds is not None and ( + mask_ds.shape[1] != ds_w or mask_ds.shape[0] != ds_h + ): + mask_ds = self._resize_mask(mask_ds, (ds_w, ds_h)) + else: + flow_ds = flow + + ds_shape = flow_ds.shape[:2] + ref_shape = (orig_h, orig_w) + self._valid_mask_hhd = mask_ds + if ( + self._hhd is None + or self._hhd_grid_shape != ds_shape + or self._hhd_ref_shape != ref_shape + ): + spacings = (18 * k, 18 * k) + self._hhd = nHHD( + cuda=self.settings.cuda, + grid=ds_shape, + spacings=spacings, + poisson_backend=self.settings.poisson_backend, + debug=self.settings.verbose, + force_reference_grid=ref_shape, + masked_poisson_solver=self.settings.masked_poisson_solver, + masked_poisson_max_boundary=self.settings.masked_poisson_max_boundary, + masked_poisson_boundary_stride=self.settings.masked_poisson_boundary_stride, + masked_poisson_max_iters=self.settings.masked_poisson_max_iters, + masked_poisson_tol=self.settings.masked_poisson_tol, + masked_poisson_reg=self.settings.masked_poisson_reg, + ) + self._hhd_grid_shape = ds_shape + self._hhd_ref_shape = ref_shape + + _, forces = self._hhd.get_force(flow_ds, valid_mask=mask_ds) + fn, fx, fy, ftor = forces + self._force_raw = (float(fn), float(fx), float(fy)) + self._torque = float(ftor) + + vnormal_ds = self._hhd.get_normal() + vshear_ds = self._hhd.get_shear() + if k > 1: + vn = self._vn_buf + vs = self._vs_buf + if vn is None or vn.shape != (orig_h, orig_w, 2) or vn.dtype != np.float32: + vn = np.empty((orig_h, orig_w, 2), dtype=np.float32) + vs = np.empty((orig_h, orig_w, 2), dtype=np.float32) + self._vn_buf = vn + self._vs_buf = vs + cv2.resize( + vnormal_ds, (orig_w, orig_h), dst=vn, interpolation=cv2.INTER_LINEAR + ) + cv2.resize( + vshear_ds, (orig_w, orig_h), dst=vs, interpolation=cv2.INTER_LINEAR + ) + self._vnormal = vn + self._vshear = vs + else: + self._vnormal = vnormal_ds.copy() + self._vshear = vshear_ds.copy() + + if valid_mask is not None: + m3 = self._mask_f32 + if m3 is None or m3.shape != (orig_h, orig_w, 1) or m3.dtype != np.float32: + m3 = np.empty((orig_h, orig_w, 1), dtype=np.float32) + self._mask_f32 = m3 + np.copyto(m3[:, :, 0], valid_mask, casting="unsafe") + np.multiply(self._vnormal, m3, out=self._vnormal) + np.multiply(self._vshear, m3, out=self._vshear) + + return ImageDecomposition( + flow=self._flow, + vnormal=self._vnormal, + vshear=self._vshear, + force_raw=self._force_raw, + torque=self._torque, + valid_mask_hhd=self._valid_mask_hhd, + ) + + def current_field(self) -> dict: + """Same-frame HHD field payload (the solver returns a private copy).""" + if self._hhd is None: + raise RuntimeError("compute_hhd() before reading the field") + return self._hhd.get_field(valid_mask=self._valid_mask_hhd) + + # ------------------------------------------------------------ measurement + def measure( + self, + image: np.ndarray, + *, + identity: FrameIdentity, + clock: FrameClock, + source_dt_s: float | None = None, + valid_mask: np.ndarray | None = None, + include: Sequence[str] = ("flow",), + want_field: bool = False, + reference_generation: int | None = None, + ) -> ImageMeasurement: + """Measure one prepared frame once and return read-only payload + T03 header. + + ``reference_generation`` is the version the owning stream claimed for this + frame; when omitted the last adopted version is used. The version is fixed + before flow is computed: a newer version adopted while this frame is in + flight does not relabel it. + """ + unknown = set(include) - _INCLUDE_FIELDS + if unknown: + raise ValueError( + f"unknown include field(s) {sorted(unknown)}; " + f"expected a subset of {sorted(_INCLUDE_FIELDS)}" + ) + started = time.perf_counter() + if reference_generation is not None: + self.set_reference_generation(reference_generation) + generated = int(self._reference_generation) + self._compute_flow_state(image, valid_mask=valid_mask) + decomposition = self.decompose(self._flow, valid_mask=valid_mask) + stamp = FrameStamp( + 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, + ) + frame = MeasurementFrame( + stamp=stamp, source_dt_s=source_dt_s, reference_generation=generated + ) + include_set = set(include) + payload = { + "flow": _readonly_copy(decomposition.flow) + if "flow" in include_set + else None, + "vnormal": _readonly_copy(decomposition.vnormal) + if "vnormal" in include_set + else None, + "vshear": _readonly_copy(decomposition.vshear) + if "vshear" in include_set + else None, + "valid_mask": _readonly_copy(decomposition.valid_mask_hhd) + if "valid_mask" in include_set + else None, + "field": _readonly_copy(self.current_field()) + if ("field" in include_set or want_field) + else None, + "image": _readonly_copy(image) if "image" in include_set else None, + } + return ImageMeasurement( + frame=frame, + force_raw=decomposition.force_raw, + torque=decomposition.torque, + flow=payload["flow"], + vnormal=payload["vnormal"], + vshear=payload["vshear"], + valid_mask=payload["valid_mask"], + field=payload["field"], + image=payload["image"], + elapsed_s=time.perf_counter() - started, + diagnostics=(("reference_generation", generated),), + ) diff --git a/src/orisys/pipeline.py b/src/orisys/pipeline.py index 0cb2acf..a40c548 100644 --- a/src/orisys/pipeline.py +++ b/src/orisys/pipeline.py @@ -636,6 +636,15 @@ class StreamPipeline: t0 = time.perf_counter() # The frame is passed explicitly: the SDK's self.img slot is shared # with the producer, so it must not be the transport. + # One reference-version source: the version claimed by this stage is + # handed to an image block that supports adopting it, so the block + # stamps the same number it will publish on the snapshot. Sensors + # without the optional capability keep the historical call. + adopt_reference = getattr( + self.sensor, "set_stream_reference_generation", None + ) + if adopt_reference is not None: + adopt_reference(reference_version) self.sensor.compute_flow(frame) flow_s = time.perf_counter() - t0 flow = getattr(self.sensor, "flow", None) diff --git a/src/orisys/sensor.py b/src/orisys/sensor.py index c13657f..6f49b6b 100644 --- a/src/orisys/sensor.py +++ b/src/orisys/sensor.py @@ -28,11 +28,11 @@ from .frame_processors import create_frame_processor from .field_calibration import FIELD_PROFILE from .get_depth import Contact3D from .hhd import nHHD +from .image_pipeline import ImagePipeline, ImageSettings # from threading import Thread, Event # from queue import Queue, Empty from .Stitcher import Stitcher -from .tracker import Tracker from .capture import ( CaptureConfig, CaptureSource, @@ -296,25 +296,38 @@ class Sensor: else None ) self.frame_processor = create_frame_processor(self.sensor_type, stitcher=self.stitcher) - self.__tracker = Tracker(adaptive=True, cuda=cuda, verbose=self.verbose, config_tracker=self.__config_tracker) + # The image block (tracker + HHD + flow EMA) lives in ImagePipeline. The + # attributes below stay as compatibility mirrors for code that reads + # them; ``__tracker``/``__hhd`` are only used by the direct (bare + # Sensor/test-double) fallback path below. + self.image_settings = ImageSettings.from_config( + self._config, + cuda=bool(cuda), + backend=backend, + verbose=self.verbose, + flow_ema_alpha=self._force_readout.flow_ema_alpha, + ) + self.image_pipeline = ImagePipeline( + self.image_settings, + context=getattr(self, "context", None), + verbose=self.verbose, + flow_ema=self._force_readout, + ) + self.__tracker = None self.__contact3d = Contact3D(config_contact=self.__config_contact, verbose=self.verbose) self.__hhd = None - # Integer >= 1: run HHD on flow downsampled by this factor for speed (1 = full res). - _raw_ds = self.__config_tracker.get("hhd_downsample", 1) - try: - self._hhd_downsample = max(1, int(_raw_ds)) if _raw_ds is not None else 1 - except (TypeError, ValueError): - self._hhd_downsample = 1 + self._hhd_downsample = self.image_settings.hhd_downsample self._hhd_grid_shape = None self._hhd_ref_shape = None - cfg_backend = self.__config_tracker.get("backend", None) - self.poisson_backend = cfg_backend if cfg_backend is not None else backend - self.masked_poisson_solver = self.__config_tracker.get("masked_poisson_solver", "sparse") - self.masked_poisson_max_boundary = self.__config_tracker.get("masked_poisson_max_boundary", 96) - self.masked_poisson_boundary_stride = self.__config_tracker.get("masked_poisson_boundary_stride", 1) - self.masked_poisson_max_iters = self.__config_tracker.get("masked_poisson_max_iters", 3) - self.masked_poisson_tol = self.__config_tracker.get("masked_poisson_tol", 1e-4) - self.masked_poisson_reg = self.__config_tracker.get("masked_poisson_reg", 1e-8) + self.poisson_backend = self.image_settings.poisson_backend + self.masked_poisson_solver = self.image_settings.masked_poisson_solver + self.masked_poisson_max_boundary = self.image_settings.masked_poisson_max_boundary + self.masked_poisson_boundary_stride = ( + self.image_settings.masked_poisson_boundary_stride + ) + self.masked_poisson_max_iters = self.image_settings.masked_poisson_max_iters + self.masked_poisson_tol = self.image_settings.masked_poisson_tol + self.masked_poisson_reg = self.image_settings.masked_poisson_reg # Legacy PyAV capture state (unused after orisys.capture refactor). # self._use_pyav = False # self._av_container = None @@ -682,7 +695,14 @@ class Sensor: image = self.img if img is None else img if image is None: raise Exception("Warning: Get image first") - + + pipeline = getattr(self, "image_pipeline", None) + if pipeline is not None: + self.flow = pipeline.compute_flow(image, valid_mask=self.valid_mask) + self._compute_flow_means(self.flow) + return + + # Bare Sensor/test-double fallback: no ImagePipeline was constructed. flow = self.__tracker.track(image, valid_mask=self.valid_mask) # flow = self.__tracker.track_raft(self.img) if hasattr(flow, 'download'): @@ -701,6 +721,45 @@ class Sensor: """``config``, ``measured`` or ``config (file)``.""" return self._force_readout.sample_rate_source + def _publish_force_sample(self) -> None: + """Publish one HHD frame's raw feature to the force side (no image work).""" + if self._measurement_session: + # Session path: publish the same-frame unfiltered feature only. Model + # inference and readout filtering belong to orisys.force_session, so + # the old Newton accessors must not report a value the session did not + # produce (they would be stale or half-processed). + self.fnormal = self.fshearx = self.fsheary = float("nan") + self._force_filtered_raw = None + self._force_unfiltered = None + self._force_raw_unfiltered = self._force_raw + else: + self._force_readout.update_sample_rate() + if self._uses_calibration_model(): + self._model_frame_id += 1 + self.compute_contact_valid = False + self.centroid_found = False + self._force_readout.apply(self._force_raw) + self._force_filtered_raw = self._force_readout.filtered_raw + self._force_raw_unfiltered = self._force_raw + else: + self._apply_force_filter() + + def _infer_hhd_model(self, field_source) -> None: + """Run the model/contact side on this frame's already computed field.""" + if self._uses_calibration_model() and not self._measurement_session: + # Spatial inference must use this HHD frame's contact, even when the + # caller did not request contact diagnostics from StreamPipeline. + # The field model needs no locator: it consumes this frame's field. + if self.calibration.input_profile == FIELD_PROFILE: + self.apply_calibration_frame( + frame_id=self._model_frame_id, + field=field_source(), + ) + elif self.calibration.input_profile == "fn_delta_autocal_uv/1": + self.compute_contact() + else: + self.apply_calibration_frame(frame_id=self._model_frame_id) + def compute_hhd(self, flow=None): """Decompose the optical flow and update the force readout. @@ -708,12 +767,32 @@ class Sensor: ``self.flow``. StreamPipeline passes its own per-frame copy so the flow stage cannot overwrite ``self.flow`` mid-decomposition. Existing callers are unaffected. + + The image half (HHD solver, grid buffers, mask handling, upsampling) is + delegated to :attr:`image_pipeline` when a real Sensor was constructed; + the force half (readout/model/contact) stays here. Bare Sensor and + test-double instances keep the historical direct path because they carry + their own solver through ``_Sensor__hhd``. """ flow = self.flow if flow is None else flow if flow is None: raise Exception("Warning: Compute flow first") + pipeline = getattr(self, "image_pipeline", None) + if pipeline is not None: + decomposition = pipeline.decompose(flow, valid_mask=self.valid_mask) + self._force_raw = decomposition.force_raw + self.ftor = decomposition.torque + self.vnormal = decomposition.vnormal + self.vshear = decomposition.vshear + self.valid_mask_hhd = decomposition.valid_mask_hhd + self.compute_deformation_valid = True + self._publish_force_sample() + self._infer_hhd_model(self.current_field) + return True + + # ---- direct fallback: the historical implementation, unchanged orig_h, orig_w = flow.shape[:2] try: k = max(1, int(self._hhd_downsample)) @@ -780,26 +859,7 @@ class Sensor: fn, fx, fy, ftor = forces self._force_raw = (float(fn), float(fx), float(fy)) self.ftor = float(ftor) - if self._measurement_session: - # Session path: publish the same-frame unfiltered feature only. Model - # inference and readout filtering belong to orisys.force_session, so - # the old Newton accessors must not report a value the session did not - # produce (they would be stale or half-processed). - self.fnormal = self.fshearx = self.fsheary = float("nan") - self._force_filtered_raw = None - self._force_unfiltered = None - self._force_raw_unfiltered = self._force_raw - else: - self._force_readout.update_sample_rate() - if self._uses_calibration_model(): - self._model_frame_id += 1 - self.compute_contact_valid = False - self.centroid_found = False - self._force_readout.apply(self._force_raw) - self._force_filtered_raw = self._force_readout.filtered_raw - self._force_raw_unfiltered = self._force_raw - else: - self._apply_force_filter() + self._publish_force_sample() vnormal_ds = self.__hhd.get_normal() vshear_ds = self.__hhd.get_shear() @@ -834,21 +894,7 @@ class Sensor: np.multiply(self.vshear, m3, out=self.vshear) self.compute_deformation_valid = True - - if self._uses_calibration_model() and not self._measurement_session: - # Spatial inference must use this HHD frame's contact, even when the - # caller did not request contact diagnostics from StreamPipeline. - # The field model needs no locator: it consumes this frame's field. - if self.calibration.input_profile == FIELD_PROFILE: - self.apply_calibration_frame( - frame_id=self._model_frame_id, - field=self.__hhd.get_field(valid_mask=mask_ds), - ) - elif self.calibration.input_profile == "fn_delta_autocal_uv/1": - self.compute_contact() - else: - self.apply_calibration_frame(frame_id=self._model_frame_id) - + self._infer_hhd_model(lambda: self.__hhd.get_field(valid_mask=mask_ds)) return True def compute_contact(self, vnormal=None): @@ -941,14 +987,34 @@ class Sensor: def measurement_session(self) -> bool: return bool(self._measurement_session) + def set_stream_reference_generation(self, generation: int) -> None: + """Adopt the driving stream's reference version for the next frame. + + Stage-0 ``StreamPipeline`` claims reference versions; the image block must + stamp that same version instead of counting its own. The pipeline calls + this on its flow thread immediately before ``compute_flow``, so each + frame keeps the version it was measured with. Calling it directly outside + a stream is allowed (the value is what ``ImagePipeline.measure`` will + stamp); it never changes tracker/EMA state. + """ + pipeline = getattr(self, "image_pipeline", None) + if pipeline is not None: + pipeline.set_reference_generation(int(generation)) + def reset_tracker(self) -> None: """Reset the optical-flow tracker at a frame boundary (session zeroing). The narrow half of :meth:`reset`: tracker and readout memory only, never the calibration baseline. ``ForceSession`` calls this on its processing thread between two frames, before the first baseline frame is measured. + The tracker/EMA history belongs to the ImagePipeline when one is bound; + bare Sensor/test-double instances keep the historical attribute path. """ - self.__tracker.reset() + pipeline = getattr(self, "image_pipeline", None) + if pipeline is not None: + pipeline.reset_tracker() + else: + self.__tracker.reset() self._force_readout.reset() self._force_unfiltered = None self._force_raw_unfiltered = None @@ -1088,6 +1154,9 @@ class Sensor: def current_field(self) -> dict: """Same-frame HHD field payload of the last decomposition (a private copy).""" + pipeline = getattr(self, "image_pipeline", None) + if pipeline is not None: + return pipeline.current_field() if self.__hhd is None: raise RuntimeError("compute_hhd() before reading the field") mask = self.valid_mask_hhd @@ -1155,7 +1224,11 @@ class Sensor: image buffers. The force half belongs to the HHD thread and is reset separately by :meth:`reset_force_state`. """ - self.__tracker.reset() + pipeline = getattr(self, "image_pipeline", None) + if pipeline is not None: + pipeline.reset_image_state() + else: + self.__tracker.reset() self._force_readout.reset_flow_ema() if self.img is not None: self.initial_img = self.img.copy() @@ -1288,6 +1361,8 @@ class Sensor: flow_ema_alpha=flow_ema_alpha, force_median_window=force_median_window, ) + # The flow-EMA coefficient/history has one owner (this readout); the image + # pipeline reads it on the next frame, so runtime configure stays live. for readout in getattr(self, "_model_readouts", {}).values(): readout.configure( lowpass_cutoff_hz=lowpass_cutoff_hz, diff --git a/tests/test_image_pipeline.py b/tests/test_image_pipeline.py new file mode 100644 index 0000000..94524d4 --- /dev/null +++ b/tests/test_image_pipeline.py @@ -0,0 +1,521 @@ +"""ImagePipeline (T05): independent image block, delegation, read-only results. + +The image block is verified three ways: the same prepared frames through the +published Sensor path and through a standalone ImagePipeline must produce +bit-identical flow/HHD/force/field results; measured arrays must be read-only +copies that later frames cannot overwrite; and the optical-flow reference +generation must be written when the image thread computes a frame (a reset +requested while a frame is in flight applies to the next frame only). +""" + +from __future__ import annotations + +import json +import threading +import time + +import numpy as np +import pytest + +from orisys import Sensor, configs +from orisys.image_pipeline import ( + ImageMeasurement, + ImagePipeline, + ImageSettings, + sensor_clock, +) +from orisys.sensors import build_sensor_context +from orisys.stream_contract import FrameIdentity, MeasurementFrame + + +def _frames(count: int = 4, size: int = 365) -> list[np.ndarray]: + """Deterministic grey frames whose pattern translates a few pixels.""" + height = width = size + yy, xx = np.mgrid[0:height, 0:width] + frames = [] + for index in range(count): + cy = 130.0 + 2.0 * index + cx = 210.0 + 3.0 * index + blob = 120.0 * np.exp(-(((xx - cx) / 25.0) ** 2 + ((yy - cy) / 21.0) ** 2)) + gradient = 40.0 + 30.0 * (xx / width) + 20.0 * (yy / height) + frames.append(np.clip(gradient + blob, 0.0, 255.0).astype(np.uint8)) + return frames + + +def _config(name: str = "box") -> dict: + return configs.load_config(name) + + +def _context(): + cfg = _config() + return build_sensor_context( + config_name="box", + config=cfg, + device_id="BOX-001", + bundle_candidates=["box"], + ) + + +def _identity(context, frame_id: int = 0) -> FrameIdentity: + return context.frame_identity( + session_id="run-1", frame_id=frame_id, source_frame_id=frame_id + ) + + +# --------------------------------------------------------------- same-run parity +def test_standalone_pipeline_matches_the_published_sensor_path() -> None: + """Old (Sensor delegation) and new (standalone) paths, same frames, same run.""" + sensor = Sensor( + "external", config_name="box", verbose=False, backend="cpu", cuda=False, mirror=False + ) + try: + settings = ImageSettings.from_config( + _config(), + cuda=False, + backend="cpu", + flow_ema_alpha=sensor._force_readout.flow_ema_alpha, + ) + assert settings.as_dict() == sensor.image_settings.as_dict() + pipeline = ImagePipeline(settings) + context = _context() + + for index, raw in enumerate(_frames()): + prepared = sensor.get_img( + frame=raw, already_processed=True, already_gray=True + ) + assert np.array_equal(prepared, raw), "same-size resize must be identity" + + sensor.compute_flow() + sensor.compute_hhd() + sensor_flow = np.array(sensor.flow, copy=True) + sensor_vnormal = np.array(sensor.vnormal, copy=True) + sensor_vshear = np.array(sensor.vshear, copy=True) + sensor_field = sensor.current_field() + sensor_force = tuple( + sensor.read_info( + sensor.info.FNORMAL_RAW, + sensor.info.FSHEARX_RAW, + sensor.info.FSHEARY_RAW, + ) + ) + + measurement = pipeline.measure( + raw, + identity=_identity(context, index), + clock=sensor_clock(1_000_000 + index, clock_id="host"), + include=("flow", "vnormal", "vshear", "field", "valid_mask"), + ) + assert np.array_equal(measurement.flow, sensor_flow), index + assert np.array_equal(measurement.vnormal, sensor_vnormal), index + assert np.array_equal(measurement.vshear, sensor_vshear), index + assert measurement.force_raw == pytest.approx(sensor_force), index + for key, value in sensor_field.items(): + if isinstance(value, np.ndarray): + assert np.array_equal(measurement.field[key], value), (index, key) + else: + assert measurement.field[key] == value, (index, key) + finally: + sensor.disconnect() + + +def test_standalone_pipeline_runs_without_constructing_a_sensor() -> None: + context = _context() + pipeline = ImagePipeline( + ImageSettings.from_config(_config(), cuda=False, backend="cpu"), context=context + ) + measurement = pipeline.measure( + _frames()[0], + identity=_identity(context), + clock=sensor_clock(123, clock_id="host"), + source_dt_s=None, + include=("flow",), + ) + assert isinstance(measurement, ImageMeasurement) + assert isinstance(measurement.frame, MeasurementFrame) + assert measurement.frame.stamp.sensor_id == "BOX-001" + assert measurement.frame.stamp.session_id == "run-1" + assert measurement.frame.reference_generation == 0 + assert measurement.frame.source_dt_s is None + # The T03 header is the serialised contract: it round-trips unchanged. + assert ( + MeasurementFrame.from_dict(json.loads(json.dumps(measurement.frame.to_dict()))) + == measurement.frame + ) + assert pipeline.context is context + + +# ---------------------------------------------------------------- read-only data +def test_measured_arrays_are_readonly_copies_that_later_frames_do_not_rewrite() -> None: + context = _context() + pipeline = ImagePipeline( + ImageSettings.from_config(_config(), cuda=False, backend="cpu"), context=context + ) + frames = _frames() + first = pipeline.measure( + frames[0], + identity=_identity(context, 0), + clock=sensor_clock(10, clock_id="host"), + include=("flow", "vnormal", "vshear", "field", "valid_mask"), + ) + seen = { + "flow": np.array(first.flow, copy=True), + "vnormal": np.array(first.vnormal, copy=True), + "field": np.array(first.field["d"], copy=True), + } + for name, array in ( + ("flow", first.flow), + ("vnormal", first.vnormal), + ("field", first.field["d"]), + ): + assert array.flags.writeable is False, name + with pytest.raises(ValueError): + array[0, 0] = 1.0 + + # Computing the next frame must not overwrite what was already published. + second = pipeline.measure( + frames[1], + identity=_identity(context, 1), + clock=sensor_clock(20, clock_id="host"), + include=("flow", "field"), + ) + assert np.array_equal(first.flow, seen["flow"]) + assert np.array_equal(first.vnormal, seen["vnormal"]) + assert np.array_equal(first.field["d"], seen["field"]) + assert not np.array_equal(second.flow, first.flow) + + +def test_include_selects_only_requested_arrays() -> None: + context = _context() + pipeline = ImagePipeline( + ImageSettings.from_config(_config(), cuda=False, backend="cpu"), context=context + ) + measurement = pipeline.measure( + _frames()[0], + identity=_identity(context), + clock=sensor_clock(1, clock_id="host"), + include=("flow",), + ) + assert measurement.flow is not None + assert measurement.vnormal is None and measurement.vshear is None + assert measurement.field is None and measurement.valid_mask is None + with pytest.raises(ValueError, match="unknown include"): + pipeline.measure( + _frames()[0], + identity=_identity(context), + clock=sensor_clock(1, clock_id="host"), + include=("not-a-field",), + ) + + +# ------------------------------------------------------ reference-generation seam +class _GatedPipeline(ImagePipeline): + """Barrier-style gate: hold a frame after its reference version was captured.""" + + def __init__(self, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + self.entered = threading.Event() + self.release = threading.Event() + self.gate_once = True + + def _compute_flow_state(self, image, *, valid_mask=None): + if self.gate_once: + self.gate_once = False + self.entered.set() + assert self.release.wait(timeout=5.0) + return super()._compute_flow_state(image, valid_mask=valid_mask) + + +def test_inflight_frame_keeps_its_own_reference_version() -> None: + """The stream can adopt a new version while an older frame is still in flight.""" + context = _context() + pipeline = _GatedPipeline( + ImageSettings.from_config(_config(), cuda=False, backend="cpu"), context=context + ) + frames = _frames() + results: list[ImageMeasurement] = [] + + def compute_first() -> None: + results.append( + pipeline.measure( + frames[0], + identity=_identity(context, 0), + clock=sensor_clock(100, clock_id="host"), + reference_generation=0, + ) + ) + + worker = threading.Thread(target=compute_first) + worker.start() + try: + assert pipeline.entered.wait(timeout=5.0), "first frame never reached the gate" + # The stream adopts version 1 while frame 0 is provably in flight. + pipeline.set_reference_generation(1) + assert pipeline.reference_generation == 1 + finally: + pipeline.release.set() + worker.join(timeout=5.0) + assert not worker.is_alive() + in_flight = results[0] + assert in_flight.frame.reference_generation == 0, "old frame was relabelled" + + next_measurement = pipeline.measure( + frames[1], + identity=_identity(context, 1), + clock=sensor_clock(200, clock_id="host"), + reference_generation=1, + ) + assert next_measurement.frame.reference_generation == 1 + assert in_flight.frame.reference_generation == 0 + + +def test_reference_version_is_adopted_not_counted_by_the_image_block() -> None: + """:class:`ImagePipeline` stamps the stream's version; a reset does not bump it.""" + context = _context() + pipeline = ImagePipeline( + ImageSettings.from_config(_config(), cuda=False, backend="cpu"), context=context + ) + frames = _frames() + first = pipeline.measure( + frames[0], + identity=_identity(context, 0), + clock=sensor_clock(1, clock_id="host"), + reference_generation=0, + ) + assert first.frame.reference_generation == 0 + pipeline.reset_image_state() + zeroed = pipeline.measure( + frames[1], + identity=_identity(context, 1), + clock=sensor_clock(2, clock_id="host"), + reference_generation=0, + ) + assert zeroed.frame.reference_generation == 0, "reset must not invent a version" + pipeline.set_reference_generation(3) + stamped = pipeline.measure( + frames[2], + identity=_identity(context, 2), + clock=sensor_clock(3, clock_id="host"), + reference_generation=3, + ) + assert stamped.frame.reference_generation == 3 + assert pipeline.reference_generation == 3 + + +def _drive_pipeline(sensor: Sensor, frames: list[np.ndarray], *, request_reset: bool): + """Run one StreamPipeline; record per-flow-frame adopted versions and snapshots.""" + from orisys.pipeline import StreamPipeline + + flow_versions: list[int] = [] + snapshots: list = [] + original = sensor.compute_flow + + def recording_flow(*args, **kwargs): + flow_versions.append(int(sensor.image_pipeline.reference_generation)) + return original(*args, **kwargs) + + sensor.compute_flow = recording_flow + queue = iter(frames) + + def produce(): + try: + raw = next(queue) + except StopIteration: + return None + return sensor.get_img(frame=raw, already_processed=True, already_gray=True) + + pipeline = StreamPipeline( + sensor, produce, include=(), on_frame=snapshots.append, verbose=False + ) + if request_reset: + pipeline.request_reset() + pipeline.start() + try: + deadline = time.perf_counter() + 10.0 + while pipeline.running and time.perf_counter() < deadline: + time.sleep(0.01) + assert pipeline.error is None + finally: + pipeline.stop() + sensor.compute_flow = original + snapshot_versions = [int(snapshot.tracker_reset_seq) for snapshot in snapshots] + return flow_versions, snapshot_versions + + +def test_pipeline_driven_reference_versions_match_every_snapshot_on_a_reused_sensor() -> None: + """One version source: StreamPipeline's claimed version reaches the image block.""" + sensor = Sensor( + "external", config_name="box", verbose=False, backend="cpu", cuda=False, mirror=False + ) + frames = _frames() + try: + sensor.reset() + flow_versions, snapshot_versions = _drive_pipeline(sensor, frames, request_reset=False) + assert flow_versions == snapshot_versions == [0, 0, 0, 0] + + flow_versions, snapshot_versions = _drive_pipeline(sensor, frames, request_reset=True) + assert flow_versions == snapshot_versions == [1, 1, 1, 1] + + # A direct Sensor.reset() between two runs must not leave the image block + # on an old private counter: the next stream's version 0 wins. + sensor.reset() + flow_versions, snapshot_versions = _drive_pipeline(sensor, frames, request_reset=False) + assert flow_versions == snapshot_versions == [0, 0, 0, 0] + assert sensor.image_pipeline.reference_generation == 0 + finally: + sensor.disconnect() + + +# ------------------------------------------------------ published flow-EMA binding +def test_runtime_flow_ema_configuration_keeps_head_behaviour() -> None: + """``Sensor.configure(flow_ema_alpha=...)`` must affect the next frame (HEAD).""" + from orisys.force_readout import ForceReadout + + context = _context() + sensor = Sensor( + "external", config_name="box", verbose=False, backend="cpu", cuda=False, mirror=False + ) + # HEAD-equivalent reference: the published Tracker/ForceReadout composition. + reference_readout = ForceReadout({"flow_ema_alpha": 1.0}) + reference = ImagePipeline( + ImageSettings.from_config( + _config(), cuda=False, backend="cpu", flow_ema_alpha=1.0 + ), + flow_ema=reference_readout, + ) + try: + for index, raw in enumerate(_frames()): + if index == 2: + sensor.configure(flow_ema_alpha=0.5) + reference_readout.configure(flow_ema_alpha=0.5) + prepared = sensor.get_img( + frame=raw, already_processed=True, already_gray=True + ) + sensor.compute_flow() + sensor.compute_hhd() + measurement = reference.measure( + prepared, + identity=_identity(context, index), + clock=sensor_clock(1000 + index, clock_id="host"), + include=("flow", "vnormal"), + reference_generation=0, + ) + assert sensor.image_pipeline.flow_ema_alpha == 0.5 if index >= 2 else True + assert np.array_equal(np.array(sensor.flow, copy=True), measurement.flow), index + assert np.array_equal( + np.array(sensor.vnormal, copy=True), measurement.vnormal + ), index + sensor_force = tuple( + sensor.read_info( + sensor.info.FNORMAL_RAW, + sensor.info.FSHEARX_RAW, + sensor.info.FSHEARY_RAW, + ) + ) + assert sensor_force == measurement.force_raw, index + finally: + sensor.disconnect() + + +def test_static_flow_ema_config_is_bound_once(tmp_path) -> None: + """A config-declared alpha<1 keeps the same flow through Sensor and a standalone pipeline.""" + config = _config() + config["filters"] = dict(config.get("filters") or {}, flow_ema_alpha=0.5) + config_path = tmp_path / "box-ema05.json" + config_path.write_text(json.dumps(config), encoding="utf-8") + + context = _context() + pipeline = ImagePipeline( + ImageSettings.from_config(config, cuda=False, backend="cpu") + ) + sensor = Sensor( + "external", + config_name=str(config_path), + verbose=False, + backend="cpu", + cuda=False, + mirror=False, + ) + try: + assert pipeline.settings.flow_ema_alpha == 0.5 + assert sensor.image_pipeline.flow_ema_alpha == 0.5 + for index, raw in enumerate(_frames()): + prepared = sensor.get_img( + frame=raw, already_processed=True, already_gray=True + ) + sensor.compute_flow() + sensor.compute_hhd() + measurement = pipeline.measure( + prepared, + identity=_identity(context, index), + clock=sensor_clock(index, clock_id="host"), + include=("flow",), + reference_generation=0, + ) + assert np.array_equal(np.array(sensor.flow, copy=True), measurement.flow), index + finally: + sensor.disconnect() + + +def test_flow_ema_matches_the_force_readout_recurrence() -> None: + from orisys.force_readout import ForceReadout + + settings = ImageSettings.from_config( + _config(), cuda=False, backend="cpu", flow_ema_alpha=0.5 + ) + pipeline = ImagePipeline(settings) + readout = ForceReadout({"flow_ema_alpha": 0.5}) + rng = np.random.default_rng(7) + for _ in range(4): + flow = rng.normal(0.0, 1.0, (6, 5, 2)).astype(np.float32) + pipeline_flow = pipeline._apply_flow_ema(flow.copy()) + readout_flow = readout.apply_flow_ema(flow.copy()) + assert np.array_equal(pipeline_flow, readout_flow) + pipeline.reset_image_state() + readout.reset_flow_ema() + flow = rng.normal(0.0, 1.0, (6, 5, 2)).astype(np.float32) + assert np.array_equal( + pipeline._apply_flow_ema(flow.copy()), readout.apply_flow_ema(flow.copy()) + ) + + +# -------------------------------------------------------------- delegation paths +def test_sensor_delegates_its_image_block_to_one_pipeline() -> None: + sensor = Sensor( + "external", config_name="box", verbose=False, backend="cpu", cuda=False, mirror=False + ) + try: + assert isinstance(sensor.image_pipeline, ImagePipeline) + assert ( + sensor.image_pipeline.settings.as_dict() == sensor.image_settings.as_dict() + ) + sensor.get_img(frame=_frames()[0], already_processed=True, already_gray=True) + sensor.compute_flow() + assert sensor.flow is sensor.image_pipeline.flow + sensor.compute_hhd() + assert sensor.vnormal is sensor.image_pipeline.vnormal + assert sensor.vshear is sensor.image_pipeline.vshear + before = sensor.image_pipeline.reference_generation + sensor.reset_image_state() + # The reference version belongs to the stream; a local reset must not + # invent one. Only an adopted stream version changes it. + assert sensor.image_pipeline.reference_generation == before + sensor.set_stream_reference_generation(7) + assert sensor.image_pipeline.reference_generation == 7 + # The published Newton/readout accessors keep working. + assert sensor.force_unit in ("N", "raw") + finally: + sensor.disconnect() + + +def test_finger_runtime_exposes_the_same_image_pipeline() -> None: + from orisys.finger.runtime import FingerRuntime + + pipeline = object() + runtime = FingerRuntime.__new__(FingerRuntime) + runtime.tactile = type( + "Tactile", (), {"raw_sensor": type("Raw", (), {"image_pipeline": pipeline})()} + )() + assert runtime.image_pipeline is pipeline + + runtime.tactile = type("Tactile", (), {"raw_sensor": object()})() + assert runtime.image_pipeline is None