diff --git a/docs/04-architecture.md b/docs/04-architecture.md index 886f0ce..658ca14 100644 --- a/docs/04-architecture.md +++ b/docs/04-architecture.md @@ -30,7 +30,7 @@ V1 的同一数据流既可接 RTSP，也可回放本地录像。V2 复用同一 | --- | --- | --- | --- | | 应用入口 | `v1/app.py` | 装配配置、窗口、线程和依赖 | 推理细节、事件判定 | | 配置 | `v1/config.py` | 解析示例和本地配置，校验非敏感字段 | 保存真实凭证 | -| 视频源 | `v1/video_source.py` | 打开、读取、重连 RTSP 或录像，附带时间戳 | Pose、报警 | +| 视频源 | `v1/video_source.py` | 打开、读取、重连 RTSP 或录像；输出帧、单调回放时间戳和显式来源状态 | Pose、报警 | | Pose 适配器 | `v1/pose.py` | 统一返回 box、关键点、置信度 | 跟踪、摔倒业务结论 | | 跟踪 | `v1/tracking.py` | 为连续人员输出 `track_id` | 根据姿态报警 | | 质量与证据 | `v1/evidence.py` | 过滤低质量点，计算水平姿态、下移和持续性证据 | GUI 状态 | @@ -85,7 +85,7 @@ RECOVERING | --- | --- | --- | | 单帧误报 | 当前 `demo/` 任一规则命中即报警 | V1 用质量门控、跟踪和时序状态机；反例录像必测。 | | 俯视关键点不稳 | 1.6 米俯视、遮挡或远距离会影响膝踝 | 全身可见前提、ROI、质量拒绝和现场回归；必要时再采集数据。 | -| RTSP 断流 | 网络抖动或摄像头重连 | 视频源显式状态、退避重连、断流不报警。 | +| RTSP 断流 | 网络抖动或摄像头重连 | 视频源显式状态、指数退避重连、断流不报警。 | | 同一事件重复报警 | CONFIRMED 状态持续多帧 | 每个事件 ID 仅执行一次报警副作用，恢复后才允许新事件。 | | Go 行为漂移 | ONNX 预后处理与 Python 不同 | 导出后跑同一录像，比较关键点、事件数量、确认时间和截图。 | | 模型误解 | Pose 指标被误当摔倒指标 | 文案仅说明姿态模型；事件级指标单独记录。 | diff --git a/docs/06-tasks.md b/docs/06-tasks.md index 1a6284a..a0fd08c 100644 --- a/docs/06-tasks.md +++ b/docs/06-tasks.md @@ -15,7 +15,7 @@ | ID | 任务 | 依赖 | 验收要点 | 状态 | | --- | --- | --- | --- | --- | | T-101 | 创建 V1 包、依赖清单、示例配置和忽略规则 | T-000 | `v1/` 可导入；真实 RTSP 凭证被拒绝提交；`pytest` 能运行。 | DONE | -| T-102 | 实现可重连的视频源与录像回放适配器 | T-101 | 有效本地录像可按时间戳产帧；无效源进入连接错误状态且不崩溃。 | DOING | +| T-102 | 实现可重连的视频源与录像回放适配器 | T-101 | 有效本地录像可按时间戳产帧；无效源进入连接错误状态且不崩溃。 | DONE | | T-103 | 实现 Pose 适配器与模型来源校验 | T-102 | 输出 person box、17 点和置信度；错误模型或哈希不符时给出明确错误。 | TODO | | T-104 | 实现人员跟踪与姿态质量门控 | T-103 | 连续人员维持 ID；低质量、缺失膝踝或空帧不会产生倒地候选。 | TODO | | T-105 | 实现按 ID 的时序摔倒状态机 | T-104 | 正例在配置秒数内确认；坐下、弯腰、短时低姿态回到 NORMAL；事件副作用只触发一次。 | TODO | diff --git a/docs/07-v1-implementation-plan.md b/docs/07-v1-implementation-plan.md index d56d47e..5d83ac0 100644 --- a/docs/07-v1-implementation-plan.md +++ b/docs/07-v1-implementation-plan.md @@ -132,7 +132,7 @@ git commit -m "feat(v1): add secure configuration baseline" - Create: `v1/video_source.py` - Create: `v1/tests/test_video_source.py` -- [ ] **Step 1: Write replay and broken-source tests** +- [x] **Step 1: Write replay and broken-source tests** ```python def test_file_source_emits_monotonic_timestamps(sample_video): @@ -146,12 +146,12 @@ def test_missing_source_returns_error_state(tmp_path): assert source.read().status == SourceStatus.ERROR ``` -- [ ] **Step 2: Run tests to verify they fail** +- [x] **Step 2: Run tests to verify they fail** Run: `python -m pytest v1/tests/test_video_source.py -v` Expected: FAIL because `VideoSource` and `SourceStatus` do not exist. -- [ ] **Step 3: Implement the source contract** +- [x] **Step 3: Implement the source contract** ```python from typing import Optional @@ -172,12 +172,12 @@ class FramePacket: `read()` returns an error packet for failures; it never emits a synthetic person or fall event. For RTSP, reconnect with bounded backoff from configuration. -- [ ] **Step 4: Run tests** +- [x] **Step 4: Run tests** Run: `python -m pytest v1/tests/test_video_source.py -v` Expected: PASS. -- [ ] **Step 5: Commit** +- [x] **Step 5: Commit** ```powershell git add v1/video_source.py v1/tests/test_video_source.py docs progress.md diff --git a/docs/api.md b/docs/api.md index aaeacd9..dd47e2c 100644 --- a/docs/api.md +++ b/docs/api.md @@ -71,6 +71,21 @@ FallEvent = { `FallEvent` 只在状态首次进入 `CONFIRMED` 时创建一次。连续帧更新 UI 状态，但不重复创建事件。 +## 视频来源帧合约 + +```text +FramePacket = { + image: ndarray | null, + timestamp_monotonic: float, + status: "connected" | "retrying" | "error" | "eof" | "closed", + error: string | null +} +``` + +- 录像优先使用容器时间戳；首帧时间戳无效或倒退时，回退为帧序号/FPS，保证回放时间单调。 +- `retrying`、`error`、`eof` 和 `closed` 都没有图像，且绝不伪造人员、姿态或摔倒事件。 +- 可重连来源以有界指数退避重新打开；断流不推进状态机的证据时间。 + ## 事件合约 | 事件 | 触发者 | 负载 | 结果 | diff --git a/docs/current-state.md b/docs/current-state.md index ee5d8a6..a889c90 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -5,13 +5,13 @@ ## 当前快照 - 日期：2026-07-21 -- 阶段：V1 工程化起步；T-102 进行中。 +- 阶段：V1 工程化起步；T-102 已验收，等待 T-103。 - 已验证环境：Windows PowerShell；Python 3.8.10；Ultralytics 8.3.205；PyQt5 可导入。 - 旧生产基线：`demo/main.py`、`demo/fall_detection_gui.py`、`demo/detect_fall.py`、`demo/best.pt`。 -- V1 代码：已建立 `v1/__init__.py`、`v1/config.py`、安全的 `config.example.json`、固定依赖清单和配置单元测试；视频源、Pose、跟踪、证据、状态机和 GUI 尚未实现。 +- V1 代码：已建立安全配置基线，以及 `v1/video_source.py` 的 OpenCV 回放/重连适配器；它输出显式状态和单调时间戳。Pose、跟踪、证据、状态机和 GUI 尚未实现。 - 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 demo` 已通过；`python -m pytest v1/tests -v` 当前有 2 个配置测试并已通过。`init.ps1` 会检查运行时依赖、编译旧基线并运行 V1 测试，但不会安装软件包。 +- 测试：`python -m compileall -q demo` 已通过；`python -m pytest v1/tests -v` 当前有 6 项配置/视频源测试并已通过。`demo/1.mp4` 的首两帧回放时间戳已验证为 0.000000 与 0.033333 秒。`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`。 @@ -32,9 +32,9 @@ ## 任务状态 -- 已完成：T-000（Harness 文档与旧基线快照）、T-101（V1 安全配置基线）。 -- 正在进行：T-102（实现可重连的视频源与录像回放适配器）。 -- 下一个可领取：完成 T-102 后为 T-103。 +- 已完成：T-000（Harness 文档与旧基线快照）、T-101（V1 安全配置基线）、T-102（视频源与录像回放）。 +- 正在进行：无。 +- 下一个可领取：T-103。 ## 当前可运行内容 diff --git a/progress.md b/progress.md index faf3f1e..4b63006 100644 --- a/progress.md +++ b/progress.md @@ -98,3 +98,12 @@ - 阻塞：无。 - 决策：视频源只返回帧、单调时间戳和来源状态；读取失败、EOF 和重连绝不生成虚构人员或摔倒事件。 - 下一步：写入时间戳单调、错误来源和有界重连的失败测试。 + +## 【2026-07-21】T-102 实现可重连的视频源与录像回放适配器（完成） + +- 状态：DONE +- 变更：新增 `v1/video_source.py`，定义 `FramePacket` 与 `connected`、`retrying`、`error`、`eof`、`closed` 来源状态。录像优先读取容器时间戳，负首帧或倒退值回退到帧序号/FPS；可重连来源采用有界指数退避，所有非连接状态返回空帧且不携带事件信息。 +- 验证：先运行 `python -m pytest v1/tests/test_video_source.py -v`，确认因缺少 `v1.video_source` 导入失败；实现后视频源测试 4 passed，完整 V1 测试在本轮最后一次运行时为 6 passed。`demo/1.mp4` 只读 smoke 成功读取首两帧，时间戳为 0.000000 与 0.033333 秒；`python -m compileall -q v1 demo` 通过。 +- 阻塞：无。 +- 决策：来源失败只表达来源状态，绝不以空帧、EOF 或重连触发人员更新或摔倒事件；本地录像测试使用临时生成的 AVI，避免将客户视频或样例录像作为 pytest 夹具。 +- 下一步：T-103，先写模型来源和 17 点 Pose 输出的失败测试。 diff --git a/v1/tests/test_video_source.py b/v1/tests/test_video_source.py new file mode 100644 index 0000000..fcb7bcd --- /dev/null +++ b/v1/tests/test_video_source.py @@ -0,0 +1,129 @@ +import cv2 +import numpy as np +import pytest + +from v1.video_source import SourceStatus, VideoSource + + +def _write_sample_video(path): + writer = cv2.VideoWriter( + str(path), cv2.VideoWriter_fourcc(*"MJPG"), 10.0, (32, 24) + ) + assert writer.isOpened() + for value in (20, 80, 140): + writer.write(np.full((24, 32, 3), value, dtype=np.uint8)) + writer.release() + + +def test_file_source_emits_monotonic_timestamps(tmp_path): + sample_video = tmp_path / "sample.avi" + _write_sample_video(sample_video) + source = VideoSource(sample_video, reconnect=False) + + first = source.read(now=10.0) + second = source.read(now=10.1) + + assert first.status is SourceStatus.CONNECTED + assert second.status is SourceStatus.CONNECTED + assert first.image is not None + assert second.image is not None + assert first.timestamp_monotonic < second.timestamp_monotonic + + +def test_missing_source_returns_error_state_without_frame(tmp_path): + source = VideoSource(tmp_path / "missing.avi", reconnect=False) + + packet = source.read(now=1.0) + + assert packet.status is SourceStatus.ERROR + assert packet.image is None + assert packet.error + + +class _ClosedCapture: + def isOpened(self): + return False + + def read(self): + return False, None + + def get(self, _property_id): + return 0.0 + + def release(self): + pass + + +class _OpenCapture: + def __init__(self): + self._read_count = 0 + + def isOpened(self): + return True + + def read(self): + self._read_count += 1 + return True, np.zeros((8, 8, 3), dtype=np.uint8) + + def get(self, _property_id): + return self._read_count * 100.0 + + def release(self): + pass + + +def test_reconnect_waits_then_reopens_with_bounded_backoff(): + captures = [_ClosedCapture(), _OpenCapture()] + + source = VideoSource( + "demo-source", + reconnect=True, + retry_initial_seconds=2.0, + retry_max_seconds=2.0, + capture_factory=lambda _source: captures.pop(0), + ) + + first = source.read(now=5.0) + waiting = source.read(now=6.0) + recovered = source.read(now=7.0) + + assert first.status is SourceStatus.RETRYING + assert waiting.status is SourceStatus.RETRYING + assert recovered.status is SourceStatus.CONNECTED + assert recovered.image is not None + + +class _NegativeFirstTimestampCapture: + def __init__(self): + self._read_count = 0 + + def isOpened(self): + return True + + def read(self): + self._read_count += 1 + return True, np.zeros((8, 8, 3), dtype=np.uint8) + + def get(self, property_id): + if property_id == cv2.CAP_PROP_POS_MSEC: + return -33.0 if self._read_count == 1 else 33.333333333333336 + if property_id == cv2.CAP_PROP_POS_FRAMES: + return float(self._read_count) + if property_id == cv2.CAP_PROP_FPS: + return 30.0 + return 0.0 + + def release(self): + pass + + +def test_negative_first_timestamp_falls_back_to_frame_index_and_fps(): + source = VideoSource( + "demo-source", reconnect=False, capture_factory=lambda _source: _NegativeFirstTimestampCapture() + ) + + first = source.read(now=10.0) + second = source.read(now=10.1) + + assert first.timestamp_monotonic == 0.0 + assert second.timestamp_monotonic == pytest.approx(1.0 / 30.0) diff --git a/v1/video_source.py b/v1/video_source.py new file mode 100644 index 0000000..f2029f2 --- /dev/null +++ b/v1/video_source.py @@ -0,0 +1,163 @@ +"""OpenCV frame acquisition with explicit replay and reconnect states.""" + +import time +from dataclasses import dataclass +from enum import Enum +from pathlib import Path +from typing import Any, Callable, Optional, Union + +import cv2 +import numpy as np + + +class SourceStatus(str, Enum): + CONNECTED = "connected" + RETRYING = "retrying" + ERROR = "error" + EOF = "eof" + CLOSED = "closed" + + +@dataclass(frozen=True) +class FramePacket: + image: Optional[np.ndarray] + timestamp_monotonic: float + status: SourceStatus + error: Optional[str] = None + + +CaptureFactory = Callable[[str], Any] + + +class VideoSource: + """Read replay or RTSP frames without treating source failures as evidence.""" + + def __init__( + self, + source: Union[str, Path], + reconnect: bool = True, + retry_initial_seconds: float = 1.0, + retry_max_seconds: float = 16.0, + capture_factory: Optional[CaptureFactory] = None, + ) -> None: + if retry_initial_seconds <= 0: + raise ValueError("retry_initial_seconds must be positive") + if retry_max_seconds < retry_initial_seconds: + raise ValueError("retry_max_seconds must not be smaller than retry_initial_seconds") + self._source = str(source) + self._reconnect = reconnect + self._retry_initial_seconds = retry_initial_seconds + self._retry_max_seconds = retry_max_seconds + self._capture_factory = capture_factory or cv2.VideoCapture + self._capture = None + self._status = SourceStatus.RETRYING if reconnect else SourceStatus.ERROR + self._retry_delay_seconds = retry_initial_seconds + self._next_retry_at = 0.0 + self._last_timestamp: Optional[float] = None + self._last_error: Optional[str] = None + self._closed = False + + @property + def status(self) -> SourceStatus: + return self._status + + def close(self) -> None: + self._release_capture() + self._closed = True + self._status = SourceStatus.CLOSED + + def read(self, now: Optional[float] = None) -> FramePacket: + """Return one frame or an explicit non-frame state. + + ``now`` is injectable for deterministic reconnect tests. It is never + interpreted as video time; replay timestamps come from OpenCV. + """ + + timestamp = time.monotonic() if now is None else float(now) + if self._closed: + return self._packet(timestamp, SourceStatus.CLOSED, "source is closed") + + 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) + + success, image = self._capture.read() + if not success or image is None: + self._release_capture() + if self._reconnect: + self._schedule_retry(timestamp, "frame read failed; retry scheduled") + return self._packet(timestamp, SourceStatus.RETRYING, self._last_error) + self._status = SourceStatus.EOF + self._last_error = "frame read reached end of source" + return self._packet(timestamp, SourceStatus.EOF, self._last_error) + + self._status = SourceStatus.CONNECTED + self._last_error = None + self._retry_delay_seconds = self._retry_initial_seconds + frame_timestamp = self._frame_timestamp(timestamp) + return FramePacket(image=image, timestamp_monotonic=frame_timestamp, status=SourceStatus.CONNECTED) + + def _open_capture(self, now: float) -> bool: + capture = self._capture_factory(self._source) + if capture is None or not capture.isOpened(): + if capture is not None: + capture.release() + if self._reconnect: + self._schedule_retry(now, "unable to open source; retry scheduled") + else: + self._status = SourceStatus.ERROR + self._last_error = "unable to open source" + return False + self._capture = capture + self._status = SourceStatus.CONNECTED + return True + + def _frame_timestamp(self, fallback_now: float) -> float: + source_seconds = float(self._capture.get(cv2.CAP_PROP_POS_MSEC)) / 1000.0 + frame_seconds = self._timestamp_from_frame_index() + if source_seconds < 0: + source_seconds = frame_seconds if frame_seconds is not None else fallback_now + elif ( + self._last_timestamp is not None + and source_seconds <= self._last_timestamp + and frame_seconds is not None + and frame_seconds > self._last_timestamp + ): + source_seconds = frame_seconds + if self._last_timestamp is not None and source_seconds <= self._last_timestamp: + source_seconds = self._last_timestamp + 0.000001 + self._last_timestamp = source_seconds + return source_seconds + + def _timestamp_from_frame_index(self) -> Optional[float]: + fps = float(self._capture.get(cv2.CAP_PROP_FPS)) + frame_index = float(self._capture.get(cv2.CAP_PROP_POS_FRAMES)) + if fps <= 0 or frame_index < 1: + return None + return (frame_index - 1.0) / fps + + def _schedule_retry(self, now: float, reason: str) -> None: + self._release_capture() + self._status = SourceStatus.RETRYING + self._last_error = reason + self._next_retry_at = now + self._retry_delay_seconds + self._retry_delay_seconds = min( + self._retry_delay_seconds * 2.0, self._retry_max_seconds + ) + + def _release_capture(self) -> None: + if self._capture is not None: + self._capture.release() + self._capture = None + + def _packet( + self, timestamp: float, status: SourceStatus, error: Optional[str] + ) -> FramePacket: + return FramePacket( + image=None, + timestamp_monotonic=timestamp, + status=status, + error=error, + )