From 6e69518742002f1a0404312c07d9b4e017f5c5d5 Mon Sep 17 00:00:00 2001 From: ila Date: Wed, 22 Jul 2026 00:04:00 +0800 Subject: [PATCH] feat(v1): latest-frame-only reader thread for low-latency RTSP Adds _FrameGrabber which decodes in a background thread and keeps only the newest frame; VideoSource drop_stale (STREAM only) returns the latest frame so a slow consumer never builds a backlog. app enables it for streams. REPLAY is unchanged. 82 tests pass. Co-Authored-By: Claude Opus 4.8 --- docs/06-tasks.md | 2 +- docs/current-state.md | 2 +- progress.md | 9 ++++ v1/app.py | 4 +- v1/tests/test_video_source.py | 70 +++++++++++++++++++++++++++++- v1/video_source.py | 82 ++++++++++++++++++++++++++++++++++- 6 files changed, 164 insertions(+), 5 deletions(-) diff --git a/docs/06-tasks.md b/docs/06-tasks.md index 8170070..e6400ff 100644 --- a/docs/06-tasks.md +++ b/docs/06-tasks.md @@ -36,7 +36,7 @@ | T-209 | 放宽俯视场景摔倒判定灵敏度（可配置） | T-208 | 判定灵敏度参数化：`require_rapid_drop`(默认关)、`require_lower_body`(默认关)、`horizontal_angle_threshold_degrees`(默认放宽)；默认下持续水平即可进入疑似，仅靠确认窗防误报；参数进 config 与 config_version，逻辑可测。 | DONE | | T-210 | 修复截图标注乱码（cv2 ASCII 文字） | T-202 | 截图标注文字改为纯 ASCII（`cv2.putText` 只支持 ASCII），去掉非 ASCII 分隔符导致的乱码；实时画面标签不变；ASCII 标签逻辑可测。 | DONE | | T-211 | 检测灵敏度参数进设置页 | T-209 | 设置页可编辑 `require_rapid_drop`/`require_lower_body`/`horizontal_angle_threshold_degrees`/`confirm_window_seconds`（下拉/数字/勾选），保存写回未跟踪 `config.local.json`、下次启动生效；写回逻辑可测。 | DONE | -| T-212 | 实时流只处理最新帧降延迟 | T-203 | STREAM 后台读帧只保留最新帧、丢弃积压，实时延迟不随推理速度增长；REPLAY 逐帧不丢；最新帧保留逻辑可测。 | DOING | +| T-212 | 实时流只处理最新帧降延迟 | T-203 | STREAM 后台读帧只保留最新帧、丢弃积压，实时延迟不随推理速度增长；REPLAY 逐帧不丢；最新帧保留逻辑可测。 | DONE | | T-213 | 截图中文标注（Pillow） | T-210 | CONFIRMED 截图可用中文标注（如“确认摔倒 P-0001”）；缺 Pillow/字体时回退 ASCII 不崩溃；标签与回退逻辑可测。 | TODO | ## Phase 3 · Go V2 迁移门槛与实现 diff --git a/docs/current-state.md b/docs/current-state.md index 4cf50d9..d6a0fb4 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -11,7 +11,7 @@ - V1 代码：已建立安全配置、显式 Replay/Stream 视频源、Pose、轻量跟踪、质量/几何证据、倒地领域规则、按 ID 四态事件机及 `v1/pipeline.py` 事件管线；运行事件带非敏感 `config_version`，缺帧/低质量/断流会中断证据确认。新增 `v1/view_model.py`（Qt-free 监控视图状态与设置草稿三份隔离）、`v1/gui.py` 薄 PyQt5 双 Tab 外壳与 `v1/app.py` 装配（`FrameWorker` 只发出已判定的 `FrameAnalysis`，窗口只渲染）。新增 `v1/alerts.py`：按 `event_id` 去重，对首次 CONFIRMED 保存带标注截图、追加 JSONL 事件行，并经可注入 `AlertSink`（Windows 侧 `QtAlertSink` 提供声音与一次性弹窗）触发一次声音/弹窗。新增 `v1/camera.py`（RTSP URL 构建 + 有界连接测试抓帧）与 `config.py` 的结构化来源（host/端口/通道/账号/密码，凭证百分号编码）、`write_local_camera_source`；设置页新增摄像头连接分组、测试连接与预览，以及传输协议/连接超时/低延迟三项抓流调优（由 `config` 存值、`app` 开流前写入 `OPENCV_FFMPEG_CAPTURE_OPTIONS`）。真实海康流已用 `VideoSource(STREAM)` 连通（1920×1080）。 - V2 代码：`v2/` 目录存在但尚无实现。 - 非代码设计工件：docs/ui/silver-pose-ui-ux-spec.md、docs/ui/2026-07-20-html-prototype-plan.md、docs/ui/silver-pose-v1-prototype.html 与 docs/ui/silver-pose-v2-prototype.html 已建立。v2 HTML 是符合正式浅色 Windows 规范的当前视觉参考：浅灰蓝底、白色卡片，红色只表示确认摔倒、其弹窗和事件证据；文件名中的 v2 只表示原型设计修订，不能理解为 Go V2 实现已开始。v1 HTML 保留为历史深色对照。两者均使用顶部双 Tab、设置草稿与状态交互，且画面、事件和时间都是模拟数据，不连接真实摄像头、模型或网络，也不改变 Phase 1 任务顺序。 -- 测试：`python -m compileall -q v1 demo` 已通过（含 `gui.py`、`app.py`、`alerts.py`、`camera.py` 语法）；`python -m pytest v1/tests -v` 当前有 79 项配置、视频源、Pose、跟踪、证据、领域规则、状态机、管线、视图模型、报警工件、摄像头/URL 构建、抓流参数拼串、每帧诊断、灵敏度放宽、截图 ASCII 标注、断流重连、事件汇总和灵敏度写回测试并已通过（含来源模式显式声明、模型置信度经适配器生效、事件号跨轮唯一、结构化凭证不入 config_version 且公开示例无凭证的守卫）。`demo/1.mp4` 的首两帧回放时间戳已验证为 0.000000 与 0.033333 秒；T-106 的真实模型/录像 smoke 在首帧得到 2 名已分析人员、第二帧得到 0 名人员且未创建事件；T-201 的视图冒烟以真实录像解码 + 真实管线 + 确定性假 Pose 适配器驱动 `build_monitor_view`，得到稳定 ID、box、14 段骨架、17/17 关键点与 NORMAL/success，断流帧 0 人且不显示摔倒标签；T-202 的报警冒烟用 `demo/1.mp4` 首帧（848×480）落盘一张可被 `cv2.imread` 读回的标注截图（480×848×3，约 330 KB）与一行 JSONL（相对截图路径、含 config_version/source_id、无 rtsp），重复派发返回 0。这些只验证管线、视图与报警工件可运行，不表示摔倒识别准确率。`init.ps1` 会检查运行时依赖、编译旧基线并运行 V1 测试，但不会安装软件包。 +- 测试：`python -m compileall -q v1 demo` 已通过（含 `gui.py`、`app.py`、`alerts.py`、`camera.py` 语法）；`python -m pytest v1/tests -v` 当前有 82 项配置、视频源、Pose、跟踪、证据、领域规则、状态机、管线、视图模型、报警工件、摄像头/URL 构建、抓流参数拼串、每帧诊断、灵敏度放宽、截图 ASCII 标注、断流重连、事件汇总、灵敏度写回和最新帧丢弃测试并已通过（含来源模式显式声明、模型置信度经适配器生效、事件号跨轮唯一、结构化凭证不入 config_version 且公开示例无凭证的守卫）。`demo/1.mp4` 的首两帧回放时间戳已验证为 0.000000 与 0.033333 秒；T-106 的真实模型/录像 smoke 在首帧得到 2 名已分析人员、第二帧得到 0 名人员且未创建事件；T-201 的视图冒烟以真实录像解码 + 真实管线 + 确定性假 Pose 适配器驱动 `build_monitor_view`，得到稳定 ID、box、14 段骨架、17/17 关键点与 NORMAL/success，断流帧 0 人且不显示摔倒标签；T-202 的报警冒烟用 `demo/1.mp4` 首帧（848×480）落盘一张可被 `cv2.imread` 读回的标注截图（480×848×3，约 330 KB）与一行 JSONL（相对截图路径、含 config_version/source_id、无 rtsp），重复派发返回 0。这些只验证管线、视图与报警工件可运行，不表示摔倒识别准确率。`init.ps1` 会检查运行时依赖、编译旧基线并运行 V1 测试，但不会安装软件包。 - 模型：`demo/best.pt` 可加载为 YOLO Pose，类别 `person`，`kpt_shape=[17, 3]`；与 `D:\PythonP\fall_detection\best.pt` 哈希一致。 - 当前标准启动：`./init.ps1`。 - 当前标准验证：`python -m compileall -q demo`。 diff --git a/progress.md b/progress.md index 9be5f97..da9c188 100644 --- a/progress.md +++ b/progress.md @@ -388,3 +388,12 @@ - 阻塞：真实线程整合需 Windows 冒烟；最新帧保留核心 `_pump` 逻辑本环境可确定性测。 - 决策：读帧解码放后台线程、消费端只取最新帧、丢弃积压，使延迟不随推理速度增长且被钳在一次推理量级；REPLAY 保持逐帧不丢；STREAM 用收帧时单调时钟。 - 下一步：写 `_FrameGrabber._pump` 只留最新帧的失败测试，再实现并在 app 对 STREAM 开启。 + +## 【2026-07-22】T-212 实时流只处理最新帧降延迟（完成） + +- 状态：DONE +- 变更：`video_source.py` 新增 `_FrameGrabber`（daemon 线程，持续 `cap.read()` 只保留最新帧、丢弃积压，读失败置 failed，读异常不崩线程）。`VideoSource` 增 `drop_stale`（仅 STREAM 生效）：懒开流后启动 grabber，`read()` 返回最新帧（无帧时 RETRYING/connecting，failed 则释放+有界退避重连），`close()` 停 grabber。`app.py` 对 STREAM 开启 `drop_stale=True`。REPLAY 路径不变（逐帧不丢）。 +- 验证：新增测试——`_FrameGrabber._pump` 连续三帧后 `take_latest` 只留最新（丢弃前两帧）、读失败置 failed；STREAM+drop_stale 集成（真实线程）在轮询内返回 CONNECTED 帧。`python3 -m pytest v1/tests -q` 为 82 passed；`python3 -m compileall -q v1` 退出码 0。 +- 阻塞：真实 RTSP 上的端到端延迟改善需 Windows 目视确认；丢弃最新帧核心逻辑与集成本环境已测。 +- 决策：解码放后台线程、消费端只取最新帧，使延迟被钳在一次推理量级、不随推理速度增长；STREAM 用收帧时单调时钟，丢帧不破坏摔倒计时（每处理帧取真实到达时间）。 +- 下一步：T-213 截图中文标注（Pillow，缺库/字体回退 ASCII）。 diff --git a/v1/app.py b/v1/app.py index 62af02b..d90f910 100644 --- a/v1/app.py +++ b/v1/app.py @@ -38,7 +38,9 @@ class FrameWorker(QtCore.QThread): # OpenCV reads this env var when it opens the stream, so set it first. os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = self._config.ffmpeg_capture_options mode = SourceMode.STREAM if self._config.source_mode == "stream" else SourceMode.REPLAY - source = VideoSource(self._config.source_url, mode=mode) + source = VideoSource( + self._config.source_url, mode=mode, drop_stale=mode is SourceMode.STREAM + ) pipeline = FallPipeline.from_config(self._config, self._pose_adapter) try: while not self._stop: diff --git a/v1/tests/test_video_source.py b/v1/tests/test_video_source.py index 1a20f29..dce7645 100644 --- a/v1/tests/test_video_source.py +++ b/v1/tests/test_video_source.py @@ -1,8 +1,10 @@ +import time + import cv2 import numpy as np import pytest -from v1.video_source import SourceMode, SourceStatus, VideoSource +from v1.video_source import SourceMode, SourceStatus, VideoSource, _FrameGrabber def _write_sample_video(path): @@ -135,6 +137,72 @@ def test_stream_recovers_after_mid_stream_drop(): assert recovered.image is not None +class _ScriptedCapture: + def __init__(self, frames, fail_after=False): + self._frames = list(frames) + self._fail_after = fail_after + + def isOpened(self): + return True + + def read(self): + if self._frames: + return True, self._frames.pop(0) + return False, None + + def get(self, _property_id): + return 0.0 + + def release(self): + pass + + +def test_frame_grabber_keeps_only_the_latest_frame(): + frame_a = np.full((4, 4, 3), 1, dtype=np.uint8) + frame_b = np.full((4, 4, 3), 2, dtype=np.uint8) + frame_c = np.full((4, 4, 3), 3, dtype=np.uint8) + grabber = _FrameGrabber(_ScriptedCapture([frame_a, frame_b, frame_c])) + + assert grabber._pump() is True + assert grabber._pump() is True + assert grabber._pump() is True + image, failed = grabber.take_latest() + + assert failed is False + assert image is frame_c # older frames dropped + + +def test_frame_grabber_marks_failed_on_read_failure(): + grabber = _FrameGrabber(_ScriptedCapture([np.zeros((4, 4, 3), dtype=np.uint8)])) + + assert grabber._pump() is True + assert grabber._pump() is False + _, failed = grabber.take_latest() + + assert failed is True + + +def test_stream_with_drop_stale_returns_a_connected_frame(): + source = VideoSource( + "rtsp://demo", + mode=SourceMode.STREAM, + drop_stale=True, + capture_factory=lambda _s: _OpenCapture(), + ) + try: + packet = None + for _ in range(40): + packet = source.read() + if packet.status is SourceStatus.CONNECTED and packet.image is not None: + break + time.sleep(0.02) + assert packet is not None + assert packet.status is SourceStatus.CONNECTED + assert packet.image is not None + finally: + source.close() + + class _NegativeFirstTimestampCapture: def __init__(self): self._read_count = 0 diff --git a/v1/video_source.py b/v1/video_source.py index 61ab92e..44cdb93 100644 --- a/v1/video_source.py +++ b/v1/video_source.py @@ -1,5 +1,6 @@ """OpenCV frame acquisition with explicit replay and reconnect states.""" +import threading import time from dataclasses import dataclass from enum import Enum @@ -10,6 +11,48 @@ import cv2 import numpy as np +class _FrameGrabber(threading.Thread): + """Continuously read a capture, keeping only the newest frame. + + Decoding runs here so a slow consumer never builds a backlog: it always gets + the latest decoded frame and older frames are dropped. This bounds live RTSP + latency to roughly one inference regardless of stream vs processing rate. + """ + + def __init__(self, capture: Any) -> None: + super().__init__(daemon=True) + self._capture = capture + self._lock = threading.Lock() + self._latest: Optional[np.ndarray] = None + self._failed = False + self._stopped = threading.Event() + + def _pump(self) -> bool: + try: + success, image = self._capture.read() + except Exception: # noqa: BLE001 - a released capture must not crash the thread + success, image = False, None + if not success or image is None: + with self._lock: + self._failed = True + return False + with self._lock: + self._latest = image + return True + + def run(self) -> None: + while not self._stopped.is_set(): + if not self._pump(): + break + + def take_latest(self): + with self._lock: + return self._latest, self._failed + + def stop(self) -> None: + self._stopped.set() + + class SourceStatus(str, Enum): CONNECTED = "connected" RETRYING = "retrying" @@ -44,6 +87,7 @@ class VideoSource: retry_initial_seconds: float = 1.0, retry_max_seconds: float = 16.0, capture_factory: Optional[CaptureFactory] = None, + drop_stale: bool = False, ) -> None: if retry_initial_seconds <= 0: raise ValueError("retry_initial_seconds must be positive") @@ -53,6 +97,8 @@ class VideoSource: raise ValueError("mode must be a SourceMode") self._source = str(source) self._mode = mode + self._drop_stale = bool(drop_stale) and mode is SourceMode.STREAM + self._grabber: Optional[_FrameGrabber] = None self._retry_initial_seconds = retry_initial_seconds self._retry_max_seconds = retry_max_seconds self._capture_factory = capture_factory or cv2.VideoCapture @@ -69,7 +115,7 @@ class VideoSource: return self._status def close(self) -> None: - self._release_capture() + self._stop_grabber() self._closed = True self._status = SourceStatus.CLOSED @@ -84,6 +130,9 @@ class VideoSource: if self._closed: return self._packet(timestamp, SourceStatus.CLOSED, "source is closed") + if self._drop_stale: + return self._read_latest(timestamp) + if self._mode is SourceMode.REPLAY and self._status is SourceStatus.EOF: return self._packet(timestamp, SourceStatus.EOF, self._last_error) @@ -109,6 +158,37 @@ class VideoSource: frame_timestamp = self._frame_timestamp(timestamp) return FramePacket(image=image, timestamp_monotonic=frame_timestamp, status=SourceStatus.CONNECTED) + def _read_latest(self, timestamp: float) -> FramePacket: + """STREAM path: return the newest grabbed frame, dropping any backlog.""" + + if self._capture is None: + if self._status is SourceStatus.RETRYING and timestamp < self._next_retry_at: + return self._packet(timestamp, SourceStatus.RETRYING, self._last_error) + if not self._open_capture(timestamp): + return self._packet(timestamp, self._status, self._last_error) + self._grabber = _FrameGrabber(self._capture) + self._grabber.start() + + image, failed = self._grabber.take_latest() + if failed: + self._stop_grabber() + self._schedule_retry(timestamp, "stream read failed; retry scheduled") + return self._packet(timestamp, SourceStatus.RETRYING, self._last_error) + if image is None: + return self._packet(timestamp, SourceStatus.RETRYING, "connecting") + + self._status = SourceStatus.CONNECTED + self._last_error = None + self._retry_delay_seconds = self._retry_initial_seconds + self._last_timestamp = timestamp + return FramePacket(image=image, timestamp_monotonic=timestamp, status=SourceStatus.CONNECTED) + + def _stop_grabber(self) -> None: + if self._grabber is not None: + self._grabber.stop() + self._grabber = None + self._release_capture() + def _open_capture(self, now: float) -> bool: capture = self._capture_factory(self._source) if capture is None or not capture.isOpened():