From e1f9f9111096a7fa6bbcf636ef1eb012bae19873 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=9C=A0=EC=B0=BD=EC=9A=B1?= Date: Sun, 12 Jul 2026 10:35:21 +0900 Subject: [PATCH] docs: P1 pilot integration plan (wire edge daemon + CPU pose + detail view) Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Ne4rU4FToTLHcBUMUvUnT1 --- .../plans/2026-07-12-p1-pilot-integration.md | 772 ++++++++++++++++++ 1 file changed, 772 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-12-p1-pilot-integration.md diff --git a/docs/superpowers/plans/2026-07-12-p1-pilot-integration.md b/docs/superpowers/plans/2026-07-12-p1-pilot-integration.md new file mode 100644 index 0000000..74814be --- /dev/null +++ b/docs/superpowers/plans/2026-07-12-p1-pilot-integration.md @@ -0,0 +1,772 @@ +# P1 파일럿 통합 (Pilot Integration) Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 이미 구현·병합된 P1-A(엣지 감지 라이브러리) + P1-B(클라우드) 사이의 **끊긴 배선을 연결**해, 노트북 웹캠으로 실행되는 **엔드투엔드 파일럿**을 만든다: 실제 프레임 → CPU pose → 감지 → 서명 POST → 클라우드 → 관리자가 **사진 보고 알림 받고 확인**. 역루프(대시보드 구역 → 엣지 적용)와 하트비트도 살린다. (실장비 Hailo·폐쇄망 번들·드릴 게이트는 별도 후속.) + +**Architecture:** 순수 배선. 기존 부품(`run_once`, EventEngine, PostClient, OfflineQueue, ZoneIntrusionDetector, 클라우드 ingest/detail/snapshot)은 그대로 두고 (1) 엣지 데몬(main 루프)이 부품을 조립·구동, (2) CloudClient가 config 폴링·하트비트, (3) MoveNet ONNX 어댑터가 PosePort 구현, (4) 클라우드 detail 뷰·Web Audio가 응답루프 완성. 모든 데몬/클라이언트 로직은 주입(fake source/pose/clock/http)으로 TDD; 실 pose는 정적이미지로 게이트 테스트. + +**Tech Stack:** Python 3.11+(dev 3.13), pytest, numpy, opencv, requests, **onnxruntime**(CPU) + **MoveNet SinglePose Lightning ONNX**(COCO-17, Apache-2). 클라우드는 기존 바닐라 PHP + JS(Web Audio, 의존성 0). 런타임 ultralytics(AGPL) 금지 유지. + +## Global Constraints + +- Python **3.11+**, 타입힌트, `edge/` 기존 패턴 준수. PHP 8.5 `declare(strict_types=1)`, 클라우드 스위트 **0 warnings/0 deprecations** 유지. +- **판정/루프 로직은 주입 타임스탬프·주입 의존성으로 TDD** — 데몬 루프 내부에서 `time.sleep`/`time.time` 직접 호출 금지(주입된 clock/sleeper 사용). 실 pose 추론만 게이트(모델/onnxruntime 없으면 `skip`). +- **폐쇄망 지속:** 새 의존성(onnxruntime)·모델(.onnx)은 **빌드머신서 받아 번들**할 것 — 후속 패키징 태스크 대상. 런타임 인터넷은 Cafe24 POST + config 폴링뿐. +- **HMAC 계약 유지:** config 폴링·하트비트도 P1-A `sign(nonce+ts+body)` 동일 서명(클라우드 T14가 이미 검증). PostClient 서명 재사용. +- **키포인트 계약:** PosePort는 `list[Person]` (COCO-17 `Keypoint`) 반환 — MoveNet 17관절을 그 순서로 매핑. +- YAGNI: 파일럿에 필요한 배선만. 멀티캠·감도·처리결과 등 P2/P3는 손대지 않음. + +## File Structure + +``` +edge/ + cloudclient.py # CloudClient: fetch_config(HMAC GET), send_heartbeat(HMAC POST) + zones.py # zone rows(dict) -> Zone objects; thresholds + runtime.py # PipelineRuntime: assemble source/pose/tracker/smoother/engine/ring/snapshotter/sink; step(); apply_zones() + daemon.py # main loop: step every frame + periodic sync/heartbeat/queue-flush; signals; main() + __main__.py # `python -m edge` -> daemon.main() + adapters/ + movenet_pose.py # MoveNetPoseAdapter(PosePort) via onnxruntime + COCO-17 remap + tools/ + fetch_movenet.py # build-machine: download MoveNet Lightning ONNX to models/ +cloud/ + src/View/event_detail.php # (replace stub) HTML detail shell + script + public/assets/detail.js # fetch GET /events/{id} -> render meta + + public/assets/app.js # (edit) link rows to /events/{id}/view, desk Web Audio on new critical + public/index.php # (edit) add GET /events/{id}/view HTML route +tests/ (edge) + cloud/tests/ +docs/pilot-runbook.md +``` + +--- + +## Task 1: CloudClient — config 폴링 + 하트비트 (HMAC 서명) + +**Files:** Create `edge/cloudclient.py`, `tests/test_cloudclient.py` + +**Interfaces:** +- Consumes: `sign` (edge/alert.py), injected `http_get`/`http_post`, `nonce_fn`, `now_fn`. +- Produces: + - `CloudClient(base_url, device_id, secret, nonce_fn, now_fn, http_get=None, http_post=None, timeout=5.0)` + - `fetch_config() -> dict | None` — GET `{base}/api/config?device_id=`, headers X-Nonce/X-Timestamp/X-Signature (sign over empty body), returns parsed `{version, zones, thresholds}` on 200 else None. + - `send_heartbeat(metrics: dict) -> bool` — POST `{base}/api/heartbeat` body `{device_id, **metrics}` signed, True on 2xx. + +- [ ] **Step 1: 실패 테스트** — `tests/test_cloudclient.py` + +```python +import json +from edge.cloudclient import CloudClient +from edge.alert import sign + +class _Resp: + def __init__(self, code, body=""): self.status_code = code; self.text = body + def json(self): return json.loads(self.text) + +def test_fetch_config_signs_and_parses(): + captured = {} + def fake_get(url, headers=None, timeout=None): + captured["url"] = url; captured["headers"] = headers + return _Resp(200, json.dumps({"version": 3, "zones": [], "thresholds": {}})) + c = CloudClient("https://x", "pi-1", "sek", nonce_fn=lambda: "n1", now_fn=lambda: 1000.0, http_get=fake_get) + cfg = c.fetch_config() + assert cfg["version"] == 3 + assert "device_id=pi-1" in captured["url"] + assert captured["headers"]["X-Signature"] == sign("sek", "n1", "1000.0", "") + +def test_fetch_config_none_on_error(): + c = CloudClient("https://x", "pi-1", "sek", lambda: "n", lambda: 1.0, http_get=lambda *a, **k: _Resp(401)) + assert c.fetch_config() is None + +def test_send_heartbeat_signs_body(): + captured = {} + def fake_post(url, data=None, headers=None, timeout=None): + captured["data"] = data; captured["headers"] = headers; return _Resp(200) + c = CloudClient("https://x", "pi-1", "sek", lambda: "n1", lambda: 1000.0, http_post=fake_post) + assert c.send_heartbeat({"fps": 12.0}) is True + body = captured["data"] + assert json.loads(body)["device_id"] == "pi-1" + assert captured["headers"]["X-Signature"] == sign("sek", "n1", "1000.0", body) +``` + +- [ ] **Step 2: 실패 확인** — `pytest tests/test_cloudclient.py -v` → FAIL (ModuleNotFoundError) + +- [ ] **Step 3: 구현** — `edge/cloudclient.py` + +```python +import json + +from edge.alert import sign + + +class CloudClient: + def __init__(self, base_url: str, device_id: str, secret: str, + nonce_fn, now_fn, http_get=None, http_post=None, timeout: float = 5.0): + self.base_url = base_url.rstrip("/") + self.device_id = device_id + self.secret = secret + self.nonce_fn = nonce_fn + self.now_fn = now_fn + self.timeout = timeout + if http_get is None or http_post is None: + import requests + http_get = http_get or requests.get + http_post = http_post or requests.post + self._get = http_get + self._post = http_post + + def _headers(self, body: str) -> dict: + nonce = self.nonce_fn() + ts = str(self.now_fn()) + return {"X-Nonce": nonce, "X-Timestamp": ts, + "X-Signature": sign(self.secret, nonce, ts, body)} + + def fetch_config(self) -> dict | None: + url = f"{self.base_url}/api/config?device_id={self.device_id}" + try: + resp = self._get(url, headers=self._headers(""), timeout=self.timeout) + except Exception: + return None + if 200 <= resp.status_code < 300: + try: + return resp.json() + except Exception: + return None + return None + + def send_heartbeat(self, metrics: dict) -> bool: + payload = {"device_id": self.device_id, **metrics} + body = json.dumps(payload, sort_keys=True) + headers = {"Content-Type": "application/json", **self._headers(body)} + try: + resp = self._post(f"{self.base_url}/api/heartbeat", data=body, + headers=headers, timeout=self.timeout) + except Exception: + return False + return 200 <= resp.status_code < 300 +``` + +- [ ] **Step 4: 통과 + 커밋** — `pytest tests/test_cloudclient.py -v` → PASS + +```bash +git add edge/cloudclient.py tests/test_cloudclient.py +git commit -m "feat(edge): CloudClient for config poll + heartbeat (HMAC-signed)" +``` + +--- + +## Task 2: 클라우드 zone 행 → Zone 객체 로딩 + +**Files:** Create `edge/zones.py`, `tests/test_zones.py` + +**Interfaces:** +- Consumes: `Zone` (edge/detectors/zone.py) +- Produces: + - `zones_from_config(config: dict) -> list[Zone]` — `config["zones"]` 각 행(`{zone_id, camera_id, level, polygon_json}`)을 `Zone(zone_id, level, polygon)`으로. `polygon_json`은 JSON 문자열(또는 이미 list) → `[(x,y),...]`. 파싱 실패/3점미만 행은 건너뜀. + - `thresholds_from_config(config: dict) -> dict` — `config["thresholds"]` 그대로 반환({}). + +- [ ] **Step 1: 실패 테스트** — `tests/test_zones.py` + +```python +from edge.zones import zones_from_config, thresholds_from_config + +def test_parses_zone_rows_json_string_polygon(): + cfg = {"zones": [ + {"zone_id": "z1", "camera_id": "cam-1", "level": "restricted", + "polygon_json": "[[0,0],[9,0],[9,9]]"}, + ], "thresholds": {"fall_prone_s": 3.0}} + zones = zones_from_config(cfg) + assert len(zones) == 1 + assert zones[0].zone_id == "z1" + assert zones[0].level == "restricted" + assert zones[0].polygon == [(0.0, 0.0), (9.0, 0.0), (9.0, 9.0)] + assert thresholds_from_config(cfg)["fall_prone_s"] == 3.0 + +def test_skips_malformed_row(): + cfg = {"zones": [{"zone_id": "bad"}], "thresholds": {}} + assert zones_from_config(cfg) == [] + +def test_empty_config(): + assert zones_from_config({}) == [] + assert thresholds_from_config({}) == {} +``` + +- [ ] **Step 2: 실패 확인** → FAIL + +- [ ] **Step 3: 구현** — `edge/zones.py` + +```python +import json + +from edge.detectors.zone import Zone + + +def zones_from_config(config: dict) -> list[Zone]: + out: list[Zone] = [] + for row in config.get("zones", []) or []: + try: + raw = row["polygon_json"] + pts = json.loads(raw) if isinstance(raw, str) else raw + polygon = [(float(p[0]), float(p[1])) for p in pts] + if len(polygon) < 3: + continue + out.append(Zone(str(row["zone_id"]), str(row["level"]), polygon)) + except (KeyError, ValueError, TypeError, json.JSONDecodeError): + continue + return out + + +def thresholds_from_config(config: dict) -> dict: + t = config.get("thresholds") + return t if isinstance(t, dict) else {} +``` + +- [ ] **Step 4: 통과 + 커밋** + +```bash +git add edge/zones.py tests/test_zones.py +git commit -m "feat(edge): parse cloud config zones/thresholds into Zone objects" +``` + +--- + +## Task 3: PipelineRuntime — 부품 조립 + zone 핫스왑 + +**Files:** Create `edge/runtime.py`, `tests/test_runtime.py` + +**Interfaces:** +- Consumes: `run_once` (edge/app.py), `IouTracker`, `TrackSmootherPool`, `EventEngine`+detectors, `RingBuffer`/`Snapshotter`/`FaceBlurrer`, `Config` (edge/config.py), `Zone`. +- Produces: + - `PipelineRuntime(config, source, pose, sink)` — 내부에 tracker/pool/engine/ring/snapshotter 구성(ZoneIntrusionDetector는 `zone_detector` 속성으로 보관). + - `step(now: float) -> list` — 한 프레임 처리(`run_once` 호출), 이벤트 반환. + - `apply_zones(zones: list[Zone]) -> None` — 러닝 중 ZoneIntrusionDetector의 zone 목록 교체(역루프). + +- [ ] **Step 1: 실패 테스트** — `tests/test_runtime.py` + +```python +import numpy as np +from edge.config import load_config +from edge.models import Person, BBox, Keypoint +from edge.detectors.zone import Zone +from edge.runtime import PipelineRuntime + +def _lying_person(): + kps = [Keypoint(150, 300, 0.9) for _ in range(17)] + kps[5] = Keypoint(100, 300, 0.9); kps[6] = Keypoint(100, 310, 0.9) + kps[11] = Keypoint(200, 300, 0.9); kps[12] = Keypoint(200, 310, 0.9) + kps[0] = Keypoint(90, 320, 0.9) + return Person(BBox(90, 290, 210, 320), kps) + +class _Source: + def read(self): return True, np.full((360, 640, 3), 128, np.uint8) +class _Pose: + def infer(self, frame): return [_lying_person()] +class _Sink: + def __init__(self): self.events = [] + def try_send(self, e, snaps): self.events.append(e); return True + +def test_step_runs_pipeline_and_emits_fall(): + cfg = load_config({"FALL_PRONE_S": "0.5"}) + sink = _Sink() + rt = PipelineRuntime(cfg, _Source(), _Pose(), sink) + rt.step(now=0.0) + rt.step(now=0.6) + from edge.events import EventType + assert any(e.event_type == EventType.FALL for e in sink.events) + +def test_apply_zones_hotswaps(): + cfg = load_config({}) + rt = PipelineRuntime(cfg, _Source(), _Pose(), _Sink()) + rt.apply_zones([Zone("z1", "restricted", [(0, 0), (9, 0), (9, 9)])]) + assert rt.zone_detector.zones[0].zone_id == "z1" +``` + +- [ ] **Step 2: 실패 확인** → FAIL + +- [ ] **Step 3: 구현** — `edge/runtime.py` + +```python +from edge.app import run_once +from edge.tracking import IouTracker +from edge.smoothing import TrackSmootherPool +from edge.detectors.fall import FallDetector +from edge.detectors.immobility import ImmobilityWatchdog +from edge.detectors.ghost import GhostIncidentDetector +from edge.detectors.zone import ZoneIntrusionDetector +from edge.engine import EventEngine +from edge.snapshot import RingBuffer, FaceBlurrer, Snapshotter + + +class PipelineRuntime: + def __init__(self, config, source, pose, sink): + self.config = config + self.source = source + self.pose = pose + self.sink = sink + self.tracker = IouTracker() + self.pool = TrackSmootherPool() + self.zone_detector = ZoneIntrusionDetector([], config.zone) + self.engine = EventEngine( + FallDetector(config.fall), + ImmobilityWatchdog(config.immobility), + GhostIncidentDetector(config.ghost), + self.zone_detector, + config.engine, + ) + self.ring = RingBuffer() + self.snapshotter = Snapshotter(FaceBlurrer()) + + def step(self, now: float) -> list: + return run_once(self.source, self.pose, self.tracker, self.pool, + self.engine, self.ring, self.snapshotter, self.sink, + now_fn=lambda: now) + + def apply_zones(self, zones: list) -> None: + self.zone_detector.zones = list(zones) +``` + +> **NOTE (implementer):** `ZoneIntrusionDetector.zones` is a public list (see edge/detectors/zone.py) — reassigning it hot-swaps the active zones without rebuilding the engine. Confirm the attribute name is `zones`. Also confirm `run_once`'s parameter order is `(source, pose, tracker, pool, engine, ring, snapshotter, sink, now_fn)` (P1-A final form after the per-track-smoother fix). + +- [ ] **Step 4: 통과 + 커밋** + +```bash +git add edge/runtime.py tests/test_runtime.py +git commit -m "feat(edge): PipelineRuntime assembles components + hot-swaps zones" +``` + +--- + +## Task 4: 데몬 — main 루프 + 주기 동기/하트비트/큐flush + +**Files:** Create `edge/daemon.py`, `edge/__main__.py`, `tests/test_daemon.py`; Modify `pyproject.toml` + +**Interfaces:** +- Consumes: `PipelineRuntime` (T3), `CloudClient` (T1), `zones_from_config` (T2), `OfflineQueue` (edge/queue.py), `PostClient` (edge/alert.py). +- Produces: + - `run_loop(runtime, client, queue, *, clock, sleep, iterations=None, sync_every=150, heartbeat_every=900, flush_every=300, frame_interval=0.0) -> dict` — `iterations` 프레임 루프(None이면 무한, 테스트는 유한). 매 프레임 `runtime.step(clock())`(예외 삼켜 카운트); sync_every 프레임마다 `client.fetch_config()`→`runtime.apply_zones(zones_from_config(...))`; heartbeat_every마다 `send_heartbeat`; flush_every마다 `queue.flush()`. 반환 카운터 dict. + - `main()` — env로 구성 후 무한 루프. `edge/__main__.py`가 호출. (`# pragma: no cover` — 실장비 엔트리) + +- [ ] **Step 1: 실패 테스트** — `tests/test_daemon.py` + +```python +from edge.daemon import run_loop + +class _Runtime: + def __init__(self): self.steps = 0; self.applied = None + def step(self, now): self.steps += 1; return [] + def apply_zones(self, zones): self.applied = zones +class _Client: + def __init__(self): self.configs = 0; self.beats = 0 + def fetch_config(self): self.configs += 1; return {"zones": [], "thresholds": {}} + def send_heartbeat(self, m): self.beats += 1; return True +class _Queue: + def __init__(self): self.flushes = 0 + def flush(self): self.flushes += 1; return 0 + +def test_loop_steps_and_periodics(): + rt, cl, q = _Runtime(), _Client(), _Queue() + ticks = iter(range(1000)) + counters = run_loop(rt, cl, q, clock=lambda: float(next(ticks)), + sleep=lambda s: None, iterations=10, + sync_every=5, heartbeat_every=5, flush_every=5) + assert rt.steps == 10 + assert cl.configs == 2 and cl.beats == 2 and q.flushes == 2 # at frames 5 and 10 + assert rt.applied == [] # zones applied from fetched config + +def test_loop_survives_step_exception(): + class Boom(_Runtime): + def step(self, now): + self.steps += 1 + if self.steps == 2: raise RuntimeError("frame glitch") + return [] + rt = Boom() + counters = run_loop(rt, _Client(), _Queue(), clock=lambda: 0.0, + sleep=lambda s: None, iterations=3, + sync_every=99, heartbeat_every=99, flush_every=99) + assert rt.steps == 3 # kept going past the exception + assert counters["errors"] == 1 +``` + +- [ ] **Step 2: 실패 확인** → FAIL + +- [ ] **Step 3: 구현** — `edge/daemon.py` + +```python +import os + + +def run_loop(runtime, client, queue, *, clock, sleep, iterations=None, + sync_every=150, heartbeat_every=900, flush_every=300, + frame_interval=0.0): + counters = {"frames": 0, "syncs": 0, "beats": 0, "flushes": 0, "errors": 0} + i = 0 + while iterations is None or i < iterations: + i += 1 + try: + runtime.step(clock()) + counters["frames"] += 1 + except Exception: + counters["errors"] += 1 + if sync_every and i % sync_every == 0: + cfg = client.fetch_config() + if cfg is not None: + from edge.zones import zones_from_config + runtime.apply_zones(zones_from_config(cfg)) + counters["syncs"] += 1 + if heartbeat_every and i % heartbeat_every == 0: + if client.send_heartbeat({"fps": 0.0, "frames": counters["frames"]}): + counters["beats"] += 1 + if flush_every and i % flush_every == 0: + queue.flush() + counters["flushes"] += 1 + if frame_interval: + sleep(frame_interval) + return counters + + +def main() -> None: # pragma: no cover (integration entrypoint, exercised on device) + import time + from edge.config import load_config + from edge.cloudclient import CloudClient + from edge.alert import PostClient + from edge.queue import OfflineQueue + from edge.runtime import PipelineRuntime + from edge.adapters.uvc_source import UvcSource + from edge.adapters.movenet_pose import MoveNetPoseAdapter + + cfg = load_config(dict(os.environ)) + nonce = lambda: os.urandom(8).hex() + now = time.time + base = cfg.cafe24_url.rstrip("/") + client = CloudClient(base, cfg.device_id, cfg.secret, nonce, now) + post = PostClient(base + "/api/events", cfg.device_id, cfg.secret, nonce, now) + queue = OfflineQueue(os.environ.get("QUEUE_DB", "queue.db"), post) + source = UvcSource(int(os.environ.get("UVC_DEVICE", "0"))) + pose = MoveNetPoseAdapter(os.environ["MOVENET_ONNX"]) + runtime = PipelineRuntime(cfg, source, pose, queue) + run_loop(runtime, client, queue, clock=now, sleep=time.sleep, iterations=None, + frame_interval=1.0 / float(os.environ.get("TARGET_FPS", "12"))) +``` + +`edge/__main__.py`: + +```python +from edge.daemon import main + +if __name__ == "__main__": + main() +``` + +- [ ] **Step 4: pyproject.toml에 스크립트 엔트리 추가** + +```toml +[project.scripts] +move-detect-edge = "edge.daemon:main" +``` + +- [ ] **Step 5: 통과 + 커밋** — `pytest tests/test_daemon.py -v` → PASS; 전체 `pytest -q` + +```bash +git add edge/daemon.py edge/__main__.py tests/test_daemon.py pyproject.toml +git commit -m "feat(edge): runnable daemon loop (step + periodic sync/heartbeat/flush)" +``` + +--- + +## Task 5: MoveNet CPU pose 어댑터 (실 pose 경로) + +**Files:** Create `edge/adapters/movenet_pose.py`, `edge/tools/fetch_movenet.py`, `tests/test_movenet_pose.py` + +**Interfaces:** +- Consumes: `Person, BBox, Keypoint` (edge/models.py), PosePort 계약. +- Produces: `MoveNetPoseAdapter(onnx_path, input_size=192, session=None, min_person_score=0.2)` implements PosePort `infer(frame) -> list[Person]`. MoveNet SinglePose Lightning: 출력 `[1,1,17,3]` = (y,x,score) 정규화(0..1). 프레임 크기로 역정규화 → COCO-17 순서 Keypoint(x,y,conf). bbox = 가시 관절(conf≥min_person_score) min/max; 가시 관절 0이면 빈 리스트. + +- [ ] **Step 1: 실패 테스트** — `tests/test_movenet_pose.py` (session 주입 — onnxruntime 불필요) + +```python +import numpy as np +from edge.adapters.movenet_pose import MoveNetPoseAdapter + +class _FakeSession: + """MoveNet 출력 형태 [1,1,17,3] (y,x,score) 정규화.""" + def get_inputs(self): + class I: name = "input" + return [I()] + def run(self, _out, feed): + kp = np.zeros((1, 1, 17, 3), dtype=np.float32) + for j in range(17): + kp[0, 0, j] = [0.5, 0.25, 0.9] # y=0.5, x=0.25, score=0.9 + return [kp] + +def test_infer_maps_movenet_to_coco17_persons(): + frame = np.zeros((400, 800, 3), dtype=np.uint8) + a = MoveNetPoseAdapter("unused.onnx", session=_FakeSession()) + persons = a.infer(frame) + assert len(persons) == 1 + p = persons[0] + assert len(p.keypoints) == 17 + # x = 0.25 * 800 = 200, y = 0.5 * 400 = 200 + assert abs(p.keypoints[0].x - 200) < 1 and abs(p.keypoints[0].y - 200) < 1 + assert p.keypoints[0].conf == 0.9 + assert p.bbox.x1 <= 200 <= p.bbox.x2 + +def test_low_score_person_dropped(): + class Low(_FakeSession): + def run(self, _o, feed): + return [np.zeros((1, 1, 17, 3), dtype=np.float32)] # all score 0 + a = MoveNetPoseAdapter("unused.onnx", session=Low()) + assert a.infer(np.zeros((10, 10, 3), np.uint8)) == [] +``` + +- [ ] **Step 2: 실패 확인** → FAIL + +- [ ] **Step 3: 구현** — `edge/adapters/movenet_pose.py` + +```python +import cv2 +import numpy as np + +from edge.models import Person, BBox, Keypoint + + +class MoveNetPoseAdapter: + """MoveNet SinglePose Lightning (COCO-17) via onnxruntime. CPU dev/pilot PosePort. + 출력 [1,1,17,3] = (y, x, score) normalized 0..1.""" + + def __init__(self, onnx_path: str, input_size: int = 192, session=None, + min_person_score: float = 0.2): + self.input_size = input_size + self.min_person_score = min_person_score + if session is None: + import onnxruntime # gated: bundled on build machine + session = onnxruntime.InferenceSession(onnx_path, providers=["CPUExecutionProvider"]) + self.session = session + self._input_name = session.get_inputs()[0].name + + def infer(self, frame) -> list: + h, w = frame.shape[:2] + img = cv2.resize(frame, (self.input_size, self.input_size)) + img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB).astype(np.int32)[None, ...] + out = self.session.run(None, {self._input_name: img})[0] # [1,1,17,3] + kp = out[0, 0] # [17,3] (y,x,score) + keypoints = [Keypoint(float(kp[j, 1] * w), float(kp[j, 0] * h), float(kp[j, 2])) + for j in range(17)] + vis = [k for k in keypoints if k.conf >= self.min_person_score] + if not vis: + return [] + xs = [k.x for k in vis]; ys = [k.y for k in vis] + bbox = BBox(min(xs), min(ys), max(xs), max(ys)) + return [Person(bbox, keypoints)] +``` + +`edge/tools/fetch_movenet.py`: + +```python +"""빌드머신 전용: MoveNet SinglePose Lightning ONNX 다운로드(폐쇄망 번들). Apache-2.0.""" +import os +import urllib.request + +# MoveNet Lightning ONNX (예시 소스; 조직 미러로 교체 가능) +URL = "https://github.com/PINTO0309/PINTO_model_zoo/raw/main/115_MoveNet/resources/movenet_singlepose_lightning_4.onnx" +DEST = os.path.join(os.path.dirname(__file__), "..", "..", "models", "movenet_lightning.onnx") + +def main(): + os.makedirs(os.path.dirname(DEST), exist_ok=True) + print(f"downloading MoveNet ONNX -> {DEST}") + urllib.request.urlretrieve(URL, DEST) + print("done") + +if __name__ == "__main__": + main() +``` + +> **NOTE (implementer):** the `fetch_movenet.py` URL is a build-machine convenience; verify the model's input size/dtype and output layout `[1,1,17,3]` against the actual downloaded file before relying on it on hardware — the fake-session unit test locks the ADAPTER's remap logic, not the specific model file. If the real model expects a different input size/dtype, adjust `input_size`/dtype accordingly; the COCO-17 remap stays. + +- [ ] **Step 4: 통과 + 커밋** — `pytest tests/test_movenet_pose.py -v` → PASS + +```bash +git add edge/adapters/movenet_pose.py edge/tools/fetch_movenet.py tests/test_movenet_pose.py +git commit -m "feat(edge): MoveNet ONNX CPU pose adapter (COCO-17 remap)" +``` + +--- + +## Task 6: 클라우드 사고 상세 뷰 — 스냅샷 표시 + +**Files:** Modify `cloud/src/View/event_detail.php`; Create `cloud/public/assets/detail.js`; Modify `cloud/public/index.php`; Modify `cloud/tests/WiringSmokeTest.php` + +**Interfaces:** +- Consumes: existing `GET /events/{id}` (JSON: event+snapshots) + `GET /snap/{uuid}` (auth image). +- Produces: `GET /events/{id}/view` → renders `event_detail.php` (HTML shell + ` +``` + +- [ ] **Step 3: detail.js** — `cloud/public/assets/detail.js` + +```javascript +(function () { + var m = location.pathname.match(/\/events\/(\d+)\/view/); + if (!m) return; + var id = m[1]; + fetch('/events/' + id, { credentials: 'same-origin' }) + .then(function (r) { return r.ok ? r.json() : Promise.reject(r.status); }) + .then(function (d) { + var ev = d.event || {}; + var meta = document.getElementById('detail-meta'); + meta.textContent = ''; + [['유형', ev.event_type], ['심각도', ev.severity], ['카메라', ev.camera_id], + ['시각', ev.occurred_at], ['지속(초)', ev.duration_s], ['확인', ev.acknowledged_by || '미확인']] + .forEach(function (row) { + var p = document.createElement('div'); + p.textContent = row[0] + ': ' + (row[1] == null ? '-' : row[1]); + meta.appendChild(p); + }); + var box = document.getElementById('detail-snapshots'); + (d.snapshots || []).forEach(function (s) { + var img = document.createElement('img'); + img.src = '/snap/' + encodeURIComponent(s.file_uuid); + img.alt = '사고 스냅샷'; + img.style.maxWidth = '100%'; + box.appendChild(img); + }); + }) + .catch(function () { + document.getElementById('detail-meta').textContent = '불러오기 실패'; + }); +})(); +``` + +- [ ] **Step 4: index.php 라우트** — 뷰 라우트 섹션에 추가: + +```php +$r->add('GET', '/events/{id}/view', fn() => $render('event_detail')); +``` + +- [ ] **Step 5: 통과 + 커밋** — 전체 `cd cloud && ./vendor/bin/phpunit --display-warnings --display-deprecations` green; `php -l cloud/public/index.php`. + +```bash +git add cloud/src/View/event_detail.php cloud/public/assets/detail.js cloud/public/index.php cloud/tests/WiringSmokeTest.php +git commit -m "feat(cloud): incident detail view renders snapshot images" +``` + +--- + +## Task 7: 응답 루프 — 목록 링크 + 데스크 Web Audio + +**Files:** Modify `cloud/public/assets/app.js` + +**Interfaces:** +- Consumes: `GET /api/events` (기존 폴링). +- Produces: 각 목록 행이 `/events/{id}/view`로 링크; **새 미확인 critical 이벤트 감지 시 데스크 Web Audio 반복 비프**(미확인 critical 존재 동안), 첫 클릭으로 audio 언락. 색맹안전 배지 유지. 기존 buildRow/폴링/ack 동작 보존. + +- [ ] **Step 1: app.js 편집** — 기존 구조에 아래를 통합(구현자는 실제 함수명에 맞춤): + +```javascript +// --- desk audio alert (no binary asset; WebAudio beep) --- +var audioCtx = null; +document.addEventListener('click', function unlock() { + if (!audioCtx) audioCtx = new (window.AudioContext || window.webkitAudioContext)(); + document.removeEventListener('click', unlock); +}); +var alertedCritical = {}; // event_id -> true (already beeped once) +var alarmTimer = null; +function beep() { + if (!audioCtx) return; + var o = audioCtx.createOscillator(), g = audioCtx.createGain(); + o.frequency.value = 880; o.connect(g); g.connect(audioCtx.destination); + g.gain.value = 0.2; o.start(); o.stop(audioCtx.currentTime + 0.4); +} +function updateAlarm(events) { + var unackedCritical = false; + events.forEach(function (e) { + if (e.severity === 'critical' && !e.acknowledged_by) { + unackedCritical = true; + if (!alertedCritical[e.event_id]) { alertedCritical[e.event_id] = true; beep(); } + } + }); + if (unackedCritical && !alarmTimer) alarmTimer = setInterval(beep, 3000); + if (!unackedCritical && alarmTimer) { clearInterval(alarmTimer); alarmTimer = null; } +} +``` + +그리고: (a) 행 빌더에서 각 행을 `/events/{event_id}/view` 링크로 감싸기 — `var a = document.createElement('a'); a.href = '/events/' + e.event_id + '/view'; a.appendChild(rowEl);` 후 컨테이너에 append. (b) 폴링 성공 콜백에서 `renderList(events)` 뒤에 `updateAlarm(events)` 호출. + +- [ ] **Step 2: 통과 + 커밋** — `node --check cloud/public/assets/app.js`(node 있으면), 전체 phpunit green(JS만 변경). + +```bash +git add cloud/public/assets/app.js +git commit -m "feat(cloud): dashboard rows link to detail + desk Web Audio on critical" +``` + +--- + +## Task 8: 파일럿 런북 + 게이트 pose 스모크 + +**Files:** Create `docs/pilot-runbook.md`, `tests/hardware/test_movenet_smoke.py` + +- [ ] **Step 1: 게이트 스모크** — `tests/hardware/test_movenet_smoke.py` + +```python +import os +import numpy as np +import pytest + +MODEL = os.path.join(os.path.dirname(__file__), "..", "..", "models", "movenet_lightning.onnx") + +@pytest.mark.skipif(not os.path.exists(MODEL), reason="MoveNet model not bundled") +def test_real_movenet_runs(): + pytest.importorskip("onnxruntime") + from edge.adapters.movenet_pose import MoveNetPoseAdapter + a = MoveNetPoseAdapter(MODEL) + persons = a.infer(np.full((480, 640, 3), 128, np.uint8)) + assert isinstance(persons, list) # runs end-to-end; each Person has 17 keypoints +``` + +- [ ] **Step 2: 파일럿 런북** — `docs/pilot-runbook.md`, 노트북 파일럿 절차: + 1. **빌드:** `pip install onnxruntime`; `python -m edge.tools.fetch_movenet` (모델 → `models/`). + 2. **클라우드:** `php -S localhost:8000 -t cloud/public` (config.php에 로컬 DB) 또는 Cafe24. admin 계정 생성 + device secret을 `devices` 테이블에 등록. + 3. **env:** `DEVICE_ID`, `DEVICE_SECRET`(클라우드와 일치), `CAFE24_URL`(예 http://localhost:8000), `MOVENET_ONNX=models/movenet_lightning.onnx`, `UVC_DEVICE=0`, `TARGET_FPS=12`. + 4. **실행:** `move-detect-edge` (또는 `python -m edge`). + 5. **데모:** 웹캠 앞 눕기 → 3초 후 fall, 10초 후 no_motion(위급, 데스크 비프) → 목록행 클릭 → 상세서 스냅샷 확인 → 확인처리(비프 멈춤). 대시보드서 구역 그리기 → 발끝 진입 → 침입 이벤트(역루프). 네트워크 끊고 이벤트 → 복구 후 큐 flush. + 6. **한계(명시):** 단일캠·CPU pose(저fps)·SMS 미연동(Web Audio만)·前中後 아닌 단일프레임. 실장비 Hailo·폐쇄망 번들·드릴 게이트·F14 컴플라이언스는 후속. + +- [ ] **Step 3: 통과 + 커밋** — `pytest -q`(hardware skip) green. + +```bash +git add docs/pilot-runbook.md tests/hardware/test_movenet_smoke.py +git commit -m "docs: pilot runbook + gated MoveNet smoke test" +``` + +--- + +## Self-Review + +**1. 최소경로 커버리지 (기능검토 1~7):** +- 엣지 데몬(1) → T4 ✓ · CPU pose(2) → T5 ✓ · edge→클라우드 흐름(3) → T4 데몬+기존 PostClient ✓ +- 사고 가시화(4) → T6 ✓ · 응답 1채널(5) → T7 Web Audio ✓ +- config-sync 역루프(6) → T1+T2+T3(apply_zones)+T4(주기 sync) ✓ · 하트비트(7) → T1+T4 ✓ · 큐 flush(rank7) → T4 ✓ +- **범위 밖(명시):** 실 Hailo(하드웨어), 폐쇄망 전체 번들(T5/T8은 fetch+게이트만; 번들·차단설치는 후속 패키징), 드릴 게이트(P1-C), F14 컴플라이언스(별도), 前中後 링버퍼(rank11 — 파일럿 단일프레임 유지). +- **의도적 미포함:** critical SMS/webhook 실연동(leak_sms 게이트 정보 필요) — T7 Web Audio로 "사람 알림" 증명, 실연동 후속. + +**2. Placeholder scan:** 데몬/클라이언트/런타임/어댑터/zone은 실코드+TDD. T5 실모델 스모크·T8은 게이트(모델 없으면 skip) — 의도적. fetch URL은 빌드 편의(구현자 노트로 검증 지시). + +**3. Type consistency:** `CloudClient.sign` = PostClient/클라우드 T14 동일(`nonce+ts+body`). `PipelineRuntime.step`가 `run_once(source,pose,tracker,pool,engine,ring,snapshotter,sink,now_fn)` — P1-A 최종형(per-track smoother 수정 후) 일치. `apply_zones`가 `ZoneIntrusionDetector.zones` 재할당. MoveNet [1,1,17,3](y,x,score)→COCO-17 Keypoint(x,y,conf) 일관. + +--- + +## Execution Handoff + +**Plan complete. 저장: `docs/superpowers/plans/2026-07-12-p1-pilot-integration.md`.** + +실행: **Subagent-Driven (추천)** 또는 Inline. 어느 방식?