e1f9f91110
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ne4rU4FToTLHcBUMUvUnT1
772 lines
34 KiB
Markdown
772 lines
34 KiB
Markdown
# 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`
|
|
|
|
```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 + `<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
|
|
<?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`
|
|
|
|
```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. 어느 방식?
|