From f80fb51a8559bd376f0627abe8c0bc03c9189a51 Mon Sep 17 00:00:00 2001 From: QiuSW Date: Tue, 21 Jul 2026 09:53:03 +0800 Subject: [PATCH] feat(v1): add verified pose adapter --- docs/04-architecture.md | 2 +- docs/06-tasks.md | 2 +- docs/07-v1-implementation-plan.md | 10 +- docs/api.md | 9 +- docs/current-state.md | 12 +-- progress.md | 9 ++ v1/pose.py | 172 ++++++++++++++++++++++++++++++ v1/tests/test_pose.py | 66 ++++++++++++ 8 files changed, 267 insertions(+), 15 deletions(-) create mode 100644 v1/pose.py create mode 100644 v1/tests/test_pose.py diff --git a/docs/04-architecture.md b/docs/04-architecture.md index 658ca14..b53114f 100644 --- a/docs/04-architecture.md +++ b/docs/04-architecture.md @@ -31,7 +31,7 @@ V1 的同一数据流既可接 RTSP，也可回放本地录像。V2 复用同一 | 应用入口 | `v1/app.py` | 装配配置、窗口、线程和依赖 | 推理细节、事件判定 | | 配置 | `v1/config.py` | 解析示例和本地配置，校验非敏感字段 | 保存真实凭证 | | 视频源 | `v1/video_source.py` | 打开、读取、重连 RTSP 或录像；输出帧、单调回放时间戳和显式来源状态 | Pose、报警 | -| Pose 适配器 | `v1/pose.py` | 统一返回 box、关键点、置信度 | 跟踪、摔倒业务结论 | +| Pose 适配器 | `v1/pose.py` | 校验锁定模型的 SHA-256、pose/person/17×3 契约，统一返回 box、关键点、置信度 | 跟踪、摔倒业务结论 | | 跟踪 | `v1/tracking.py` | 为连续人员输出 `track_id` | 根据姿态报警 | | 质量与证据 | `v1/evidence.py` | 过滤低质量点，计算水平姿态、下移和持续性证据 | GUI 状态 | | 状态机 | `v1/fall_state.py` | 管理每个 ID 的 NORMAL、SUSPECT、CONFIRMED、RECOVERING | 播放声音或存文件 | diff --git a/docs/06-tasks.md b/docs/06-tasks.md index 306e408..fdc6ef1 100644 --- a/docs/06-tasks.md +++ b/docs/06-tasks.md @@ -16,7 +16,7 @@ | --- | --- | --- | --- | --- | | T-101 | 创建 V1 包、依赖清单、示例配置和忽略规则 | T-000 | `v1/` 可导入；真实 RTSP 凭证被拒绝提交；`pytest` 能运行。 | DONE | | T-102 | 实现可重连的视频源与录像回放适配器 | T-101 | 有效本地录像可按时间戳产帧；无效源进入连接错误状态且不崩溃。 | DONE | -| T-103 | 实现 Pose 适配器与模型来源校验 | T-102 | 输出 person box、17 点和置信度；错误模型或哈希不符时给出明确错误。 | DOING | +| T-103 | 实现 Pose 适配器与模型来源校验 | T-102 | 输出 person box、17 点和置信度；错误模型或哈希不符时给出明确错误。 | DONE | | 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 5d83ac0..d2c5406 100644 --- a/docs/07-v1-implementation-plan.md +++ b/docs/07-v1-implementation-plan.md @@ -190,7 +190,7 @@ git commit -m "feat(v1): add replayable video source" - Create: `v1/pose.py` - Create: `v1/tests/test_pose.py` -- [ ] **Step 1: Write adapter shape tests** +- [x] **Step 1: Write adapter shape tests** ```python def test_pose_adapter_rejects_non_pose_model(tmp_path): @@ -202,12 +202,12 @@ def test_person_pose_has_seventeen_keypoints(fake_yolo_result): assert len(poses[0].keypoints) == 17 ``` -- [ ] **Step 2: Run tests to verify failure** +- [x] **Step 2: Run tests to verify failure** Run: `python -m pytest v1/tests/test_pose.py -v` Expected: FAIL because `PoseAdapter` is missing. -- [ ] **Step 3: Implement only the adapter contract** +- [x] **Step 3: Implement only the adapter contract** ```python from typing import Sequence, Tuple @@ -226,12 +226,12 @@ class PoseAdapter: At construction, hash the model, require task `pose`, class `person`, and exactly 17 three-value keypoints. -- [ ] **Step 4: Run tests and a real model smoke** +- [x] **Step 4: Run tests and a real model smoke** Run: `python -m pytest v1/tests/test_pose.py -v; python -c "from ultralytics import YOLO; assert YOLO('demo/best.pt').task == 'pose'"` Expected: PASS and no assertion error. -- [ ] **Step 5: Commit** +- [x] **Step 5: Commit** ```powershell git add v1/pose.py v1/tests/test_pose.py docs progress.md diff --git a/docs/api.md b/docs/api.md index dd47e2c..ce33210 100644 --- a/docs/api.md +++ b/docs/api.md @@ -40,12 +40,15 @@ ```text Keypoint = { x: float, y: float, confidence: float } PersonPose = { - track_id: string, - detected_at_monotonic: float, box_xyxy: [float, float, float, float], box_confidence: float, keypoints: Keypoint[17] } +TrackedPersonPose = { + track_id: string, + detected_at_monotonic: float, + pose: PersonPose +} PoseQuality = { accepted: bool, reason: string, @@ -71,6 +74,8 @@ FallEvent = { `FallEvent` 只在状态首次进入 `CONFIRMED` 时创建一次。连续帧更新 UI 状态，但不重复创建事件。 +`PersonPose` 是 T-103 的纯模型输出，不带人员 ID；T-104 的跟踪模块产生 `TrackedPersonPose` 后，才允许事件证据按人员连续积累。 + ## 视频来源帧合约 ```text diff --git a/docs/current-state.md b/docs/current-state.md index 4829cbb..2d48c2e 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -5,13 +5,13 @@ ## 当前快照 - 日期：2026-07-21 -- 阶段：V1 工程化起步；T-103 进行中。 +- 阶段：V1 工程化起步；T-103 已验收，等待 T-104。 - 已验证环境：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/video_source.py` 的 OpenCV 回放/重连适配器；它输出显式状态和单调时间戳。Pose、跟踪、证据、状态机和 GUI 尚未实现。 +- V1 代码：已建立安全配置基线、OpenCV 回放/重连适配器，以及 `v1/pose.py` 的 SHA-256 锁定 Pose 适配器；真实模型 smoke 已验证首帧输出 person box 和 17 点。跟踪、证据、状态机和 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` 当前有 6 项配置/视频源测试并已通过。`demo/1.mp4` 的首两帧回放时间戳已验证为 0.000000 与 0.033333 秒。`init.ps1` 会检查运行时依赖、编译旧基线并运行 V1 测试，但不会安装软件包。 +- 测试：`python -m compileall -q demo` 已通过；`python -m pytest v1/tests -v` 当前有 9 项配置/视频源/Pose 测试并已通过。`demo/1.mp4` 的首两帧回放时间戳已验证为 0.000000 与 0.033333 秒，首帧 Pose smoke 得到 2 名人员、每人 17 点。`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-103（实现 Pose 适配器与模型来源校验）。 -- 下一个可领取：完成 T-103 后为 T-104。 +- 已完成：T-000（Harness 文档与旧基线快照）、T-101（V1 安全配置基线）、T-102（视频源与录像回放）、T-103（Pose 适配器与模型校验）。 +- 正在进行：无。 +- 下一个可领取：T-104。 ## 当前可运行内容 diff --git a/progress.md b/progress.md index b9050c6..38ecac7 100644 --- a/progress.md +++ b/progress.md @@ -116,3 +116,12 @@ - 阻塞：无。 - 决策：构造适配器时必须校验模型 SHA-256、任务为 pose、类别为 person、关键点形状为 17×3；不把任何输出命名为摔倒概率。 - 下一步：写入模型来源错误与 person Pose 输出形状的失败测试。 + +## 【2026-07-21】T-103 实现 Pose 适配器与模型来源校验（完成） + +- 状态：DONE +- 变更：新增 `v1/pose.py`，以 SHA-256 锁定模型来源，要求任务为 pose、存在 person 类且关键点形状严格为 17×3；输出不含跟踪 ID 的 `PersonPose`（box、box 置信度、17 个三元关键点）。 +- 验证：先运行 `python -m pytest v1/tests/test_pose.py -v`，确认因缺少 `v1.pose` 导入失败；实现后 Pose 测试 3 passed，完整 V1 测试在本轮最后一次运行时为 9 passed。对 `demo/best.pt` 的只读 smoke 使用实际 SHA-256 加载模型，并从 `demo/1.mp4` 首帧得到 2 名人员、每人 17 点；`python -m compileall -q v1 demo` 通过。 +- 阻塞：无。 +- 决策：模型输出绝不称为摔倒概率；人员 ID 由 T-104 在 Pose 输出之后分配，模型不匹配或结果形状损坏时必须显式报错而不是继续推理。 +- 下一步：T-104，先写缺失下肢关键点拒绝和水平姿态只作为证据的失败测试。 diff --git a/v1/pose.py b/v1/pose.py new file mode 100644 index 0000000..c1ca5b0 --- /dev/null +++ b/v1/pose.py @@ -0,0 +1,172 @@ +"""Verified Ultralytics Pose adapter for the V1 event pipeline.""" + +import hashlib +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable, List, Optional, Sequence, Tuple + +import numpy as np +from ultralytics import YOLO + + +class ModelValidationError(ValueError): + """Raised when a model cannot satisfy the locked V1 Pose contract.""" + + +@dataclass(frozen=True) +class Keypoint: + x: float + y: float + confidence: float + + +@dataclass(frozen=True) +class PersonPose: + box_xyxy: Tuple[float, float, float, float] + box_confidence: float + keypoints: Sequence[Keypoint] + + +ModelFactory = Callable[[str], Any] +_SHA256 = re.compile(r"^[0-9a-fA-F]{64}$") + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with Path(path).open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +class PoseAdapter: + """Load only the locked 17-keypoint person Pose model.""" + + def __init__( + self, + model_path: Path, + expected_sha256: str, + confidence_threshold: float = 0.25, + model_factory: Optional[ModelFactory] = None, + ) -> None: + self._model_path = Path(model_path) + self._confidence = float(confidence_threshold) + if not 0.0 <= self._confidence <= 1.0: + raise ModelValidationError("confidence threshold must be between 0 and 1") + self._validate_hash(expected_sha256) + factory = model_factory or YOLO + try: + self._model = factory(str(self._model_path)) + except Exception as exc: + raise ModelValidationError("unable to load Pose model: {0}".format(exc)) + self._person_class_ids = self._validate_model_contract(self._model) + + @property + def model_path(self) -> Path: + return self._model_path + + def infer(self, image: np.ndarray) -> Sequence[PersonPose]: + results = self._model(image, conf=self._confidence, verbose=False) + return self.from_results(results, self._person_class_ids) + + @classmethod + def from_results( + cls, results: Any, person_class_ids: Optional[Sequence[int]] = None + ) -> Sequence[PersonPose]: + """Convert Ultralytics result objects into framework-independent poses.""" + + if not isinstance(results, (list, tuple)): + result_items = [results] + else: + result_items = results + poses: List[PersonPose] = [] + for result in result_items: + boxes = getattr(result, "boxes", None) + keypoints = getattr(result, "keypoints", None) + if boxes is None or keypoints is None: + raise ModelValidationError("Pose result is missing boxes or keypoints") + names = getattr(result, "names", {}) + allowed_ids = ( + set(person_class_ids) + if person_class_ids is not None + else cls._person_ids_from_names(names) + ) + box_rows = cls._rows(getattr(boxes, "xyxy", None), "boxes.xyxy") + confidence_rows = cls._rows(getattr(boxes, "conf", None), "boxes.conf") + class_rows = cls._rows(getattr(boxes, "cls", None), "boxes.cls") + keypoint_rows = cls._rows(getattr(keypoints, "data", None), "keypoints.data") + if not ( + len(box_rows) + == len(confidence_rows) + == len(class_rows) + == len(keypoint_rows) + ): + raise ModelValidationError("Pose result arrays have inconsistent lengths") + for index, box in enumerate(box_rows): + class_id = int(class_rows[index]) + if class_id not in allowed_ids: + continue + if len(box) != 4: + raise ModelValidationError("person box must contain four coordinates") + points = keypoint_rows[index] + if len(points) != 17: + raise ModelValidationError("person Pose result must contain exactly 17 keypoints") + parsed_points = [] + for point in points: + if len(point) != 3: + raise ModelValidationError("keypoint must contain x, y and confidence") + parsed_points.append( + Keypoint(float(point[0]), float(point[1]), float(point[2])) + ) + poses.append( + PersonPose( + box_xyxy=tuple(float(value) for value in box), + box_confidence=float(confidence_rows[index]), + keypoints=tuple(parsed_points), + ) + ) + return tuple(poses) + + def _validate_hash(self, expected_sha256: str) -> None: + if not _SHA256.match(expected_sha256 or ""): + raise ModelValidationError("expected SHA-256 must be a 64-character hex value") + if not self._model_path.is_file(): + raise ModelValidationError("Pose model file does not exist: {0}".format(self._model_path)) + actual_sha256 = sha256_file(self._model_path) + if actual_sha256.lower() != expected_sha256.lower(): + raise ModelValidationError("Pose model SHA-256 does not match expected value") + + @classmethod + def _validate_model_contract(cls, model: Any) -> Sequence[int]: + if getattr(model, "task", None) != "pose": + raise ModelValidationError("model task must be pose") + person_ids = cls._person_ids_from_names(getattr(model, "names", {})) + raw_shape = getattr(getattr(model, "model", None), "kpt_shape", None) + if raw_shape is None: + raw_shape = getattr(model, "kpt_shape", None) + if tuple(raw_shape or ()) != (17, 3): + raise ModelValidationError("model keypoint shape must be exactly (17, 3)") + return tuple(person_ids) + + @staticmethod + def _person_ids_from_names(names: Any) -> Sequence[int]: + items = names.items() if isinstance(names, dict) else enumerate(names or ()) + person_ids = [int(index) for index, name in items if str(name).lower() == "person"] + if not person_ids: + raise ModelValidationError("model must expose a person class") + return tuple(person_ids) + + @staticmethod + def _rows(value: Any, field_name: str) -> List[Any]: + if value is None: + raise ModelValidationError("Pose result is missing {0}".format(field_name)) + if hasattr(value, "cpu"): + value = value.cpu() + if hasattr(value, "numpy"): + value = value.numpy() + if isinstance(value, np.ndarray): + return value.tolist() + if hasattr(value, "tolist"): + return value.tolist() + return list(value) diff --git a/v1/tests/test_pose.py b/v1/tests/test_pose.py new file mode 100644 index 0000000..271ae42 --- /dev/null +++ b/v1/tests/test_pose.py @@ -0,0 +1,66 @@ +import hashlib + +import numpy as np +import pytest + +from v1.pose import ModelValidationError, PoseAdapter + + +class _FakeKeypoints: + def __init__(self, data): + self.data = data + + +class _FakeBoxes: + def __init__(self): + self.xyxy = np.array([[10.0, 20.0, 50.0, 90.0], [1.0, 2.0, 3.0, 4.0]]) + self.conf = np.array([0.9, 0.8]) + self.cls = np.array([0, 1]) + + +class _FakeResult: + names = {0: "person", 1: "chair"} + + def __init__(self): + person = [[float(index), float(index + 1), 0.9] for index in range(17)] + chair = [[float(index), float(index + 1), 0.8] for index in range(17)] + self.boxes = _FakeBoxes() + self.keypoints = _FakeKeypoints(np.array([person, chair])) + + +def test_pose_adapter_rejects_hash_mismatch_before_loading_model(tmp_path): + model_path = tmp_path / "pose.pt" + model_path.write_bytes(b"not a real model") + + with pytest.raises(ModelValidationError, match="SHA-256"): + PoseAdapter(model_path, expected_sha256="0" * 64) + + +def test_from_results_extracts_person_box_confidence_and_seventeen_keypoints(): + poses = PoseAdapter.from_results(_FakeResult()) + + assert len(poses) == 1 + assert poses[0].box_xyxy == (10.0, 20.0, 50.0, 90.0) + assert poses[0].box_confidence == 0.9 + assert len(poses[0].keypoints) == 17 + assert poses[0].keypoints[5].confidence == 0.9 + + +def test_pose_adapter_rejects_non_pose_model_after_hash_validation(tmp_path): + model_path = tmp_path / "model.pt" + model_path.write_bytes(b"model bytes") + expected_sha256 = hashlib.sha256(model_path.read_bytes()).hexdigest() + + class _WrongTaskModel: + task = "detect" + names = {0: "person"} + + class model: + kpt_shape = (17, 3) + + with pytest.raises(ModelValidationError, match="task"): + PoseAdapter( + model_path, + expected_sha256=expected_sha256, + model_factory=lambda _path: _WrongTaskModel(), + )