move_detect/docs/superpowers/plans/2026-07-12-p1-pilot-integration.md

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 + <img src=/snap/{uuid}>
  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

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

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

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

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

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: 통과 + 커밋

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

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

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: 통과 + 커밋

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

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

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:

from edge.daemon import main

if __name__ == "__main__":
    main()

• Step 4: pyproject.toml에 스크립트 엔트리 추가

[project.scripts]
move-detect-edge = "edge.daemon:main"

• Step 5: 통과 + 커밋 — pytest tests/test_daemon.py -v → PASS; 전체 pytest -q

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 불필요)

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

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:

"""빌드머신 전용: 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

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 + <script src="/assets/detail.js">). detail.js parses the id from the URL, fetches GET /events/{id}, renders meta + each snapshot as <img src="/snap/{uuid}"> (createElement, XSS-safe).

• Step 1: 스모크 테스트 갱신 — cloud/tests/WiringSmokeTest.php의 paths 배열에 ['GET','/events/{id}/view'] 추가. cd cloud && ./vendor/bin/phpunit tests/WiringSmokeTest.php — (index 라우트 추가 전이라 실제 앱엔 없지만 스모크는 자체 라우터에 등록하므로) 통과 유지 확인. (스모크는 라우터 매칭만 검증.)

• Step 2: event_detail.php (스텁 교체)

<?php /* rendered at GET /events/{id}/view; data loaded by detail.js */ ?>
<h1>사고 상세</h1>
<div id="detail-meta" role="status">불러오는 중…</div>
<div id="detail-snapshots"></div>
<p><a href="/">← 목록</a></p>
<script src="/assets/detail.js" defer></script>

• Step 3: detail.js — cloud/public/assets/detail.js

(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 라우트 — 뷰 라우트 섹션에 추가:

$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.

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 편집 — 기존 구조에 아래를 통합(구현자는 실제 함수명에 맞춤):

// --- 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만 변경).

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

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.

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→COCO-17 Keypoint(x,y,conf) 일관.

---

Execution Handoff

Plan complete. 저장: docs/superpowers/plans/2026-07-12-p1-pilot-integration.md.

실행: Subagent-Driven (추천) 또는 Inline. 어느 방식?