diff --git a/apps/studio/config/locales/en_US.json b/apps/studio/config/locales/en_US.json index 66000ac..c2f6f5a 100644 --- a/apps/studio/config/locales/en_US.json +++ b/apps/studio/config/locales/en_US.json @@ -155,7 +155,8 @@ "units_newton": "Newton", "units_raw": "Raw", "settings": "Settings", - "settings_tip": "Show or hide the unit, record-mode and preview controls" + "settings_tip": "Show or hide the unit, record-mode and preview controls", + "stopped_analysis": "analysis" }, "units": { "newton_short": "N", @@ -218,7 +219,10 @@ "mesh3d_rejected": "The 3D scene was refused: {message}\nOnly the 3D view is affected; measuring, recording and the charts work as usual.", "mesh3d_unavailable_gl_reason": "3D is not available on this machine: {reason}\nMeasuring, recording and the charts work as usual.", "mesh3d_settings_tip": "3D display settings (view, housing, colour scale, frame rate)", - "mesh3d_reset_tip": "Reset view: back to the framing the scene itself uses" + "mesh3d_reset_tip": "Reset view: back to the framing the scene itself uses", + "analysis_pipeline": "Analysis result (one finished pass)", + "analysis_pipeline_tip": "This frame's arrows come from a stored analysis result rather than a chain running now; changing the threshold only changes which stored samples count as active.", + "analysis_empty": "the analysis result holds no frame here" }, "chart": { "fn": "FN", @@ -278,7 +282,16 @@ "warn_config_missing": "no config hash was recorded, so recomputation cannot be verified", "warn_rows_skipped": "some unparsable CSV rows were skipped", "warn_frames_dropped": "frames were lost to a full queue; only their index rows remain (shown as \"not recorded\")", - "warn_out_of_order": "this recording was written by an older version: its rows are not in time order, so the replay re-sorts them and window edges may differ" + "warn_out_of_order": "this recording was written by an older version: its rows are not in time order, so the replay re-sorts them and window edges may differ", + "analysis_empty": "this analysis result holds no frame to show", + "analysis_unreadable": "this frame cannot be read ({error})", + "analysis_complete": "replaying from the analysis result: {frames} frames, cursor on signal time (the recording's own frames.csv)", + "analysis_reused": "an analysis result was already stored and is being replayed: {frames} frames", + "analysis_failed": "the analysis did not finish ({reason}): playback is not served. Re-run it, or turn 「从视频重算」 off to read the recorded forces.csv", + "analysis_cancelled": "the analysis was cancelled: no complete result was written, so no playback is served", + "analysis_no_source": "this recording has no video segment to analyse", + "analysis_busy": "an analysis is already running: wait for it, or cancel it", + "analysis_grid_mismatch": "the current display settings moved the sample grid ({keys}); the stored candidate locations cannot serve it, so the replay keeps drawing the result's own grid" }, "record": { "label": "Record", @@ -497,6 +510,27 @@ "severity_info": "Note", "severity_warning": "Warning", "severity_error": "Error", - "stats": "shown {displayed} · display drops {dropped} · compose {compose} ms · paint {paint} ms" + "stats": "shown {displayed} · display drops {dropped} · compose {compose} ms · paint {paint} ms", + "source_analysis": "Analysis result: one finished pass, read by frame", + "source_analysis_tip": "The picture comes from a stored analysis result: the chain has already run and its output is on disk, so scrubbing and playing are file reads. Changing a threshold or a colour does not recompute." + }, + "analysis": { + "title": "Analysing the recording", + "subtitle": "once only: scrubbing, playing and stepping afterwards are reads", + "stage_open": "open chain", + "stage_frames": "frame by frame", + "stage_write": "write result", + "frames": "{written} / {expected} frames", + "segment": "segment {index} / {count}", + "elapsed": "{seconds:.1f} s elapsed", + "cancel": "Cancel", + "cancelling": "Cancelling…", + "cancel_tip": "Cancel this pass: the frames already written stay, but the result is marked incomplete and is never played back as if it were whole", + "reused": "a stored result is already there — nothing to recompute", + "complete": "analysis finished: {frames} frames in {seconds:.1f} s", + "cancelled": "analysis cancelled ({frames} frames written; the result is marked incomplete)", + "failed": "analysis failed: {reason}", + "incomplete": "the result is incomplete: {reason}", + "blocked": "a session is running: stop it before analysing" } } diff --git a/apps/studio/config/locales/zh_CN.json b/apps/studio/config/locales/zh_CN.json index 66cbb35..0f07aac 100644 --- a/apps/studio/config/locales/zh_CN.json +++ b/apps/studio/config/locales/zh_CN.json @@ -155,7 +155,8 @@ "units_newton": "牛顿", "units_raw": "原始", "settings": "设置", - "settings_tip": "展开或收起单位、记录模式与预览" + "settings_tip": "展开或收起单位、记录模式与预览", + "stopped_analysis": "分析" }, "units": { "newton_short": "N", @@ -218,7 +219,10 @@ "mesh3d_rejected": "三维场景被拒绝：{message}\n只影响三维视图：采集、记录与力曲线照常工作。", "mesh3d_unavailable_gl_reason": "本机无法使用三维显示：{reason}\n采集、记录与力曲线照常工作。", "mesh3d_settings_tip": "三维显示设置（视角、外壳、色标、显示频率）", - "mesh3d_reset_tip": "复位视角：回到场景自带的取景" + "mesh3d_reset_tip": "复位视角：回到场景自带的取景", + "analysis_pipeline": "分析结果（已跑完的一遍）", + "analysis_pipeline_tip": "这一帧的箭头来自存下来的分析结果，不是此刻在跑的链路；换阈值只改变哪些样本算「有效」，不会重算。", + "analysis_empty": "分析结果里没有这一帧" }, "chart": { "fn": "FN", @@ -278,7 +282,16 @@ "warn_config_missing": "找不到配置哈希，无法校验重算一致性", "warn_rows_skipped": "有无法解析的 CSV 行被跳过", "warn_frames_dropped": "有帧在队列溢出时丢失，仅保留索引记录（回放显示为“该帧未录制”）", - "warn_out_of_order": "该录制由旧版本写入，行序不是时间序；回放已按时间重排，区间边界可能与当时不同" + "warn_out_of_order": "该录制由旧版本写入，行序不是时间序；回放已按时间重排，区间边界可能与当时不同", + "analysis_empty": "这个分析结果里没有可显示的帧", + "analysis_unreadable": "这一帧读不出来（{error}）", + "analysis_complete": "已用分析结果回放：{frames} 帧，游标按信号时间（录制自己的 frames.csv）", + "analysis_reused": "找到了现成的分析结果，直接回放：{frames} 帧", + "analysis_failed": "分析未完成（{reason}）：不提供回放。可以重新分析，或关闭「从视频重算」看录制的 forces.csv", + "analysis_cancelled": "分析已取消：未写出完整结果，因此不提供回放", + "analysis_no_source": "这段录制没有可用于分析的视频分段", + "analysis_busy": "分析进行中：先等待或取消当前这一遍", + "analysis_grid_mismatch": "当前显示设置改了采样网格（{keys}），存下来的候选位置无法服务它：回放仍按结果自己的网格绘制" }, "record": { "label": "记录", @@ -497,6 +510,27 @@ "severity_info": "提示", "severity_warning": "警告", "severity_error": "错误", - "stats": "已显示 {displayed} · 显示跳帧 {dropped} · 合成 {compose} ms · 绘制 {paint} ms" + "stats": "已显示 {displayed} · 显示跳帧 {dropped} · 合成 {compose} ms · 绘制 {paint} ms", + "source_analysis": "分析结果：已跑完的一遍，按帧读取", + "source_analysis_tip": "画面来自存下来的分析结果：算法链路已经跑完并写入缓存，拖动时间轴与播放都只是读文件。换阈值或配色不会重算。" + }, + "analysis": { + "title": "正在分析录像", + "subtitle": "只跑一遍：之后拖动时间轴、播放与单步都只是读结果", + "stage_open": "打开链路", + "stage_frames": "逐帧分析", + "stage_write": "写入结果", + "frames": "{written} / {expected} 帧", + "segment": "分段 {index} / {count}", + "elapsed": "已用 {seconds:.1f} s", + "cancel": "取消", + "cancelling": "正在取消…", + "cancel_tip": "取消这一遍：已完成的帧会留下，但结果会被标记为未完成，永远不会被当作有效结果回放", + "reused": "已有结果，无需重算", + "complete": "分析完成：{frames} 帧，用时 {seconds:.1f} s", + "cancelled": "分析已取消（{frames} 帧已写完，结果标记为未完成）", + "failed": "分析失败：{reason}", + "incomplete": "结果未完成：{reason}", + "blocked": "正在采集：停止后再分析" } } diff --git a/apps/studio/docs/replay-analysis-part-b.md b/apps/studio/docs/replay-analysis-part-b.md new file mode 100644 index 0000000..cbba90d --- /dev/null +++ b/apps/studio/docs/replay-analysis-part-b.md @@ -0,0 +1,125 @@ +# 回放分析结果（Part B：从结果回放） + +本文是 [replay-analysis.md](replay-analysis.md) 的续篇：那一篇定义结果本身，这一篇定义播放端 +怎么读它。打开录像即取用或跑出结果，之后播放、拖动、单步与跳转**全部是读文件**。实现分布在 +`replay_analysis.py`（结果 → 时间轴与画面）、`analysis_worker.py`（pass 的线程边界与取消）、 +`ui/analysis_progress.py`（进度/取消面板）、`replay.py` 与 `replay_controller.py`（一个 +游标、两个模式），以及 `app_window.py` 的接线（第 12 节）。各节编号承接 Part A。 + +## 9. 一个时间轴，两个模式 + +游标只有一个：`ReplayEngine.position()` 的 `t_rel_s`。模式说明**画面从哪里读**： + +| 模式 | 画面来源 | 说明 | +|---|---|---| +| `recording` | 录像自己的 `raw_*.mp4`，按需解码 | 只有原始相机图像：没有光流，也没有深度 | +| `analysis` | 存下来的分析结果，按帧读 | 有 pass 记录下来的光流样本与候选网格 | + +图表、二维画布、三维面板都读这一个游标，模式由 `ReplayEngine.mode()` 回答；「从视频重算」 +因此不再是第三种来源，而是**显式重跑**这一遍的开关（第 10 节）。时间基准是**信号时间** +（Part A 第 5 节），进度、位置、单步与跳转都说它。 + +**帧身份只来自录像自己的计数器。** 证据录像 id 是 20…52 而非 0…32，所以 +`AnalysisSession.frame(frame_id)` 对结果里没有的 id 报 `AnalysisNotFoundError`：把索引当 id +传进去不会被"顺手改成能读的那一帧"，那会返回一张看起来完全正确的图。按时间取帧走 +`frame_id_at(t)`，取**该时刻或之前**的那一帧 —— 与顺序播放经过的是同一帧。 + +## 10. 打开即分析 + +打开录像后固定做一件事：**有兼容结果就用，没有就跑**。`stored_result()` 命中是一次文件读， +面板显示"已有结果，无需重算"；没有命中就在**工作线程**上跑 pass，`Progress` 按 0.1 s 合流后 +经 Qt 信号回到 GUI 线程，面板显示 `已写/预计`、分段号与已用时间。pass 期间时间轴留在 +`recording` 模式，完成后才切到 `analysis` —— 半截结果永远不上屏。面板的取消按钮置一次取消 +事件，服务在检查点停下，目录带 `INCOMPLETE` 标记留下，**不提供回放**（第 13 节）。「从视频 +重算」语义不变，与打开路径走同一个 `on_analysis_requested()`，因此不可能出现两条并行模式。 + +分析的四条终态消息**都不写状态栏**，只更新进度面板：pass 是异步的，可能在操作员做了别的事 +几秒后才结束，一条活得比原因久的状态栏消息会把无关操作说成失败。原因留在面板上，也由 +`analysis_error()` 提供。 + +**并发与工作目录。** 开链路会往当前目录写 2 MB 预览图（Part A 第 6 节），所以 pass 跑在**本 +进程**的一个线程里，启动前要求没有采集会话、没有预览、也没有第二遍分析，并且**本模块从不改 +工作目录**。锁只覆盖开链路那一小段（`_working_directory` 在 `_OpenChain.open()` 内进出）， +逐帧处理在锁之外，GUI 线程这期间正常刷新。关窗时先取消并等待 pass。 + +## 11. 三视图如何被同一个位置喂 + +| 视图 | 读什么 | 怎么读 | +|---|---|---| +| 图表 | `forces.csv`（记录值） | `charts.set_cursor(t)` 用同一游标；结果里不存力 | +| 二维画布 | 结果的 `flow` 与 `valid_mask` | `ImageFrame(source="analysis")` 走窗口唯一的 2D 出口 `_show_image_frame`，箭头、暗点、配色与域边界与实时/重算**同一套代码** | +| 三维面板 | 同一帧的 `SdkPacket` | `session.packet(frame)` → `Scene3DView.submit_packet`，走同一个 adapter、域掩码 latch 与有效性求交 | + +**画面是重建的，并且是精确重建。** 二维需要的 `[H,W,2]` 流场是把存下来的 `dx`/`dy` 写回 +**记录下来的候选位置**、其余格子填 `NaN`（选区器把非有限值当作非法，邻居格子因此不可能贡献 +出 pass 没记录过的样本）。写回要除以 +`DISPLACEMENT_PX_PER_FLOW × flow_arrow_scale × flow_length_margin`。 + +`dx`/`dy` 是**画出来的**位移，已经带上长度余量那一步的缩放：只用前两个因子会让每根箭头短 +20%（finger 余量 0.8），330 个样本里 319 个不同、活动集也不同 —— 实测的，不是推导的。余量补 +上后，重建流场重新过一遍共享选区器，样本集与存下来的**逐元素相等**（33 帧全覆盖）。三维面板 +直接拿存下来的 `FlowSamples`，所以"两张图是同一个测量"由构造决定。 + +## 12. 显示设置是读 + +阈值、配色范围、暗点开关都不在失效键里（Part A 第 3 节），所以改动它们必须**只重画**：阈值 +经 `AnalysisReader.samples(policy=…)` 作用在存下来的 `magnitude` 上，配色范围与暗点开关只 +存在于绘制调用里。走线是 `Scene3DPanel.settingsChanged` → `on_display_settings()` → +`session.serve_policy()` → `ReplayEngine.redraw()`，重画不移动游标、不写文件。 + +**栅格类设置被拒绝**：`flow_grid_spacing`、`flow_edge_margin`、`max_arrows`、 +`flow_length_margin`、`flow_arrow_scale` 决定候选位置**在哪里**，而记录下来的位置是 pass 当时 +测的那些。`serve_policy()` 返回 `False` 并提示哪一项动了，画面继续按结果自己的栅格绘制 —— 画 +一份"看起来一样、其实采在别处"的箭头才是真正错误的选项。 + +`app_window.py` 的接线只有三处，供 Part C 使用：`_analysis_service(recording, cancelled, *, +on_progress)`（`chain_spec_for` 从录像自己的 `session.json` 取链路，因此结果可复现）、 +`_show_analysis_frame(packet)`、`_connect_analysis_display(mesh_panel)`（把 +`Scene3DPanel.settingsChanged` 接到控制器；三维视图首次使用时才建，用有界的 40 × 250 ms 重试 +等它出现）。工具栏、复位与 transport 控件、`display_prefs.py`、`_on_reset_view`/F8 未改。 + +## 13. 未完成与取消 + +`AnalysisSession.__init__` 在 `reader.complete` 为假时抛 `ResultPolicyError`； +`ResultLibrary.open(..., allow_incomplete=True)` 也拒绝 ——「可查看」与「可回放」是两个不同的 +断言。面板的 `cancelled` 与 `failed` 是两个独立状态，都不会显示成完成；这两种情况下时间轴留 +在 `recording` 模式。 + +## 14. 验收证据 + +`apps/studio/tests/test_replay_analysis.py`（15 项，真实素材）： + +* **跳转等于顺序播放**：顺序走过全部 33 帧记下每帧，再按 `time_of(frame_id)` 直接跳到 5 个 + 采样位置。容差是零：帧 id、`rows/cols/dx/dy/active` 逐元素相等，流场按位相等（非候选格按 + 构造是 `NaN`，用 `equal_nan` 比较）。跳转精确的原因是给它的时间就是那一帧自己的 `t_rel_s`。 +* **显示设置不写结果**：`scan_results(cache)` 的目录集合在改阈值（0.2 与 1e9）、改显示增益、 + 重复同一栅格前后完全相同。 +* **取消不被当完整**：pass 在第一帧后取消 → 面板报 `cancelled`、`analysis_active()` 为假、 + 引擎停在 `recording` 模式、目录里 `complete=false` 且默认读端抛 + `AnalysisIncompleteError`。 +* **回归**：`apps/studio/tests` 440 项 / 0 失败 / 18 跳过（原 425 项：新增 15 项，既有断言一条 + 未改）。把分析终态写进状态栏时， + `test_the_replay_reset_is_the_display_reset_and_never_the_device` 失败过 —— 合成录像的配置 + 缺失让 pass 在断言之前结束并覆盖了 `复位显示` 的提示；修的是消息落在哪里。 + +真实 GL 快照（`QT_QPA_PLATFORM=windows` + `QT_OPENGL=desktop`，真实窗口路径、pass 写入 +scratch cache、游标停在存下来的帧 id 上）： + +| 文件 | 内容 | +|---|---| +| `.scratch/replay-analysis-b/snapshots/analysis-frame-21-2d.png` | 帧 21（t=0.0641 s）二维画布，308×331 | +| `.scratch/replay-analysis-b/snapshots/analysis-frame-21-3d.png` | 同一帧三维面板，676×610 | +| `.scratch/replay-analysis-b/snapshots/analysis-frame-36-2d.png` | 帧 36（t=0.9601 s）二维画布，308×331 | +| `.scratch/replay-analysis-b/snapshots/analysis-frame-36-3d.png` | 同一帧三维面板，676×594 | + +二维画面是候选网格上的暗点阵（该帧活动样本很少），三维是手指岛面加同一套样本点；"同一个样本 +集"由上面的逐元素断言保证，快照证明这条链路真的走到了 GL。 + +## 15. 已知限制 + +* 结果里没有力，也没有深度：力曲线始终读记录的 `forces.csv`，深度视图在回放模式下不可用（没有 + 可显示的数据，面板按空处理而不是画一张假的）。 +* 结果覆盖的帧是 pass 走过的那些：录像比结果长时游标可以离开结果范围（面板说"这个分析结果里 + 没有可显示的帧"），不会静默重复最后一帧。 +* 栅格类显示设置不生效（第 12 节）；换栅格就是重跑一遍，那是**另一个键**。 +* 一个进程同时只有一遍分析在跑；采集会话运行期间不启动分析。 diff --git a/apps/studio/src/orisys_studio/analysis_worker.py b/apps/studio/src/orisys_studio/analysis_worker.py new file mode 100644 index 0000000..c9caf18 --- /dev/null +++ b/apps/studio/src/orisys_studio/analysis_worker.py @@ -0,0 +1,201 @@ +"""Running the analysis pass off the GUI thread, with progress and a cancel. + +Part A's :class:`~orisys_app_shared.analysis.service.AnalysisService` is a blocking +call that costs about half a wall-second per video-second. This module is the thread +boundary around it: the pass runs on a plain Python thread, its ``Progress`` callback +is coalesced and re-emitted through Qt signals (delivered to the GUI thread by Qt's +queued connections), and ``cancel()`` sets the service's own cancellation event. + +**A plain thread, not a ``QThread``.** The pass holds no Qt and no widgets, and the +signals are emitted through a :class:`~PySide6.QtCore.QObject` that stays on the GUI +thread. Moving a worker object onto a ``QThread`` would buy nothing here and cost a +C++ lifetime to get right on teardown — including the teardown that happens while a +pass is still running. + +**In this process, never concurrently.** The one global side effect is stated by the +service itself and respected here: constructing a box ``Sensor`` writes about 2 MB of +``config/compound_eye.png`` into the *process working directory*, so that open owns the +CWD under a process-wide lock for its duration. This module therefore never changes the +working directory, and it refuses to start a second pass while one is running; the +controller additionally refuses to start while a session or a recompute pass is live +(see ``ReplayController.on_analysis_requested``). +""" + +from __future__ import annotations + +import logging +import threading +import time +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from orisys_app_shared.analysis.service import AnalysisOutcome, Progress +from PySide6.QtCore import QObject, Signal + +logger = logging.getLogger(__name__) + +#: How often the pass's per-frame progress may reach the GUI. The pass writes one frame +#: per call and a ten-minute recording holds ~18,000 of them; a signal per frame would +#: spend more time in the event loop than in the analysis. +PROGRESS_INTERVAL_S = 0.1 + + +@dataclass(frozen=True) +class AnalysisReport: + """What one pass did, in the terms the progress surface shows.""" + + outcome: AnalysisOutcome | None + cancelled: bool + failed: bool + reason: str + frames_written: int + frames_expected: int + elapsed_s: float + directory: Path | None + + @property + def complete(self) -> bool: + return bool( + self.outcome is not None + and not self.cancelled + and not self.failed + and self.outcome.complete + ) + + +def report_from(outcome: AnalysisOutcome | None, *, cancelled: bool) -> AnalysisReport: + """Turn a service outcome into the surface's report.""" + if outcome is None: + return AnalysisReport( + outcome=None, + cancelled=bool(cancelled), + failed=not cancelled, + reason="cancelled" if cancelled else "no_result", + frames_written=0, + frames_expected=0, + elapsed_s=0.0, + directory=None, + ) + return AnalysisReport( + outcome=outcome, + cancelled=bool(cancelled), + failed=not bool(outcome.complete) and not cancelled, + reason=str(outcome.reason or ""), + frames_written=int(outcome.frames_written), + frames_expected=int(outcome.frames_expected), + elapsed_s=float(outcome.elapsed_s), + directory=Path(outcome.directory), + ) + + +class AnalysisWorker(QObject): + """One analysis pass, on its own thread, reporting progress as Qt signals. + + ``progressed`` and ``finished`` are declared on this object, which lives on the GUI + thread; emitting them from the pass's thread crosses the thread boundary, so Qt + queues them and the slots run on the GUI thread. That is the whole reason this + class exists rather than a bare ``threading.Thread`` at the call site. + """ + + #: Emitted on the GUI thread with each coalesced ``Progress``. + progressed = Signal(object) + #: Emitted once, on the GUI thread, with the ``AnalysisReport``. + finished = Signal(object) + + def __init__( + self, + service_factory: Callable[[threading.Event, Callable[[Progress], None]], Any], + *, + parent: QObject | None = None, + ) -> None: + super().__init__(parent) + self._service_factory = service_factory + self._thread: threading.Thread | None = None + self._cancel = threading.Event() + self._last_emit = 0.0 + self._lock = threading.Lock() + self._service: Any = None + self._started_at = 0.0 + self._failure = "" + + # ------------------------------------------------------------------ control + def running(self) -> bool: + thread = self._thread + return thread is not None and thread.is_alive() + + def start(self) -> bool: + """Start the pass. Returns ``False`` when one is already running.""" + if self.running(): + return False + self._cancel.clear() + self._last_emit = 0.0 + self._failure = "" + self._started_at = time.perf_counter() + thread = threading.Thread(target=self._run, name="analysis-pass", daemon=True) + self._thread = thread + thread.start() + return True + + def cancel(self) -> None: + """Ask the pass to stop at its next checkpoint. + + The service checks the event between frames and between segments, so this stops + within one frame's work. Whatever was written stays on disk *marked incomplete*, + which is what keeps a cancelled pass from ever being played back. + """ + self._cancel.set() + + def cancelled(self) -> bool: + return self._cancel.is_set() + + def service(self) -> Any: + """The live service, for a caller that wants its dependency report.""" + with self._lock: + return self._service + + def wait(self, timeout_s: float = 10.0) -> bool: + """Block until the pass's thread has finished. For teardown and for tests.""" + thread = self._thread + if thread is None: + return True + thread.join(float(timeout_s)) + return not thread.is_alive() + + # ------------------------------------------------------------------- thread + def _run(self) -> None: + outcome: AnalysisOutcome | None = None + try: + service = self._service_factory(self._cancel, self._on_progress) + with self._lock: + self._service = service + outcome = service.run() + except Exception as exc: # noqa: BLE001 - reported, never swallowed + logger.exception("the analysis pass raised") + self._failure = f"{type(exc).__name__}: {exc}" + finally: + with self._lock: + self._service = None + cancelled = self._cancel.is_set() + report = report_from(outcome, cancelled=cancelled) + if outcome is None and not cancelled: + report = AnalysisReport( + outcome=None, + cancelled=False, + failed=True, + reason=self._failure or "failed", + frames_written=0, + frames_expected=0, + elapsed_s=time.perf_counter() - self._started_at, + directory=None, + ) + self.finished.emit(report) + + def _on_progress(self, progress: Progress) -> None: + """Runs on the pass's thread: coalesce, then hand the value across.""" + now = time.perf_counter() + if now - self._last_emit < PROGRESS_INTERVAL_S: + return + self._last_emit = now + self.progressed.emit(progress) diff --git a/apps/studio/src/orisys_studio/app_window.py b/apps/studio/src/orisys_studio/app_window.py index a7697bf..5be0e83 100644 --- a/apps/studio/src/orisys_studio/app_window.py +++ b/apps/studio/src/orisys_studio/app_window.py @@ -58,6 +58,7 @@ from orisys_studio.scene3d.prefs import settings_to_prefs from orisys_studio.session_controller import SessionController from orisys_studio.ui import record_status as record_states from orisys_studio.ui.about_dialog import AboutDialog +from orisys_studio.ui.analysis_progress import AnalysisProgress from orisys_studio.ui.arrow_panel import ArrowPanel, bgr_to_qimage from orisys_studio.ui.brand_bar import BrandBar from orisys_studio.ui.cal_panel import CalPanel @@ -121,6 +122,13 @@ VIEW_SHOWS_3D = ("custom", "live", "replay") #: "not in the table" must not mean "standing at its minimum width". DEFAULT_3D_PANE_WIDTH = 620 +#: How long the window waits for the 3D view (built on first use) before giving up on +#: binding a display-settings change to the replay: 40 × 250 ms. Nothing depends on the +#: retry succeeding — a display change then simply redraws on the next cursor move — so +#: it is a bounded convenience, never a requirement. +ANALYSIS_DISPLAY_ATTEMPTS = 40 +ANALYSIS_DISPLAY_INTERVAL_MS = 250 + #: Kept under its historical name for ``tools/ui_snapshot.py``, which imports it from #: here. There is one conversion in the application (``ui.arrow_panel.bgr_to_qimage``). @@ -251,6 +259,11 @@ class StudioMainWindow(QMainWindow): self.toolbar = ToolbarWidget(theme=self._theme) root.addWidget(self.toolbar) self.status_chip = self.toolbar.status_chip + # Part B: the analysis pass a replay is served from. The surface sits directly + # under the rail, so "what is it doing and can I stop it" is answered where the + # transport is; it is hidden until a pass (or a cache hit) has something to say. + self.analysis_progress = AnalysisProgress(theme=self._theme) + root.addWidget(self.analysis_progress) self.content_splitter = QSplitter(Qt.Orientation.Horizontal) self.mesh3d_panel = Mesh3dPanel(theme=self._theme) @@ -284,9 +297,15 @@ class StudioMainWindow(QMainWindow): charts=self.force_plots, image_panel=self.arrow_panel, mesh_panel=self.mesh3d_panel, + analysis_progress=self.analysis_progress, parent=self, ) self.replay_ctl.wire() + # Part B: a display-settings change while a replay is served from an analysis + # result is a *redraw*, never a new pass. The scene panel owns those controls, + # so its own signal is the one place that sees every one of them (threshold, + # colour range, idle dots, arrows). + self._await_analysis_display() # How the numbers are shown (units, chart range, arrows, the fps readout) is a # preference of its own: the dialog that changes the chart settings stays here, # the values and what they do to the panels live there. @@ -2706,6 +2725,91 @@ class StudioMainWindow(QMainWindow): return self.arrow_panel.show_frame(bgr_to_qimage(rendered)) + # ------------------------------------------------------ analysis (Part B) + # What this window owes the replay controller, and nothing more. The *policy* + # (when a pass runs, what is done with its result, what a display change means) + # lives in ``ReplayController``; these are the three window-level things it cannot + # own: the sensor chain a pass is built from, the 3D pane's producer, and the + # scene panel's settings signal. + def _analysis_service( + self, recording: Path, cancelled=None, *, on_progress=None + ) -> Any: + """One ``AnalysisService`` over ``recording``, ready to run. + + The chain comes from the recording's own ``session.json`` + (``chain_spec_for``), which is what makes a stored result reproducible: the + configuration the footage was taken under, not whatever happens to be installed + now. ``cancelled`` is the pass's cancellation event and ``on_progress`` its + progress callback; both are ``None`` for a caller that only wants to ask whether + a compatible result is already stored. + """ + from orisys_app_shared.analysis.service import AnalysisService, chain_spec_for + from orisys_app_shared.record import RecordingReader + + session = RecordingReader(Path(recording)).session() + chain = chain_spec_for(session) + return AnalysisService( + Path(recording), + chain=chain, + cancelled=cancelled, + on_progress=on_progress, + ) + + def _show_analysis_frame(self, packet) -> None: + """Hand one stored analysis frame to the 3D pane's producer. + + The pane snapshots a packet through its ordinary path, so a replay's figures go + through the same adapter, domain latch and mask intersection as a live run. The + producer is O(1) and Qt-free, and it is the pane that decides whether it is on. + """ + view = self.mesh3d_panel.view + if view is None: + return + self._connect_analysis_display(self.mesh3d_panel) + view.submit_packet(packet) + + def _connect_analysis_display(self, mesh_panel) -> bool: + """Route every display-settings change through the replay controller. + + Connected to the scene panel's ``settingsChanged`` rather than to each control, + so a new control cannot be added without the replay noticing it. The controller + decides what a change *means*: a redraw from the stored frame, and never a new + analysis pass. + + The scene panel is built with the 3D view, which is built on first use, so this + is idempotent and answers whether the seam exists yet; the caller retries (see + ``_await_analysis_display``) rather than reaching into another controller's + widgets. + """ + if getattr(self, "_analysis_display_connected", False): + return True + view = mesh_panel.view + panel = getattr(view, "panel", None) if view is not None else None + if panel is None or not hasattr(panel, "settingsChanged"): + return False + panel.settingsChanged.connect(self.replay_ctl.on_display_settings) + self._analysis_display_connected = True + return True + + def _await_analysis_display(self) -> None: + """Wait, briefly and off the critical path, for the 3D view to exist. + + The display-settings signal is the only part of Part B that needs the 3D view, + and that view is created the first time the pane is used — possibly long after + this window is built. A short retry chain is the honest way to reach it from + here: the alternative is to reach into ``ViewLayoutController`` and create the + view early, which would change when GL is initialised for every user. + """ + if self._connect_analysis_display(self.mesh3d_panel): + return + self._analysis_display_attempts = ( + int(getattr(self, "_analysis_display_attempts", 0)) + 1 + ) + if self._analysis_display_attempts > ANALYSIS_DISPLAY_ATTEMPTS: + logger.debug("the 3D display settings stayed unavailable; no signal bound") + return + QTimer.singleShot(ANALYSIS_DISPLAY_INTERVAL_MS, self._await_analysis_display) + def _restore_geometry(self) -> None: geo = self.prefs.get("window_geometry") if isinstance(geo, str) and geo: @@ -2752,6 +2856,13 @@ class StudioMainWindow(QMainWindow): # Detach the sinks first: a worker that is still winding down must not keep a # reference to this window (and through it to widgets Qt is about to destroy). self.session.detach_sinks() + # Part B: an analysis pass opens an SDK chain in this process, so it is stopped + # and waited for before the window goes — the same rule the acquisition session + # follows below. The partial result keeps its INCOMPLETE marker, so it will not + # be served on the next open either. + if self.replay_ctl.cancel_analysis(): + self.replay_ctl.wait_for_analysis() + self.replay_ctl.results.close() # Blocking: on the way out we must not leave a device handle or a # half-written experiment behind. self._stop_run(reason="user_stop", wait=True) diff --git a/apps/studio/src/orisys_studio/replay.py b/apps/studio/src/orisys_studio/replay.py index ce91f6f..fbe636c 100644 --- a/apps/studio/src/orisys_studio/replay.py +++ b/apps/studio/src/orisys_studio/replay.py @@ -1,10 +1,10 @@ """Replay engine: one recording, one cursor, one frame at a time. -This is the headless half of the replay view. It owns the recording reader and -the cursor and it *shows* nothing: the main window renders the frame into the 2D -panel, the curves read the recording through -:class:`~orisys_app_shared.force.source.RecordingCurveSource`, and the transport -on the chart header drives :meth:`set_position`. +This is the headless half of the replay view. It owns the recording reader, the +analysis session and the cursor, and it *shows* nothing: the main window renders the +frame into the 2D panel, the curves read the recording through +:class:`~orisys_app_shared.force.source.RecordingCurveSource`, and the transport on the +chart header drives :meth:`set_position`. It is an *engine*, not a controller: it answers "what is at t?" and knows nothing about the transport's policy (when to start, what the end of the timeline means, whether a @@ -12,9 +12,25 @@ recompute pass is running) or about the widgets. That policy lives in ``ReplayController`` (the window-level controller for this view), which is why the two names are deliberately different — see ``apps/studio/docs/shell-lifecycle.md``. -There is no depth recomputation here: depth is not stored with a recording, the -replay view shows the recorded frames, and re-running the pipeline over a video -is the acquisition path's job (see the studio's "recompute" mode). +**One timeline, two modes.** The cursor is a single ``t_rel_s`` on the **signal** time +base in both modes — the recording's own ``frames.csv`` time, which is what +``RecordingReader`` and ``AnalysisReader`` both index by. The mode says what the frame +under the cursor is *read from*: + +``"recording"`` + the recorded video's own frames, decoded on demand (``RecordingReader``); no flow, + no depth, because a recording stores neither. This is the mode a finished record + opens in, and the one the charts read ``forces.csv`` for. +``"analysis"`` + a stored analysis result (Part A): flow samples and the candidate grid the pass + recorded, read by frame id from the cache. A seek here is a file read — no pass, + no video, no SDK. + +There is no third mode and there are no per-view sources: the charts, the 2D canvas and +the 3D pane all read this one cursor, in whichever mode it is in. Re-running the vision +chain over a video is *not* a mode of this engine: it is the acquisition pipeline's run +(see the studio's 「从视频重算」), which produces its own frames and, when it finishes, +can leave a result this engine then reads. """ from __future__ import annotations @@ -27,8 +43,15 @@ from orisys_app_shared.i18n import t from orisys_app_shared.record import RecordingReader, VideoOnlySource from PySide6.QtCore import QObject, Signal +from orisys_studio.replay_analysis import AnalysisSession, AnalysedFrame + logger = logging.getLogger(__name__) +#: The cursor reads the recording's own frames (decoded on demand). +MODE_RECORDING = "recording" +#: The cursor reads a stored analysis result (a file read per frame, no pass). +MODE_ANALYSIS = "analysis" + def _sdk_version() -> str: """The SDK version in use, or "" when the SDK is not importable.""" @@ -52,18 +75,29 @@ class ReplayEngine(QObject): status_changed = Signal() #: A recording was opened (True) or the previous one was dropped (False). recording_changed = Signal(bool) + #: The timeline's mode changed (``MODE_RECORDING`` / ``MODE_ANALYSIS``), or a + #: result was adopted or dropped while staying in the same mode. Views that cache + #: what they drew must drop it; the number of readers did change. + timeline_changed = Signal() def __init__(self, parent=None) -> None: super().__init__(parent) self._reader: RecordingReader | None = None #: A video with no recording beside it: playable, but it has no measurements. self._video: VideoOnlySource | None = None + #: The stored analysis result the cursor reads, when the mode is analysis. + self._analysis: AnalysisSession | None = None + self._mode = MODE_RECORDING self._unit = "raw" self._position = 0.0 #: The frame cache: one video image per cursor position. self._video_position: float | None = None self._video_bgr: np.ndarray | None = None self._video_reason = "" + #: The analysed frame under the cursor, and the id it was built for. + self._analysis_id: int | None = None + self._analysis_frame: AnalysedFrame | None = None + self._analysis_reason = "" self._warnings = "" # ------------------------------------------------------------- recording @@ -90,6 +124,7 @@ class ReplayEngine(QObject): """Open a bare video for playback (no forces, no session metadata).""" self._release_reader() self._release_video() + self._release_analysis() try: self._video = VideoOnlySource(Path(path)) except Exception: # noqa: BLE001 - reported through the status text @@ -101,7 +136,9 @@ class ReplayEngine(QObject): return False self._reset_frames() self._position = 0.0 + self._mode = MODE_RECORDING self.recording_changed.emit(True) + self.timeline_changed.emit() self._refresh_warnings() self.set_position(0.0) return True @@ -124,8 +161,14 @@ class ReplayEngine(QObject): return RecordingCurveSource(self._reader) def open_recording(self, directory) -> bool: - """Open a recording directory.""" + """Open a recording directory. + + A new recording drops any analysis result that was open with the old one: the + result belongs to a recording, and keeping the cursor on another recording's + frames is how a picture ends up describing footage that is not on screen. + """ self._release_reader() + self._release_analysis() try: self._reader = RecordingReader(directory) self._reader.info() @@ -138,7 +181,9 @@ class ReplayEngine(QObject): return False self._reset_frames() self._position = 0.0 + self._mode = MODE_RECORDING self.recording_changed.emit(True) + self.timeline_changed.emit() self._refresh_warnings() self.set_position(0.0) return True @@ -147,8 +192,11 @@ class ReplayEngine(QObject): """Drop whatever is open (the decoder is released with it).""" self._release_reader() self._release_video() + self._release_analysis() self._reset_frames() + self._mode = MODE_RECORDING self.recording_changed.emit(False) + self.timeline_changed.emit() self.status_changed.emit() def _release_reader(self) -> None: @@ -167,32 +215,129 @@ class ReplayEngine(QObject): except Exception: # noqa: BLE001 - releasing is best effort logger.debug("Could not close the video", exc_info=True) + def _release_analysis(self) -> None: + """Forget the borrowed session. It does **not** close it. + + The reader belongs to the controller's ``ResultLibrary``: closing it here would + leave the controller holding a session whose file handle this engine shut. + """ + self._analysis = None + self._analysis_id = None + self._analysis_frame = None + self._analysis_reason = "" + def _reset_frames(self) -> None: self._video_position = None self._video_bgr = None self._video_reason = "" - # ---------------------------------------------------------------- cursor - def position(self) -> float: - return self._position + # --------------------------------------------------------------- timeline + def mode(self) -> str: + """What the frame under the cursor is read from: the video, or a result.""" + return self._mode + + def analysis(self) -> AnalysisSession | None: + return self._analysis + + def use_analysis(self, session: AnalysisSession) -> None: + """Read the cursor from a stored result instead of decoding the video. + + The session is *borrowed*, not owned: its reader belongs to whoever opened it + (``ResultLibrary`` in the controller), which is what lets a mode change be a + cheap switch rather than a file handle handed between two owners. + """ + self._analysis = session + self._mode = MODE_ANALYSIS + self._analysis_id = None + self._analysis_frame = None + self._analysis_reason = "" + # The cursor keeps its *time*: both timelines are the recording's own signal + # time, so the frame under it is the one the operator was already looking at + # (a different frame id may resolve, and that is the honest answer: the result + # covers the frames the pass wrote). + self.timeline_changed.emit() + self.set_position(self._position) + self._refresh_warnings() + + def use_recording(self) -> None: + """Read the cursor from the recording's own video again.""" + if self._mode != MODE_ANALYSIS: + return + self._release_analysis() + self._mode = MODE_RECORDING + self.timeline_changed.emit() + self.set_position(self._position) + self._refresh_warnings() def duration_s(self) -> float: + if self._mode == MODE_ANALYSIS and self._analysis is not None: + return self._analysis.duration_s() if self._reader is not None: return float(self._reader.info().duration_s) if self._video is not None: return float(self._video.duration_s) return 0.0 + # ---------------------------------------------------------------- cursor + def position(self) -> float: + return self._position + + def frame_id(self) -> int | None: + """The frame id under the cursor, or ``None`` when nothing is open. + + In analysis mode this is the **recording's own frame id** — never an index into + the result. A recording whose ids run 20…52 answers 20…52 here, and a caller + that needs a specific frame asks :meth:`frame_id_for_time` for it. + """ + if self._mode == MODE_ANALYSIS and self._analysis is not None: + return self._analysis.frame_id_at(self._position) + if self._reader is not None: + image, frame = self._reader.video_frame_at(self._position) + return None if frame is None else int(frame.frame_id) + return None + + def frame_id_for_time(self, t_rel_s: float) -> int | None: + """The frame id a position resolves to, without moving the cursor.""" + if self._mode == MODE_ANALYSIS and self._analysis is not None: + return self._analysis.frame_id_at(float(t_rel_s)) + if self._reader is None: + return None + _image, frame = self._reader.video_frame_at(float(t_rel_s)) + return None if frame is None else int(frame.frame_id) + + def time_of_frame(self, frame_id: int) -> float | None: + """The signal time of a frame **id**, or ``None`` when it is not in the result. + + An id the recording does not hold is refused, not coerced: a caller that passes + an index gets ``None`` rather than the time of the frame at that offset. + """ + if self._mode == MODE_ANALYSIS and self._analysis is not None: + return self._analysis.time_of(int(frame_id)) + return None + def set_position(self, t_rel_s: float) -> None: """Move the single cursor: the video frame and the charts follow it.""" - if not self.has_video(): + if not self.has_video() and self._analysis is None: return duration = self.duration_s() self._position = max(0.0, min(float(t_rel_s), duration if duration > 0 else 0.0)) self._refresh_video() + self._refresh_analysis() self.position_changed.emit(self._position) self.frames_changed.emit() + def redraw(self) -> None: + """Re-emit the frame under the cursor without moving it. + + What a display-settings change does: the picture is rebuilt from the same + stored frame, no result is written and no position moves. Anything that caches + what it drew must drop that cache, so the producers behind it are reset too. + """ + self._reset_frames() + self._analysis_id = None + self._analysis_frame = None + self.frames_changed.emit() + def unit(self) -> str: return self._unit @@ -207,15 +352,32 @@ class ReplayEngine(QObject): return self._unit # ----------------------------------------------------------------- frames + def analysed_frame(self) -> AnalysedFrame | None: + """The stored frame under the cursor, or ``None`` outside analysis mode.""" + if self._mode != MODE_ANALYSIS: + return None + self._refresh_analysis() + return self._analysis_frame + def video_frame_bgr(self) -> np.ndarray | None: """The recorded frame under the cursor, as BGR. The engine hands over pixels, not a prepared widget image: the single 2D - backend (``acquisition.image_backend``) decides what the picture is. + backend (``acquisition.image_backend``) decides what the picture is. In + analysis mode the pixels are the result's own canvas — the flow field the pass + recorded, drawn on the domain grid — not a video frame. """ + if self._mode == MODE_ANALYSIS: + frame = self.analysed_frame() + return None if frame is None else frame.base_bgr self._refresh_video() return self._video_bgr + def analysis_frame_bgr(self) -> np.ndarray | None: + """The same pixels :meth:`video_frame_bgr` returns in analysis mode.""" + frame = self.analysed_frame() + return None if frame is None else frame.base_bgr + def video_image(self): """The same frame as a ``QImage`` (kept for callers that predate the plan).""" from orisys_studio.ui.arrow_panel import bgr_to_qimage @@ -224,6 +386,9 @@ class ReplayEngine(QObject): return None if frame is None else bgr_to_qimage(frame) def video_message(self) -> str: + if self._mode == MODE_ANALYSIS: + self._refresh_analysis() + return self._analysis_reason self._refresh_video() return self._video_reason @@ -257,12 +422,44 @@ class ReplayEngine(QObject): self._video_reason = "" self._video_bgr = image + def _refresh_analysis(self) -> None: + """Read the stored frame under the cursor, once per frame id.""" + session = self._analysis + if session is None or self._mode != MODE_ANALYSIS: + return + frame_id = session.frame_id_at(self._position) + if frame_id is None: + self._analysis_frame = None + self._analysis_id = None + self._analysis_reason = t("playback.analysis_empty") + return + if self._analysis_frame is not None and self._analysis_id == frame_id: + return + try: + frame = session.frame(frame_id) + except Exception as exc: # noqa: BLE001 - a broken result is reported, not raised + logger.exception("Could not read frame %s from the analysis result", frame_id) + self._analysis_frame = None + self._analysis_id = None + self._analysis_reason = t( + "playback.analysis_unreadable", error=type(exc).__name__ + ) + return + self._analysis_frame = frame + self._analysis_id = frame_id + self._analysis_reason = "" + # ---------------------------------------------------------------- notes def warnings_text(self) -> str: return self._warnings def _refresh_warnings(self) -> None: self._warnings = "" + if self._mode == MODE_ANALYSIS and self._analysis is not None: + # Nothing is stale about a result just because the SDK moved: that is + # reported, not hidden (``AnalysisReader.stale_reasons``), and the caller + # that opened the session is the one holding the facts to compare with. + return if self._reader is None: if self._video is not None: # Not a warning: a statement of what this file *is*, so nobody reads an diff --git a/apps/studio/src/orisys_studio/replay_analysis.py b/apps/studio/src/orisys_studio/replay_analysis.py new file mode 100644 index 0000000..206c614 --- /dev/null +++ b/apps/studio/src/orisys_studio/replay_analysis.py @@ -0,0 +1,461 @@ +"""Serving a replay from an analysis result: one position, one frame, no pass. + +``replay.py`` is the *engine* for a recording: a reader, a cursor, and the recorded +picture under it. This module is the same thing for a **stored analysis result** — +Part A's output — and it exists so that playing a recording stops costing 0.48–0.51 +wall-seconds per video-second. + +Two decisions worth stating out loud, because both are load-bearing: + +**Position is signal time, and frames are addressed by frame id.** A stored frame is +located by its ``t_rel_s`` (``reader.frame_at``), and the frame it resolves to carries +the recording's own ``frame_id`` — the acquisition session's counter, which does not +start at 0 or 1. Every ``frame_id`` this module is handed is checked against the result +and refused when it is absent (:meth:`AnalysisSession.frame`), so a caller that passes a +*position* where an *id* belongs gets an error rather than the frame at that offset. +Container video time is never derived from ``t_rel_s`` and never converted into it: the +two time bases are different numbers for the same footage (33 container frames at 30 fps +is 1.100 s; the signal those frames carry is 1.920 s at 16.664 fps) and this module only +ever speaks the second one. + +**The result is read, never re-derived — but the display policy is still honoured.** +``flow_on_px`` arrives from the display settings and is applied to the stored +``magnitude`` by ``AnalysisReader.samples``, which is why a threshold change is a read +and never a new pass. The colour range and the idle-dot switch are not even inputs +here: they live in the draw call, so changing one redraws and writes nothing. + +The picture is rebuilt from the stored samples: ``dx``/``dy`` are the canvas +displacements the pass selected (uv pixels, already divided by the policy's arrow +scale), so the flow field is recovered by dividing them back out by +:data:`~orisys_app_shared.scene3d.visual_policy.DISPLACEMENT_PX_PER_FLOW` and writing +those values at the recorded candidate locations. The 2D canvas then runs its ordinary +path — the shared selector, the same idle dots, the same colour ramp — over that field. +This is exact, not approximate, and ``apps/studio/tests/test_replay_analysis.py`` +asserts it against a pass over real footage: re-selecting the recovered field through +the display policy reproduces the stored sample set element for element. + +The 3D pane is handed the **stored** sample set directly (:attr:`AnalysedFrame.samples`) +— the object Part A documents as "what the canvas and the composer already draw" — so +the figure the pane composes from a replay is the figure the pass recorded, however the +display policy is set. +""" + +from __future__ import annotations + +import logging +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.acquisition.models import SdkPacket +from orisys_app_shared.analysis.reader import AnalysisReader +from orisys_app_shared.scene3d.visual_policy import ( + DISPLACEMENT_PX_PER_FLOW, + FlowSamples, + VisualPolicy, +) + +logger = logging.getLogger(__name__) + +__all__ = [ + "AnalysedFrame", + "AnalysisSession", + "ResultLibrary", + "ResultPolicyError", + "policy_divergence", +] + +#: The UV domain to rebuild the flow field in when the result carries no validity mask +#: keyframe (the box chain has none). Only the recorded candidate locations are ever +#: read back out of it, so its size is a canvas for the samples rather than a claim +#: about the device — but it has to hold every candidate, which +#: :meth:`AnalysisSession._recover_flow` checks. +DEFAULT_DOMAIN_HW = (1080, 1920) + +#: The policy keys that decide *where* the candidate locations are. A display setting +#: that moves one of these cannot be served from a stored result: the recorded grid is +#: the pass's, and re-gridding it would draw samples nothing measured while looking +#: exactly like the stored result. ``flow_on_px``/``flow_off_px`` are deliberately not +#: among them — they decide which recorded samples are active, and ``magnitude`` is +#: stored precisely so that question can be asked again later. +GRID_KEYS = ( + "flow_grid_spacing", + "flow_edge_margin", + "max_arrows", + "flow_length_margin", + "flow_arrow_scale", +) + + +class ResultPolicyError(RuntimeError): + """A result that cannot be served: incomplete, or not in this display's domain.""" + + +@dataclass(frozen=True) +class AnalysedFrame: + """One frame of a stored result, ready for both views. + + ``flow``/``valid_mask`` are what the 2D canvas draws from; ``samples`` is the + stored selection the 3D pane overlays. Both describe the *same* frame id at the + *same* ``t_rel_s``, which is what makes one position drive two figures. + """ + + frame_id: int + t_rel_s: float + flow: np.ndarray + valid_mask: np.ndarray | None + samples: FlowSamples + base_bgr: np.ndarray + depth_nonzero: int = 0 + dropped: bool = False + + @property + def domain_hw(self) -> tuple[int, int]: + return (int(self.flow.shape[0]), int(self.flow.shape[1])) + + @property + def active_count(self) -> int: + return int(self.samples.active_count) + + @property + def candidates(self) -> int: + return int(self.samples.rows.size) + + +def policy_divergence(stored: dict[str, Any], display: VisualPolicy) -> list[str]: + """The location settings a display policy moves away from the stored ones. + + Empty means the stored grid can serve this display: the candidate locations are + the recorded ones and only the activity verdict changes. A non-empty list is the + reason a replay cannot honour a display setting, named so the message can say + which one instead of "unsupported". + """ + wanted = display.normalised() + moved: list[str] = [] + for key in GRID_KEYS: + if key not in stored: + continue + recorded = float(stored[key]) + if abs(recorded - float(getattr(wanted, key))) > 1e-9: + moved.append(key) + return moved + + +class AnalysisSession: + """A stored result, read as a timeline: t → frame id → the two views' inputs.""" + + def __init__( + self, + reader: AnalysisReader, + *, + policy: VisualPolicy | None = None, + display_policy: VisualPolicy | None = None, + session_id: int = 0, + recording: Path | None = None, + ) -> None: + if not bool(getattr(reader, "complete", True)): + # A cancelled or failed pass must never be played from. The reader refuses + # such a result unless ``allow_incomplete=True`` was passed, and this + # refuses it even then: "inspectable" and "playable" are different claims. + raise ResultPolicyError( + "this analysis result is incomplete and cannot be played from" + ) + self.reader = reader + self.policy = (policy or VisualPolicy()).normalised() + self.display_policy = (display_policy or self.policy).normalised() + self.session_id = int(session_id) + self.recording = None if recording is None else Path(recording) + self._frame_ids = np.asarray(reader.frame_ids, dtype=np.int64) + self._times = np.asarray(reader.times, dtype=np.float64) + self._mask: np.ndarray | None = None + self._mask_loaded = False + self._cache_id: int | None = None + self._cache: AnalysedFrame | None = None + + # ------------------------------------------------------------------ identity + @property + def frame_count(self) -> int: + return int(self._frame_ids.size) + + @property + def first_frame_id(self) -> int: + return int(self._frame_ids[0]) if self._frame_ids.size else 0 + + @property + def last_frame_id(self) -> int: + return int(self._frame_ids[-1]) if self._frame_ids.size else 0 + + @property + def frame_ids(self) -> np.ndarray: + """Every stored frame id, in stored order (the recording's own counter).""" + return self._frame_ids + + @property + def result_key(self) -> str: + return str(self.reader.key) + + @property + def directory(self) -> Path: + return Path(self.reader.directory) + + def has(self, frame_id: int) -> bool: + """Whether the result holds this frame id. Never an index lookup.""" + return bool(self._frame_ids.size) and bool( + np.any(self._frame_ids == int(frame_id)) + ) + + def duration_s(self) -> float: + """The result's span in **signal** time, from ``t_rel_s`` (never the container).""" + if self._times.size == 0: + return 0.0 + return float(max(0.0, self._times[-1] - self._times[0])) + + def origin_s(self) -> float: + return float(self._times[0]) if self._times.size else 0.0 + + def stored_policy(self) -> dict[str, Any]: + """The selection policy the pass recorded, for comparison with the display's.""" + return dict(getattr(self.reader.provenance, "policy", {}) or {}) + + # ------------------------------------------------------------------- policy + def serve_policy(self, display_policy: VisualPolicy) -> bool: + """Adopt a display policy, if the stored grid can still serve it. + + Returns ``False`` — and changes nothing — when the policy moves a location + setting (see :data:`GRID_KEYS`). Those decide where the candidates are, and the + recorded locations are the ones the pass measured. The threshold is not one of + them: it decides which recorded samples are active. + """ + if policy_divergence(self.stored_policy(), display_policy): + return False + self.display_policy = display_policy.normalised() + return True + + def threshold(self) -> float: + return float(self.display_policy.flow_on_px) + + # ------------------------------------------------------------------ looking + def frame_id_at(self, t_rel_s: float) -> int | None: + """The stored frame at or just before ``t_rel_s``, by *signal* time.""" + if self._times.size == 0: + return None + if bool(np.all(np.diff(self._times) >= 0)): + slot = int(np.searchsorted(self._times, float(t_rel_s), side="right")) - 1 + slot = max(0, min(slot, self._times.size - 1)) + else: # a foreign result whose times are not ordered: scan rather than guess + candidates = np.flatnonzero(self._times <= float(t_rel_s)) + slot = int(candidates[-1]) if candidates.size else 0 + return int(self._frame_ids[slot]) + + def time_of(self, frame_id: int) -> float | None: + """The stored ``t_rel_s`` of one frame id, or ``None`` when it is absent.""" + hits = np.flatnonzero(self._frame_ids == int(frame_id)) + if not hits.size: + return None + return float(self._times[int(hits[0])]) + + def frame(self, frame_id: int) -> AnalysedFrame: + """One frame by its recording id. + + Raises :class:`~orisys_app_shared.analysis.reader.AnalysisNotFoundError` for an + id the result does not hold. That refusal is the point: a frame *index* is not + a frame *id*, and serving the frame at that offset would answer a question + nobody asked with a picture that looks right. + """ + wanted = int(frame_id) + if self._cache is not None and self._cache_id == wanted: + return self._cache + payload = self.reader.frame(wanted) + built = self._compose(payload) + self._cache = built + self._cache_id = wanted + return built + + def frame_at(self, t_rel_s: float) -> AnalysedFrame | None: + """One frame by signal time, or ``None`` for an empty result.""" + frame_id = self.frame_id_at(t_rel_s) + return None if frame_id is None else self.frame(frame_id) + + def frames(self) -> Iterator[AnalysedFrame]: + """Every stored frame, in stored order (a pass over a result, not a replay).""" + for frame_id in self._frame_ids: + yield self.frame(int(frame_id)) + + # ------------------------------------------------------------------ packets + def packet(self, frame: AnalysedFrame, *, t_rel_s: float | None = None) -> SdkPacket: + """One frame in the shape the 3D pane's producer consumes. + + The pane is fed through its ordinary path — it snapshots a packet — so a + replay's figures go through the same adapter, the same domain latch and the + same validity intersection as a live or recomputed frame. Feeding it anything + else would make "the pane looks the same in every mode" an accident of two code + paths happening to agree. + + The forces are zero and ``forces_valid`` is ``False``: the result stores no + force, ``forces.csv`` is where that measurement lives, and a zero here would be + a measurement nothing made. + """ + moment = float(frame.t_rel_s if t_rel_s is None else t_rel_s) + return SdkPacket( + raw_frame=frame.base_bgr, + processed_frame=frame.base_bgr, + timestamp=moment, + mono_t=moment, + fps=0.0, + process_ms=0.0, + session_id=self.session_id, + frame_id=int(frame.frame_id), + valid=True, + stale_forces=True, + forces_valid=False, + normal_force=0.0, + shear_x=0.0, + shear_y=0.0, + flow=frame.flow, + valid_mask=frame.valid_mask, + flow_valid=True, + flow_frame_id=int(frame.frame_id), + depth=None, + depth_valid=False, + depth_frame_id=None, + ) + + # ------------------------------------------------------------------ building + def mask(self) -> np.ndarray | None: + """The validity mask the result was built with, read once.""" + if not self._mask_loaded: + self._mask_loaded = True + self._mask = self.reader.mask() + return self._mask + + def domain_hw(self) -> tuple[int, int]: + mask = self.mask() + if mask is None: + return DEFAULT_DOMAIN_HW + return (int(mask.shape[0]), int(mask.shape[1])) + + def _compose(self, payload: Any) -> AnalysedFrame: + mask = self.mask() + shape_hw = self.domain_hw() + samples = payload.samples(threshold=self.threshold()) + flow = self._recover_flow(samples, shape_hw) + base = np.zeros((shape_hw[0], shape_hw[1], 3), dtype=np.uint8) + return AnalysedFrame( + frame_id=int(payload.frame_id), + t_rel_s=float(payload.t_rel_s), + flow=flow, + valid_mask=mask, + samples=samples, + base_bgr=base, + depth_nonzero=int(getattr(payload, "depth_nonzero", 0) or 0), + dropped=bool(getattr(payload, "dropped", False)), + ) + + def _recover_flow( + self, samples: FlowSamples, shape_hw: tuple[int, int] + ) -> np.ndarray: + """Write the stored displacements back at the recorded candidate locations. + + Only those locations carry a value; every other cell is ``NaN``, which + :func:`~orisys_app_shared.scene3d.visual_policy.select_flow_samples` treats as + invalid, so no neighbouring cell can contribute a sample the pass did not + record. + + ``dx``/``dy`` are the **drawn** displacements, which the selector produced as + ``DISPLACEMENT_PX_PER_FLOW * flow / flow_arrow_scale`` and then scaled by + ``flow_length_margin`` when the arrow was longer than its cell. Recovering the + field is therefore a division by the arrow scale, the display constant *and* + that margin — measured, not assumed: dividing by the first two alone leaves every + arrow 20% short (the finger margin is 0.8), which changes 319 of 330 samples and + one frame's active set. With the margin included the selector's own + ``np.where(length > spacing * margin, ...)`` branch is idempotent, so + re-selecting the recovered field reproduces the stored samples exactly. + """ + rows = np.asarray(samples.rows, np.int64) + cols = np.asarray(samples.cols, np.int64) + height, width = int(shape_hw[0]), int(shape_hw[1]) + if rows.size and (int(rows.max()) >= height or int(cols.max()) >= width): + raise ResultPolicyError( + "the result's candidate locations do not fit the domain it declares " + f"({height}x{width}); the mask keyframe and the grid keyframe disagree" + ) + divisor = ( + DISPLACEMENT_PX_PER_FLOW + * float(self.policy.flow_arrow_scale) + * float(self.policy.flow_length_margin) + ) + flow = np.full((height, width, 2), np.nan, dtype=np.float64) + with np.errstate(invalid="ignore"): + # ``NaN`` is the point of the mask, and the selector's tip arithmetic on a + # discarded cell is defined to be discarded: say so here rather than let + # numpy warn about an invalid cast on every frame. + flow[rows, cols, 0] = np.asarray(samples.dx, np.float64) / divisor + flow[rows, cols, 1] = np.asarray(samples.dy, np.float64) / divisor + return flow + + +class ResultLibrary: + """Opening stored results as playable sessions, one at a time. + + The controller keeps one of these. It owns the *reader lifecycle* — the previous + session's file handle is closed when another is opened — so nothing else has to + remember that a result is a file on disk with a handle behind it. + """ + + def __init__(self) -> None: + self._session: AnalysisSession | None = None + + @property + def session(self) -> AnalysisSession | None: + return self._session + + def open( + self, + directory: str | Path, + *, + policy: VisualPolicy | None = None, + display_policy: VisualPolicy | None = None, + session_id: int = 0, + recording: Path | None = None, + allow_incomplete: bool = False, + ) -> AnalysisSession: + """Open a result directory as a session, closing whatever was open. + + ``allow_incomplete`` is for *inspection* only and never reaches this class's + own session: an incomplete result raises + :class:`~orisys_app_shared.analysis.reader.AnalysisIncompleteError` here even + when the caller asked to look at one, because a partial pass that plays back + like a whole one is exactly the silent wrong answer Part A's marker exists to + prevent. + """ + reader = AnalysisReader(directory, allow_incomplete=allow_incomplete) + session = AnalysisSession( + reader, + policy=policy, + display_policy=display_policy, + session_id=session_id, + recording=recording, + ) + self.open_session(session) + return session + + def open_session(self, session: AnalysisSession) -> AnalysisSession: + """Adopt a session that was built elsewhere, closing whatever was open. + + The engine *borrows* its session (a mode switch must not hand a file handle + between two owners), so the library is the single owner and this is the only + place a reader is closed. + """ + self.close() + self._session = session + return session + + def close(self) -> None: + session, self._session = self._session, None + if session is not None: + try: + session.reader.close() + except Exception: # noqa: BLE001 - releasing is best effort + logger.debug("could not close the analysis reader", exc_info=True) diff --git a/apps/studio/src/orisys_studio/replay_controller.py b/apps/studio/src/orisys_studio/replay_controller.py index af7aa91..f62d399 100644 --- a/apps/studio/src/orisys_studio/replay_controller.py +++ b/apps/studio/src/orisys_studio/replay_controller.py @@ -28,12 +28,20 @@ from pathlib import Path from typing import Any, Protocol import numpy as np -from orisys_app_shared.acquisition.image_backend import SOURCE_REPLAY, ImageFrame +from orisys_app_shared.acquisition.image_backend import SOURCE_ANALYSIS, SOURCE_REPLAY, ImageFrame from orisys_app_shared.i18n import t from orisys_app_shared.paths import recordings_dir from PySide6.QtCore import QObject, QTimer from PySide6.QtWidgets import QFileDialog +from orisys_studio.analysis_worker import AnalysisReport, AnalysisWorker +from orisys_studio.replay import MODE_ANALYSIS +from orisys_studio.replay_analysis import ( + AnalysisSession, + ResultLibrary, + ResultPolicyError, +) + logger = logging.getLogger(__name__) #: Playback tick while a recording is playing. 33 ms is one frame at 30 fps; the @@ -72,6 +80,15 @@ class ReplayHost(Protocol): def _show_image_frame(self, frame: ImageFrame) -> None: ... + def _show_analysis_frame(self, packet: Any) -> None: + """Hand one stored analysis frame to the 3D pane's producer. + + The pane is fed through its ordinary path (it snapshots a packet), so a replay + goes through the same adapter, the same domain latch and the same validity + intersection as a live run. The window owns the pane, so it owns this call. + """ + ... + def _current_source_kind(self) -> str: ... def _drop_render_marker(self) -> None: ... @@ -86,6 +103,19 @@ class ReplayHost(Protocol): def _refresh_status_rail(self) -> None: ... + def _analysis_service( + self, recording: Path, cancelled: Any, *, on_progress: Any = None + ) -> Any: + """A ready-to-run ``AnalysisService`` over ``recording``. + + The window owns this because the chain specification comes from the app's + sensor configuration (``chain_spec_for`` over the recording's session), which is + window-level knowledge; the controller owns *when* a pass runs and what is done + with its result. ``cancelled`` is the pass's cancellation event (``None`` when + the caller only wants to ask whether a result is already stored). + """ + ... + @dataclass class ReplayActions: @@ -110,6 +140,7 @@ class ReplayController(QObject): charts: Any, image_panel: Any, mesh_panel: Any, + analysis_progress: Any | None = None, parent: QObject | None = None, ) -> None: super().__init__(parent) @@ -119,15 +150,27 @@ class ReplayController(QObject): self._charts = charts self._image_panel = image_panel self._mesh_panel = mesh_panel + #: The progress and cancel surface for an analysis pass. Optional so a caller + #: with no place to put it (a headless controller, a test) still gets the + #: behaviour; a pass then runs without a visible bar, and that is the caller's + #: choice rather than a silent failure. + self._progress = analysis_progress self._actions: ReplayActions | None = None #: True while a recompute pass is running: it is a *run* over a recording, so #: the acquire pipeline owns the panels and the transport stops the pass. self._recompute_running = False + #: The stored results the timeline can read from, and the pass in flight. + self.results = ResultLibrary() + self._analysis_worker: AnalysisWorker | None = None + self._analysis_target: Path | None = None + self._analysis_error = "" self._play_from = 0.0 self._play_started_at = 0.0 self.play_timer = QTimer(self) self.play_timer.setInterval(PLAYBACK_TICK_MS) self.play_timer.timeout.connect(self.on_play_tick) + if self._progress is not None: + self._progress.cancel_requested.connect(self.on_analysis_cancel) # ------------------------------------------------------------------ wiring def wire(self) -> None: @@ -139,7 +182,6 @@ class ReplayController(QObject): # Dragging the shared cursor on the charts scrubs the recording too. self._charts.cursor_moved.connect(self.on_cursor_moved) self._charts.timeline_taken_over.connect(self.pause_playback) - def bind_actions( self, *, @@ -354,6 +396,9 @@ class ReplayController(QObject): self._host._notify(t("playback.open_failed", error=Path(directory).name)) return False self.refresh_state() + # Analysis on open: serve the stored result when there is a compatible one, run + # the pass with a progress and cancel surface when there is not. + self.on_analysis_requested() return True def open_latest_recording(self) -> None: @@ -568,20 +613,328 @@ class ReplayController(QObject): # frames to show again now that the pass is over. self.render_frame() + # ---------------------------------------------------------------- analysis + def analysis_session(self) -> AnalysisSession | None: + """The stored result the timeline is reading, if any.""" + session = self._engine.analysis() + if session is not None: + return session + return self.results.session + + def analysis_active(self) -> bool: + """True while the cursor is read from a stored result. + + Asked with ``getattr`` because the engine is a seam a test can stand in for: a + stand-in that predates the analysable timeline has no mode, and "no mode" means + "no result to read", which is the safe answer rather than an error. + """ + mode = getattr(self._engine, "mode", None) + return ( + callable(mode) + and mode() == MODE_ANALYSIS + and self._engine.analysis() is not None + ) + + def analysis_running(self) -> bool: + worker = self._analysis_worker + return worker is not None and worker.running() + + def analysis_error(self) -> str: + """Why the last pass did not leave a playable result ("" when it did).""" + return self._analysis_error + + def on_display_settings(self, settings: Any) -> None: + """A display setting changed: redraw from the stored frame, write nothing. + + This is the property the result key was designed for. The threshold is applied + to the stored ``magnitude`` (so it is a *read*), the colour range and the idle + dots live in the draw call, and none of them is an input to the pass — so + nothing here touches the cache, and the assertion in + ``test_replay_analysis`` that the result directory set is unchanged is a test of + this method, not of the panel. + + A policy that moves the sample *grid* cannot be served by the stored candidate + locations; :meth:`AnalysisSession.serve_policy` says so, the message names the + key, and the frame is still redrawn from the result's own grid, because + re-gridding it would draw samples nothing measured. + """ + session = self.analysis_session() + if session is None or not self.analysis_active(): + return + from orisys_app_shared.scene3d.visual_policy import policy_for_settings + + policy = policy_for_settings(settings) + moved = not session.serve_policy(policy) + if moved: + self._host._notify(t("playback.analysis_grid_mismatch", keys="·".join(moved))) + self._engine.redraw() + + def on_analysis_requested(self) -> None: + """Run the pass over what is open, unless a compatible result already exists. + + The operator-facing entry for "re-run": the recompute switch calls this once it + has been turned on, and opening a recording calls it when no result is stored. + """ + if self.analysis_running(): + self._host._notify(t("playback.analysis_busy")) + return + if self._host._running or self._host._preview_open or self._host._session_active: + # The pass opens an SDK chain in this process, and so does a session. One + # at a time, and the working directory stays untouched while it runs. + self._host._notify(t("analysis.blocked")) + self._set_recompute_checked(False) + return + recording = self.recording_directory() + if recording is None: + self._host._notify(t("playback.analysis_no_source")) + self._set_recompute_checked(False) + return + self._analysis_error = "" + if self._activate_stored_result(recording, notify_reuse=True): + return + self._start_pass(recording) + + def recording_directory(self) -> Path | None: + """The recording directory an analysis pass should read.""" + directory = self._engine.directory() + if directory is not None: + return Path(directory) + named = self._toolbar.video_path() + if named: + folder = Path(named) + if (folder / "session.json").is_file(): + return folder + if (folder.parent / "session.json").is_file(): + return folder.parent + return None + + def _activate_stored_result( + self, recording: Path, *, notify_reuse: bool = False, reanalyse: bool = False + ) -> bool: + """Point the timeline at a stored result, if one is compatible. + + Returns ``False`` when there is nothing to read — no result, or a result the + reader refuses (an incomplete one), which is a *reason to run a pass* rather + than something to serve. The refusal is deliberate and is the whole point of + Part A's marker: a cancelled pass must never play back like a finished one. + """ + service = self._host._analysis_service(recording, None) + reader = None + try: + reader = service.stored_result() + except Exception as exc: # noqa: BLE001 - an unreadable result means "run one" + logger.info("the stored result is not usable: %s", exc) + self._analysis_error = str(exc) + reader = None + if reader is None: + return False + try: + session = AnalysisSession( + reader, + policy=getattr(service, "policy", None), + display_policy=self._display_policy(), + session_id=self._session_number(recording), + recording=recording, + ) + except ResultPolicyError as exc: + # Incomplete (or otherwise unplayable): close it and run a fresh pass. + logger.info("refusing to play an incomplete result: %s", exc) + self._analysis_error = str(exc) + try: + reader.close() + except Exception: # noqa: BLE001 - releasing is best effort + logger.debug("could not close the refused reader", exc_info=True) + if reanalyse and self._progress is not None: + self._progress.failed(reason=str(exc)) + return False + self.results.close() + self._adopt_session(session) + frames = session.frame_count + if notify_reuse: + # The rail's own notification (the 3D pane's source line and the 2D footer say + # "analysis result"), not the transient status line: this is a state, and it + # stays true for as long as the timeline is in it. + self._host._refresh_status_rail() + if self._progress is not None: + self._progress.reused(frames=frames) + return True + + def _adopt_session(self, session: AnalysisSession) -> None: + """One timeline: the cursor, the charts and both figures read this result.""" + self.results.open_session(session) + self._engine.use_analysis(session) + self._charts.set_recording_source(self._engine.reader()) + self._image_panel.set_source_kind("analysis") + self._mesh_panel.set_source_kind("analysis") + if self._actions is not None: + self._actions.recompute.setEnabled(True) + self.render_frame() + + def _display_policy(self) -> Any: + from orisys_app_shared.scene3d.visual_policy import policy_for_settings + + settings = getattr(self._mesh_panel, "display_settings", None) + return policy_for_settings(settings() if callable(settings) else None) + + def _session_number(self, recording: Path) -> int: + reader = self._engine.reader() + if reader is not None: + try: + return int(dict(reader.session().get("session_id") or 0)) + except Exception: # noqa: BLE001 - a session number is a nicety + logger.debug("could not read the session id", exc_info=True) + return 0 + + def _start_pass(self, recording: Path) -> None: + """Run the analysis pass on a worker thread, with progress and a cancel.""" + worker = AnalysisWorker( + lambda event, on_progress: self._host._analysis_service( + recording, event, on_progress=on_progress + ), + parent=self, + ) + worker.progressed.connect(self.on_analysis_progress) + worker.finished.connect(self.on_analysis_finished) + self._analysis_worker = worker + self._analysis_target = Path(recording) + if self._progress is not None: + self._progress.start() + if not worker.start(): # pragma: no cover - guarded by ``analysis_running`` + logger.warning("an analysis pass was already running") + self._analysis_worker = None + + def on_analysis_progress(self, progress: Any) -> None: + if self._progress is None: + return + self._progress.progress( + written=int(getattr(progress, "frames_written", 0)), + expected=int(getattr(progress, "frames_expected", 0)), + segment_index=int(getattr(progress, "segment_index", 0)), + segment_count=int(getattr(progress, "segment_count", 0)), + frame_id=int(getattr(progress, "frame_id", 0)), + elapsed_s=float(getattr(progress, "elapsed_s", 0.0)), + ) + + def on_analysis_finished(self, report: AnalysisReport) -> None: + """The pass ended: serve it only when it is complete, and say so when it is not. + + Nothing here writes to the transient status line, deliberately. An analysis pass + is asynchronous and can end seconds after the operator did something else, and a + status message that outlives its cause made an unrelated action look like it + failed — measured on ``test_the_replay_reset_is_the_display_reset_and_never_the_device``, + whose "display reset" line a failing pass overwrote. The progress surface stays + up with the outcome (complete, cancelled, or failed *with the reason*), and + :meth:`analysis_error` carries the same reason for a caller that asks. + """ + self._analysis_worker = None + target = self._analysis_target + if report.cancelled: + self._analysis_error = "cancelled" + if self._progress is not None: + self._progress.cancelled(frames=report.frames_written) + self._set_recompute_checked(False) + return + if not report.complete: + self._analysis_error = report.reason or "incomplete" + if self._progress is not None: + self._progress.failed( + reason=self._analysis_error, frames=report.frames_written + ) + self._set_recompute_checked(False) + return + if target is None or not self._activate_stored_result(target): + # A complete pass that cannot be opened is a bug, not a state to show as + # success: report it rather than leaving the operator with a progress bar + # that says "finished" over a recording that is not being served. + self._analysis_error = "unreadable" + if self._progress is not None: + self._progress.failed(reason="unreadable") + return + if self._progress is not None: + self._progress.complete( + frames=report.frames_written, elapsed_s=report.elapsed_s + ) + # The rail states what the timeline is reading from (the source label), so a + # completed pass says so where the state lives rather than in a line that expires. + self._host._refresh_status_rail() + self.refresh_state() + + def on_analysis_cancel(self) -> None: + """The operator pressed cancel: the pass stops and its result stays incomplete.""" + worker = self._analysis_worker + if worker is None: + return + worker.cancel() + + def cancel_analysis(self) -> bool: + """Ask a running pass to stop. Returns whether there was one.""" + worker = self._analysis_worker + if worker is None or not worker.running(): + return False + worker.cancel() + return True + + def wait_for_analysis(self, timeout_s: float = 10.0) -> bool: + """Block until a pass has finished. For teardown and for tests.""" + worker = self._analysis_worker + if worker is None: + return True + return worker.wait(timeout_s) + + def _set_recompute_checked(self, value: bool) -> None: + if self._actions is None: + return + if self._actions.recompute.isChecked() == bool(value): + return + self._actions.recompute.blockSignals(True) + self._actions.recompute.setChecked(bool(value)) + self._actions.recompute.blockSignals(False) + self._toolbar.set_recompute_checked(bool(value)) + # ------------------------------------------------------------------ drawing def render_frame(self) -> None: """Draw what the 2D panel should show for the current replay cursor. - A recording stores the raw camera image and nothing else, so the frame this - builds has no flow and no depth — the plan then says so, in one line, and the - panel offers exactly the one view the frame can be. The processed picture of the - same footage is what 「从视频重算」 produces, through the *same* backend. + One method for both timeline modes, because it is one cursor: in analysis mode + the picture is the stored frame's own flow field (the pass's candidate + locations, on the domain grid), and in recording mode it is the decoded video + frame, which carries no flow and no depth — so the plan says so in one line and + the panel offers exactly the one view the frame can be. The window's single 2D + sink draws both through the *same* backend, so the arrows, the idle dots, the + colour range and the domain boundary behave identically in the two modes. + + Nothing here writes anything: a display change reaches this method through the + engine's ``redraw()`` and rebuilds the picture from the same stored frame. """ if not self._host._in_replay_view() or not self._engine.has_video(): return if self._recompute_running: # The pipeline is drawing its own frames (see ``owns_image``). return + if self.analysis_active(): + frame = self._engine.analysed_frame() + if frame is None: + self._image_panel.clear_preview(self._engine.video_message()) + return + self._host._show_image_frame( + ImageFrame( + frame.base_bgr, + flow=frame.flow, + valid_mask=frame.valid_mask, + source=SOURCE_ANALYSIS, + frame_id=int(frame.frame_id), + t_rel_s=float(frame.t_rel_s), + flow_valid=True, + flow_frame_id=int(frame.frame_id), + depth_valid=False, + ) + ) + session = self._engine.analysis() + if session is not None: + # The same frame, in the shape the 3D pane's producer consumes: one + # cursor, two figures, one frame id. + self._host._show_analysis_frame(session.packet(frame)) + return image = self._engine.video_frame_bgr() if image is None: self._image_panel.clear_preview(self._engine.video_message()) diff --git a/apps/studio/src/orisys_studio/ui/analysis_progress.py b/apps/studio/src/orisys_studio/ui/analysis_progress.py new file mode 100644 index 0000000..ef69232 --- /dev/null +++ b/apps/studio/src/orisys_studio/ui/analysis_progress.py @@ -0,0 +1,275 @@ +"""The progress and cancel surface for an analysis pass. + +One widget, four states, and one button. It exists because analysis-on-open is the +approved default: opening a recording runs a pass that costs about half a wall-second +per video-second, so the operator has to be able to see that it is happening, see how +far along it is, and stop it. + +The state it shows is deliberately the **result's** state, not the thread's: + +``reused`` + a compatible stored result was already there and is being played; nothing ran. +``running`` + the pass is going. The bar is ``frames written / frames expected`` and the line + under it names the segment. +``complete`` + the pass finished and the result may be played from. +``cancelled`` / ``failed`` + the pass did **not** finish. Both say so in the same place, and neither is dressed + as success: a cancelled or failed pass leaves an incomplete result on disk, and the + reader refuses it, so "it looks finished" is exactly the state this surface must + never show for one. + +Part C owns the rail's transport and reset controls; this widget owns nothing outside +its own frame, and its cancel button is enabled only while a pass is running. +""" + +from __future__ import annotations + +from orisys_app_shared.i18n import t +from orisys_app_shared.theme import color +from PySide6.QtCore import Qt, Signal +from PySide6.QtWidgets import ( + QFrame, + QHBoxLayout, + QLabel, + QProgressBar, + QPushButton, + QSizePolicy, + QVBoxLayout, +) + +#: The states the surface can be in. ``idle`` is "nothing to say" and hides the widget. +IDLE = "idle" +REUSED = "reused" +RUNNING = "running" +COMPLETE = "complete" +CANCELLED = "cancelled" +FAILED = "failed" + +#: Which theme token colours the headline in each state. +_STATE_TOKEN = { + IDLE: "hint_text", + REUSED: "info", + RUNNING: "accent", + COMPLETE: "success", + CANCELLED: "warning", + FAILED: "danger", +} + +#: The progress bar's thousandths (Qt's own percentage unit, 0–1000). +_FULL = 1000 + + +class AnalysisProgress(QFrame): + """Shows one analysis pass: how far, which segment, and how to stop it.""" + + #: The operator asked to cancel the running pass. + cancel_requested = Signal() + + def __init__(self, parent=None, *, theme: str = "dark") -> None: + super().__init__(parent) + self._theme = str(theme) + self._state = IDLE + self._detail = "" + self._cancelling = False + self.setObjectName("analysisProgress") + self.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Maximum) + + self.title_label = QLabel(t("analysis.title"), self) + self.title_label.setObjectName("analysisTitle") + self.detail_label = QLabel("", self) + self.detail_label.setObjectName("analysisDetail") + self.detail_label.setWordWrap(True) + self.progress_bar = QProgressBar(self) + self.progress_bar.setObjectName("analysisBar") + self.progress_bar.setRange(0, _FULL) + self.progress_bar.setValue(0) + self.progress_bar.setTextVisible(False) + self.progress_bar.setFixedHeight(6) + self.stats_label = QLabel("", self) + self.stats_label.setObjectName("analysisStats") + self.cancel_button = QPushButton(t("analysis.cancel"), self) + self.cancel_button.setObjectName("analysisCancel") + self.cancel_button.setToolTip(t("analysis.cancel_tip")) + self.cancel_button.clicked.connect(self._on_cancel_clicked) + + footer = QHBoxLayout() + footer.setContentsMargins(0, 0, 0, 0) + footer.setSpacing(8) + footer.addWidget(self.stats_label, 1) + footer.addWidget(self.cancel_button, 0) + + layout = QVBoxLayout(self) + layout.setContentsMargins(10, 8, 10, 8) + layout.setSpacing(4) + layout.addWidget(self.title_label) + layout.addWidget(self.progress_bar) + layout.addLayout(footer) + layout.addWidget(self.detail_label) + + self._refresh() + self.setVisible(False) + + # --------------------------------------------------------------------- API + def state(self) -> str: + return self._state + + def detail(self) -> str: + return self._detail + + def cancelling(self) -> bool: + """True between the cancel click and the pass's own report.""" + return self._cancelling + + def start(self) -> None: + """A pass began: show the surface empty and cancellable.""" + self._state = RUNNING + self._cancelling = False + self._detail = "" + self.progress_bar.setValue(0) + self.stats_label.setText(t("analysis.frames", written=0, expected=0)) + self.setVisible(True) + self._refresh() + + def progress( + self, + *, + written: int, + expected: int, + segment_index: int = 0, + segment_count: int = 0, + frame_id: int = 0, + elapsed_s: float = 0.0, + ) -> None: + """One coalesced update from the pass.""" + if self._state != RUNNING: + return + fraction = 0.0 if expected <= 0 else min(1.0, written / expected) + self.progress_bar.setValue(int(round(fraction * _FULL))) + self.stats_label.setText( + t("analysis.frames", written=int(written), expected=int(expected)) + ) + parts: list[str] = [] + if segment_count > 0: + parts.append( + t( + "analysis.segment", + index=int(segment_index), + count=int(segment_count), + ) + ) + if frame_id: + parts.append(f"#{int(frame_id)}") + if elapsed_s > 0: + parts.append(t("analysis.elapsed", seconds=float(elapsed_s))) + self._detail = " · ".join(parts) + self._refresh() + + def reused(self, *, frames: int = 0) -> None: + """A stored result was already there: nothing ran, and that is the good news.""" + self._state = REUSED + self._cancelling = False + self.progress_bar.setValue(_FULL) + self.stats_label.setText(t("analysis.frames", written=frames, expected=frames)) + self._detail = "" + self.setVisible(True) + self._refresh() + + def complete(self, *, frames: int = 0, elapsed_s: float = 0.0) -> None: + """The pass finished and the result may be played from.""" + self._state = COMPLETE + self._cancelling = False + self.progress_bar.setValue(_FULL) + self.stats_label.setText(t("analysis.frames", written=frames, expected=frames)) + self._detail = t("analysis.complete", frames=int(frames), seconds=float(elapsed_s)) + self.setVisible(True) + self._refresh() + + def cancelled(self, *, frames: int = 0) -> None: + """The pass was stopped: the result is incomplete and will not be played.""" + self._state = CANCELLED + self._cancelling = False + self.progress_bar.setValue(0) + self.stats_label.setText(t("analysis.frames", written=frames, expected=frames)) + self._detail = t("analysis.cancelled", frames=int(frames)) + self.setVisible(True) + self._refresh() + + def failed(self, *, reason: str = "", frames: int = 0, cancelled: bool = False) -> None: + """The pass did not finish, whether by cancellation or by failure.""" + if cancelled: + self.cancelled(frames=frames) + return + self._state = FAILED + self._cancelling = False + self.progress_bar.setValue(0) + self.stats_label.setText(t("analysis.frames", written=frames, expected=0)) + self._detail = t("analysis.failed", reason=reason or "?") + self.setVisible(True) + self._refresh() + + def clear(self) -> None: + """Hide the surface: no pass has anything to say.""" + self._state = IDLE + self._cancelling = False + self._detail = "" + self.progress_bar.setValue(0) + self.setVisible(False) + + def set_theme(self, theme: str) -> None: + self._theme = str(theme) + self._refresh() + + def retranslate(self) -> None: + self.title_label.setText(t("analysis.title")) + self.cancel_button.setText( + t("analysis.cancelling") if self._cancelling else t("analysis.cancel") + ) + self.cancel_button.setToolTip(t("analysis.cancel_tip")) + self._refresh() + + # ------------------------------------------------------------------ private + def _on_cancel_clicked(self) -> None: + if self._state != RUNNING or self._cancelling: + return + self._cancelling = True + self.retranslate() + self.cancel_requested.emit() + + def _refresh(self) -> None: + """Repaint from the state: colour, which chrome is offered, and the tooltip. + + The title is one string in every state — what a pass is doing does not change + its name — so the *state* reaches the operator through the colour token, the + detail line and whether cancel is offered, not through a second vocabulary. + """ + token = _STATE_TOKEN.get(self._state, "hint_text") + running = self._state == RUNNING + self.cancel_button.setVisible(running) + self.cancel_button.setEnabled(running and not self._cancelling) + self.progress_bar.setVisible(running) + self.title_label.setText(t("analysis.title")) + self.title_label.setStyleSheet( + f"color: {color(self._theme, token)}; font-weight: 600;" + ) + self.detail_label.setText(self._detail) + self.detail_label.setVisible(bool(self._detail)) + self.stats_label.setStyleSheet(f"color: {color(self._theme, 'text_secondary')};") + self.detail_label.setStyleSheet( + f"color: {color(self._theme, 'text_secondary')};" + ) + self.setStyleSheet( + "QFrame#analysisProgress {" + f" background: {color(self._theme, 'card_bg')};" + f" border: 1px solid {color(self._theme, 'card_border')};" + " border-radius: 6px; }" + "QProgressBar {" + f" background: {color(self._theme, 'input_bg')};" + f" border: 1px solid {color(self._theme, 'input_border')};" + " border-radius: 3px; }" + "QProgressBar::chunk {" + f" background: {color(self._theme, 'accent')}; border-radius: 2px; }}" + ) + self.setToolTip(self._detail) + self.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True) diff --git a/apps/studio/tests/test_replay_analysis.py b/apps/studio/tests/test_replay_analysis.py new file mode 100644 index 0000000..f4f8094 --- /dev/null +++ b/apps/studio/tests/test_replay_analysis.py @@ -0,0 +1,906 @@ +"""Part B: playback served from an analysis result. + +What is being defended, in the order the plan asked for it: + +1. **one timeline.** The charts, the 2D canvas and the 3D pane read one cursor, and a + direct seek to frame *N* produces the same frame as playing forward to *N*; +2. **analysis on open.** A recording opens on a compatible stored result, and runs the + pass (with a progress and cancel surface) only when there is none; +3. **a display change writes nothing.** The flow threshold, the colour range and the + idle dots are reads: the result directory set is unchanged; +4. **an incomplete result is never served.** A cancelled or failed pass is reported as + cancelled or failed, and playback is not offered from it; +5. **frames are addressed by id.** An id the recording does not hold is refused, never + coerced into an index. + +The fixtures use real footage — the 33-frame box recording under ``dist/`` and the +shipped finger configuration — driven through the **finger** chain, which is what gives +the result a validity mask and therefore a static candidate grid. The pass is O(5 s) +and is shared per module. Where the SDK or the footage is absent the tests skip rather +than fake a result: a synthetic result would not defend the id span (20…52), the +keyframe identity or the sample round trip. +""" + +from __future__ import annotations + +import threading +from pathlib import Path + +import numpy as np +import pytest +from orisys_app_shared.analysis.cache import ResultCache, scan_results +from orisys_app_shared.analysis.reader import ( + AnalysisIncompleteError, + AnalysisNotFoundError, + AnalysisReader, +) +from orisys_app_shared.analysis.service import AnalysisService, ChainSpec +from orisys_app_shared.i18n import t +from orisys_app_shared.scene3d.contracts import DisplaySettings +from orisys_app_shared.scene3d.visual_policy import ( + policy_for_settings, + select_flow_samples, +) +from orisys_studio.analysis_worker import AnalysisWorker, report_from +from orisys_studio.replay import MODE_ANALYSIS, MODE_RECORDING +from orisys_studio.replay_analysis import ( + AnalysisSession, + ResultLibrary, + ResultPolicyError, +) + +REPO = Path(__file__).resolve().parents[3] +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" + +#: The evidence recording's own frame ids. Deliberately not 0…32: an implementation +#: that used an index would pass every test written against a 0-based recording. +FIRST_FRAME_ID = 20 +LAST_FRAME_ID = 52 + + +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 _service(root: Path, **kwargs) -> AnalysisService: + return AnalysisService(RECORDING, chain=_finger_chain(), cache=ResultCache(root), **kwargs) + + +# ------------------------------------------------------------------------ fixtures +@pytest.fixture(scope="module") +def pass_root(tmp_path_factory) -> Path: + """One real analysis pass over the evidence recording, in its own cache root.""" + _require_inputs() + + root = tmp_path_factory.mktemp("replay-analysis-b") + outcome = _service(root).run() + assert outcome.complete, outcome.reason + return root + + +@pytest.fixture(scope="module") +def reader(pass_root: Path): + reader = AnalysisReader(_result_directory(pass_root)) + yield reader + reader.close() + + +def _result_directory(root: Path) -> Path: + directories = [entry.directory for entry in scan_results(root) if entry.complete] + assert directories, f"no complete result under {root}" + return directories[0] + + +@pytest.fixture(scope="module") +def session(reader) -> AnalysisSession: + service = _service(Path(reader.directory).parent) + return AnalysisSession( + reader, + policy=service.policy, + display_policy=policy_for_settings(DisplaySettings()), + session_id=2, + recording=RECORDING, + ) + + +def _same_samples(left, right) -> dict[str, bool]: + """Compare two ``FlowSamples`` (or two frames that carry one) array by array.""" + left = getattr(left, "samples", left) + right = getattr(right, "samples", right) + return { + name: bool( + np.array_equal( + np.asarray(getattr(left, name)), np.asarray(getattr(right, name)) + ) + ) + for name in ("rows", "cols", "dx", "dy", "active") + } + + +# ------------------------------------------------------------------- fakes (Qt) +class _Signal: + def __init__(self) -> None: + self.slots: list = [] + + def connect(self, slot) -> None: + self.slots.append(slot) + + +class _Action: + def __init__(self, checked: bool = False) -> None: + self._checked = bool(checked) + self.enabled: bool | None = None + self.tooltips: list[str] = [] + self.triggered = _Signal() + self.toggled = _Signal() + + def isChecked(self) -> bool: # noqa: N802 - Qt's spelling + return self._checked + + def setChecked(self, value: bool) -> None: # noqa: N802 + self._checked = bool(value) + + def setEnabled(self, value: bool) -> None: # noqa: N802 + self.enabled = bool(value) + + def setToolTip(self, text: str) -> None: # noqa: N802 + self.tooltips.append(str(text)) + + def blockSignals(self, flag: bool) -> None: # noqa: N802 + pass + + +class _Button: + def __init__(self) -> None: + self.checked = False + self.enabled = True + + def isChecked(self) -> bool: # noqa: N802 + return self.checked + + def isEnabled(self) -> bool: # noqa: N802 + return self.enabled + + def setChecked(self, value: bool) -> None: # noqa: N802 + self.checked = bool(value) + + +class _Toolbar: + def __init__(self, *, video_path: str = "") -> None: + self.btn_run = _Button() + self._video_path = video_path + self.run_checked: list[bool] = [] + self.source_modes: list[str] = [] + self.recompute_enabled: list[bool] = [] + self.recompute_checked: list[bool] = [] + + def run_role(self) -> str: + return "playback" + + def video_path(self) -> str: + return self._video_path + + def set_video_path(self, path: str) -> None: + self._video_path = str(path) + + def set_run_checked(self, value: bool) -> None: + self.run_checked.append(bool(value)) + + def refresh_labels(self) -> None: + pass + + def set_source_mode(self, mode: str) -> None: + self.source_modes.append(str(mode)) + + def set_recompute_enabled(self, value: bool) -> None: + self.recompute_enabled.append(bool(value)) + + def set_recompute_checked(self, value: bool) -> None: + self.recompute_checked.append(bool(value)) + + +class _Charts: + def __init__(self) -> None: + self.cursors: list[float] = [] + self._recording = False + self.cleared = 0 + self.unit_value = "raw" + + def unit(self) -> str: + return self.unit_value + + def set_cursor(self, position: float) -> None: + self.cursors.append(float(position)) + + def set_recording_source(self, reader) -> None: + self._recording = reader is not None + + def clear_recording_source(self) -> None: + self.cleared += 1 + self._recording = False + + def is_recording_source(self) -> bool: + return self._recording + + def follow_latest(self) -> None: + pass + + def show_all(self) -> None: + pass + + +class _ImagePanel: + def __init__(self) -> None: + self.kinds: list[str] = [] + self.cleared: list[str] = [] + + def set_source_kind(self, kind: str) -> None: + self.kinds.append(str(kind)) + + def clear_preview(self, message: str) -> None: + self.cleared.append(str(message)) + + +class _Mesh: + def __init__(self) -> None: + self.kinds: list[str] = [] + self.reference_only = 0 + self._settings = DisplaySettings() + + def set_source_kind(self, kind: str) -> None: + self.kinds.append(str(kind)) + + def show_reference_only(self) -> None: + self.reference_only += 1 + + def display_settings(self) -> DisplaySettings: + return self._settings + + def set_display_settings(self, settings: DisplaySettings) -> None: + self._settings = settings + + +class _Progress: + """The progress surface as the controller uses it, recording every transition.""" + + def __init__(self) -> None: + self.cancel_requested = _Signal() + self.states: list[str] = [] + self.updates: list[dict] = [] + self.cancels = 0 + + def start(self) -> None: + self.states.append("running") + + def progress(self, **kwargs) -> None: + self.updates.append(dict(kwargs)) + + def reused(self, **kwargs) -> None: + self.states.append("reused") + + def complete(self, **kwargs) -> None: + self.states.append("complete") + + def cancelled(self, **kwargs) -> None: + self.states.append("cancelled") + + def failed(self, **kwargs) -> None: + self.states.append("failed") + + def clear(self) -> None: + self.states.append("idle") + + +class _Host: + def __init__(self, service_factory=None) -> None: + self.prefs: dict = {} + self.view = "replay" + self._running = False + self._preview_open = False + self._session_active = False + self.notes: list[str] = [] + self.frames: list = [] + self.analysis_packets: list = [] + self.units: list[str] = [] + self.presets: list[str] = [] + self.service_factory = service_factory + self.services_built = 0 + + def _in_replay_view(self) -> bool: + return self.view == "replay" + + def _current_source_kind(self) -> str: + return "replay" + + def _apply_view_preset(self, preset: str) -> None: + self.presets.append(preset) + self.view = preset + + def _set_display_unit(self, unit: str) -> None: + self.units.append(unit) + + def _set_source_mode(self, mode: str) -> None: + pass + + def _show_image_frame(self, frame) -> None: + self.frames.append(frame) + + def _show_analysis_frame(self, packet) -> None: + self.analysis_packets.append(packet) + + def _drop_render_marker(self) -> None: + pass + + def _notify(self, text: str, ms: int = 0) -> None: + self.notes.append(str(text)) + + def _start_run(self) -> None: + pass + + def _stop_run(self, *, reason: str = "user_stop", wait: bool = False) -> None: + pass + + def _update_run_enabled(self) -> None: + pass + + def _refresh_status_rail(self) -> None: + pass + + def _analysis_service(self, recording: Path, cancelled=None, *, on_progress=None): + self.services_built += 1 + if self.service_factory is None: + raise AssertionError("the test did not provide an analysis service") + return self.service_factory(recording, cancelled, on_progress) + + +class _RecordingReader: + """The recording reader as the engine and the controller use it.""" + + def __init__(self, directory: Path) -> None: + self.directory = Path(directory) + self.closed = False + + def session(self) -> dict: + return {"session_id": 2, "device": {"sensor_type": "finger"}} + + def segments(self) -> list[Path]: + return [self.directory / "raw_0001.mp4"] + + def info(self): + return type("Info", (), {"duration_s": 1.92, "newtons": None, "decimated": False})() + + def video_frame_at(self, t_rel_s: float): + from orisys_app_shared.record import RecordingFrame + + index = int(t_rel_s * 30) + frame = RecordingFrame( + frame_id=FIRST_FRAME_ID + index, + t_rel_s=float(t_rel_s), + in_video=True, + dropped=False, + segment=1, + ) + return np.zeros((4, 4, 3), dtype=np.uint8), frame + + def close(self) -> None: + self.closed = True + + +def _controller(engine=None, host=None, **kwargs): + from orisys_studio.replay_controller import ReplayController + + engine = engine if engine is not None else _Engine() + host = host if host is not None else _Host() + toolbar = kwargs.pop("toolbar", None) or _Toolbar() + charts = kwargs.pop("charts", None) or _Charts() + panel = kwargs.pop("image_panel", None) or _ImagePanel() + mesh = kwargs.pop("mesh_panel", None) or _Mesh() + progress = kwargs.pop("progress", None) or _Progress() + controller = ReplayController( + engine=engine, + host=host, + toolbar=toolbar, + charts=charts, + image_panel=panel, + mesh_panel=mesh, + analysis_progress=progress, + ) + actions = { + "open_video": _Action(), + "playback": _Action(), + "latest": _Action(), + "play": _Action(), + "recompute": _Action(), + } + controller.bind_actions(**actions) + return controller, host, engine, (toolbar, charts, panel, mesh, progress, actions) + + +class _Engine: + """The engine as the controller uses it, with the timeline's mode explicit.""" + + def __init__(self, *, duration: float = 1.92, position: float = 0.0) -> None: + self._duration = duration + self._position = position + self._mode = MODE_RECORDING + self._session: AnalysisSession | None = None + self._reader = _RecordingReader(RECORDING) + self._has_video = True + self.positions: list[float] = [] + self.redraws = 0 + + def has_video(self) -> bool: + return self._has_video + + def is_loaded(self) -> bool: + return True + + def is_video_only(self) -> bool: + return False + + def duration_s(self) -> float: + return self._duration + + def position(self) -> float: + return self._position + + def set_position(self, position: float) -> None: + self._position = float(position) + self.positions.append(float(position)) + + def reader(self): + return self._reader + + def directory(self): + return self._reader.directory + + def video_path(self): + return None + + def video_frame_bgr(self): + return np.zeros((4, 4, 3), dtype=np.uint8) + + def video_message(self) -> str: + return "" + + def set_unit(self, unit: str) -> None: + pass + + def mode(self) -> str: + return self._mode + + def analysis(self): + return self._session + + def use_analysis(self, session) -> None: + self._session = session + self._mode = MODE_ANALYSIS + + def use_recording(self) -> None: + self._session = None + self._mode = MODE_RECORDING + + def analysed_frame(self): + if self._session is None: + return None + return self._session.frame_at(self._position) + + def redraw(self) -> None: + self.redraws += 1 + + def open_recording(self, directory) -> bool: + self._reader = _RecordingReader(Path(directory)) + self._session = None + self._mode = MODE_RECORDING + return True + + def open_video(self, path) -> bool: + return False + + +# ------------------------------------------------------------------ the timeline +def test_a_stored_result_is_served_with_the_recording_s_own_frame_ids(session): + """Frame identity is the recording's counter (20…52), never an index (0…32).""" + assert session.frame_count == 33 + assert int(session.frame_ids[0]) == FIRST_FRAME_ID + assert int(session.frame_ids[-1]) == LAST_FRAME_ID + assert not session.has(0), "an index was accepted as a frame id" + assert session.has(FIRST_FRAME_ID) + + +def test_a_frame_id_the_recording_does_not_hold_is_refused(session): + """A *position* passed where an *id* belongs gets an error, not a picture. + + This is the trap the plan names: the result's ids run 20…52, so a caller that + passes an index reads a frame that exists and looks right. The reader refuses an id + it does not hold, and the session does not repair the request into one it can serve. + """ + with pytest.raises(AnalysisNotFoundError): + session.frame(int(session.frame_ids[0]) + 1000) + with pytest.raises(AnalysisNotFoundError): + session.frame(0) + + +def test_position_is_signal_time_and_resolves_to_a_stored_frame_id(session): + """The cursor's time base is ``t_rel_s`` from the recording's own ``frames.csv``.""" + span = session.duration_s() + assert span == pytest.approx(1.920, abs=1e-3), ( + "the result's span must be signal time; 1.100 s would be the container's" + ) + middle = int(session.frame_ids[len(session.frame_ids) // 2]) + moment = session.time_of(middle) + assert moment is not None + assert session.frame_id_at(moment) == middle + assert session.frame_id_at(0.0) == FIRST_FRAME_ID + assert session.time_of(0) is None, "an absent id answered with a time" + + +def test_the_recovered_field_reselects_to_the_stored_samples(session): + """The 2D canvas and the 3D pane draw the same samples, frame for frame. + + The canvas takes a flow field and re-runs the shared selector; the pane takes the + stored selection. This asserts they agree on every stored frame — the locations, the + drawn offsets and the activity verdict — so "one timeline, two figures" is a + measurement rather than a hope. Magnitude is compared as float32-derived truth: it + differs from the stored value by at most one ulp, which does not move a sample. + """ + policy = session.display_policy + checked = 0 + for frame_id in session.frame_ids: + frame = session.frame(int(frame_id)) + reselected = select_flow_samples(frame.flow, frame.valid_mask, policy) + agreement = _same_samples(frame.samples, reselected) + assert all(agreement.values()), f"frame {frame_id}: {agreement}" + assert np.array_equal( + np.asarray(frame.samples.active), np.asarray(reselected.active) + ) + checked += 1 + assert checked == 33 + + +def test_a_seek_equals_playing_forward_to_the_same_frame(session): + """Seek to *N* and play to *N* must produce the same picture. + + Tolerance: none on the frame's identity (the frame id and every drawn array are + compared exactly). The cursor is a float in seconds, so the *time* handed to a seek + is the stored ``t_rel_s`` read back from the result: the comparison is exact because + the seek is given the frame's own time, and the frame id it resolves to is then the + frame that was played to. Positions between stored frames resolve to the frame at or + before them, which is what playing forward passes through as well. + """ + ids = [int(value) for value in session.frame_ids] + # Play forward from the first frame, remembering what each frame looked like. + played: dict[int, object] = {} + for frame_id in ids: + played[frame_id] = session.frame(frame_id) + # Now seek directly to a sample of positions, in a fresh session so nothing is + # answered from the sequential walk's own cache. + fresh = AnalysisSession( + AnalysisReader(_result_directory(Path(session.directory).parent)), + policy=session.policy, + display_policy=session.display_policy, + ) + try: + for frame_id in (ids[0], ids[1], ids[len(ids) // 2], ids[-2], ids[-1]): + moment = fresh.time_of(frame_id) + assert moment is not None + seeked = fresh.frame_at(moment) + assert seeked is not None + assert seeked.frame_id == frame_id + assert all(_same_samples(played[frame_id], seeked).values()) + assert np.array_equal( + np.asarray(played[frame_id].flow), + np.asarray(seeked.flow), + equal_nan=True, + ) + finally: + fresh.reader.close() + + +def test_the_3d_pane_is_fed_the_same_frame_the_canvas_draws(session): + """One cursor, two figures: the packet's id, flow and mask are the frame's own.""" + frame = session.frame_at(session.time_of(LAST_FRAME_ID - 2)) + assert frame is not None + packet = session.packet(frame) + assert int(packet.frame_id) == int(frame.frame_id) + # ``equal_nan``: every cell that is not a recorded candidate is NaN by construction, + # in both the frame and the packet, and that is the contract rather than a defect. + assert np.array_equal( + np.asarray(packet.flow), np.asarray(frame.flow), equal_nan=True + ) + assert np.array_equal(np.asarray(packet.valid_mask), np.asarray(frame.valid_mask)) + assert packet.flow_valid is True and packet.flow_frame_id == int(frame.frame_id) + assert packet.forces_valid is False, "a result stores no force, so none may be claimed" + assert float(packet.mono_t) == pytest.approx(frame.t_rel_s) + + +# --------------------------------------------------------------- display settings +def test_a_display_change_is_a_read_and_writes_no_result(pass_root, session): + """The result key is blind to display settings, so a change must not write. + + The assertion is the strongest one available: the set of result directories is + compared before and after the threshold, the colour range and the idle dots are + changed, and a fresh service (the object that would write) is asked whether a + compatible result exists. Nothing may appear. + """ + before = {entry.directory.name for entry in scan_results(pass_root)} + recorded = int(session.reader.frame(int(session.frame_ids[0])).active_count) + + for settings in ( + DisplaySettings(flow_on_px=0.2), + DisplaySettings(flow_on_px=1e9), + # The colour range and the exaggeration are the same class of setting: display + # only, so they must not reach the cache either. ``color_scale`` needs a + # ``FieldDescriptor`` and a LUT id, so the field that is constructible on its own + # stands in for it here; the threshold above is the one that also changes what is + # drawn. + DisplaySettings(display_gain=4.0), + DisplaySettings(flow_grid_spacing=10), + ): + assert session.serve_policy(policy_for_settings(settings)) is True + + after = {entry.directory.name for entry in scan_results(pass_root)} + assert after == before, "a display change wrote a new analysis result" + # The same cache, asked the same question, still answers with the stored result: + # nothing about the display entered the key. + service = _service(pass_root) + assert service.stored_result() is not None + # And the change really did change the picture: a threshold of 1e9 activates nothing. + session.serve_policy(policy_for_settings(DisplaySettings(flow_on_px=1e9))) + assert session.frame(int(session.frame_ids[0])).active_count == 0 + session.serve_policy(policy_for_settings(DisplaySettings(flow_on_px=0.4))) + assert session.frame(int(session.frame_ids[0])).active_count == recorded + + +def test_a_display_change_that_moves_the_grid_is_refused_not_served(session): + """The stored candidate locations are the pass's; a new grid cannot be served.""" + assert session.serve_policy(policy_for_settings(DisplaySettings(max_arrows=50))) is False + assert session.threshold() == pytest.approx(0.4), "a refused policy changed the session" + # The threshold is the one setting that is served from the stored magnitude. + assert session.serve_policy(policy_for_settings(DisplaySettings(flow_on_px=0.3))) is True + + +def _flush(qt_app=None) -> None: + """Deliver the worker's queued signals: they arrive through the GUI event loop.""" + from PySide6.QtWidgets import QApplication + + QApplication.processEvents() + + +def test_the_controller_redraws_on_a_display_change_and_runs_no_pass(pass_root): + """Through the controller: the surface shows the redraw and no service is built. + + Its own reader and session, so neither the threshold this test sets nor the frame + cache is shared with another test. + """ + reader = AnalysisReader(_result_directory(pass_root)) + session = AnalysisSession( + reader, display_policy=policy_for_settings(DisplaySettings()) + ) + try: + controller, host, engine, parts = _controller() + controller.results.open_session(session) + engine.use_analysis(session) + engine.redraws = 0 + built = host.services_built + controller.on_display_settings(DisplaySettings(flow_on_px=0.5)) + assert engine.redraws == 1, "the display change did not redraw" + assert host.services_built == built, "a display change built an analysis service" + # 0.5, not the 0.5 that arrived: ``flow_off_px`` is 0.3 and ``normalised()`` keeps + # on >= off, so a threshold below the off band is raised to it rather than served + # as an inverted band. The point here is that the session *adopted* the value. + assert session.threshold() == pytest.approx(0.5) + # The frame is rebuilt from the same stored record, not from a new one. + assert session.frame(int(session.frame_ids[0])).frame_id == FIRST_FRAME_ID + finally: + session.reader.close() + + +# ------------------------------------------------------------ analysis on open +def test_opening_a_recording_with_a_stored_result_runs_no_pass(pass_root): + """Analysis on open reuses a compatible result: no progress, no pass.""" + _require_inputs() + stored = _result_directory(pass_root) + calls: list[str] = [] + + def factory(recording: Path, cancelled, on_progress): + calls.append(str(recording)) + return _service(pass_root) + + host = _Host(service_factory=factory) + controller, host, engine, parts = _controller(host=host) + assert controller.load_recording(RECORDING) is True + assert calls == [str(RECORDING)], f"the service was built {len(calls)} times" + assert controller.analysis_active(), "the timeline did not adopt the stored result" + assert engine.mode() == MODE_ANALYSIS + assert parts[2].kinds[-1] == "analysis", "the 2D panel was not told it is a result" + assert parts[3].kinds[-1] == "analysis", "the 3D pane was not told it is a result" + assert parts[4].states[-1] == "reused", "the surface did not report the reuse" + assert stored.is_dir() + + +def test_opening_a_recording_without_a_result_starts_the_pass(tmp_path, qt_app): + """No compatible result: the pass runs behind a progress surface with a cancel.""" + _require_inputs() + empty = tmp_path / "cache" + empty.mkdir() + + def factory(recording: Path, cancelled, on_progress): + return _service(empty, cancelled=cancelled, on_progress=on_progress) + + host = _Host(service_factory=factory) + controller, host, engine, parts = _controller(host=host) + assert controller.load_recording(RECORDING) is True + assert parts[4].states[:1] == ["running"], "the progress surface did not start" + assert controller.analysis_running(), "no pass was started" + assert controller.wait_for_analysis(120) + _flush() + assert parts[4].states[-1] == "complete", parts[4].states + assert controller.analysis_active(), "a complete result was not adopted" + assert engine.mode() == MODE_ANALYSIS + + +def test_a_cancelled_pass_is_reported_and_never_served(tmp_path, qt_app): + """Cancel mid-pass: the surface says so and playback is not offered from it.""" + _require_inputs() + empty = tmp_path / "cache" + empty.mkdir() + + def factory(recording: Path, cancelled, on_progress): + # Cancel on the first frame, which is what the operator's button does. + def cancel_after_one(progress): + if progress.frames_written >= 1: + cancelled.set() + if on_progress is not None: + on_progress(progress) + + return _service(empty, cancelled=cancelled, on_progress=cancel_after_one) + + host = _Host(service_factory=factory) + controller, host, engine, parts = _controller(host=host) + controller.load_recording(RECORDING) + assert controller.wait_for_analysis(120) + _flush() + assert parts[4].states[-1] == "cancelled", parts[4].states + assert not controller.analysis_active(), "an incomplete result was served" + assert engine.mode() == MODE_RECORDING + assert controller.analysis_error() == "cancelled" + assert not any( + t("playback.analysis_cancelled") in note for note in host.notes + ), "a terminal analysis state was written to the transient status line" + # The partial result is on disk and refuses to be read as complete. + directories = [entry.directory for entry in scan_results(empty)] + assert directories, "the cancelled pass wrote nothing at all" + assert all( + not entry.complete for entry in scan_results(empty) + ), "a cancelled result claims to be complete" + with pytest.raises(AnalysisIncompleteError): + AnalysisReader(directories[0]) + assert AnalysisReader(directories[0], allow_incomplete=True).complete is False + + +def test_an_incomplete_result_on_disk_is_not_adopted(tmp_path): + """The refusal is the reader's, and the session never dresses it as playable.""" + _require_inputs() + empty = tmp_path / "cache" + empty.mkdir() + cancelled = threading.Event() + + def cancel_after_one(progress): + if progress.frames_written >= 1: + cancelled.set() + + service = _service(empty, cancelled=cancelled, on_progress=cancel_after_one) + outcome = service.run() + assert outcome.complete is False + assert outcome.frames_written >= 1, "the pass wrote nothing, so nothing was refused" + with pytest.raises(AnalysisIncompleteError): + AnalysisReader(outcome.directory) + reader = AnalysisReader(outcome.directory, allow_incomplete=True) + try: + with pytest.raises(ResultPolicyError): + AnalysisSession(reader) + finally: + reader.close() + # And the service's own "is one stored?" answer is not a playable one: the directory + # exists, so it answers by refusing — which is what makes the controller run a pass + # instead of serving a partial result. + with pytest.raises(AnalysisIncompleteError): + service.stored_result() + + +def test_the_library_never_hands_out_an_incomplete_session(tmp_path): + """``ResultLibrary.open`` refuses a partial result even when asked to inspect one.""" + _require_inputs() + empty = tmp_path / "cache" + empty.mkdir() + cancelled = threading.Event() + + def cancel_after_one(progress): + if progress.frames_written >= 1: + cancelled.set() + + outcome = _service(empty, cancelled=cancelled, on_progress=cancel_after_one).run() + library = ResultLibrary() + with pytest.raises(ResultPolicyError): + library.open(outcome.directory, allow_incomplete=True) + assert library.session is None + + +def test_the_worker_reports_progress_and_a_cancelled_outcome(): + """The thread boundary: a pass reports progress, and a cancel is a cancelled report.""" + seen: list = [] + + def factory(cancelled, on_progress): + def cancel_after(progress): + on_progress(progress) + if progress.frames_written >= 2: + cancelled.set() + + from orisys_app_shared.analysis.service import Progress + + class _Fake: + def run(self): + for index in range(5): + cancel_after( + Progress( + frames_written=index + 1, + frames_expected=5, + segment_index=1, + segment_count=1, + frame_id=FIRST_FRAME_ID + index, + elapsed_s=0.1 * index, + ) + ) + from orisys_app_shared.analysis.service import AnalysisOutcome + from orisys_app_shared.analysis.model import ResultKey + + return AnalysisOutcome( + key=ResultKey( + recording="r", algorithm="a", sensor_config="s", calibration="", reference_frame=None + ), + directory=Path("."), + complete=False, + reason="cancelled", + frames_written=2, + frames_expected=5, + elapsed_s=0.3, + ) + + return _Fake() + + from PySide6.QtWidgets import QApplication + + worker = AnalysisWorker(factory) + worker.progressed.connect(seen.append) + worker.finished.connect(lambda report: seen.append(("report", report))) + assert worker.start() is True + assert worker.wait(30) + QApplication.processEvents() + reports = [item for item in seen if isinstance(item, tuple)] + assert reports, "the worker never reported" + report = reports[-1][1] + assert report.cancelled is True and report.complete is False + assert report.frames_written == 2 + assert report_from(None, cancelled=True).cancelled is True diff --git a/shared/src/orisys_app_shared/acquisition/image_backend.py b/shared/src/orisys_app_shared/acquisition/image_backend.py index 51766b5..200e98a 100644 --- a/shared/src/orisys_app_shared/acquisition/image_backend.py +++ b/shared/src/orisys_app_shared/acquisition/image_backend.py @@ -55,6 +55,11 @@ SOURCE_LIVE = "live" SOURCE_RECOMPUTE = "recompute" SOURCE_REPLAY = "replay" SOURCE_PREVIEW = "preview" +#: A frame read from a stored analysis result: the vision chain is *not* running, and +#: the fields are the ones a finished pass recorded. Derived like ``live``, because a +#: stored frame does carry flow — but never conflated with it, because "the pipeline is +#: computing this now" and "this was computed once and read back" are different claims. +SOURCE_ANALYSIS = "analysis" #: Shown when the only thing a frame can be is a raw picture. RAW_NOTE_KEY = "panels.replay_raw_note" @@ -64,7 +69,7 @@ RAW_NOTE_KEY = "panels.replay_raw_note" MISSING_NOTE_KEY = "panels.frame_missing_note" #: Sources whose frames carry derived fields. -DERIVED_SOURCES = (SOURCE_LIVE, SOURCE_RECOMPUTE) +DERIVED_SOURCES = (SOURCE_LIVE, SOURCE_RECOMPUTE, SOURCE_ANALYSIS) #: How much a frame's domain mask must overlap the latched one before the latched one #: is kept. Measured on the shipped finger island (32937 px, 39.2% of the 300x280