From b6f6a06851c13ca9946ca14272a52e409d9ebfdc Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Sun, 4 Oct 2026 13:37:58 +0800 Subject: [PATCH] Add an offline analysis pass and result cache for replay Replay re-ran the SDK chain on every seek: 0.48-0.51 wall-seconds per video second, about 4.8 minutes for a ten-minute recording. The new orisys_app_shared.analysis package runs that pass once and stores the flow samples a replay draws - 4,034 B per frame at the shipped finger grid instead of the chain's 1,176,000 B; the result contract is in apps/studio/docs/replay-analysis.md. Verified: shared/tests 416 passed / 1 skipped (15 new) and apps/studio/tests unchanged at 425 items / 0 failed / 18 skipped; tools/lint_gate.py clean; a 264-frame pass costs 0.5036 wall-s per video-s against 0.5052 for the bare chain measured the same day. --- apps/studio/docs/replay-analysis.md | 117 ++ .../orisys_app_shared/analysis/__init__.py | 27 + .../src/orisys_app_shared/analysis/cache.py | 624 ++++++++++ .../src/orisys_app_shared/analysis/model.py | 526 ++++++++ .../src/orisys_app_shared/analysis/reader.py | 567 +++++++++ .../src/orisys_app_shared/analysis/service.py | 1067 +++++++++++++++++ shared/tests/test_replay_analysis.py | 615 ++++++++++ 7 files changed, 3543 insertions(+) create mode 100644 apps/studio/docs/replay-analysis.md create mode 100644 shared/src/orisys_app_shared/analysis/__init__.py create mode 100644 shared/src/orisys_app_shared/analysis/cache.py create mode 100644 shared/src/orisys_app_shared/analysis/model.py create mode 100644 shared/src/orisys_app_shared/analysis/reader.py create mode 100644 shared/src/orisys_app_shared/analysis/service.py create mode 100644 shared/tests/test_replay_analysis.py diff --git a/apps/studio/docs/replay-analysis.md b/apps/studio/docs/replay-analysis.md new file mode 100644 index 0000000..5503d26 --- /dev/null +++ b/apps/studio/docs/replay-analysis.md @@ -0,0 +1,117 @@ +# 回放分析结果（Part A） + +本文定义 `orisys_app_shared.analysis` 这个包：一次离线分析 pass 存下什么、怎么编码、 +什么时候作废、播放端怎么读。它是 [record-channel.md](record-channel.md) 所描述的记录格式的 +下游消费者。播放端（时间轴、跳转、单步、倍速）是它的续篇 +[replay-analysis-part-b.md](replay-analysis-part-b.md)：结果里没有的播放语义都在那一篇。 + +回放过去等于每移动一次光标就重跑一遍 SDK 链路：**每个视频秒 0.48–0.51 秒墙钟**，十分钟录像 +一次约 4.8 分钟。Part A 把这一遍跑一次，只存回放真正要画的东西。 + +## 1. 结果模型 + +原始输出每帧 **1,176,000 B**（`flow` 672,000 + `depth_map` 336,000 + `valid_mask` 84,000 + +灰度 `img` 84,000），30 fps 十分钟 21.2 GB —— 不可缓存。结果里没有这些字段，只有共享选择器 +`select_flow_samples` 选出的样本。 + +**每帧记录**（`frame_id`、`t_rel_s`、网格 keyframe 编号、候选数、深度统计、`dx`、`dy`、 +`magnitude`、位打包 `active`、掩码摘要）在 shipped finger 网格下是 **4,034 B/帧**： +`24 + 330×12 + ⌈330/8⌉ + 8`；十分钟 30 fps 约 70 MB。选这个编码而不是实测的另外两种 +（float16 的 2,022 B/帧、丢掉 `magnitude` 的 4,002 B/帧），是因为往返必须**逐位相等**，而 +`magnitude` 正是"同一份结果换个阈值也能用"的依据（第 3 节）。`dx`/`dy`/`magnitude` 存 +float32，与实时选择器比对时后者也要转成 float32 才是精确比较。 + +**不存的**：力（`forces.csv` 已是图表读的内容，180.9 B/帧）；深度图只存每帧一个 +`depth_nonzero` 计数（4 B）加汇总统计 —— 这段素材上深度几乎全零（264 帧中 241 帧全零，最大 +336 px / 84,000），统计记下来"深度视图是空的"才有解释。 + +`provenance` 随结果存：算法版本、传感器配置及其内容哈希、标定 id、单位、参考帧 id、首个输出 +帧 id、解码偏移、逐通道的"重算/记录"标记（`flow`、`depth` 重算，`forces` 记录）、`complete` +标志与计数。 + +## 2. Keyframe：候选位置与有效掩码 + +330 个候选**位置**在整段录像上逐字节相同（两个视频、327 帧共用一个摘要），所以它是 +**keyframe**，只存一次（int16 rows + int16 cols 共 1,320 B，规则网格 zlib 后 132 B）； +有效掩码同样静态（327 帧中 0 帧变化），也存 keyframe（位打包 10,500 B，zlib level 6 后 +687 B）。两者加两个 36 B 的 blob 头，`keyframes.bin` 一共 891 B。 + +每帧仍带 8 字节掩码摘要：某一帧的掩码若真的变了，读端能**发现**，而不是拿错误的掩码把这一帧 +静默重建出来。测试用 Step 0 记录的 blake2b-16 摘要 `d76349a3b901aea50915eb1308901c2e` +校验存回来的掩码，所以 keyframe 是被验证过的副本。 + +box 链路没有 `valid_mask`，`max_arrows=400` 的上限因此按幅值排序，位置集会逐帧移动。格式对此 +只有一条代码路径：**每个位置集都是 keyframe，按摘要去重**。finger 一段录像一个；box 的 33 帧 +录像 26 个，每帧 1,600 B —— 正好等于把位置内联存储的代价。 + +## 3. 失效键 + +键是五个分量的摘要（`model.result_key`）：**录像身份、算法版本、传感器配置、标定、参考帧**； +目录名就是这个摘要。 + +**显示设置刻意不在键里。** 阈值（`flow_on_px`/`flow_off_px`）变化时读端从存下来的 +`magnitude` 重新推导 `active`，不需要重建结果；只有采样**网格**（`flow_grid_spacing`、 +`flow_edge_margin`、`max_arrows`）变化才会移动候选位置，那份策略记在 provenance 里供比对而 +不塞进键里。SDK 版本同样不在键里（补丁版本不该让键变化），它记在 provenance 的 `versions`， +由 `AnalysisReader.stale_reasons()` 报出。 + +## 4. 参考帧规则 + +`Sensor.reset()` 清空光流跟踪器，所以**重置在指定的那一帧处理之前施加**：参考帧本身就是新参考 +的第一帧，位移恰好为零。实测在真实录像第 36 帧重置后该帧 `|flow|.mean()` = 0.000000000，而 +上一帧是 0.5354；第 36 帧之前的每一帧与不重置的那一遍逐字节相同。 + +provenance 记 `reference_frame_id`（跟踪器被清空的那一帧）与 `first_output_frame_id`（结果 +第一帧）；没有显式重置时两者相等，所以这个字段永远不是"未知"，`reference_is_explicit` 区分 +两种情况。 + +帧 id 是采集会话自己的计数器，**不从 0 或 1 开始**：证据里的 box 录像是 20…52。所以请求一个 +录像里不存在的参考帧会**拒绝运行**（`ValueError`，不留下目录），而不是默默按"没有重置"跑完 +—— 那会存下一个与要求不符的 `reference_frame_id`。 + +## 5. 两个时间基准 + +**视频时间 ≠ 信号时间。** 容器 33 帧 @30 fps 是 1.100 s，这 33 帧对应的信号是 1.920 s（实测 +16.664 fps）。`t_rel_s` 来自 `frames.csv`，是**信号时间**，也是读端 `frame_at()` 使用的唯一 +时间轴；容器时长永远不派生自它，也不与它互相换算。有序段索引由 `session.json` 的 +`files.video`、`stats.segments`、`prereoll_segment` 加 `RecordingReader.segments()` 的 +`raw_*.mp4` 排序推导；两者不一致时 pass **不修复**：声明了却缺失的段让结果 `complete=false`。 + +## 6. box 的偏移陷阱 + +在视频上构造 box `Sensor` 时，拼接探测**悄悄吃掉最前面的 7 个已解码帧**（33 帧视频只剩 26 帧 +可处理，调用方读到的第一帧是视频第 8 帧）；finger 链路吃掉 0 帧。偏移不是猜的：构造后直接读 +底层 capture 的 `CAP_PROP_POS_FRAMES`（box = 7，finger = 0），记进 provenance 的 +`decode_order_offsets`。因此**帧身份只来自录像自己的 `frames.csv`**：可读帧 d 对应 `in_video` +行 `d + offset`，`frame_id`、`t_rel_s`、`segment` 全部取自该行。后备项（后端报不出位置时按常量 +7）把 `decode_order_offset_source` 记成 `declared`。计数把"少掉的帧"分开记：`skipped`（被偏移 +吃掉的索引行）、`unmapped`（解码出来却没有索引行）、`frames_short`（有索引行却没有解码帧）。 + +**构造 box `Sensor` 还会往进程当前目录写约 2 MB 的 `config/compound_eye.png`**，所以每次开 +链路都在临时工作目录里、由进程级锁串行化：这是本设计唯一无法消除的全局副作用，SDK 没有提供 +改写这个路径的入口。 + +## 7. 完整性与淘汰 + +结果目录里有一个显式 `INCOMPLETE` 标记，从写入第一个字节起存在，直到元数据声明完成之后才 +删除。默认读端只接受同时满足"元数据 `complete=true`"、"provenance 也 `complete=true`"、 +"没有 `INCOMPLETE` 标记"的目录，三者在读取时互相校验，不一致直接报格式错误。 + +取消、SDK 失败、任意异常走同一条路：写完已完成的块与 keyframe，写上 `complete=false` 和停止 +原因，保留标记 —— 被中断的一遍是**可查看的**（`allow_incomplete=True` 能读到已写完的帧），但 +绝不会被默认当作有效结果。一帧的链路调用失败就结束整遍，而不是把过期或空测量写进缓存。 + +写入分块：帧先攒在内存里，满了整块落到 `frames-*.bin.part` 再原子改名，内存上限是 +`chunk_frames × 每帧记录大小`（256 帧约 1.0 MB）。淘汰按明确的容量上限整份删除结果（默认 +1 GiB），先删不可用的未完成结果，再按时间从旧到新，永不触碰调用方 `keep` 的那一份。 + +## 8. 读端 + +`AnalysisReader` 按帧 id 和按时间戳随机访问：`meta.json` 的块清单定位块，块尾偏移表定位帧， +一次读一帧记录。seek 不需要重跑 pass，也不需要录像、SDK 或 Qt —— 测试把结果目录整体复制到 +别处再读，逐帧比对相同。这一遍不比裸链路贵：264 帧 / 8.8 s 上三次计时均值 0.5036 墙钟秒每 +视频秒，同一天用 Step 0 自己的 `bench_chain.py` 重测裸链路为 0.5052。 + +两个建议给 SDK（本文档不实施）：给外部帧模式一个公开的逐帧读取方法，替代 +`Sensor._read_frame`；给拼接预览路径一个可配置目标，替代相对当前目录的 +`config/compound_eye.png`。 diff --git a/shared/src/orisys_app_shared/analysis/__init__.py b/shared/src/orisys_app_shared/analysis/__init__.py new file mode 100644 index 0000000..23440f8 --- /dev/null +++ b/shared/src/orisys_app_shared/analysis/__init__.py @@ -0,0 +1,27 @@ +"""Offline analysis of a recording: run the chain once, read the result many times. + +Why this exists: replaying a recording used to mean re-running the SDK chain every time +the operator moved the cursor, which costs 0.48–0.51 wall-seconds per video-second — +about 4.8 minutes for a ten-minute recording, every time. Part A runs that pass **once** +and stores the flow samples a replay draws, so seeking is a file read. + +What is stored is deliberately *not* the chain's output. One frame of raw output is +1,176,000 B (``flow`` [300,280,2] float32, ``depth_map``, ``valid_mask``, the gray +image), which is 21.2 GB for ten minutes at 30 fps. The selected samples cost 4,034 B +per frame. The rest is recomputed — that is what makes the cache a cache. + +The three modules have one job each: + +``model`` + The result and its byte encoding: the key, the provenance, the frame record. +``cache`` + Writing it without ever leaving a partial result that looks finished, and evicting + whole results against a size cap. +``reader`` + Serving it: by frame id, by timestamp, without replaying anything. +``service`` + The pass itself — the only place that imports the SDK. + +Nothing here imports Qt, so the pass and the reader are testable headless, and a +caller that only reads a stored result never loads the SDK at all. +""" diff --git a/shared/src/orisys_app_shared/analysis/cache.py b/shared/src/orisys_app_shared/analysis/cache.py new file mode 100644 index 0000000..b5288f4 --- /dev/null +++ b/shared/src/orisys_app_shared/analysis/cache.py @@ -0,0 +1,624 @@ +"""Writing and evicting analysis results: chunked, atomic, and never half-served. + +A pass over a ten-minute recording produces tens of thousands of frame records. Two +properties matter more than speed: + +* **A reader must never see a partial result as a finished one.** Every file is built + under a temporary name and moved into place with :func:`os.replace`, the metadata + that declares the result complete is written *last*, and an explicit ``INCOMPLETE`` + marker exists from the first byte written until that move. A cancelled or crashed + pass leaves a directory that refuses to open by default. +* **Memory must not grow with the recording.** Frames accumulate into a fixed-size + chunk and the chunk is written once. The bound is + ``chunk_frames * frame_payload_bytes(max candidates)`` — 256 frames of the shipped + finger grid is about 1.0 MB — and ``chunk_frames`` is clamped so a caller cannot + raise it without bound. + +Storage is not free, so :func:`evict` removes whole results against a stated size cap. +It only ever deletes a direct child of the cache root that carries a ``meta.json``, so +a wrong path cannot turn into a recursive delete of something else. +""" + +from __future__ import annotations + +import json +import logging +import os +import shutil +import struct +import time +import zlib +from collections.abc import Iterable +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import numpy as np + +from orisys_app_shared.analysis.model import ( + ANALYSIS_FORMAT, + ANALYSIS_FORMAT_VERSION, + CHUNK_HEADER, + CHUNK_MAGIC, + CHUNK_TRAILER, + CODEC_RAW, + CODEC_ZLIB, + KEYFRAME_GRID, + KEYFRAME_HEADER, + KEYFRAME_MAGIC, + KEYFRAME_MASK, + RESULT_DIRNAME, + KeyframeRef, + Provenance, + ResultKey, + digest8, + keyframe_payload, +) + +logger = logging.getLogger(__name__) + +__all__ = [ + "DEFAULT_MAX_BYTES", + "INCOMPLETE_MARKER", + "META_NAME", + "AnalysisWriter", + "ResultCache", + "ResultSummary", + "evict", + "scan_results", +] + +#: The file that declares a finished result. Its presence is not enough — the +#: provenance inside must also say ``complete``. +META_NAME = "meta.json" + +#: Written before the first data byte and removed after the last. A reader refuses a +#: directory that has one, whatever the metadata claims. +INCOMPLETE_MARKER = "INCOMPLETE" + +CHUNK_PREFIX = "frames-" +CHUNK_SUFFIX = ".bin" +KEYFRAMES_NAME = "keyframes.bin" + +DEFAULT_CHUNK_FRAMES = 256 + +#: A chunk is built in memory, so this is a real memory bound, not a tuning knob. +MAX_CHUNK_FRAMES = 4096 + +#: Stated cap for :func:`evict`. A ten-minute finger recording is about 70 MB, so this +#: holds roughly fourteen of them; the operator can lower it per call. +DEFAULT_MAX_BYTES = 1 << 30 + +#: Above this many remembered grid digests the writer stops deduplicating and gives a +#: deviating frame its own keyframe. Correct either way; the map exists to keep the +#: common case (one grid for the whole recording) at one keyframe. +MAX_GRID_DEDUP = 4096 + +_KIND_NAMES = {KEYFRAME_GRID: "grid", KEYFRAME_MASK: "mask"} +_KIND_CODES = {name: code for code, name in _KIND_NAMES.items()} +_CODEC_NAMES = {CODEC_RAW: "raw", CODEC_ZLIB: "zlib"} +_CODEC_CODES = {name: code for code, name in _CODEC_NAMES.items()} + + +def _atomic_write_json(path: Path, payload: Any) -> None: + """Write JSON through a temporary sibling, so a reader sees all of it or none.""" + text = json.dumps( + payload, indent=2, sort_keys=False, ensure_ascii=False, default=str + ) + temporary = path.with_name(path.name + ".part") + with temporary.open("w", encoding="utf-8", newline="\n") as handle: + handle.write(text) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + + +def _atomic_replace_file(path: Path, data: bytes | bytearray) -> None: + temporary = path.with_name(path.name + ".part") + with temporary.open("wb") as handle: + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + + +@dataclass(frozen=True) +class ResultSummary: + """One result directory as :func:`scan_results` found it.""" + + directory: Path + key: str + complete: bool + bytes_used: int + modified: float + stopped_reason: str = "" + + +def _directory_bytes(directory: Path) -> int: + total = 0 + for path in directory.rglob("*"): + try: + if path.is_file(): + total += path.stat().st_size + except OSError: + # A file that vanished between listing and sizing is not a reason to stop + # measuring the rest of the directory. + logger.debug("could not size %s", path, exc_info=True) + return total + + +def scan_results(root: str | Path) -> list[ResultSummary]: + """Every result directly under ``root``, newest last. Unreadable entries are listed + as incomplete rather than skipped: a directory nobody can read is worth seeing.""" + base = Path(root) + found: list[ResultSummary] = [] + if not base.is_dir(): + return found + for child in sorted(base.iterdir()): + if not child.is_dir(): + continue + meta_path = child / META_NAME + complete = False + key = child.name + reason = "" + modified = 0.0 + if meta_path.is_file(): + try: + meta = json.loads(meta_path.read_text(encoding="utf-8")) + complete = bool(meta.get("complete")) + key = str(meta.get("key") or child.name) + reason = str(meta.get("stopped_reason") or "") + modified = meta_path.stat().st_mtime + except (OSError, ValueError): + logger.warning( + "unreadable result metadata at %s", meta_path, exc_info=True + ) + if not modified: + try: + modified = child.stat().st_mtime + except OSError: + modified = 0.0 + if (child / INCOMPLETE_MARKER).exists(): + complete = False + found.append( + ResultSummary( + directory=child, + key=key, + complete=complete, + bytes_used=_directory_bytes(child), + modified=modified, + stopped_reason=reason, + ) + ) + return found + + +def evict( + root: str | Path, + *, + max_bytes: int = DEFAULT_MAX_BYTES, + keep: Iterable[str | Path] = (), + dry_run: bool = False, +) -> list[Path]: + """Remove whole results under ``root`` until the total fits ``max_bytes``. + + Incomplete results go first — they are already unreadable by default, so they are + the cheapest thing to lose — and then complete ones oldest first. Any directory in + ``keep`` is never touched, which is how a caller protects the result it is using. + + Only a direct child of ``root`` carrying a ``meta.json`` is a candidate, so a bad + argument cannot delete an arbitrary tree. + """ + base = Path(root).resolve() + protected = {Path(item).resolve() for item in keep} + limit = max(0, int(max_bytes)) + summaries = scan_results(base) + total = sum(item.bytes_used for item in summaries) + removed: list[Path] = [] + order = sorted( + (item for item in summaries if item.directory.resolve() not in protected), + key=lambda item: (item.complete, item.modified), + ) + for item in order: + if total <= limit: + break + if not (item.directory / META_NAME).is_file(): + continue + if item.directory.parent.resolve() != base: + continue + removed.append(item.directory) + total -= item.bytes_used + if not dry_run: + shutil.rmtree(item.directory, ignore_errors=False) + if removed: + logger.info( + "analysis cache: %s %d result(s), %d bytes used against a %d byte cap", + "would remove" if dry_run else "removed", + len(removed), + total, + limit, + ) + return removed + + +class AnalysisWriter: + """Append-only writer for one result directory. + + Usage is a fixed order — ``begin``, any number of keyframes and frames, then + either ``commit`` or ``abandon`` — and every exit path leaves the directory + honest: ``commit`` is the only thing that produces a result a reader will serve. + """ + + def __init__( + self, + directory: str | Path, + *, + key: ResultKey, + chunk_frames: int = DEFAULT_CHUNK_FRAMES, + ) -> None: + self.directory = Path(directory) + self.key = key + self.chunk_frames = max(1, min(int(chunk_frames), MAX_CHUNK_FRAMES)) + self._keyframe_file = None + self._keyframe_offset = 0 + self._keyframes: list[KeyframeRef] = [] + self._grid_ids: dict[bytes, int] = {} + self._mask_id: int | None = None + self._next_keyframe_id = 0 + self._chunk_buffer = bytearray() + self._chunk_offsets: list[int] = [] + self._chunk_ids: list[int] = [] + self._chunk_times: list[float] = [] + self._chunk_index = 0 + self._chunks: list[dict[str, Any]] = [] + self._frame_count = 0 + self.bytes_written = 0 + self._committed = False + self._abandoned = False + self._started_at = "" + + # ------------------------------------------------------------------- lifecycle + def begin(self) -> None: + """Claim the directory: drop an older result, then mark this one incomplete. + + The marker is written *before* the old data is removed, so an interruption + between the two steps still leaves a directory that will not be served. + """ + self.directory.mkdir(parents=True, exist_ok=True) + self._started_at = time.strftime("%Y-%m-%dT%H:%M:%S%z") + self._write_marker("running") + for stale in list(self.directory.glob(CHUNK_PREFIX + "*" + CHUNK_SUFFIX)): + stale.unlink() + for stale in list(self.directory.glob("*.part")): + stale.unlink() + keyframes = self.directory / KEYFRAMES_NAME + if keyframes.exists(): + keyframes.unlink() + self._keyframe_file = (self.directory / (KEYFRAMES_NAME + ".part")).open("wb") + self._write_meta(complete=False, provenance=None, counts={}, reason="running") + + def _write_marker(self, reason: str) -> None: + _atomic_write_json( + self.directory / INCOMPLETE_MARKER, + {"key": self.key.digest, "started_at": self._started_at, "reason": reason}, + ) + + def _write_meta( + self, + *, + complete: bool, + provenance: Provenance | None, + counts: dict[str, Any], + reason: str, + ) -> None: + record = dict(provenance.to_dict()) if provenance is not None else {} + record["complete"] = bool(complete) + record["stopped_reason"] = str(reason) + _atomic_write_json( + self.directory / META_NAME, + { + "format": ANALYSIS_FORMAT, + "format_version": ANALYSIS_FORMAT_VERSION, + "key": self.key.digest, + "key_components": self.key.components, + "complete": bool(complete), + "stopped_reason": str(reason), + "provenance": record, + "keyframes": [item.to_dict() for item in self._keyframes], + "chunks": list(self._chunks), + "counts": dict(counts), + }, + ) + + def commit(self, *, provenance: Provenance, counts: dict[str, Any]) -> Path: + """Finish the result: close the chunk and keyframes, then declare it complete. + + The completion flag is written before the marker is removed, so an interruption + in between leaves a *complete-looking* directory that still refuses to open. + Failing closed is the only acceptable direction. + """ + if self._committed: + raise RuntimeError("this result was already committed") + if self._abandoned: + raise RuntimeError("this result was abandoned and cannot be committed") + self._flush_chunk() + if self._keyframe_file is not None: + self._keyframe_file.flush() + os.fsync(self._keyframe_file.fileno()) + self._keyframe_file.close() + self._keyframe_file = None + part = self.directory / (KEYFRAMES_NAME + ".part") + if part.exists(): + os.replace(part, self.directory / KEYFRAMES_NAME) + self._write_meta(complete=True, provenance=provenance, counts=counts, reason="") + marker = self.directory / INCOMPLETE_MARKER + if marker.exists(): + marker.unlink() + self._committed = True + return self.directory + + def abandon(self, reason: str, *, provenance: Provenance | None = None) -> Path: + """Stop without a result. Whatever was written stays, and stays unusable. + + The buffered chunk and the open keyframes are closed into real files rather than + dropped, so an interrupted pass is *inspectable* (a caller passing + ``allow_incomplete=True`` can read the frames that were finished) while still + being refused by every default reader. Keeping the frames costs one chunk and + buys the only evidence of how far the pass got; discarding them would leave a + directory whose metadata claims five frames and whose files hold none. + + The provenance is kept when the caller has one: a cancelled pass still has to say + which input it was working on. + """ + if self._committed: + raise RuntimeError("this result was already committed") + self._abandoned = True + try: + self._flush_chunk() + except OSError: + logger.warning( + "could not write the partial chunk of a stopped pass", exc_info=True + ) + if self._keyframe_file is not None: + try: + self._keyframe_file.flush() + os.fsync(self._keyframe_file.fileno()) + self._keyframe_file.close() + except OSError: + logger.debug("could not close the keyframe file", exc_info=True) + self._keyframe_file = None + part = self.directory / (KEYFRAMES_NAME + ".part") + try: + if part.exists(): + os.replace(part, self.directory / KEYFRAMES_NAME) + except OSError: + logger.debug("could not publish the partial keyframes", exc_info=True) + if self.directory.is_dir(): + self._write_marker(reason or "failed") + counts = {"frames_written": self._frame_count} + if provenance is not None: + counts = dict(provenance.counts) + self._write_meta( + complete=False, + provenance=provenance, + counts=counts, + reason=reason or "failed", + ) + return self.directory + + def __enter__(self) -> AnalysisWriter: + self.begin() + return self + + def __exit__(self, exc_type, exc, tb) -> bool: + if exc_type is not None and not self._committed: + reason = f"{exc_type.__name__}: {exc}" if exc else "failed" + self.abandon(reason) + return False + + # ------------------------------------------------------------------- keyframes + def _append_keyframe( + self, kind: int, payload: bytes, *, count: int, height: int, width: int + ) -> int: + digest = digest8(payload) + codec = CODEC_ZLIB + stored = zlib.compress(payload, 6) + if len(stored) >= len(payload): + codec = CODEC_RAW + stored = payload + identifier = self._next_keyframe_id + self._next_keyframe_id += 1 + header = KEYFRAME_HEADER.pack( + KEYFRAME_MAGIC, + kind, + codec, + int(count), + int(height), + int(width), + 0, + 0, + len(payload), + digest, + ) + if self._keyframe_file is None: + raise RuntimeError("the writer is not open; call begin() first") + self._keyframe_file.write(header) + self._keyframe_file.write(stored) + shape = (int(count),) if kind == KEYFRAME_GRID else (int(height), int(width)) + self._keyframes.append( + KeyframeRef( + id=identifier, + kind=_KIND_NAMES[kind], + digest=digest.hex(), + shape=shape, + codec=_CODEC_NAMES[codec], + raw_bytes=len(payload), + stored_bytes=len(stored), + offset=self._keyframe_offset, + ) + ) + self._keyframe_offset += len(header) + len(stored) + self.bytes_written += len(header) + len(stored) + return identifier + + def grid_keyframe(self, rows: np.ndarray, cols: np.ndarray) -> int: + """The candidate location set as a keyframe, reused when it repeats. + + The finger chain produces one of these for a whole recording because its + validity mask is static; a chain whose candidate set moves gets one per + distinct set, which is the same cost as storing the locations inline. + """ + rows = np.ascontiguousarray(rows, dtype="= MAX_GRID_DEDUP: + for stale in list(self._grid_ids)[: MAX_GRID_DEDUP // 2]: + del self._grid_ids[stale] + identifier = self._append_keyframe( + KEYFRAME_GRID, payload, count=int(rows.size), height=0, width=0 + ) + self._grid_ids[digest] = identifier + return identifier + + def mask_keyframe(self, mask: np.ndarray, digest: bytes) -> int: + """The validity mask as a keyframe. One per result: it is static when it is not.""" + array = np.asarray(mask, dtype=bool) + if array.ndim != 2: + raise ValueError(f"a validity mask is 2-D, got shape {array.shape}") + payload = keyframe_payload(KEYFRAME_MASK, array) + stored_digest = digest8(payload) + if stored_digest != digest: + raise ValueError("the mask digest does not describe the stored mask") + if self._mask_id is None: + self._mask_id = self._append_keyframe( + KEYFRAME_MASK, + payload, + count=0, + height=int(array.shape[0]), + width=int(array.shape[1]), + ) + return self._mask_id + + # ---------------------------------------------------------------------- frames + def add_frame(self, record: bytes, *, frame_id: int, t_rel_s: float) -> None: + """Append one encoded frame. The chunk is written when it is full.""" + if self._committed or self._abandoned: + raise RuntimeError("this result is closed") + # Absolute from the start of the file, like the offset table below it: a reader + # seeks once and reads one record, with no header arithmetic in between. + self._chunk_offsets.append(CHUNK_HEADER.size + len(self._chunk_buffer)) + self._chunk_ids.append(int(frame_id)) + self._chunk_times.append(float(t_rel_s)) + self._chunk_buffer += record + self._frame_count += 1 + if len(self._chunk_ids) >= self.chunk_frames: + self._flush_chunk() + + def _flush_chunk(self) -> None: + if not self._chunk_ids: + return + offsets = struct.pack(f"<{len(self._chunk_offsets)}Q", *self._chunk_offsets) + ids = struct.pack(f"<{len(self._chunk_ids)}i", *self._chunk_ids) + times = struct.pack(f"<{len(self._chunk_times)}d", *self._chunk_times) + trailer = offsets + ids + times + # Absolute, from the start of the file: the reader seeks to it directly. + offset_table_offset = CHUNK_HEADER.size + len(self._chunk_buffer) + body = bytes(self._chunk_buffer) + header = CHUNK_HEADER.pack( + CHUNK_MAGIC, + ANALYSIS_FORMAT_VERSION, + 0, + self._chunk_index, + len(self._chunk_ids), + CHUNK_HEADER.size, + ) + payload = ( + header + + body + + trailer + + CHUNK_TRAILER.pack(offset_table_offset, len(self._chunk_ids), CHUNK_MAGIC) + ) + name = f"{CHUNK_PREFIX}{self._chunk_index:04d}{CHUNK_SUFFIX}" + _atomic_replace_file(self.directory / name, payload) + self._chunks.append( + { + "index": self._chunk_index, + "file": name, + "frames": len(self._chunk_ids), + "first_frame_id": int(self._chunk_ids[0]), + "last_frame_id": int(self._chunk_ids[-1]), + "bytes": len(payload), + } + ) + self.bytes_written += len(payload) + self._chunk_index += 1 + self._chunk_buffer = bytearray() + self._chunk_offsets = [] + self._chunk_ids = [] + self._chunk_times = [] + + @property + def frames_written(self) -> int: + return self._frame_count + + +class ResultCache: + """The results belonging to one recording, addressed by key. + + ``root`` is usually ``/analysis``: a result then travels with the + recording it describes, and deleting the recording deletes its cache. Nothing here + writes to the current working directory — that matters because opening a *box* + ``Sensor`` drops a ~2 MB ``config/compound_eye.png`` relative to it (see + ``service.py``). + """ + + def __init__(self, root: str | Path, *, max_bytes: int = DEFAULT_MAX_BYTES) -> None: + self.root = Path(root) + self.max_bytes = int(max_bytes) + + @classmethod + def for_recording(cls, recording_dir: str | Path, **kwargs: Any) -> ResultCache: + return cls(Path(recording_dir) / RESULT_DIRNAME, **kwargs) + + def directory(self, key: ResultKey) -> Path: + return self.root / key.digest + + def exists(self, key: ResultKey) -> bool: + return (self.directory(key) / META_NAME).is_file() + + def is_complete(self, key: ResultKey) -> bool: + directory = self.directory(key) + if (directory / INCOMPLETE_MARKER).exists(): + return False + meta = directory / META_NAME + if not meta.is_file(): + return False + try: + return bool(json.loads(meta.read_text(encoding="utf-8")).get("complete")) + except (OSError, ValueError): + return False + + def reader(self, key: ResultKey, *, allow_incomplete: bool = False): + """Open the stored result. Imported lazily so this module stays import-light.""" + from orisys_app_shared.analysis.reader import AnalysisReader + + return AnalysisReader(self.directory(key), allow_incomplete=allow_incomplete) + + def writer( + self, key: ResultKey, *, chunk_frames: int = DEFAULT_CHUNK_FRAMES + ) -> AnalysisWriter: + return AnalysisWriter(self.directory(key), key=key, chunk_frames=chunk_frames) + + def evict( + self, *, keep: Iterable[str | Path] = (), dry_run: bool = False + ) -> list[Path]: + return evict(self.root, max_bytes=self.max_bytes, keep=keep, dry_run=dry_run) + + def total_bytes(self) -> int: + return sum(item.bytes_used for item in scan_results(self.root)) diff --git a/shared/src/orisys_app_shared/analysis/model.py b/shared/src/orisys_app_shared/analysis/model.py new file mode 100644 index 0000000..37df696 --- /dev/null +++ b/shared/src/orisys_app_shared/analysis/model.py @@ -0,0 +1,526 @@ +"""What one offline analysis pass stores, and the byte encoding it uses. + +The pass re-runs the SDK chain over the recording's ``raw_*.mp4`` segments and keeps +only what a replay draws: the flow samples the shared selector returns, not the flow +field they came from. The fields a frame produces are far too large to cache — the +measured cost is 1,176,000 B per frame (``flow`` 672,000 + ``depth_map`` 336,000 + +``valid_mask`` 84,000 + the gray ``img`` 84,000), which is 21.2 GB for ten minutes at +30 fps. The samples themselves cost 4,034 B per frame at the shipped finger grid. + +Three structural facts decide the layout, and each one is a measured result rather +than a guess: + +* the candidate **location** set is byte-identical on every frame of a finger + recording (330 candidates, one digest across two videos), so it is a **keyframe** + and the per-frame record carries only values; +* the **validity mask** is likewise static (0 of 327 frames differed), so it is a + keyframe too — and the per-frame record still carries an 8-byte digest of the mask + it was computed with, so a frame that *did* change is detectable instead of being + silently reconstructed against the wrong mask; +* a location set that is *not* static (the box chain has no valid mask, so its + ``max_arrows`` cap orders by magnitude and moves frame to frame) gets its own + keyframe, deduplicated by digest. The format has one code path, and "stored once" + is true by construction wherever it can be. + +Display settings are deliberately **not** part of the invalidation key. The record +keeps ``magnitude`` as well as the packed ``active`` bits, so a threshold change is +served by re-deriving activity from magnitude; only a change to the sample *grid* +(``flow_grid_spacing``, ``flow_edge_margin``, ``max_arrows``) moves the candidate +locations, and that policy is recorded in the provenance for comparison instead of +being baked into the key. +""" + +from __future__ import annotations + +import hashlib +import json +import struct +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from typing import Any + +import numpy as np + +from orisys_app_shared.scene3d.visual_policy import FlowSamples + +__all__ = [ + "ALGORITHM_VERSION", + "ANALYSIS_FORMAT", + "ANALYSIS_FORMAT_VERSION", + "CHUNK_HEADER", + "CHUNK_MAGIC", + "CHUNK_TRAILER", + "CODEC_RAW", + "CODEC_ZLIB", + "FRAME_FLAG_DROPPED", + "FRAME_FLAG_VALIDITY", + "FRAME_HEADER", + "KEYFRAME_GRID", + "KEYFRAME_HEADER", + "KEYFRAME_MAGIC", + "KEYFRAME_MASK", + "UNIT", + "FramePayload", + "KeyframeRef", + "Provenance", + "ResultKey", + "decode_frame", + "encode_frame", + "frame_payload_bytes", + "keyframe_payload", + "recording_identity", + "result_key", + "unpack_mask", +] + +#: Directory name a result lives under, beside the recording's ``frames.csv``. +RESULT_DIRNAME = "analysis" + +#: On-disk result format. Bump only for an incompatible layout change; a change to +#: what the *chain computes* belongs in :data:`ALGORITHM_VERSION` instead. +ANALYSIS_FORMAT = "orisys-replay-analysis" +ANALYSIS_FORMAT_VERSION = 1 + +#: Identity of the analysis the pass runs: the SDK calls it makes, in order. +#: ``compute_deformation(decompose=True, check_motion=False)`` then +#: ``compute_contact()`` — the same pair the live acquisition loop and the offline +#: depth replay use. A change to either call, or to the chain they drive, must change +#: this string or two different results would share one cache entry. +ALGORITHM_VERSION = "flow-hhd-contact-1" + +#: Unit of ``dx``/``dy``/``magnitude``: UV pixels of arrow-tip displacement, the same +#: unit :class:`~orisys_app_shared.scene3d.visual_policy.FlowSamples` documents. +UNIT = "uv_px" + +#: ``frame_id``, ``t_rel_s``, grid keyframe id, flags, candidate count, depth statistic. +#: The four trailing pad bytes keep the value arrays 8-byte aligned inside a record. +FRAME_HEADER = struct.Struct(" bytes: + """An 8-byte blake2b over the concatenation of ``parts``.""" + hasher = hashlib.blake2b(digest_size=8) + for part in parts: + hasher.update(part) + return hasher.digest() + + +def canonical_digest8(payload: Any) -> bytes: + """An 8-byte digest of a JSON-representable structure, independent of key order.""" + text = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + return digest8(text.encode("utf-8")) + + +# --------------------------------------------------------------------------- the key +@dataclass(frozen=True) +class ResultKey: + """The five facts that decide whether a cached result still describes this input. + + Display settings are absent on purpose: they are a property of the viewer, not of + the measurement, and the record keeps enough to serve a differently-thresholded + view (see the module docstring). + """ + + recording: str + algorithm: str + sensor_config: str + calibration: str + reference_frame: int | None + + @property + def components(self) -> dict[str, str]: + return { + "recording": self.recording, + "algorithm": self.algorithm, + "sensor_config": self.sensor_config, + "calibration": self.calibration, + "reference_frame": "" + if self.reference_frame is None + else str(self.reference_frame), + } + + @property + def digest(self) -> str: + """The directory name. Short because it is a name, not a security boundary.""" + return canonical_digest8(self.components).hex() + + +def result_key( + *, + recording: str, + sensor_config: str, + calibration: str, + reference_frame: int | None = None, + algorithm: str = ALGORITHM_VERSION, +) -> ResultKey: + """Build the invalidation key. A changed argument is a different key, by construction.""" + return ResultKey( + recording=str(recording), + algorithm=str(algorithm), + sensor_config=str(sensor_config), + calibration=str(calibration), + reference_frame=None if reference_frame is None else int(reference_frame), + ) + + +def recording_identity( + session: Mapping[str, Any], segments: Sequence[tuple[str, int]] +) -> str: + """Identity of one recorded session, from its own metadata and segment sizes. + + Deliberately not the directory path: a copied recording directory is the same + recording, and a re-used path is not. Segment *sizes* are included so a truncated + or extended ``raw_*.mp4`` cannot inherit an earlier result; modification times are + not, because copying a tree rewrites them. + """ + device = dict(session.get("device") or {}) + stats = dict(session.get("stats") or {}) + payload = { + "session_id": session.get("session_id"), + "started_at": session.get("started_at"), + "ended_at": session.get("ended_at"), + "sensor_type": device.get("sensor_type"), + "config_ref": device.get("config_ref"), + "config_sha256": device.get("config_sha256"), + "frames": stats.get("frames"), + "video_frames": stats.get("video_frames"), + "segments": stats.get("segments"), + "files": sorted((str(name), int(size)) for name, size in segments), + } + return canonical_digest8(payload).hex() + + +# ------------------------------------------------------------------- frame encoding +def encode_frame( + *, + frame_id: int, + t_rel_s: float, + grid_id: int, + dx: np.ndarray, + dy: np.ndarray, + magnitude: np.ndarray, + active: np.ndarray, + validity_digest: bytes | None = None, + depth_nonzero: int = 0, + dropped: bool = False, +) -> bytes: + """One frame record: a fixed header, then the value arrays. + + Arrays are little-endian regardless of host order, so a result written on one + machine reads back identically on another. ``dx``/``dy``/``magnitude`` are stored + as float32 — the precision the display needs and the reason a stored payload is + compared to a live selection *cast to float32* rather than in float64. + """ + dx = np.ascontiguousarray(dx, dtype=" 0xFFFF: + raise ValueError(f"{count} candidates do not fit the frame header") + flags = 0 + if validity_digest is not None: + flags |= FRAME_FLAG_VALIDITY + if dropped: + flags |= FRAME_FLAG_DROPPED + parts = [ + FRAME_HEADER.pack( + int(frame_id), + float(t_rel_s), + int(grid_id), + flags, + count, + max(0, int(depth_nonzero)), + ), + dx.tobytes(), + dy.tobytes(), + magnitude.tobytes(), + np.packbits(active).tobytes(), + ] + if validity_digest is not None: + digest = bytes(validity_digest) + if len(digest) != 8: + raise ValueError(f"validity digest must be 8 bytes, got {len(digest)}") + parts.append(digest) + return b"".join(parts) + + +@dataclass(frozen=True) +class DecodedFrame: + """The raw fields of one frame record, before the grid keyframe is applied.""" + + frame_id: int + t_rel_s: float + grid_id: int + count: int + depth_nonzero: int + dropped: bool + dx: np.ndarray + dy: np.ndarray + magnitude: np.ndarray + active: np.ndarray + validity_digest: bytes | None + + +def decode_frame( + buffer: bytes | bytearray | memoryview, offset: int = 0 +) -> DecodedFrame: + """Decode one frame record starting at ``offset``. Ignores any trailing bytes.""" + view = memoryview(buffer) + frame_id, t_rel, grid_id, flags, count, depth_nonzero = FRAME_HEADER.unpack_from( + view, offset + ) + cursor = offset + FRAME_HEADER.size + arrays = [] + for _ in range(3): + end = cursor + count * 4 + arrays.append(np.frombuffer(view[cursor:end], dtype=" int: + """Stored bytes of one frame record at ``candidates`` sample locations.""" + total = FRAME_HEADER.size + int(candidates) * 12 + (int(candidates) + 7) // 8 + return total + 8 if with_validity else total + + +def keyframe_payload(kind: int, array: np.ndarray) -> bytes: + """The uncompressed payload of a grid or mask keyframe. + + A grid is ``rows`` (int16) followed by ``cols`` (int16); a mask is bit-packed + along its flattened order. Both are the exact bytes the digest is taken over, so + a reader can verify a keyframe before trusting it. + """ + if kind == KEYFRAME_GRID: + return array.astype(" np.ndarray: + """Bit-packed mask payload back to a ``(height, width)`` boolean array.""" + count = int(height) * int(width) + bits = np.unpackbits(np.frombuffer(payload, dtype=np.uint8), count=count) + return bits.astype(bool).reshape(int(height), int(width)) + + +# ------------------------------------------------------------------- decoded frames +@dataclass(frozen=True) +class FramePayload: + """One analysed frame, as the reader serves it. + + ``rows``/``cols`` come from the frame's grid keyframe, so they are identical for + every frame that shares one; ``dx``/``dy``/``magnitude``/``active`` are this + frame's own values. Arrays are float32 and int32 because that is what is stored — + a comparison against a live selection must cast to float32 to be exact. + """ + + frame_id: int + t_rel_s: float + grid_id: int + rows: np.ndarray + cols: np.ndarray + dx: np.ndarray + dy: np.ndarray + magnitude: np.ndarray + active: np.ndarray + validity_digest: bytes | None = None + depth_nonzero: int = 0 + dropped: bool = False + + @property + def candidates(self) -> int: + return int(self.rows.size) + + @property + def active_count(self) -> int: + return int(np.count_nonzero(self.active)) + + def samples(self, *, threshold: float | None = None) -> FlowSamples: + """This frame as the shared selector's own type. + + ``threshold=None`` returns the activity the pass recorded. Any other value + re-derives it from ``magnitude``, which is what lets a display-settings change + be served without invalidating the result. + """ + active = self.active + if threshold is not None: + active = np.asarray(self.magnitude >= float(threshold), dtype=bool) + return FlowSamples( + rows=self.rows.copy(), + cols=self.cols.copy(), + dx=self.dx.copy(), + dy=self.dy.copy(), + magnitude=self.magnitude.copy(), + active=active.copy(), + ) + + def canonical_bytes(self) -> bytes: + """A byte image of the payload, for exact comparison between two results.""" + return encode_frame( + frame_id=self.frame_id, + t_rel_s=self.t_rel_s, + grid_id=self.grid_id, + dx=self.dx, + dy=self.dy, + magnitude=self.magnitude, + active=self.active, + validity_digest=self.validity_digest, + depth_nonzero=self.depth_nonzero, + dropped=self.dropped, + ) + + +@dataclass(frozen=True) +class KeyframeRef: + """Inventory entry for one keyframe blob inside ``keyframes.bin``.""" + + id: int + kind: str + digest: str + shape: tuple[int, ...] + codec: str + raw_bytes: int + stored_bytes: int + offset: int + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "kind": self.kind, + "digest": self.digest, + "shape": list(self.shape), + "codec": self.codec, + "raw_bytes": self.raw_bytes, + "stored_bytes": self.stored_bytes, + "offset": self.offset, + } + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> KeyframeRef: + return cls( + id=int(data["id"]), + kind=str(data["kind"]), + digest=str(data["digest"]), + shape=tuple(int(v) for v in data.get("shape") or ()), + codec=str(data["codec"]), + raw_bytes=int(data["raw_bytes"]), + stored_bytes=int(data["stored_bytes"]), + offset=int(data["offset"]), + ) + + +# ------------------------------------------------------------------- provenance +@dataclass +class Provenance: + """Everything needed to say what a stored result is and what it is not. + + ``complete`` is duplicated from the result's top-level metadata on purpose: one is + the gate a reader checks, the other travels with the record. They are written from + a single source and compared on read, so they cannot disagree silently. + """ + + format_version: int = ANALYSIS_FORMAT_VERSION + algorithm: str = ALGORITHM_VERSION + unit: str = UNIT + key: str = "" + key_components: dict[str, str] = field(default_factory=dict) + recording: str = "" + recording_identity: str = "" + session_id: int = 0 + sensor_type: str = "" + chain: str = "" + sensor_config_ref: str = "" + sensor_config_sha256: str = "" + calibration_id: str = "" + #: The frame whose processing started from a clean tracker reference. Equal to + #: ``first_output_frame_id`` when no explicit reset was requested. + reference_frame_id: int | None = None + reference_is_explicit: bool = False + first_output_frame_id: int | None = None + last_output_frame_id: int | None = None + #: Decoded frames the chain consumed *before* the first frame a caller can read. + #: Measured per segment; ``None`` when the checked segments disagree. + decode_order_offset: int | None = None + decode_order_offsets: dict[str, int] = field(default_factory=dict) + #: ``measured`` when the offset came from the capture's own decode position, + #: ``declared`` when the backend could not report one and the chain's constant was + #: used instead. A result built on a declared offset is weaker evidence. + decode_order_offset_source: str = "measured" + segments: list[int] = field(default_factory=list) + segments_declared: int = 0 + segments_missing: list[int] = field(default_factory=list) + #: The selection policy used while recording samples. Not part of the key. + policy: dict[str, Any] = field(default_factory=dict) + #: Per channel: ``recomputed`` (re-run from the video) or ``recorded`` (read from a + #: CSV the recorder wrote). No channel is both. + sources: dict[str, str] = field(default_factory=dict) + versions: dict[str, str] = field(default_factory=dict) + config: dict[str, Any] = field(default_factory=dict) + open_working_directory: str = "" + debug_files_written: list[str] = field(default_factory=list) + started_at: str = "" + finished_at: str = "" + elapsed_s: float = 0.0 + #: Wall-clock cost of opening the SDK chain, summed over segments. It is separate + #: from ``elapsed_s`` because a box open spends 1.6 s before the first frame and + #: that time is not throughput. + open_seconds: float = 0.0 + warnings: list[str] = field(default_factory=list) + complete: bool = False + stopped_reason: str = "" + counts: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return dict(self.__dict__) + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> Provenance: + known = {f for f in cls().__dict__} + return cls(**{key: value for key, value in dict(data).items() if key in known}) diff --git a/shared/src/orisys_app_shared/analysis/reader.py b/shared/src/orisys_app_shared/analysis/reader.py new file mode 100644 index 0000000..a4a7ef9 --- /dev/null +++ b/shared/src/orisys_app_shared/analysis/reader.py @@ -0,0 +1,567 @@ +"""Reading a stored result: by frame id, by timestamp, and nothing else. + +A seek must not re-run the pass. Two levels of index make that true: + +* ``meta.json`` lists the chunks in order with their first and last frame id, so a + frame id picks a chunk by binary search; +* each chunk ends with a trailer holding the frame's own byte offset, frame id and + timestamp, so a position inside that chunk is another binary search and one read. + +The trailer is read the first time a chunk is needed and then cached, which costs +20 bytes per frame of memory — 360 KB for a ten-minute recording at 30 fps. Everything +else (the sample grid, the validity mask) is loaded from its keyframe on demand and +verified against the digest recorded for it. +""" + +from __future__ import annotations + +import json +import logging +import zlib +from collections import OrderedDict +from collections.abc import Iterator +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import numpy as np + +from orisys_app_shared.analysis.cache import INCOMPLETE_MARKER, META_NAME +from orisys_app_shared.analysis.model import ( + ANALYSIS_FORMAT, + ANALYSIS_FORMAT_VERSION, + CHUNK_HEADER, + CHUNK_MAGIC, + CHUNK_TRAILER, + CODEC_RAW, + CODEC_ZLIB, + KEYFRAME_GRID, + KEYFRAME_HEADER, + KEYFRAME_MAGIC, + KEYFRAME_MASK, + FramePayload, + KeyframeRef, + Provenance, + decode_frame, + digest8, + unpack_mask, +) +from orisys_app_shared.scene3d.visual_policy import FlowSamples + +logger = logging.getLogger(__name__) + +__all__ = [ + "AnalysisFormatError", + "AnalysisIncompleteError", + "AnalysisNotFoundError", + "AnalysisReader", + "AnalysisResultError", +] + +#: How many keyframe payloads to keep decoded. One grid and one mask is the norm; the +#: box chain needs more, and re-reading a 1.3 KB blob is cheap when it falls out. +_GRID_CACHE = 16 + + +class AnalysisResultError(RuntimeError): + """Base for every reason a stored result cannot be served.""" + + +class AnalysisNotFoundError(AnalysisResultError): + """There is no result at that path.""" + + +class AnalysisIncompleteError(AnalysisResultError): + """The result exists but was cancelled, failed, or never finished. + + Raised by default rather than served with a warning: a partial result that looks + finished is worse than no result, because every chart drawn from it would be + missing an unknown number of frames without saying so. + """ + + +class AnalysisFormatError(AnalysisResultError): + """The result is not in a format this build can read, or it is corrupt.""" + + +@dataclass +class _ChunkIndex: + """One chunk's trailer, loaded on first use.""" + + index: int + path: Path + offsets: np.ndarray # uint64, per frame, relative to the file start + frame_ids: np.ndarray # int32 + times: np.ndarray # float64 + sizes: np.ndarray # uint64, per frame record length + first_frame_id: int + last_frame_id: int + + +class AnalysisReader: + """Random access into one analysis result directory.""" + + def __init__( + self, directory: str | Path, *, allow_incomplete: bool = False + ) -> None: + self.directory = Path(directory) + self.allow_incomplete = bool(allow_incomplete) + meta_path = self.directory / META_NAME + if not meta_path.is_file(): + raise AnalysisNotFoundError(f"no analysis result at {self.directory}") + try: + meta = json.loads(meta_path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise AnalysisFormatError(f"{meta_path} is unreadable: {exc}") from exc + if str(meta.get("format") or "") != ANALYSIS_FORMAT: + raise AnalysisFormatError( + f"{meta_path} is not an {ANALYSIS_FORMAT} result " + f"(format={meta.get('format')!r})" + ) + version = int(meta.get("format_version") or 0) + if version != ANALYSIS_FORMAT_VERSION: + raise AnalysisFormatError( + f"result format version {version} cannot be read by this build " + f"(it reads {ANALYSIS_FORMAT_VERSION})" + ) + self.meta: dict[str, Any] = dict(meta) + self.key: str = str(meta.get("key") or self.directory.name) + self.key_components: dict[str, str] = dict(meta.get("key_components") or {}) + self.provenance = Provenance.from_dict(meta.get("provenance") or {}) + marker = self.directory / INCOMPLETE_MARKER + declared = bool(meta.get("complete")) + if declared != bool(self.provenance.complete): + raise AnalysisFormatError( + f"{meta_path} disagrees with its own provenance about completeness" + ) + if marker.exists() or not declared: + if not self.allow_incomplete: + reason = self.provenance.stopped_reason or "unfinished" + raise AnalysisIncompleteError( + f"{self.directory} is not a complete result ({reason}); " + "pass allow_incomplete=True only to inspect a partial one" + ) + self.complete = False + else: + self.complete = True + self.counts: dict[str, Any] = dict(meta.get("counts") or {}) + self.keyframes: dict[int, KeyframeRef] = { + item.id: item + for item in ( + KeyframeRef.from_dict(entry) for entry in meta.get("keyframes") or [] + ) + } + self.chunks: list[dict[str, Any]] = [ + dict(entry) for entry in meta.get("chunks") or [] + ] + self._chunk_index: dict[int, _ChunkIndex] = {} + self._mask_cache: np.ndarray | None = None + self._mask_loaded = False + self._grid_cache: OrderedDict[int, tuple[np.ndarray, np.ndarray]] = ( + OrderedDict() + ) + self._handle = None + self._handle_chunk = -1 + + # ------------------------------------------------------------------- lifecycle + def close(self) -> None: + if self._handle is not None: + try: + self._handle.close() + finally: + self._handle = None + self._handle_chunk = -1 + + def __enter__(self) -> AnalysisReader: + return self + + def __exit__(self, exc_type, exc, tb) -> bool: + self.close() + return False + + # -------------------------------------------------------------------- indexing + def __len__(self) -> int: + return int(sum(int(entry.get("frames") or 0) for entry in self.chunks)) + + @property + def frame_ids(self) -> np.ndarray: + """Every stored frame id, in stored order.""" + parts = [self._chunk(chunk["index"]).frame_ids for chunk in self.chunks] + if not parts: + return np.empty(0, dtype=np.int64) + return np.concatenate(parts).astype(np.int64) + + @property + def times(self) -> np.ndarray: + """Every stored ``t_rel_s``, in stored order.""" + parts = [self._chunk(chunk["index"]).times for chunk in self.chunks] + if not parts: + return np.empty(0, dtype=np.float64) + return np.concatenate(parts) + + @property + def frame_record_bytes(self) -> int: + """Stored bytes of the frame records alone. + + Excludes the chunk offset tables, the keyframes and the metadata, so a caller can + compare it with :func:`~orisys_app_shared.analysis.model.frame_payload_bytes` + instead of guessing what the overhead was. + """ + return int( + sum(int(self._chunk(entry["index"]).sizes.sum()) for entry in self.chunks) + ) + + @property + def bytes_per_frame(self) -> float: + """Frame-record bytes divided by frames — the number to quote per frame.""" + count = len(self) + return 0.0 if not count else self.frame_record_bytes / count + + @property + def keyframe_counts(self) -> dict[str, int]: + """How many keyframes of each kind the result holds.""" + counts: dict[str, int] = {} + for reference in self.keyframes.values(): + counts[reference.kind] = counts.get(reference.kind, 0) + 1 + return counts + + def _chunk(self, index: int) -> _ChunkIndex: + cached = self._chunk_index.get(index) + if cached is not None: + return cached + entry = next( + (item for item in self.chunks if int(item.get("index", -1)) == int(index)), + None, + ) + if entry is None: + raise AnalysisFormatError(f"chunk {index} is not in the result inventory") + path = self.directory / str(entry["file"]) + if not path.is_file(): + raise AnalysisFormatError(f"chunk file {path} is missing") + data = path.read_bytes() + if len(data) < CHUNK_HEADER.size + CHUNK_TRAILER.size: + raise AnalysisFormatError(f"{path} is shorter than a chunk header") + magic, version, _reserved, chunk_index, count, header_size = ( + CHUNK_HEADER.unpack_from(data) + ) + if magic != CHUNK_MAGIC: + raise AnalysisFormatError(f"{path} is not a frame chunk") + if version != ANALYSIS_FORMAT_VERSION or int(chunk_index) != int(index): + raise AnalysisFormatError( + f"{path} declares version {version} chunk {chunk_index}, expected " + f"{ANALYSIS_FORMAT_VERSION} and {index}" + ) + declared = int(entry.get("frames") or 0) + if declared != int(count): + raise AnalysisFormatError( + f"{path} holds {count} frames, the inventory says {declared}" + ) + table_offset, trailer_count, trailer_magic = CHUNK_TRAILER.unpack_from( + data, len(data) - CHUNK_TRAILER.size + ) + if trailer_magic != CHUNK_MAGIC or int(trailer_count) != int(count): + raise AnalysisFormatError(f"{path} has a damaged trailer") + offsets = np.frombuffer( + data, dtype=" int(table_offset): + raise AnalysisFormatError( + f"{path} declares a header past its own frame data" + ) + sizes = np.empty(count, dtype=np.uint64) + if count: + sizes[:-1] = offsets[1:] - offsets[:-1] + sizes[-1] = np.uint64(int(table_offset) - int(offsets[-1])) + record = _ChunkIndex( + index=int(index), + path=path, + offsets=offsets, + frame_ids=frame_ids, + times=times, + sizes=sizes, + first_frame_id=int(frame_ids[0]) if count else 0, + last_frame_id=int(frame_ids[-1]) if count else 0, + ) + self._chunk_index[int(index)] = record + return record + + def _chunk_for_frame(self, frame_id: int) -> _ChunkIndex | None: + """The chunk holding ``frame_id``. + + The inventory is ordered by frame id for every result this code writes, so the + binary search answers in ``log(chunks)``. A hand-edited or foreign result whose + ranges are not monotonic falls back to a scan rather than answering with the + wrong chunk. + """ + low, high = 0, len(self.chunks) - 1 + while low <= high: + middle = (low + high) // 2 + record = self._chunk(int(self.chunks[middle]["index"])) + if not len(record.frame_ids): + low = middle + 1 + continue + if frame_id < record.first_frame_id: + high = middle - 1 + elif frame_id > record.last_frame_id: + low = middle + 1 + else: + return record + for entry in self.chunks: + record = self._chunk(int(entry["index"])) + if len(record.frame_ids) and bool( + np.any(record.frame_ids == int(frame_id)) + ): + return record + return None + + # ------------------------------------------------------------------ keyframes + def _keyframe_payload(self, reference: KeyframeRef) -> bytes: + path = self.directory / "keyframes.bin" + if not path.is_file(): + raise AnalysisFormatError("keyframes.bin is missing from the result") + with path.open("rb") as handle: + handle.seek(reference.offset) + header = handle.read(KEYFRAME_HEADER.size) + if len(header) != KEYFRAME_HEADER.size: + raise AnalysisFormatError(f"keyframe {reference.id} is truncated") + ( + magic, + kind, + codec, + count, + height, + width, + _flags, + _reserved, + raw_bytes, + digest, + ) = KEYFRAME_HEADER.unpack(header) + if magic != KEYFRAME_MAGIC: + raise AnalysisFormatError(f"keyframe {reference.id} has no magic") + stored = handle.read(reference.stored_bytes) + if len(stored) != reference.stored_bytes: + raise AnalysisFormatError(f"keyframe {reference.id} is truncated") + payload = zlib.decompress(stored) if codec == CODEC_ZLIB else stored + if len(payload) != int(raw_bytes): + raise AnalysisFormatError( + f"keyframe {reference.id} decoded to {len(payload)} bytes, " + f"its header says {raw_bytes}" + ) + if digest8(payload).hex() != reference.digest: + raise AnalysisFormatError( + f"keyframe {reference.id} does not match its recorded digest" + ) + expected_kind = {KEYFRAME_GRID: "grid", KEYFRAME_MASK: "mask"}.get(kind) + if expected_kind != reference.kind: + raise AnalysisFormatError( + f"keyframe {reference.id} is a {expected_kind}, the inventory says " + f"{reference.kind}" + ) + if codec not in (CODEC_RAW, CODEC_ZLIB): + raise AnalysisFormatError(f"keyframe {reference.id} uses codec {codec}") + declared = (int(height), int(width)) if kind == KEYFRAME_MASK else (int(count),) + if reference.kind == "grid" and declared != reference.shape: + raise AnalysisFormatError( + f"keyframe {reference.id} shape disagrees with it" + ) + return payload + + def grid(self, grid_id: int) -> tuple[np.ndarray, np.ndarray]: + """The candidate locations of one grid keyframe, as ``(rows, cols)`` int32.""" + cached = self._grid_cache.get(grid_id) + if cached is not None: + self._grid_cache.move_to_end(grid_id) + return cached + reference = self.keyframes.get(int(grid_id)) + if reference is None or reference.kind != "grid": + raise AnalysisFormatError(f"no grid keyframe {grid_id} in this result") + payload = self._keyframe_payload(reference) + count = int(reference.shape[0]) if reference.shape else 0 + if len(payload) != count * 4: + raise AnalysisFormatError( + f"grid keyframe {grid_id} holds {len(payload)} bytes for {count} locations" + ) + rows = np.frombuffer(payload, dtype=" _GRID_CACHE: + self._grid_cache.popitem(last=False) + return rows, cols + + def mask(self) -> np.ndarray | None: + """The validity mask this result was built with, or ``None`` when there was none.""" + if self._mask_loaded: + return self._mask_cache + self._mask_loaded = True + self._mask_cache = None + for reference in self.keyframes.values(): + if reference.kind != "mask": + continue + payload = self._keyframe_payload(reference) + height = int(reference.shape[0]) if len(reference.shape) > 0 else 0 + width = int(reference.shape[1]) if len(reference.shape) > 1 else 0 + self._mask_cache = unpack_mask(payload, height, width) + break + return self._mask_cache + + @property + def mask_digest(self) -> str | None: + for reference in self.keyframes.values(): + if reference.kind == "mask": + return reference.digest + return None + + # ----------------------------------------------------------------- frame access + def _read_record(self, chunk: _ChunkIndex, slot: int) -> bytes: + if self._handle_chunk != chunk.index or self._handle is None: + if self._handle is not None: + self._handle.close() + self._handle = chunk.path.open("rb") + self._handle_chunk = chunk.index + self._handle.seek(int(chunk.offsets[slot])) + return self._handle.read(int(chunk.sizes[slot])) + + def _slot_for(self, chunk: _ChunkIndex, frame_id: int) -> int: + ids = chunk.frame_ids + if ids.size and bool(np.all(np.diff(ids) > 0)): + return int(np.searchsorted(ids, int(frame_id))) + hits = np.flatnonzero(ids == int(frame_id)) + if not hits.size: + raise AnalysisNotFoundError( + f"frame {frame_id} is not in chunk {chunk.path.name}" + ) + return int(hits[0]) + + def has(self, frame_id: int) -> bool: + return self._chunk_for_frame(int(frame_id)) is not None + + def frame(self, frame_id: int) -> FramePayload: + """One frame by its recorded id. Random access; no replay of the pass.""" + chunk = self._chunk_for_frame(int(frame_id)) + if chunk is None: + raise AnalysisNotFoundError(f"frame {frame_id} is not in this result") + slot = self._slot_for(chunk, int(frame_id)) + decoded = decode_frame(self._read_record(chunk, slot)) + if decoded.frame_id != int(frame_id): + raise AnalysisFormatError( + f"frame {frame_id} resolved to a record for {decoded.frame_id}" + ) + rows, cols = self.grid(decoded.grid_id) + if rows.size != decoded.count: + raise AnalysisFormatError( + f"frame {frame_id} has {decoded.count} values but grid " + f"{decoded.grid_id} has {rows.size} locations" + ) + return FramePayload( + frame_id=decoded.frame_id, + t_rel_s=decoded.t_rel_s, + grid_id=decoded.grid_id, + rows=rows, + cols=cols, + dx=decoded.dx, + dy=decoded.dy, + magnitude=decoded.magnitude, + active=decoded.active, + validity_digest=decoded.validity_digest, + depth_nonzero=decoded.depth_nonzero, + dropped=decoded.dropped, + ) + + def frame_at(self, t_rel_s: float) -> FramePayload | None: + """The stored frame at or just before ``t_rel_s``, or ``None`` for an empty result. + + Uses the recording's own time base, which is *signal* time: the container + duration and the recorded duration are different numbers and neither is + derived from the other. + """ + times = self.times + if times.size == 0: + return None + if bool(np.all(np.diff(times) >= 0)): + slot = int(np.searchsorted(times, float(t_rel_s), side="right")) - 1 + slot = max(0, min(slot, times.size - 1)) + else: + candidates = np.flatnonzero(times <= float(t_rel_s)) + slot = int(candidates[-1]) if candidates.size else 0 + ids = self.frame_ids + return self.frame(int(ids[slot])) + + def samples(self, frame_id: int, *, policy: Any | None = None) -> FlowSamples: + """The frame's flow samples, as the shared selector would return them. + + A ``policy`` is honoured only for its activation threshold: the candidate + *locations* are the ones recorded, because those are what the pass selected. + Change the grid and this result no longer describes the view — compare + :attr:`provenance.policy` rather than expecting the reader to re-grid. + """ + threshold = None if policy is None else float(policy.normalised().flow_on_px) + return self.frame(int(frame_id)).samples(threshold=threshold) + + def frames( + self, *, start: int | None = None, stop: int | None = None + ) -> Iterator[FramePayload]: + """Frames in stored order, optionally bounded by frame id (both inclusive).""" + for entry in self.chunks: + chunk = self._chunk(int(entry["index"])) + for slot in range(len(chunk.frame_ids)): + frame_id = int(chunk.frame_ids[slot]) + if start is not None and frame_id < int(start): + continue + if stop is not None and frame_id > int(stop): + continue + yield self.frame(frame_id) + + # ------------------------------------------------------------------- freshness + def stale_reasons( + self, + *, + sensor_config: str = "", + algorithm: str = "", + calibration: str = "", + sdk_version: str = "", + ) -> list[str]: + """Why this result may no longer describe what would be computed now. + + The key covers the recording, the algorithm, the sensor configuration, the + calibration and the reference frame. The SDK version is deliberately **not** in + the key — a patch release does not change the key, so it would otherwise be + served silently. It is reported here instead, next to the caller's other + warnings, exactly as the replay already reports a changed recorded config. + """ + reasons: list[str] = [] + recorded = self.key_components + if ( + sensor_config + and recorded.get("sensor_config") + and sensor_config != recorded["sensor_config"] + ): + reasons.append("sensor_config_changed") + if ( + algorithm + and recorded.get("algorithm") + and algorithm != recorded["algorithm"] + ): + reasons.append("algorithm_changed") + if ( + calibration + and recorded.get("calibration") + and calibration != recorded["calibration"] + ): + reasons.append("calibration_changed") + if sdk_version: + stored = str(self.provenance.versions.get("sdk") or "") + if stored and stored != sdk_version: + reasons.append("sdk_changed") + if not self.complete: + reasons.append("incomplete") + return reasons diff --git a/shared/src/orisys_app_shared/analysis/service.py b/shared/src/orisys_app_shared/analysis/service.py new file mode 100644 index 0000000..7fe1f8d --- /dev/null +++ b/shared/src/orisys_app_shared/analysis/service.py @@ -0,0 +1,1067 @@ +"""The offline analysis pass: drive the SDK over a recording and store the samples. + +The pass is the only part of this package that touches the SDK, and it drives it the +way the Step 0 measurement did — ``FingerRuntime.open(video=…)`` → ``read_source()`` → +``process_frame()`` → :func:`~orisys_app_shared.acquisition.sdk_worker.process_chain`, +which is the same deformation + contact pair the live acquisition loop and the offline +depth replay call. + +Two traps decide the design, both measured on the real box recording: + +**Frame identity comes from ``frames.csv``, never from decode order.** Constructing a +box ``Sensor`` on a video silently consumes the first 7 decoded frames (33-frame video +→ 26 processed; the first frame a caller can read is video frame #8), while the finger +path consumes 0. The offset is *measured* from the capture's own decode position after +the open rather than assumed, and it is recorded in the provenance so a reader can see +which frames the pass could not have produced. Frame id, timestamp, segment and +``in_video`` all come from the recording's own frame table. + +**The process working directory is not a scratch space.** Opening a box ``Sensor`` +writes a ~2 MB ``config/compound_eye.png`` relative to the current directory, so every +chain open runs inside a temporary working directory under a process-wide lock. That +makes this pass and any *other* user of the process working directory mutually +exclusive for the duration of an open — the one global side effect this design cannot +avoid, because the SDK offers no way to redirect the path (see the report's proposed +SDK change). + +One generator, :meth:`AnalysisService.recompute`, produces every frame both +:meth:`AnalysisService.run` and a verification caller consume, so the stored result and +a fresh re-run cannot drift apart in *how* they drive the chain. + +A frame whose chain call fails ends the pass as an incomplete result rather than being +stored. Anything else would put a stale or empty measurement into a cache that claims +to describe the recording, and every chart drawn from it would inherit the hole +silently. +""" + +from __future__ import annotations + +import contextlib +import hashlib +import importlib +import json +import logging +import os +import shutil +import tempfile +import threading +import time +from collections.abc import Callable, Iterator, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import numpy as np + +from orisys_app_shared.analysis import model as model_module +from orisys_app_shared.analysis.cache import ( + DEFAULT_MAX_BYTES, + AnalysisWriter, + ResultCache, +) +from orisys_app_shared.analysis.model import ( + ALGORITHM_VERSION, + KEYFRAME_MASK, + Provenance, + ResultKey, + digest8, + encode_frame, + keyframe_payload, + recording_identity, + result_key, +) +from orisys_app_shared.record.playback import RecordingFrame, RecordingReader +from orisys_app_shared.scene3d.visual_policy import ( + FlowSamples, + VisualPolicy, + select_flow_samples, +) + +logger = logging.getLogger(__name__) + +__all__ = [ + "AnalysisOutcome", + "AnalysisService", + "ChainSpec", + "LiveFrame", + "Progress", + "chain_spec_for", + "recording_config_sha256", + "resolve_finger_export_dir", +] + +#: Serialises every change of the process working directory. The SDK's box sensor +#: writes its stitch preview relative to the CWD; two passes must not interleave. +_CWD_LOCK = threading.Lock() + +#: Frames the box chain is known to consume at construction when the capture backend +#: cannot report its decode position. Used only as a fallback, and recorded as +#: ``decode_order_offset_source="declared"`` so the weaker evidence is visible. +_DECLARED_OFFSETS = {"box": 7, "finger": 0} + +#: Default frame records per chunk. A chunk is built in memory, so this times the frame +#: record size (about 1.0 MB at the shipped finger grid) is the memory bound. +DEFAULT_CHUNK_FRAMES = 256 + + +def _import_sdk() -> Any: + return importlib.import_module("orisys") + + +def _version_of(module_name: str) -> str: + try: + module = importlib.import_module(module_name) + except Exception: # noqa: BLE001 - a missing version is a fact, not a failure + return "" + return str(getattr(module, "__version__", "") or "") + + +def recording_config_sha256(config_ref: str | Path | None) -> str: + """SHA-256 of the sensor configuration file, or "" when it cannot be read. + + Computed here rather than imported from the Studio app: ``shared`` must not depend + on an application package, and the fact belongs to the file either way. + """ + if not config_ref: + return "" + try: + path = Path(str(config_ref)) + if path.is_file(): + return hashlib.sha256(path.read_bytes()).hexdigest() + except OSError: + # An unreadable config must not stop the pass; the empty hash is recorded, and + # the key then differs from a pass that could read it — the safe direction. + logger.debug("could not hash the sensor config %s", config_ref, exc_info=True) + return "" + + +def resolve_finger_export_dir(config_ref: str | Path) -> Path | None: + """``assets/30fps-0703`` beside a finger ``sensor.json``, when it is complete. + + The SDK's own default points at a path that does not exist in a source checkout, + and its built-in ``finger`` config disagrees with the shipped Studio one + (``source_scale`` 0.4 vs 0.42, ``hhd_downsample`` 2 vs 3). The shipped config is + what the footage was recorded with, so the pass uses the bundle beside it — the + measurement's own rule. + """ + from orisys_app_shared.acquisition.finger_worker import ( + resolve_finger_export_dir as resolve, + ) + + try: + return resolve(config_ref) + except FileNotFoundError: + logger.warning( + "finger export bundle beside %s is incomplete", config_ref, exc_info=True + ) + return None + + +@dataclass(frozen=True) +class ChainSpec: + """Which sensor chain to drive, and the configuration that defines it.""" + + sensor_type: str + config_ref: Path | None = None + export_dir: Path | None = None + flip_source_vertical: bool = True + mirror: bool = False + + def __post_init__(self) -> None: + if self.sensor_type not in ("finger", "box"): + raise ValueError(f"unsupported sensor type {self.sensor_type!r}") + + @property + def config_sha256(self) -> str: + """Content hash of the sensor configuration, or "" when it cannot be read.""" + return recording_config_sha256(self.config_ref) + + def identity(self) -> str: + """The sensor-configuration half of the invalidation key. + + The file *content* is the identity, because that is what decides the chain: two + installs whose ``sensor.json`` differ compute different numbers even when the + path is the same. The export bundle is named too — the finger's UV map changes + the result without touching ``sensor.json``. + """ + payload = { + "sensor_type": self.sensor_type, + "config_sha256": self.config_sha256, + "export_dir": "" if self.export_dir is None else str(self.export_dir), + "flip_source_vertical": self.flip_source_vertical, + "mirror": self.mirror, + } + return digest8(repr(sorted(payload.items())).encode("utf-8")).hex() + + +def chain_spec_for( + session: dict[str, Any], + *, + sensor_type: str = "", + config_ref: str | Path | None = None, + export_dir: str | Path | None = None, +) -> ChainSpec: + """The chain a recording was made with, unless the caller names a different one. + + Defaulting from ``session.json`` is what makes a stored result reproducible: the + recorded ``device.config_ref`` is the configuration the footage was taken under, + not whatever happens to be installed now. + """ + device = dict(session.get("device") or {}) + kind = str(sensor_type or device.get("sensor_type") or "") + if kind not in ("finger", "box"): + raise ValueError( + f"the recording does not name a supported sensor type ({kind!r}); " + "pass one explicitly" + ) + reference = config_ref if config_ref is not None else device.get("config_ref") + resolved = Path(str(reference)) if reference else None + bundle: Path | None = None + if export_dir is not None: + bundle = Path(str(export_dir)) + elif kind == "finger" and resolved is not None: + bundle = resolve_finger_export_dir(resolved) + return ChainSpec(sensor_type=kind, config_ref=resolved, export_dir=bundle) + + +@contextlib.contextmanager +def _working_directory(path: Path) -> Iterator[Path]: + """Own the process working directory for the duration of an SDK open.""" + with _CWD_LOCK: + previous = Path.cwd() + path.mkdir(parents=True, exist_ok=True) + os.chdir(path) + try: + yield path + finally: + os.chdir(previous) + + +def _capture_position(capture: Any) -> int | None: + """Frames the capture has already handed out, or ``None`` when it cannot say. + + This is the measurement that makes the decode-order offset a fact instead of a + constant: a freshly opened file capture reports 0, and a box ``Sensor`` + construction leaves it at 7 because the stitch probe read that many frames. + """ + if capture is None: + return None + try: + import cv2 + + number = float(capture.get(cv2.CAP_PROP_POS_FRAMES)) + except Exception: # noqa: BLE001 - a backend that cannot answer is not an error + return None + if number != number or number < 0: + return None + return int(round(number)) + + +class _OpenChain: + """One open SDK chain over one segment's video file.""" + + def __init__(self, spec: ChainSpec, video: Path, working: Path) -> None: + self.spec = spec + self.video = Path(video) + self.working = Path(working) + self.sensor: Any = None + self.open_seconds = 0.0 + self.decode_position: int | None = None + self.debug_files: list[str] = [] + self._runtime: Any = None + self._read: Callable[[], tuple[bool, Any]] | None = None + self._feed: Callable[[Any], Any] | None = None + + def open(self) -> _OpenChain: + started = time.perf_counter() + with _working_directory(self.working): + if self.spec.sensor_type == "finger": + from orisys.finger import FingerRuntime + + kwargs: dict[str, Any] = { + "video": str(self.video), + "verbose": False, + "flip_source_vertical": bool(self.spec.flip_source_vertical), + "mirror": bool(self.spec.mirror), + } + if self.spec.config_ref is not None: + kwargs["config"] = str(self.spec.config_ref) + if self.spec.export_dir is not None: + kwargs["export_dir"] = str(self.spec.export_dir) + self._runtime = FingerRuntime.open(**kwargs) + self.sensor = self._runtime.tactile + self._read = self._runtime.read_source + self._feed = self._runtime.process_frame + capture = getattr(self._runtime, "capture", None) + else: + sensor = _import_sdk().Sensor( + str(self.video), + config_name=str(self.spec.config_ref) + if self.spec.config_ref + else "box", + verbose=False, + backend="auto", + ) + self.sensor = sensor + raw_read = getattr(sensor, "_read_frame", None) + if not callable(raw_read): + raise RuntimeError( + "this SDK's Sensor has no ``_read_frame``; the offline pass needs " + "a public per-frame read in external-frame mode" + ) + self._read = raw_read + self._feed = lambda frame: sensor.get_img(frame=frame) + capture = getattr(sensor, "cap", None) + self.open_seconds = time.perf_counter() - started + self.decode_position = _capture_position(capture) + self.debug_files = sorted( + str(path.relative_to(self.working)) + for path in self.working.rglob("*") + if path.is_file() + ) + return self + + def read(self) -> tuple[bool, Any]: + if self._read is None: + return False, None + return self._read() + + def feed(self, frame: Any) -> Any: + if self._feed is None: + return None + return self._feed(frame) + + def close(self) -> None: + try: + if self._runtime is not None: + self._runtime.close() + elif self.sensor is not None: + self.sensor.disconnect() + except Exception: + # Teardown must not mask the pass result; the failure is reported instead. + logger.warning( + "could not close the %s chain", self.spec.sensor_type, exc_info=True + ) + finally: + self._runtime = None + self.sensor = None + + def __enter__(self) -> _OpenChain: + return self.open() + + def __exit__(self, exc_type, exc, tb) -> bool: + self.close() + return False + + +@dataclass(frozen=True) +class LiveFrame: + """One frame exactly as the chain computes it now. + + This is the reference a stored result is checked against: the same sensor, the same + selection policy, the same frames in the same order. It is *not* a playback path — + replaying from it would cost the 0.48 s of wall clock per video second that the + stored result exists to avoid. + """ + + row: RecordingFrame + samples: FlowSamples + mask: np.ndarray | None + depth_nonzero: int + depth_abs_sum: float + reset_applied: bool + segment: int + + def payload_matches(self, stored: Any) -> bool: + """True when a stored payload equals this frame's selection, in stored precision. + + ``dx``/``dy``/``magnitude`` are compared as float32 because that is how they are + stored; the rest is exact. + """ + live = self.samples + return bool( + np.array_equal( + np.asarray(stored.rows, np.int32), np.asarray(live.rows, np.int32) + ) + and np.array_equal( + np.asarray(stored.cols, np.int32), np.asarray(live.cols, np.int32) + ) + and np.array_equal( + np.asarray(stored.dx, np.float32), np.asarray(live.dx, np.float32) + ) + and np.array_equal( + np.asarray(stored.dy, np.float32), np.asarray(live.dy, np.float32) + ) + and np.array_equal( + np.asarray(stored.magnitude, np.float32), + np.asarray(live.magnitude, np.float32), + ) + and np.array_equal( + np.asarray(stored.active, bool), np.asarray(live.active, bool) + ) + ) + + +@dataclass(frozen=True) +class Progress: + """A running description of the pass, for a caller that wants to show it.""" + + frames_written: int + frames_expected: int + segment_index: int + segment_count: int + frame_id: int + elapsed_s: float + + @property + def fraction(self) -> float: + if self.frames_expected <= 0: + return 0.0 + return min(1.0, self.frames_written / self.frames_expected) + + +@dataclass(frozen=True) +class AnalysisOutcome: + """What a pass did, whether or not it finished.""" + + key: ResultKey + directory: Path + complete: bool + reason: str + frames_written: int + frames_expected: int + elapsed_s: float + provenance: Provenance = field(default_factory=Provenance) + + @property + def result_key(self) -> str: + return self.key.digest + + +@dataclass +class _PassState: + """Bookkeeping shared by the generator, the writer and the provenance.""" + + provenance: Provenance + segments: list[int] + grouped: dict[int, list[RecordingFrame]] + working: Path + plan: dict[str, int] = field( + default_factory=lambda: {"decoded": 0, "unmapped": 0, "skipped": 0} + ) + offsets: dict[str, int] = field(default_factory=dict) + sources: set[str] = field(default_factory=set) + reasons: list[str] = field(default_factory=list) + frames_expected: int = 0 + frames_written: int = 0 + started: float = 0.0 + + +class AnalysisService: + """Run the offline analysis over one recording and store the result. + + One call to :meth:`run` owns one open SDK chain at a time and one writer. It never + leaves a directory that a reader would treat as finished unless every segment was + written and the metadata was committed. + """ + + def __init__( + self, + recording: str | Path, + *, + chain: ChainSpec, + cache: ResultCache | None = None, + policy: VisualPolicy | None = None, + reference_frame: int | None = None, + chunk_frames: int = DEFAULT_CHUNK_FRAMES, + max_bytes: int = DEFAULT_MAX_BYTES, + working_directory: str | Path | None = None, + on_progress: Callable[[Progress], None] | None = None, + cancelled: threading.Event | None = None, + ) -> None: + self.recording = Path(recording) + self.chain = chain + self.cache = cache or ResultCache.for_recording( + self.recording, max_bytes=max_bytes + ) + #: The policy the samples are selected with. Recorded, never keyed (see + #: ``model.py``): the shipped defaults make a stored result independent of the + #: operator's display preferences. + self.policy = policy or VisualPolicy() + self.reference_frame = None if reference_frame is None else int(reference_frame) + self.chunk_frames = int(chunk_frames) + self.working_directory = ( + None if working_directory is None else Path(working_directory) + ) + self.on_progress = on_progress + self.cancelled = cancelled + self._reader: RecordingReader | None = None + + # ------------------------------------------------------------------- recording + def reader(self) -> RecordingReader: + if self._reader is None: + self._reader = RecordingReader(self.recording) + return self._reader + + def segments(self) -> tuple[list[int], list[int]]: + """The ordered segment index, from the recording's own declaration and files. + + ``session.json`` says how many segments there should be and how they are named; + the directory says which ones exist. The disagreement is *returned* rather than + resolved, because a silently shortened segment list produces a result that + claims to describe a recording it never read. + """ + reader = self.reader() + session = reader.session() + declared = int(dict(session.get("stats") or {}).get("segments") or 0) + present: list[int] = [] + for path in reader.segments(): + suffix = path.stem.rsplit("_", 1)[-1] + if suffix.isdigit(): + present.append(int(suffix)) + present = sorted(set(present)) + if not declared: + declared = len(present) + missing = [number for number in range(1, declared + 1) if number not in present] + return present, missing + + def rows_by_segment( + self, frames: Sequence[RecordingFrame] + ) -> dict[int, list[RecordingFrame]]: + """The frames each segment actually holds, in ``segment_frame`` order. + + Only ``in_video`` rows have a video frame — a row written by ``write_gap`` has + ``in_video=0`` and exists precisely to record a hole — so this list *is* the + decode order of the segment once the offset is applied. + """ + grouped: dict[int, list[RecordingFrame]] = {} + for frame in frames: + if not frame.in_video or frame.segment <= 0: + continue + grouped.setdefault(int(frame.segment), []).append(frame) + for rows in grouped.values(): + rows.sort(key=lambda item: item.segment_frame) + return grouped + + def dependency_report(self) -> dict[str, Any]: + """Everything the key and the provenance need from the recording, without running.""" + reader = self.reader() + info = reader.info() + session = reader.session() + segments, missing = self.segments() + sizes: list[tuple[str, int]] = [] + for number in segments: + path = reader.segment_path(number) + if path is not None: + sizes.append((path.name, path.stat().st_size)) + identity = recording_identity(session, sizes) + calibration = str( + session.get("calibration_id") + or dict(session.get("calibration") or {}).get("version") + or "" + ) + key = result_key( + recording=identity, + sensor_config=self.chain.identity(), + calibration=calibration, + reference_frame=self.reference_frame, + ) + return { + "session": session, + "info": info, + "segments": segments, + "missing_segments": missing, + "segment_sizes": sizes, + "recording_identity": identity, + "calibration_id": calibration, + "key": key, + } + + def stored_result(self, *, allow_incomplete: bool = False): + """A reader for the result this configuration would produce, if one exists.""" + facts = self.dependency_report() + if not self.cache.exists(facts["key"]): + return None + return self.cache.reader(facts["key"], allow_incomplete=allow_incomplete) + + def _require_reference(self, grouped: dict[int, list[RecordingFrame]]) -> None: + """Refuse a reference frame the recording cannot supply. + + Frame ids are the acquisition session's own counter and do **not** start at 0 or + 1 — the box recording in this repository's evidence runs 20…52. A reference + frame that is absent therefore cannot be applied, and a pass that ignored it + would store a ``reference_frame_id`` other than the one asked for while claiming + the reset happened. + """ + if self.reference_frame is None: + return + available = [int(row.frame_id) for rows in grouped.values() for row in rows] + if int(self.reference_frame) in available: + return + span = f"{min(available)}...{max(available)}" if available else "no frames" + raise ValueError( + f"reference frame {self.reference_frame} is not among the recording's video " + f"frames ({span}); a reset that cannot be applied is not a result" + ) + + # ------------------------------------------------------------------------ pass + def recompute(self) -> Iterator[LiveFrame]: + """Re-run the chain over every segment and yield what it computes now. + + The reference implementation of the pass, sharing its frame generator with + :meth:`run` so a verification cannot drift from what was stored. It needs a + working directory of its own and leaves no result behind. + """ + facts = self.dependency_report() + key: ResultKey = facts["key"] + grouped = self.rows_by_segment(facts["info"].frames) + self._require_reference(grouped) + working = self._make_working_directory(key) + state = _PassState( + provenance=self._provenance(key, facts), + segments=list(facts["segments"]), + grouped=grouped, + working=working, + started=time.perf_counter(), + ) + if facts["missing_segments"]: + state.reasons.append("segments_missing") + if not state.segments: + state.reasons.append("no_segments") + try: + yield from self._pass_frames(state) + finally: + self._discard_working_directory(working) + + def run(self) -> AnalysisOutcome: + """Analyse the recording and store the result. + + Cancellation, an SDK failure and an ordinary Python exception all end the same + way: the directory keeps an ``INCOMPLETE`` marker and metadata that says so, and + :class:`~orisys_app_shared.analysis.reader.AnalysisReader` refuses it. Only a + pass that reached the end of every segment commits. + """ + started = time.perf_counter() + facts = self.dependency_report() + key: ResultKey = facts["key"] + grouped = self.rows_by_segment(facts["info"].frames) + # Checked before anything is written, so a caller error does not leave a + # directory behind at all. + self._require_reference(grouped) + working = self._make_working_directory(key) + writer = AnalysisWriter( + self.cache.directory(key), key=key, chunk_frames=self.chunk_frames + ) + provenance = self._provenance(key, facts) + state = _PassState( + provenance=provenance, + segments=list(facts["segments"]), + grouped=grouped, + working=working, + started=started, + ) + state.frames_expected = sum( + len(state.grouped.get(number, ())) for number in state.segments + ) + if facts["missing_segments"]: + state.reasons.append("segments_missing") + if not state.segments: + state.reasons.append("no_segments") + if self._stop_requested(): + state.reasons.append("cancelled_before_start") + writer.begin() + try: + for live in self._pass_frames(state): + if state.reasons: + break + failure = self._write_frame(writer, live, state) + if failure: + state.reasons.append(failure) + break + state.frames_written += 1 + except Exception as exc: + # Reported, not swallowed: the directory is abandoned below, so a reader + # sees an incomplete result rather than a short one presented as whole. + logger.exception("analysis pass failed") + state.reasons.append(f"{type(exc).__name__}: {exc}") + finally: + self._discard_working_directory(working) + counts = provenance.counts + counts.update(state.plan) + counts["frames_written"] = state.frames_written + counts["frames_expected"] = state.frames_expected + counts["frames_short"] = max(0, state.frames_expected - state.frames_written) + counts["result_bytes"] = writer.bytes_written + if state.frames_written: + counts["bytes_per_frame"] = round( + writer.bytes_written / state.frames_written, 2 + ) + counts["open_seconds"] = provenance.open_seconds + provenance.decode_order_offsets = dict(state.offsets) + distinct = sorted(set(state.offsets.values())) + provenance.decode_order_offset = distinct[0] if len(distinct) == 1 else None + provenance.decode_order_offset_source = ( + "declared" if "declared" in state.sources else "measured" + ) + provenance.finished_at = time.strftime("%Y-%m-%dT%H:%M:%S%z") + provenance.elapsed_s = round(time.perf_counter() - started, 4) + provenance.counts = counts + provenance.warnings = list(dict.fromkeys(provenance.warnings + state.reasons)) + if state.reasons: + provenance.complete = False + provenance.stopped_reason = state.reasons[-1] + writer.abandon(state.reasons[-1], provenance=provenance) + return AnalysisOutcome( + key=key, + directory=self.cache.directory(key), + complete=False, + reason=state.reasons[-1], + frames_written=state.frames_written, + frames_expected=state.frames_expected, + elapsed_s=provenance.elapsed_s, + provenance=provenance, + ) + provenance.complete = True + provenance.stopped_reason = "" + writer.commit(provenance=provenance, counts=counts) + return AnalysisOutcome( + key=key, + directory=self.cache.directory(key), + complete=True, + reason="", + frames_written=state.frames_written, + frames_expected=state.frames_expected, + elapsed_s=provenance.elapsed_s, + provenance=provenance, + ) + + # ------------------------------------------------------------------ generator + def _pass_frames(self, state: _PassState) -> Iterator[LiveFrame]: + """Walk every segment, yield each frame's live selection, and track the plan.""" + from orisys_app_shared.acquisition.sdk_worker import process_chain + + provenance = state.provenance + for position, number in enumerate(state.segments): + if state.reasons or self._stop_requested(): + if self._stop_requested(): + state.reasons.append("cancelled") + return + path = self.reader().segment_path(number) + if path is None: + state.reasons.append(f"segment_file_missing:{number}") + return + rows = state.grouped.get(number, ()) + if any(row.segment_frame != index + 1 for index, row in enumerate(rows)): + provenance.warnings.append(f"segment_frame_gap:{number}") + with _OpenChain(self.chain, path, state.working) as chain: + offset, measured = self._segment_offset(chain, number, state.offsets) + state.offsets[str(number)] = offset + state.sources.add("measured" if measured else "declared") + state.plan["skipped"] += min(offset, len(rows)) + provenance.open_seconds = round( + provenance.open_seconds + chain.open_seconds, 4 + ) + for name in chain.debug_files: + if name not in provenance.debug_files_written: + provenance.debug_files_written.append(name) + state.frames_expected = self._expected_total( + state.segments, position, state.grouped, state.offsets + ) + ordinal = 0 + while True: + if self._stop_requested(): + state.reasons.append("cancelled") + return + ok, frame = chain.read() + if not ok or frame is None: + break + index = ordinal + int(offset) + ordinal += 1 + state.plan["decoded"] += 1 + if index >= len(rows): + state.plan["unmapped"] += 1 + continue + row = rows[index] + unrolled = chain.feed(frame) + if unrolled is None: + state.reasons.append(f"unroll_failed:{row.frame_id}") + return + reset_applied = bool( + self.reference_frame is not None + and int(row.frame_id) == int(self.reference_frame) + ) + if reset_applied: + reset_started = time.perf_counter() + chain.sensor.reset() + provenance.counts["reset_seconds"] = round( + time.perf_counter() - reset_started, 6 + ) + provenance.counts["reset_count"] = ( + provenance.counts.get("reset_count", 0) + 1 + ) + depth, error = process_chain(chain.sensor) + if error: + provenance.warnings.append(f"chain:{error}") + state.reasons.append(f"chain_failed:{row.frame_id}") + return + flow = getattr(chain.sensor, "flow", None) + if flow is None or not getattr(flow, "size", 0): + state.reasons.append(f"no_flow:{row.frame_id}") + return + flow = np.asarray(flow) + mask = getattr(chain.sensor, "valid_mask", None) + mask = None if mask is None else np.asarray(mask, dtype=bool) + if mask is not None and tuple(mask.shape[:2]) != tuple( + flow.shape[:2] + ): + provenance.warnings.append( + f"mask_shape_mismatch:{row.frame_id}" + ) + mask = None + samples = select_flow_samples(flow, mask, self.policy) + depth_array = None if depth is None else np.asarray(depth) + yield LiveFrame( + row=row, + samples=samples, + mask=mask, + depth_nonzero=( + 0 + if depth_array is None + else int(np.count_nonzero(depth_array)) + ), + depth_abs_sum=( + 0.0 + if depth_array is None + else float(np.abs(depth_array, dtype=np.float64).sum()) + ), + reset_applied=reset_applied, + segment=int(number), + ) + + # ------------------------------------------------------------------- internals + def _make_working_directory(self, key: ResultKey) -> Path: + if self.working_directory is not None: + self.working_directory.mkdir(parents=True, exist_ok=True) + return self.working_directory + return Path(tempfile.mkdtemp(prefix=f"orisys-analysis-{key.digest[:8]}-")) + + def _discard_working_directory(self, working: Path) -> None: + if self.working_directory is not None: + return + shutil.rmtree(working, ignore_errors=True) + + def _stop_requested(self) -> bool: + return self.cancelled is not None and self.cancelled.is_set() + + def _segment_offset( + self, chain: _OpenChain, number: int, offsets: dict[str, int] + ) -> tuple[int, bool]: + """Frames consumed before the first readable one, measured where possible.""" + measured = chain.decode_position is not None + offset = ( + int(chain.decode_position) + if measured + else int(_DECLARED_OFFSETS.get(self.chain.sensor_type, 0)) + ) + if offsets: + previous = next(iter(offsets.items())) + if offset != previous[1]: + logger.warning( + "segment %d reports a decode offset of %d, segment %s reported %d", + number, + offset, + previous[0], + previous[1], + ) + return offset, measured + + @staticmethod + def _expected_total( + segments: Sequence[int], + position: int, + grouped: dict[int, list[RecordingFrame]], + offsets: dict[str, int], + ) -> int: + """A running estimate of the frame count, so progress means something early.""" + last = offsets.get(str(segments[position]), 0) + total = 0 + for number in segments: + offset = offsets.get(str(number), last) + total += max(0, len(grouped.get(number, ())) - offset) + return total + + def _write_frame( + self, writer: AnalysisWriter, live: LiveFrame, state: _PassState + ) -> str | None: + """Store one live frame. Returns a failure reason, or ``None`` when it was stored.""" + provenance = state.provenance + row = live.row + samples = live.samples + if live.reset_applied: + provenance.reference_frame_id = int(row.frame_id) + if provenance.reference_frame_id is None: + # No explicit reset: the pass itself starts from a clean tracker, so the + # first stored frame is the reference. Recording it always means the field + # never has to be interpreted as "unknown". + provenance.reference_frame_id = int(row.frame_id) + if provenance.first_output_frame_id is None: + provenance.first_output_frame_id = int(row.frame_id) + provenance.last_output_frame_id = int(row.frame_id) + grid_id = writer.grid_keyframe( + np.asarray(samples.rows, dtype=np.int32), + np.asarray(samples.cols, dtype=np.int32), + ) + validity_digest = None + if live.mask is not None: + validity_digest = digest8(keyframe_payload(KEYFRAME_MASK, live.mask)) + writer.mask_keyframe(live.mask, validity_digest) + self._accumulate( + provenance.counts, samples, live.depth_nonzero, live.depth_abs_sum + ) + record = encode_frame( + frame_id=row.frame_id, + t_rel_s=row.t_rel_s, + grid_id=grid_id, + dx=samples.dx, + dy=samples.dy, + magnitude=samples.magnitude, + active=samples.active, + validity_digest=validity_digest, + depth_nonzero=live.depth_nonzero, + dropped=row.dropped, + ) + writer.add_frame(record, frame_id=row.frame_id, t_rel_s=row.t_rel_s) + if self.on_progress is not None: + self.on_progress( + Progress( + frames_written=state.frames_written + 1, + frames_expected=max( + state.frames_expected, state.frames_written + 1 + ), + segment_index=state.segments.index(live.segment) + 1, + segment_count=len(state.segments), + frame_id=int(row.frame_id), + elapsed_s=time.perf_counter() - state.started, + ) + ) + return None + + @staticmethod + def _accumulate( + counts: dict[str, Any], + samples: FlowSamples, + depth_nonzero: int, + depth_abs_sum: float, + ) -> None: + candidates = int(len(samples.rows)) + active = int(np.count_nonzero(samples.active)) + counts["candidates_min"] = min( + counts.get("candidates_min", candidates), candidates + ) + counts["candidates_max"] = max( + counts.get("candidates_max", candidates), candidates + ) + counts["candidates_last"] = candidates + counts["active_zero_frames"] = counts.get("active_zero_frames", 0) + ( + 1 if active == 0 else 0 + ) + counts["active_total"] = counts.get("active_total", 0) + active + counts["frames_with_samples"] = counts.get("frames_with_samples", 0) + 1 + counts["depth_nonzero_max"] = max( + counts.get("depth_nonzero_max", depth_nonzero), depth_nonzero + ) + counts["depth_nonzero_frames"] = counts.get("depth_nonzero_frames", 0) + ( + 1 if depth_nonzero else 0 + ) + counts["depth_abs_sum"] = round( + float(counts.get("depth_abs_sum", 0.0)) + float(depth_abs_sum), 6 + ) + + def _provenance(self, key: ResultKey, facts: dict[str, Any]) -> Provenance: + session = facts["session"] + policy = self.policy.normalised() + return Provenance( + format_version=model_module.ANALYSIS_FORMAT_VERSION, + algorithm=ALGORITHM_VERSION, + unit=model_module.UNIT, + key=key.digest, + key_components=key.components, + recording=str(self.recording), + recording_identity=facts["recording_identity"], + session_id=int(session.get("session_id") or 0), + sensor_type=self.chain.sensor_type, + chain=f"{self.chain.sensor_type}:{ALGORITHM_VERSION}", + sensor_config_ref=( + "" if self.chain.config_ref is None else str(self.chain.config_ref) + ), + sensor_config_sha256=self.chain.config_sha256, + calibration_id=facts["calibration_id"], + reference_is_explicit=self.reference_frame is not None, + segments=list(facts["segments"]), + segments_declared=int( + dict(session.get("stats") or {}).get("segments") + or len(facts["segments"]) + ), + segments_missing=list(facts["missing_segments"]), + policy={ + "flow_on_px": policy.flow_on_px, + "flow_off_px": policy.flow_off_px, + "flow_grid_spacing": policy.flow_grid_spacing, + "flow_edge_margin": policy.flow_edge_margin, + "flow_length_margin": policy.flow_length_margin, + "max_arrows": policy.max_arrows, + }, + sources={ + "flow": "recomputed", + "depth": "recomputed", + "forces": "recorded", + }, + versions={ + "sdk": _version_of("orisys"), + "shared": _version_of("orisys_app_shared"), + "numpy": str(np.__version__), + "algorithm": ALGORITHM_VERSION, + }, + config=self._config_summary(), + started_at=time.strftime("%Y-%m-%dT%H:%M:%S%z"), + counts={}, + ) + + def _config_summary(self) -> dict[str, Any]: + """The configuration values that decide what the chain computes. + + Recorded so the SDK-versus-shipped divergence is visible in the result instead + of being rediscovered: the SDK's built-in ``finger`` config says + ``source_scale`` 0.4 and ``hhd_downsample`` 2, while the shipped Studio config + says 0.42 and 3. + """ + reader = self.reader() + summary: dict[str, Any] = { + "sensor_type": self.chain.sensor_type, + "export_dir": "" + if self.chain.export_dir is None + else str(self.chain.export_dir), + "recording_frames": len(reader.info().frames), + "video_frames": int( + dict(reader.session().get("stats") or {}).get("video_frames") or 0 + ), + } + if self.chain.config_ref is not None and Path(self.chain.config_ref).is_file(): + try: + raw = json.loads( + Path(self.chain.config_ref).read_text(encoding="utf-8") + ) + except (OSError, ValueError): + raw = {} + camera = dict(raw.get("camera") or {}) + tracker = dict(raw.get("tracker") or {}) + summary.update( + { + "source_scale": camera.get("source_scale"), + "hhd_downsample": tracker.get("hhd_downsample"), + "dis_mode": tracker.get("dis_mode"), + "resolution_stitch": camera.get("resolution_stitch"), + "fps": camera.get("fps"), + } + ) + return summary diff --git a/shared/tests/test_replay_analysis.py b/shared/tests/test_replay_analysis.py new file mode 100644 index 0000000..cc918a0 --- /dev/null +++ b/shared/tests/test_replay_analysis.py @@ -0,0 +1,615 @@ +"""Part A: the offline analysis pass and its result cache, checked on real footage. + +Three things are being defended here, and each one is a way the feature could be +silently wrong rather than visibly broken: + +* the stored payload is **what the live selector returns for that frame** — not a + close approximation of it, and not a different frame; +* a result is either complete or unreadable, so a cancelled pass can never be drawn; +* the pass still costs what it cost when it was measured, so a cache that is correct + but slower than re-running the chain is caught. + +The inputs are real: the 33-frame box recording under ``dist/`` (its own ``frames.csv``, +``session.json`` and ``raw_0001.mp4``) and the shipped finger configuration and asset +bundle. The pass is driven with the **box** chain on it for the offset and round-trip +checks, and with the **finger** chain on the same footage for the keyframe, reset and +size checks — the finger chain over box footage is exactly what the Step 0 measurement +did, and it is the only way to get a validity mask (and therefore a static candidate +grid) out of this recording. +""" + +from __future__ import annotations + +import hashlib +import inspect +import json +import os +import shutil +import threading +from pathlib import Path + +import numpy as np +import pytest +from orisys_app_shared.analysis import model as analysis_model +from orisys_app_shared.analysis.cache import ( + INCOMPLETE_MARKER, + META_NAME, + ResultCache, + evict, + scan_results, +) +from orisys_app_shared.analysis.model import frame_payload_bytes, result_key +from orisys_app_shared.analysis.reader import ( + AnalysisIncompleteError, + AnalysisReader, +) +from orisys_app_shared.analysis.service import ( + AnalysisService, + ChainSpec, + chain_spec_for, +) +from orisys_app_shared.scene3d.visual_policy import VisualPolicy + +REPO = Path(__file__).resolve().parents[2] +STUDIO = REPO / "dist" / "OrisysStudio" +RECORDING = ( + STUDIO + / "outputs" + / "recordings" + / "20260930T015028Z_session0002_40cb8e7441434feab48bb646afdba011" +) +FINGER_CONFIG = STUDIO / "config" / "sensors" / "finger" / "sensor.json" +FINGER_EXPORT = STUDIO / "config" / "sensors" / "finger" / "assets" / "30fps-0703" + +#: Step 0's digest of the finger validity mask, over ``np.packbits(mask)`` with +#: blake2b-16: one value across three videos and 327 frames. Reproduced here from the +#: stored keyframe, which is what makes the mask keyframe a *verified* copy rather than +#: a plausible-looking one. (The result stores a blake2b-8 digest of the same bytes; the +#: longer digest is the comparison because that is the one on record.) +STEP0_MASK_DIGEST = "d76349a3b901aea50915eb1308901c2e" + +#: The frame id the reset is applied at. The recording's own ids run 20…52, so this is +#: deliberately not a small number: an off-by-one in the reset rule would land on a +#: frame that does not exist and be invisible. +REFERENCE_FRAME = 36 + + +def _require_inputs() -> None: + missing = [ + str(path) + for path in ( + RECORDING / "frames.csv", + RECORDING / "raw_0001.mp4", + FINGER_CONFIG, + ) + if not path.exists() + ] + if missing: + pytest.skip("analysis inputs are not present: " + ", ".join(missing)) + pytest.importorskip("orisys", reason="the SDK is needed to run the analysis pass") + + +def _finger_chain() -> ChainSpec: + return ChainSpec( + sensor_type="finger", config_ref=FINGER_CONFIG, export_dir=FINGER_EXPORT + ) + + +def _box_service(root: Path, **kwargs) -> AnalysisService: + from orisys_app_shared.record.playback import RecordingReader + + chain = chain_spec_for(RecordingReader(RECORDING).session()) + return AnalysisService(RECORDING, chain=chain, cache=ResultCache(root), **kwargs) + + +def _finger_service(root: Path, **kwargs) -> AnalysisService: + return AnalysisService( + RECORDING, chain=_finger_chain(), cache=ResultCache(root), **kwargs + ) + + +def _directory_bytes(directory: Path) -> int: + return sum(path.stat().st_size for path in directory.rglob("*") if path.is_file()) + + +# --------------------------------------------------------------------------- fixtures +@pytest.fixture(scope="module") +def box_pass(tmp_path_factory): + _require_inputs() + root = tmp_path_factory.mktemp("analysis-box") + service = _box_service(root) + outcome = service.run() + return service, outcome + + +@pytest.fixture(scope="module") +def finger_pass(tmp_path_factory): + _require_inputs() + root = tmp_path_factory.mktemp("analysis-finger") + service = _finger_service(root) + outcome = service.run() + return service, outcome + + +@pytest.fixture(scope="module") +def finger_pass_again(tmp_path_factory): + """A second pass over the same input, in its own cache root.""" + _require_inputs() + root = tmp_path_factory.mktemp("analysis-finger-again") + service = _finger_service(root) + return service, service.run() + + +@pytest.fixture(scope="module") +def finger_reset_pass(tmp_path_factory): + _require_inputs() + root = tmp_path_factory.mktemp("analysis-finger-reset") + service = _finger_service(root, reference_frame=REFERENCE_FRAME) + return service, service.run() + + +# --------------------------------------------------------------- frame identity +def test_the_box_chain_consumes_frames_before_the_first_one_a_caller_can_read(box_pass): + """The offset is measured, recorded, and applied — not assumed. + + Constructing a box ``Sensor`` reads seven frames through its stitch probe, so the + first frame the pass can read is the recording's *eighth* video frame. Nothing about + the file says so, which is why the offset comes from the capture's own decode + position and why the resulting frame identity is checked against ``frames.csv``. + """ + _, outcome = box_pass + assert outcome.complete, outcome.reason + assert outcome.provenance.decode_order_offsets == {"1": 7} + assert outcome.provenance.decode_order_offset == 7 + assert outcome.provenance.decode_order_offset_source == "measured" + assert outcome.provenance.counts["skipped"] == 7 + + frames = _box_service(Path(outcome.directory).parent).reader().info().frames + rows = sorted( + (frame for frame in frames if frame.in_video and frame.segment == 1), + key=lambda frame: frame.segment_frame, + ) + assert len(rows) == 33 + assert outcome.provenance.first_output_frame_id == rows[7].frame_id == 27 + assert outcome.provenance.last_output_frame_id == rows[-1].frame_id == 52 + assert outcome.frames_written == 26 + assert outcome.frames_expected == 26 + + +def test_the_box_result_round_trips_against_a_fresh_run_of_the_chain(box_pass): + """Frame for frame, the store equals what the chain computes now.""" + service, outcome = box_pass + assert outcome.complete, outcome.reason + reader = AnalysisReader(outcome.directory) + try: + checked = 0 + for live in service.recompute(): + stored = reader.frame(live.row.frame_id) + assert stored.t_rel_s == live.row.t_rel_s + assert live.payload_matches(stored), f"frame {live.row.frame_id} differs" + checked += 1 + assert checked == outcome.frames_written == len(reader) == 26 + finally: + reader.close() + + +# ------------------------------------------------------------------- keyframes +def test_the_finger_result_stores_the_candidate_grid_and_the_mask_once(finger_pass): + """330 candidates on every frame, from one grid keyframe and one mask keyframe. + + This is the measurement the format rests on: the candidate *location* set does not + move, so it is stored once and the per-frame record carries only values. + """ + _, outcome = finger_pass + assert outcome.complete, outcome.reason + reader = AnalysisReader(outcome.directory) + try: + assert reader.keyframe_counts == {"grid": 1, "mask": 1} + payloads = list(reader.frames()) + assert len(payloads) == 33 + assert {payload.candidates for payload in payloads} == {330} + assert len({payload.grid_id for payload in payloads}) == 1 + + mask = reader.mask() + assert mask is not None and mask.shape == (300, 280) + packed = np.packbits(mask, axis=None).tobytes() + assert hashlib.blake2b(packed, digest_size=16).hexdigest() == STEP0_MASK_DIGEST + + assert outcome.provenance.counts["candidates_min"] == 330 + assert outcome.provenance.counts["candidates_max"] == 330 + assert outcome.provenance.counts["active_zero_frames"] == 5 + assert outcome.provenance.counts["active_total"] == 152 + finally: + reader.close() + + +def test_every_stored_frame_reconstructs_what_the_live_selector_returns(finger_pass): + """The keyframe path is exact: the grid is the one the selector chose, per frame.""" + service, outcome = finger_pass + assert outcome.complete, outcome.reason + reader = AnalysisReader(outcome.directory) + try: + policy = service.policy + assert policy.flow_grid_spacing == 10 and policy.max_arrows == 400 + for live in service.recompute(): + stored = reader.frame(live.row.frame_id) + assert live.payload_matches(stored), f"frame {live.row.frame_id} differs" + rebuilt = reader.samples(live.row.frame_id) + assert np.array_equal(rebuilt.rows, np.asarray(live.samples.rows, np.int32)) + assert np.array_equal(rebuilt.cols, np.asarray(live.samples.cols, np.int32)) + assert np.array_equal( + np.asarray(rebuilt.magnitude, np.float32), + np.asarray(live.samples.magnitude, np.float32), + ) + assert np.array_equal(rebuilt.active, np.asarray(live.samples.active, bool)) + finally: + reader.close() + + +def test_a_threshold_change_is_served_without_rebuilding_the_result(finger_pass): + """Display settings are not in the key because the record can answer a new threshold. + + Activity is stored, but magnitude is stored too — so a viewer with a different + ``flow_on_px`` gets a correct answer from the same file. + """ + _, outcome = finger_pass + reader = AnalysisReader(outcome.directory) + try: + frame_id = int(reader.frame_ids[0]) + recorded = reader.samples(frame_id) + impossible = reader.samples(frame_id, policy=VisualPolicy(flow_on_px=1e9)) + assert impossible.active_count == 0 + permissive = reader.samples( + frame_id, policy=VisualPolicy(flow_on_px=0.0, flow_off_px=0.0) + ) + assert permissive.active_count >= recorded.active_count + assert np.array_equal(permissive.magnitude, recorded.magnitude) + assert np.array_equal( + permissive.active, np.asarray(permissive.magnitude >= 0.0, bool) + ) + finally: + reader.close() + + +# ------------------------------------------------------------------ determinism +def test_two_passes_over_the_same_input_store_byte_identical_payloads( + finger_pass, finger_pass_again +): + """Re-running the pass produces the same bytes, so a cache hit cannot drift. + + The determinism contract is the **frame records** and the **keyframe digests**. The + metadata is deliberately outside it: ``elapsed_s``, ``open_seconds``, ``started_at`` + and ``finished_at`` describe the run, not the measurement, and two runs of the same + pass cannot share them. + """ + _, first = finger_pass + _, second = finger_pass_again + assert first.complete and second.complete + reader_a = AnalysisReader(first.directory) + reader_b = AnalysisReader(second.directory) + try: + ids = reader_a.frame_ids + assert np.array_equal(ids, reader_b.frame_ids) + assert reader_a.frame_record_bytes == reader_b.frame_record_bytes + for frame_id in ids: + assert ( + reader_a.frame(int(frame_id)).canonical_bytes() + == reader_b.frame(int(frame_id)).canonical_bytes() + ), f"frame {frame_id} is not reproducible" + assert {key: value.digest for key, value in reader_a.keyframes.items()} == { + key: value.digest for key, value in reader_b.keyframes.items() + } + assert reader_a.mask_digest == reader_b.mask_digest + finally: + reader_a.close() + reader_b.close() + + +# ------------------------------------------------------------------ cancellation +def test_an_interrupted_pass_is_incomplete_and_is_not_served(tmp_path): + """A cancelled pass leaves a directory that refuses to open by default.""" + _require_inputs() + root = tmp_path / "cache" + cancelled = threading.Event() + seen: list[int] = [] + + def on_progress(progress): + seen.append(progress.frame_id) + if len(seen) >= 5: + cancelled.set() + + service = _finger_service(root, on_progress=on_progress, cancelled=cancelled) + outcome = service.run() + + assert outcome.complete is False + assert outcome.reason == "cancelled" + assert 0 < outcome.frames_written < 33 + assert (outcome.directory / INCOMPLETE_MARKER).is_file() + meta = json.loads((outcome.directory / META_NAME).read_text(encoding="utf-8")) + assert meta["complete"] is False + assert meta["stopped_reason"] == "cancelled" + assert meta["provenance"]["complete"] is False + assert [item.complete for item in scan_results(root)] == [False] + + with pytest.raises(AnalysisIncompleteError): + AnalysisReader(outcome.directory) + + # The partial data is still inspectable on purpose, and says what it is. + partial = AnalysisReader(outcome.directory, allow_incomplete=True) + try: + assert partial.complete is False + assert len(partial) == outcome.frames_written + assert "incomplete" in partial.stale_reasons() + finally: + partial.close() + + +def test_a_declared_segment_that_is_missing_makes_the_result_incomplete(tmp_path): + """A segment the session declares and the directory lacks is a failure, not a short read.""" + _require_inputs() + recording = tmp_path / "recording" + recording.mkdir() + shutil.copy2(RECORDING / "frames.csv", recording / "frames.csv") + shutil.copy2(RECORDING / "forces.csv", recording / "forces.csv") + shutil.copy2(RECORDING / "events.csv", recording / "events.csv") + session = json.loads((RECORDING / "session.json").read_text(encoding="utf-8")) + session["stats"]["segments"] = 2 + (recording / "session.json").write_text(json.dumps(session), encoding="utf-8") + try: + os.link(RECORDING / "raw_0001.mp4", recording / "raw_0001.mp4") + except OSError: + shutil.copy2(RECORDING / "raw_0001.mp4", recording / "raw_0001.mp4") + + service = AnalysisService( + recording, chain=_finger_chain(), cache=ResultCache(tmp_path / "cache") + ) + outcome = service.run() + assert outcome.complete is False + assert outcome.reason == "segments_missing" + assert outcome.frames_written == 0 + assert outcome.provenance.segments == [1] + assert outcome.provenance.segments_declared == 2 + assert outcome.provenance.segments_missing == [2] + with pytest.raises(AnalysisIncompleteError): + AnalysisReader(outcome.directory) + + +def test_a_reference_frame_outside_the_recording_is_refused(tmp_path): + """The reset is applied at a named frame or the pass does not run. + + Frame ids come from the acquisition session and need not start at zero or one: this + recording's run 20…52. A requested reference of 16 therefore names nothing, and + ignoring that would store a result whose ``reference_frame_id`` is not the frame + that was asked for. + """ + _require_inputs() + service = _finger_service(tmp_path / "cache", reference_frame=16) + with pytest.raises(ValueError, match="reference frame 16"): + service.run() + assert not (tmp_path / "cache").exists(), ( + "a refused pass must not leave a directory" + ) + + +# ------------------------------------------------------------------ reference reset +def test_resetting_the_reference_records_the_frame_and_clears_the_tracker( + finger_pass, finger_reset_pass +): + """The reset is exact at the named frame and touches nothing before it. + + ``Sensor.reset`` clears the optical-flow tracker, so the frame the reset is applied + at measures against an empty reference: its displacement is exactly zero — the + Step 0 signature. The frames before it are byte-identical to a pass with no reset, + which is what makes "reference frame" a statement about a position rather than about + a slightly different run. + """ + _, plain = finger_pass + _, reset = finger_reset_pass + assert reset.complete, reset.reason + assert reset.provenance.reference_is_explicit is True + assert reset.provenance.reference_frame_id == REFERENCE_FRAME + assert reset.provenance.first_output_frame_id == 20 + assert reset.provenance.counts["reset_count"] == 1 + assert reset.provenance.counts["reset_seconds"] >= 0.0 + # With no explicit reset the reference is the pass's own clean start. + assert plain.provenance.reference_is_explicit is False + assert plain.provenance.reference_frame_id == plain.provenance.first_output_frame_id + + reader = AnalysisReader(plain.directory) + reset_reader = AnalysisReader(reset.directory) + try: + before = int(REFERENCE_FRAME) - 1 + assert reset_reader.frame(REFERENCE_FRAME).magnitude.max() == 0.0 + assert reader.frame(REFERENCE_FRAME).magnitude.max() > 0.0 + assert reset_reader.frame(before).magnitude.max() > 0.0 + + for frame_id in reader.frame_ids: + if int(frame_id) < REFERENCE_FRAME: + assert ( + reader.frame(int(frame_id)).canonical_bytes() + == reset_reader.frame(int(frame_id)).canonical_bytes() + ), f"frame {frame_id} before the reference was disturbed by the reset" + assert ( + reader.frame(int(REFERENCE_FRAME)).canonical_bytes() + != reset_reader.frame(int(REFERENCE_FRAME)).canonical_bytes() + ) + finally: + reader.close() + reset_reader.close() + + +# ------------------------------------------------------------------ the key +def test_the_key_follows_the_inputs_that_change_the_result_and_no_others(finger_pass): + """Five components, and display settings are deliberately not one of them.""" + base = dict( + recording="rec-a", + sensor_config="config-a", + calibration="cal-a", + reference_frame=36, + ) + digest = result_key(**base).digest + assert result_key(**base).digest == digest + assert result_key(**{**base, "sensor_config": "config-b"}).digest != digest + assert result_key(**{**base, "algorithm": "other-chain"}).digest != digest + assert result_key(**{**base, "calibration": "cal-b"}).digest != digest + assert result_key(**{**base, "reference_frame": None}).digest != digest + assert result_key(**{**base, "recording": "rec-b"}).digest != digest + + # Nothing in the signature can carry a display setting, so a change to one cannot + # move the key even by accident. + parameters = set(inspect.signature(result_key).parameters) + assert not {name for name in parameters if "policy" in name or "display" in name} + assert "policy" not in parameters + + # And the concrete consequence: two services that differ only in the policy the + # samples are selected with address the same stored result. + service = _finger_service(finger_pass[1].directory.parent) + other = _finger_service(finger_pass[1].directory.parent) + other.policy = VisualPolicy(flow_on_px=1.5, max_arrows=1) + assert ( + service.dependency_report()["key"].digest + == other.dependency_report()["key"].digest + == finger_pass[1].key.digest + ) + recorded = finger_pass[1].provenance.policy + assert recorded["flow_on_px"] == 0.4 and recorded["max_arrows"] == 400 + + +# ------------------------------------------------------------------ size +def test_the_stored_size_is_the_measured_encoding(box_pass, finger_pass): + """State the encoding, then hold it: 4,034 B per frame at the finger grid. + + Implicit candidate grid (a keyframe) + float32 ``dx``/``dy``/``magnitude`` + + bit-packed ``active`` + a 24-byte header + the 8-byte validity digest: + ``24 + 330*12 + 42 + 8 = 4,034`` B. The alternative measured encodings are float16 + ``dxy``+magnitude (2,022 B) and dropping magnitude (4,002 B with the grid, no + header); float32 is chosen because the round trip has to be *exact*, and the + magnitude is what serves a changed threshold. + """ + assert frame_payload_bytes(330, with_validity=True) == 4034 + assert frame_payload_bytes(330, with_validity=False) == 4026 + assert frame_payload_bytes(400, with_validity=False) == 4874 + + _, finger = finger_pass + reader = AnalysisReader(finger.directory) + try: + assert reader.bytes_per_frame == 4034.0 + assert reader.frame_record_bytes == 33 * 4034 + assert reader.bytes_per_frame <= 4096, "the stated per-frame bound" + total_per_frame = _directory_bytes(finger.directory) / len(reader) + assert total_per_frame <= 4600, "keyframes and metadata included" + finally: + reader.close() + + _, box = box_pass + box_reader = AnalysisReader(box.directory) + try: + # The box chain has no validity mask, so there is no digest and every frame + # carries its own grid keyframe: 4,874 B of record per frame at 400 candidates. + assert box_reader.frame_record_bytes == 26 * 4874 + assert box_reader.keyframe_counts == {"grid": 26} + assert box_reader.mask() is None + finally: + box_reader.close() + + +# ------------------------------------------------------------------ reader +def test_a_stored_result_is_served_without_the_recording_or_a_replay( + finger_pass, tmp_path +): + """A seek is a file read: the result answers with the recording and SDK out of reach.""" + _, outcome = finger_pass + copy = tmp_path / "copied-result" + shutil.copytree(outcome.directory, copy) + + original = AnalysisReader(outcome.directory) + copied = AnalysisReader(copy) + try: + assert np.array_equal(original.frame_ids, copied.frame_ids) + for frame_id in original.frame_ids: + assert ( + original.frame(int(frame_id)).canonical_bytes() + == copied.frame(int(frame_id)).canonical_bytes() + ) + # Random access by frame id and by timestamp both land on the right frame. + times = copied.times + middle = len(times) // 2 + assert copied.frame(int(copied.frame_ids[middle])).t_rel_s == times[middle] + assert copied.frame_at(float(times[middle])).frame_id == int( + copied.frame_ids[middle] + ) + assert copied.frame_at(float(times[middle]) + 1e-6).frame_id == int( + copied.frame_ids[middle] + ) + assert copied.frame_at(-1.0).frame_id == int(copied.frame_ids[0]) + assert copied.frame_at(float(times[-1]) + 100.0).frame_id == int( + copied.frame_ids[-1] + ) + window = list( + copied.frames(start=int(copied.frame_ids[1]), stop=int(copied.frame_ids[2])) + ) + assert [payload.frame_id for payload in window] == [ + int(copied.frame_ids[1]), + int(copied.frame_ids[2]), + ] + assert copied.has(int(copied.frame_ids[0])) + assert not copied.has(10_000) + finally: + original.close() + copied.close() + + +def test_eviction_removes_whole_results_against_the_cap(tmp_path): + """The cap is enforced by removing complete results, oldest first, never a mistake.""" + root = tmp_path / "cache" + root.mkdir() + made: list[Path] = [] + for index in range(3): + directory = root / f"result-{index}" + directory.mkdir() + (directory / META_NAME).write_text( + json.dumps({"complete": True, "key": f"result-{index}", "format": "x"}), + encoding="utf-8", + ) + (directory / "blob.bin").write_bytes(b"\0" * 4096) + os.utime(directory / META_NAME, (1_600_000_000 + index, 1_600_000_000 + index)) + made.append(directory) + + def size_of(path: Path) -> int: + return sum(item.stat().st_size for item in path.rglob("*") if item.is_file()) + + kept = made[2] + # A cap that one removal satisfies: the oldest complete result goes, and the + # protected one is never a candidate whatever the arithmetic says. + cap = sum(size_of(item) for item in made) - size_of(made[0]) + removed = evict(root, max_bytes=cap, keep=[kept]) + assert removed == [made[0]] + assert made[0].exists() is False + assert kept.is_dir() and made[1].is_dir() + assert [item.directory.name for item in scan_results(root)] == [ + "result-1", + "result-2", + ] + + # A dry run reports without deleting, and still never proposes the protected result. + assert evict(root, max_bytes=1, dry_run=True, keep=[kept]) == [made[1]] + assert made[1].is_dir() + assert kept.is_dir() + + +# ------------------------------------------------------------------ incompleteness gate +def test_a_result_without_metadata_is_never_served(tmp_path): + """The only thing that opens is a directory whose metadata says it finished.""" + directory = tmp_path / "empty-result" + directory.mkdir() + from orisys_app_shared.analysis.reader import AnalysisNotFoundError + + with pytest.raises(AnalysisNotFoundError): + AnalysisReader(directory) + + (directory / META_NAME).write_text( + json.dumps({"format": analysis_model.ANALYSIS_FORMAT, "format_version": 1}), + encoding="utf-8", + ) + with pytest.raises(AnalysisIncompleteError): + AnalysisReader(directory)