From a4f2ead2112c1511669791dd9281fe5ae7ea47a6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=9C=A0=EC=B0=BD=EC=9A=B1?= Date: Sat, 11 Jul 2026 20:08:01 +0900 Subject: [PATCH] docs: workplace safety AI CCTV design spec v2.0 + P1-A edge core plan MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 설계 스펙 v2.0 (다관점 리뷰 반영: 기절경로 분리, 시점강건 낙상, 인증/RBAC, 평가셋, fps 3티어, 컴플라이언스, 증거 무결성) - P1-A 엣지 감지 코어 구현계획 (16 태스크 TDD) Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Ne4rU4FToTLHcBUMUvUnT1 --- .gitignore | 27 + .../2026-07-11-p1a-edge-detection-core.md | 2211 +++++++++++++++++ ...6-07-11-workplace-safety-ai-cctv-design.md | 456 ++++ 3 files changed, 2694 insertions(+) create mode 100644 .gitignore create mode 100644 docs/superpowers/plans/2026-07-11-p1a-edge-detection-core.md create mode 100644 docs/superpowers/specs/2026-07-11-workplace-safety-ai-cctv-design.md diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..6d52361 --- /dev/null +++ b/.gitignore @@ -0,0 +1,27 @@ +# Python +__pycache__/ +*.py[cod] +*.egg-info/ +.venv/ +venv/ +.pytest_cache/ +.mypy_cache/ +build/ +dist/ + +# Edge runtime artifacts +*.db +*.hef +snapshots/ +queue.db + +# Secrets / local config +.env +*.local + +# OS +.DS_Store +Thumbs.db + +# Tool caches +.impeccable/ diff --git a/docs/superpowers/plans/2026-07-11-p1a-edge-detection-core.md b/docs/superpowers/plans/2026-07-11-p1a-edge-detection-core.md new file mode 100644 index 0000000..fe56e40 --- /dev/null +++ b/docs/superpowers/plans/2026-07-11-p1a-edge-detection-core.md @@ -0,0 +1,2211 @@ +# P1-A 엣지 감지 코어 (Edge Detection Core) 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:** 작업장 안전 AI CCTV의 "감지 브레인"을 순수 Python으로 구현 — 자세추정 결과에서 넘어짐·자세무관 부동·유령사고(트랙소실)·위험구역 침입을 판정하고, 스냅샷과 함께 Cafe24로 POST(오프라인 큐 포함). Hailo 추론·카메라는 어댑터 인터페이스 뒤에 두어 하드웨어 없이 전 로직을 TDD로 완결한다. + +**Architecture:** 프레임 → PosePort(어댑터) → 스무딩 → TrackerPort → EventEngine(FallDetector·ImmobilityWatchdog·GhostIncidentDetector·ZoneIntrusionDetector) → Snapshotter(+FaceBlur) → PostClient/OfflineQueue. 모든 판정 로직은 주입된 타임스탬프(`now: float`)로 동작(결정론적 테스트). 임계값은 Config(env)에서 per-camera/zone. + +**Tech Stack:** Python 3.11, pytest, numpy, opencv-python(cv2), stdlib(sqlite3·hmac·hashlib·json·collections·dataclasses). POST는 requests. **런타임 ultralytics(AGPL) 금지.** 점-다각형·One-Euro·IoU 추적은 경량 자체 구현(어댑터로 ByteTrack 교체 가능). + +## Global Constraints + +- Python **3.11** 고정. 타입힌트 필수. +- **폐쇄망(air-gap):** 런타임 인터넷은 Cafe24 POST뿐. 의존성은 벤더링/핀 고정(오프라인 설치). `.hef`·휠 사전 번들. +- **런타임 `ultralytics` 금지** (AGPL). pose는 Hailo HEF 어댑터로만. +- **불변 지향:** 데이터 모델은 `@dataclass(frozen=True)`. 새 객체 생성, 제자리 변경 지양(가변 상태는 detector 내부 dict에 격리). +- **타임스탬프 주입:** 판정 로직 내부에서 `time.*` 호출 금지. `now: float`(초, monotonic) 인자로 받는다. (테스트 결정론) +- **픽셀 임계는 bbox 높이로 정규화.** 원시 픽셀 임계 금지. +- **단일 공유 pose 모델** 가정: 어댑터는 프레임당 `list[Person]`(박스+관절17) 반환. +- **스냅샷만**: 영상/클립 없음. 이벤트당 JPEG 1~3장. +- **POST 서명:** `HMAC-SHA256(nonce + timestamp + body, per_device_secret)` + nonce/timestamp(replay 방지). +- **임계값 하드코딩 금지:** Config(env)에서 로드, per-(device,camera,zone) 오버라이드. +- 파일은 작고 단일책임(200~400줄). 함수 <50줄. 커밋 자주. +- COCO-17 관절 인덱스 표준. 이미지 좌표계 y는 아래로 증가. + +--- + +## File Structure + +``` +edge/ + __init__.py + models.py # Keypoint, BBox, Person, COCO 인덱스 상수 + geometry.py # com, torso_angle_deg, head_below_hips, visible_count, core_visible, normalize, height_of + smoothing.py # OneEuroFilter, KeypointSmoother + ports.py # PosePort, SourcePort (Protocol) — 하드웨어 경계 + tracking.py # TrackFrame, Track, TrackerPort, IouTracker(경량) + events.py # EventType, Severity, Event, severity_for + detectors/ + __init__.py + fall.py # FallDetector (STANDING→FALLING→FALLEN→NO_MOTION, OR로직, baseline freeze) + immobility.py # ImmobilityWatchdog (자세무관 부동) + ghost.py # GhostIncidentDetector (트랙소실 공간사고) + zone.py # point_in_polygon, ZoneIntrusionDetector (발끝/디바운스) + engine.py # EventEngine (detector 오케스트레이션, dedup, cooldown) + snapshot.py # RingBuffer, FaceBlurrer, Snapshotter + alert.py # PostClient (HMAC+nonce) + queue.py # OfflineQueue (sqlite) + config.py # Config, load_config (env) + app.py # 메인 루프 배선 (source→pose→track→engine→snapshot→alert/queue) +tests/ + test_geometry.py test_smoothing.py test_tracking.py + test_fall.py test_immobility.py test_ghost.py test_zone.py + test_events.py test_engine.py test_snapshot.py + test_alert.py test_queue.py test_config.py test_app.py + hardware/ # 실장비 전용(여기서 미실행) +pyproject.toml +``` + +--- + +## Task 1: 프로젝트 스캐폴드 + 데이터 모델 + +**Files:** +- Create: `pyproject.toml`, `edge/__init__.py`, `edge/models.py`, `tests/test_models.py` + +**Interfaces:** +- Consumes: (없음) +- Produces: `Keypoint(x,y,conf)`, `BBox(x1,y1,x2,y2)` (props: `width,height,aspect,cx,cy`), `Person(bbox, keypoints: list[Keypoint])`; 상수 `NOSE=0, L_SHOULDER=5, R_SHOULDER=6, L_HIP=11, R_HIP=12, L_ANKLE=15, R_ANKLE=16, CORE_JOINTS=(5,6,11,12)`. + +- [ ] **Step 1: pyproject.toml 작성** + +```toml +[project] +name = "move-detect-edge" +version = "0.1.0" +requires-python = ">=3.11" +dependencies = ["numpy", "opencv-python", "requests"] + +[project.optional-dependencies] +dev = ["pytest"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-q" +``` + +- [ ] **Step 2: 실패 테스트 작성** — `tests/test_models.py` + +```python +from edge.models import Keypoint, BBox, Person, L_SHOULDER, R_HIP, CORE_JOINTS + +def test_bbox_derived_properties(): + b = BBox(10, 20, 30, 80) + assert b.width == 20 + assert b.height == 60 + assert b.aspect == 20 / 60 + assert b.cx == 20 and b.cy == 50 + +def test_bbox_zero_height_aspect_is_zero(): + assert BBox(0, 0, 10, 0).aspect == 0.0 + +def test_person_holds_17_keypoints(): + kps = [Keypoint(0, 0, 0.9) for _ in range(17)] + p = Person(BBox(0, 0, 1, 1), kps) + assert len(p.keypoints) == 17 + assert p.keypoints[L_SHOULDER].conf == 0.9 + +def test_coco_index_constants(): + assert CORE_JOINTS == (5, 6, 11, 12) + assert R_HIP == 12 +``` + +- [ ] **Step 3: 실패 확인** + +Run: `pytest tests/test_models.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'edge.models'` + +- [ ] **Step 4: 구현** — `edge/__init__.py`(빈 파일), `edge/models.py` + +```python +from dataclasses import dataclass + +# COCO-17 keypoint indices +NOSE = 0 +L_EYE, R_EYE, L_EAR, R_EAR = 1, 2, 3, 4 +L_SHOULDER, R_SHOULDER = 5, 6 +L_ELBOW, R_ELBOW, L_WRIST, R_WRIST = 7, 8, 9, 10 +L_HIP, R_HIP = 11, 12 +L_KNEE, R_KNEE, L_ANKLE, R_ANKLE = 13, 14, 15, 16 +CORE_JOINTS = (L_SHOULDER, R_SHOULDER, L_HIP, R_HIP) + + +@dataclass(frozen=True) +class Keypoint: + x: float + y: float + conf: float + + +@dataclass(frozen=True) +class BBox: + x1: float + y1: float + x2: float + y2: float + + @property + def width(self) -> float: + return self.x2 - self.x1 + + @property + def height(self) -> float: + return self.y2 - self.y1 + + @property + def aspect(self) -> float: + return self.width / self.height if self.height else 0.0 + + @property + def cx(self) -> float: + return (self.x1 + self.x2) / 2 + + @property + def cy(self) -> float: + return (self.y1 + self.y2) / 2 + + +@dataclass(frozen=True) +class Person: + bbox: BBox + keypoints: list # list[Keypoint], length 17 +``` + +- [ ] **Step 5: 통과 확인 + 커밋** + +Run: `pytest tests/test_models.py -v` → Expected: PASS + +```bash +git add pyproject.toml edge/__init__.py edge/models.py tests/test_models.py +git commit -m "feat(edge): data models (Keypoint, BBox, Person) + COCO indices" +``` + +--- + +## Task 2: 기하 유틸 (geometry) + +**Files:** +- Create: `edge/geometry.py`, `tests/test_geometry.py` + +**Interfaces:** +- Consumes: `Keypoint, BBox, CORE_JOINTS, NOSE, L_SHOULDER, R_SHOULDER, L_HIP, R_HIP` (Task 1) +- Produces: + - `com(keypoints) -> tuple[float, float]` — 어깨2+엉덩이2 평균 + - `torso_angle_deg(keypoints) -> float` — 몸통벡터의 수직 대비 각(0=수직/서있음, 90=수평/누움) + - `head_below_hips(keypoints) -> bool` + - `visible_count(keypoints, thr=0.2) -> int` + - `core_visible(keypoints, thr=0.5) -> bool` + - `height_of(bbox) -> float` + - `normalize(value, bbox_height) -> float` + +- [ ] **Step 1: 실패 테스트** — `tests/test_geometry.py` + +```python +import math +from edge.models import Keypoint, BBox +from edge.geometry import ( + com, torso_angle_deg, head_below_hips, visible_count, + core_visible, normalize, +) + +def _kps(overrides): + kps = [Keypoint(0, 0, 0.0) for _ in range(17)] + for i, (x, y, c) in overrides.items(): + kps[i] = Keypoint(x, y, c) + return kps + +def test_com_is_mean_of_shoulders_and_hips(): + kps = _kps({5: (0, 0, 1), 6: (10, 0, 1), 11: (0, 10, 1), 12: (10, 10, 1)}) + assert com(kps) == (5.0, 5.0) + +def test_torso_angle_upright_near_zero(): + # shoulders above hips (smaller y), vertical torso + kps = _kps({5: (0, 0, 1), 6: (10, 0, 1), 11: (0, 100, 1), 12: (10, 100, 1)}) + assert torso_angle_deg(kps) < 10 + +def test_torso_angle_lying_near_ninety(): + # shoulders and hips at same height, horizontal torso + kps = _kps({5: (0, 50, 1), 6: (0, 55, 1), 11: (100, 50, 1), 12: (100, 55, 1)}) + assert torso_angle_deg(kps) > 80 + +def test_head_below_hips_true_when_nose_lower_on_screen(): + kps = _kps({0: (5, 120, 1), 11: (0, 100, 1), 12: (10, 100, 1)}) + assert head_below_hips(kps) is True + +def test_visible_count_and_core_visible(): + kps = _kps({5: (0, 0, 0.9), 6: (1, 0, 0.9), 11: (0, 1, 0.9), 12: (1, 1, 0.9)}) + assert visible_count(kps, thr=0.2) == 4 + assert core_visible(kps, thr=0.5) is True + +def test_normalize_divides_by_bbox_height(): + assert normalize(30, 60) == 0.5 + assert normalize(30, 0) == 0.0 +``` + +- [ ] **Step 2: 실패 확인** + +Run: `pytest tests/test_geometry.py -v` → Expected: FAIL — `ModuleNotFoundError: edge.geometry` + +- [ ] **Step 3: 구현** — `edge/geometry.py` + +```python +import math + +from edge.models import ( + Keypoint, BBox, NOSE, L_SHOULDER, R_SHOULDER, L_HIP, R_HIP, CORE_JOINTS, +) + + +def _mean(points: list[tuple[float, float]]) -> tuple[float, float]: + n = len(points) + return (sum(p[0] for p in points) / n, sum(p[1] for p in points) / n) + + +def com(keypoints: list[Keypoint]) -> tuple[float, float]: + pts = [(keypoints[i].x, keypoints[i].y) for i in CORE_JOINTS] + return _mean(pts) + + +def _shoulder_center(kps: list[Keypoint]) -> tuple[float, float]: + return _mean([(kps[L_SHOULDER].x, kps[L_SHOULDER].y), + (kps[R_SHOULDER].x, kps[R_SHOULDER].y)]) + + +def _hip_center(kps: list[Keypoint]) -> tuple[float, float]: + return _mean([(kps[L_HIP].x, kps[L_HIP].y), + (kps[R_HIP].x, kps[R_HIP].y)]) + + +def torso_angle_deg(keypoints: list[Keypoint]) -> float: + """몸통(어깨중심→엉덩이중심) 벡터가 수직축과 이루는 각. 0=수직, 90=수평.""" + sx, sy = _shoulder_center(keypoints) + hx, hy = _hip_center(keypoints) + vx, vy = hx - sx, hy - sy + # 수직축(0,1) 대비 편차. 위/아래 무관하게 |성분| 사용. + return math.degrees(math.atan2(abs(vx), abs(vy))) if (vx or vy) else 0.0 + + +def head_below_hips(keypoints: list[Keypoint]) -> bool: + _, hy = _hip_center(keypoints) + return keypoints[NOSE].y > hy + + +def visible_count(keypoints: list[Keypoint], thr: float = 0.2) -> int: + return sum(1 for k in keypoints if k.conf >= thr) + + +def core_visible(keypoints: list[Keypoint], thr: float = 0.5) -> bool: + return all(keypoints[i].conf >= thr for i in CORE_JOINTS) + + +def height_of(bbox: BBox) -> float: + return bbox.height + + +def normalize(value: float, bbox_height: float) -> float: + return value / bbox_height if bbox_height else 0.0 +``` + +- [ ] **Step 4: 통과 확인** + +Run: `pytest tests/test_geometry.py -v` → Expected: PASS + +- [ ] **Step 5: 커밋** + +```bash +git add edge/geometry.py tests/test_geometry.py +git commit -m "feat(edge): pose geometry utils (torso angle, com, normalization)" +``` + +--- + +## Task 3: 키포인트 스무딩 (One-Euro) + +**Files:** +- Create: `edge/smoothing.py`, `tests/test_smoothing.py` + +**Interfaces:** +- Consumes: `Keypoint` (Task 1) +- Produces: + - `OneEuroFilter(min_cutoff=1.0, beta=0.0)` with `filter(value: float, t: float) -> float` + - `KeypointSmoother(min_cutoff=1.0, beta=0.007)` with `smooth(keypoints: list[Keypoint], t: float) -> list[Keypoint]` (관절별 x,y 스무딩, conf 통과) + +- [ ] **Step 1: 실패 테스트** — `tests/test_smoothing.py` + +```python +from edge.models import Keypoint +from edge.smoothing import OneEuroFilter, KeypointSmoother + +def test_one_euro_first_sample_returns_input(): + f = OneEuroFilter(min_cutoff=1.0, beta=0.0) + assert f.filter(5.0, t=0.0) == 5.0 + +def test_one_euro_attenuates_jitter(): + f = OneEuroFilter(min_cutoff=0.5, beta=0.0) + f.filter(0.0, t=0.0) + out = f.filter(10.0, t=0.1) # big jump at low cutoff -> damped below input + assert 0.0 < out < 10.0 + +def test_keypoint_smoother_preserves_conf_and_length(): + s = KeypointSmoother() + kps = [Keypoint(float(i), float(i), 0.8) for i in range(17)] + out = s.smooth(kps, t=0.0) + assert len(out) == 17 + assert out[3].conf == 0.8 +``` + +- [ ] **Step 2: 실패 확인** + +Run: `pytest tests/test_smoothing.py -v` → Expected: FAIL — `ModuleNotFoundError: edge.smoothing` + +- [ ] **Step 3: 구현** — `edge/smoothing.py` + +```python +import math + +from edge.models import Keypoint + + +class OneEuroFilter: + """1€ filter (Casiez et al.). 저지연 지터 감쇠.""" + + def __init__(self, min_cutoff: float = 1.0, beta: float = 0.0, d_cutoff: float = 1.0): + self.min_cutoff = min_cutoff + self.beta = beta + self.d_cutoff = d_cutoff + self._x_prev: float | None = None + self._dx_prev: float = 0.0 + self._t_prev: float | None = None + + @staticmethod + def _alpha(cutoff: float, dt: float) -> float: + tau = 1.0 / (2 * math.pi * cutoff) + return 1.0 / (1.0 + tau / dt) + + def filter(self, value: float, t: float) -> float: + if self._x_prev is None or self._t_prev is None or t <= self._t_prev: + self._x_prev, self._t_prev, self._dx_prev = value, t, 0.0 + return value + dt = t - self._t_prev + dx = (value - self._x_prev) / dt + a_d = self._alpha(self.d_cutoff, dt) + dx_hat = a_d * dx + (1 - a_d) * self._dx_prev + cutoff = self.min_cutoff + self.beta * abs(dx_hat) + a = self._alpha(cutoff, dt) + x_hat = a * value + (1 - a) * self._x_prev + self._x_prev, self._dx_prev, self._t_prev = x_hat, dx_hat, t + return x_hat + + +class KeypointSmoother: + """관절 17개 각각 x,y에 독립 1€ 필터.""" + + def __init__(self, min_cutoff: float = 1.0, beta: float = 0.007): + self._fx = [OneEuroFilter(min_cutoff, beta) for _ in range(17)] + self._fy = [OneEuroFilter(min_cutoff, beta) for _ in range(17)] + + def smooth(self, keypoints: list[Keypoint], t: float) -> list[Keypoint]: + out = [] + for i, k in enumerate(keypoints): + out.append(Keypoint(self._fx[i].filter(k.x, t), + self._fy[i].filter(k.y, t), k.conf)) + return out +``` + +- [ ] **Step 4: 통과 확인** + +Run: `pytest tests/test_smoothing.py -v` → Expected: PASS + +- [ ] **Step 5: 커밋** + +```bash +git add edge/smoothing.py tests/test_smoothing.py +git commit -m "feat(edge): One-Euro keypoint smoothing (INT8 jitter)" +``` + +--- + +## Task 4: 추적 (Track 모델 + 경량 IoU 추적기 + 공간 바인딩) + +**Files:** +- Create: `edge/tracking.py`, `edge/ports.py`, `tests/test_tracking.py` + +**Interfaces:** +- Consumes: `Person, BBox, Keypoint` (Task 1) +- Produces: + - `ports.PosePort` (Protocol): `infer(frame) -> list[Person]` + - `ports.SourcePort` (Protocol): `read() -> tuple[bool, object]` (ok, frame) + - `TrackFrame(t: float, bbox: BBox, keypoints: list[Keypoint])` + - `Track(track_id: int, frames: deque[TrackFrame])` — `latest` 프로퍼티 + - `TrackerPort` (Protocol): `update(persons: list[Person], t: float) -> list[Track]` + - `IouTracker(iou_thr=0.3, max_age=1.0, history=90)` implements TrackerPort; **소실 트랙은 `lost()` 로 노출**(공간 바인딩·유령사고용): `lost() -> list[Track]` + +- [ ] **Step 1: 실패 테스트** — `tests/test_tracking.py` + +```python +from edge.models import Person, BBox, Keypoint +from edge.tracking import IouTracker + +def _person(x1, y1, x2, y2): + kps = [Keypoint((x1 + x2) / 2, (y1 + y2) / 2, 0.9) for _ in range(17)] + return Person(BBox(x1, y1, x2, y2), kps) + +def test_new_person_gets_track_id(): + tr = IouTracker() + tracks = tr.update([_person(0, 0, 10, 20)], t=0.0) + assert len(tracks) == 1 + assert tracks[0].track_id == 1 + +def test_overlapping_person_keeps_same_id(): + tr = IouTracker(iou_thr=0.3) + tr.update([_person(0, 0, 10, 20)], t=0.0) + tracks = tr.update([_person(1, 1, 11, 21)], t=0.1) # high overlap + assert tracks[0].track_id == 1 + +def test_disappeared_track_is_reported_lost(): + tr = IouTracker(max_age=0.2) + tr.update([_person(0, 0, 10, 20)], t=0.0) + tr.update([], t=0.5) # nothing detected, past max_age + lost = tr.lost() + assert len(lost) == 1 + assert lost[0].track_id == 1 + +def test_history_accumulates_frames(): + tr = IouTracker() + tr.update([_person(0, 0, 10, 20)], t=0.0) + tracks = tr.update([_person(0, 0, 10, 20)], t=0.1) + assert len(tracks[0].frames) == 2 + assert tracks[0].latest.t == 0.1 +``` + +- [ ] **Step 2: 실패 확인** + +Run: `pytest tests/test_tracking.py -v` → Expected: FAIL — `ModuleNotFoundError: edge.tracking` + +- [ ] **Step 3: 구현** — `edge/ports.py` + +```python +from typing import Protocol + + +class SourcePort(Protocol): + def read(self) -> tuple[bool, object]: + """(ok, frame). ok=False면 프레임 없음.""" + ... + + +class PosePort(Protocol): + def infer(self, frame) -> list: # list[Person] + ... +``` + +그리고 `edge/tracking.py`: + +```python +from collections import deque +from dataclasses import dataclass + +from edge.models import Person, BBox, Keypoint + + +@dataclass +class TrackFrame: + t: float + bbox: BBox + keypoints: list # list[Keypoint] + + +@dataclass +class Track: + track_id: int + frames: deque + + @property + def latest(self) -> TrackFrame: + return self.frames[-1] + + +def _iou(a: BBox, b: BBox) -> float: + ix1, iy1 = max(a.x1, b.x1), max(a.y1, b.y1) + ix2, iy2 = min(a.x2, b.x2), min(a.y2, b.y2) + iw, ih = max(0.0, ix2 - ix1), max(0.0, iy2 - iy1) + inter = iw * ih + union = a.width * a.height + b.width * b.height - inter + return inter / union if union else 0.0 + + +class IouTracker: + """경량 IoU 그리디 매칭 추적기. 실장비는 ByteTrack 어댑터로 교체 가능(동일 TrackerPort).""" + + def __init__(self, iou_thr: float = 0.3, max_age: float = 1.0, history: int = 90): + self.iou_thr = iou_thr + self.max_age = max_age + self.history = history + self._tracks: dict[int, Track] = {} + self._last_seen: dict[int, float] = {} + self._lost: list[Track] = [] + self._next_id = 1 + + def update(self, persons: list[Person], t: float) -> list[Track]: + self._lost = [] + unmatched = list(persons) + # 기존 트랙과 그리디 IoU 매칭 + for tid, track in list(self._tracks.items()): + best, best_iou = None, self.iou_thr + for p in unmatched: + iou = _iou(track.latest.bbox, p.bbox) + if iou >= best_iou: + best, best_iou = p, iou + if best is not None: + unmatched.remove(best) + track.frames.append(TrackFrame(t, best.bbox, best.keypoints)) + self._last_seen[tid] = t + # 신규 트랙 + for p in unmatched: + tid = self._next_id + self._next_id += 1 + self._tracks[tid] = Track(tid, deque([TrackFrame(t, p.bbox, p.keypoints)], + maxlen=self.history)) + self._last_seen[tid] = t + # 소실 판정 + for tid in list(self._tracks.keys()): + if t - self._last_seen[tid] > self.max_age: + self._lost.append(self._tracks.pop(tid)) + self._last_seen.pop(tid, None) + return list(self._tracks.values()) + + def lost(self) -> list[Track]: + return list(self._lost) +``` + +- [ ] **Step 4: 통과 확인** + +Run: `pytest tests/test_tracking.py -v` → Expected: PASS + +- [ ] **Step 5: 커밋** + +```bash +git add edge/ports.py edge/tracking.py tests/test_tracking.py +git commit -m "feat(edge): Track model + IoU tracker with lost-track reporting" +``` + +--- + +## Task 5: 이벤트 모델 (Event, 심각도) + +**Files:** +- Create: `edge/events.py`, `tests/test_events.py` + +**Interfaces:** +- Consumes: (없음) +- Produces: + - `EventType` (str enum): `FALL, NO_MOTION, IMMOBILITY, GHOST, ZONE_WARNING, ZONE_RESTRICTED, PINCH, MONITORING_DEGRADED, DEVICE_FAULT` + - `Severity` (str enum): `CRITICAL, HIGH, WARNING, INFO` + - `Event(event_type, severity, camera_id, track_id, zone_id, occurred_at, duration_s, metrics: dict, spatial: tuple|None)` — `@dataclass(frozen=True)`, `to_payload(device_id) -> dict` (§7.3 계약 형태, snapshots 제외) + - `severity_for(event_type: EventType) -> Severity` + +- [ ] **Step 1: 실패 테스트** — `tests/test_events.py` + +```python +from edge.events import EventType, Severity, Event, severity_for + +def test_severity_mapping(): + assert severity_for(EventType.NO_MOTION) == Severity.CRITICAL + assert severity_for(EventType.ZONE_RESTRICTED) == Severity.HIGH + assert severity_for(EventType.FALL) == Severity.HIGH + assert severity_for(EventType.IMMOBILITY) == Severity.WARNING + assert severity_for(EventType.ZONE_WARNING) == Severity.WARNING + +def test_event_to_payload_shape(): + e = Event(EventType.FALL, Severity.HIGH, camera_id="cam-01", track_id=7, + zone_id=None, occurred_at=1000.0, duration_s=3.2, + metrics={"torso_angle": 78.0}, spatial=None) + p = e.to_payload(device_id="pi-01") + assert p["device_id"] == "pi-01" + assert p["event_type"] == "fall" + assert p["severity"] == "high" + assert p["track_id"] == 7 + assert p["metrics"]["torso_angle"] == 78.0 + assert "snapshots" not in p # snapshots는 Alerter가 붙임 +``` + +- [ ] **Step 2: 실패 확인** + +Run: `pytest tests/test_events.py -v` → Expected: FAIL — `ModuleNotFoundError: edge.events` + +- [ ] **Step 3: 구현** — `edge/events.py` + +```python +from dataclasses import dataclass +from enum import Enum + + +class EventType(str, Enum): + FALL = "fall" + NO_MOTION = "no_motion" + IMMOBILITY = "immobility" + GHOST = "ghost" + ZONE_WARNING = "zone_warning" + ZONE_RESTRICTED = "zone_restricted" + PINCH = "pinch" + MONITORING_DEGRADED = "monitoring_degraded" + DEVICE_FAULT = "device_fault" + + +class Severity(str, Enum): + CRITICAL = "critical" + HIGH = "high" + WARNING = "warning" + INFO = "info" + + +_SEVERITY = { + EventType.NO_MOTION: Severity.CRITICAL, + EventType.GHOST: Severity.CRITICAL, + EventType.FALL: Severity.HIGH, + EventType.ZONE_RESTRICTED: Severity.HIGH, + EventType.PINCH: Severity.HIGH, + EventType.IMMOBILITY: Severity.WARNING, + EventType.ZONE_WARNING: Severity.WARNING, + EventType.MONITORING_DEGRADED: Severity.WARNING, + EventType.DEVICE_FAULT: Severity.WARNING, +} + + +def severity_for(event_type: EventType) -> Severity: + return _SEVERITY.get(event_type, Severity.INFO) + + +@dataclass(frozen=True) +class Event: + event_type: EventType + severity: Severity + camera_id: str + track_id: int | None + zone_id: str | None + occurred_at: float + duration_s: float + metrics: dict + spatial: tuple | None # (x, y) 유령사고 등 공간 기준 + + def to_payload(self, device_id: str) -> dict: + return { + "device_id": device_id, + "camera_id": self.camera_id, + "event_type": self.event_type.value, + "severity": self.severity.value, + "track_id": self.track_id, + "zone_id": self.zone_id, + "occurred_at": self.occurred_at, + "duration_s": round(self.duration_s, 2), + "metrics": self.metrics, + } +``` + +- [ ] **Step 4: 통과 확인** + +Run: `pytest tests/test_events.py -v` → Expected: PASS + +- [ ] **Step 5: 커밋** + +```bash +git add edge/events.py tests/test_events.py +git commit -m "feat(edge): Event model + severity mapping (§7.3 payload)" +``` + +--- + +## Task 6: FallDetector (상태머신 + OR로직 + baseline freeze + 낙상후 미동) + +**Files:** +- Create: `edge/detectors/__init__.py`(빈 파일), `edge/detectors/fall.py`, `tests/test_fall.py` + +**Interfaces:** +- Consumes: `Track, TrackFrame` (Task 4); `torso_angle_deg, head_below_hips, core_visible, visible_count, normalize, com` (Task 2) +- Produces: + - `FallConfig(falling_angle=45, fallen_angle=70, aspect_ratio=1.25, height_collapse=0.35, prone_confirm_s=3.0, no_motion_s=10.0, motion_thr=0.02, min_conf=0.5, min_visible=10)` + - `FallDetector(cfg: FallConfig)` with `update(track: Track, now: float) -> tuple[str|None, dict]` + - 반환: `("fall", metrics)` (FALLEN 확정), `("no_motion", metrics)` (낙상후 미동 ≥ no_motion_s), `("recover", {})` (자세 복원), 아니면 `(None, {})` + - 내부 상태: `STANDING/FALLING/FALLEN/NO_MOTION` (track_id별 `_State` dict, `_State.name` 노출), incident 활성 중 baseline(정상 키높이) 동결 + +- [ ] **Step 1: 실패 테스트** — `tests/test_fall.py` + +```python +from collections import deque +from edge.models import Keypoint, BBox +from edge.tracking import Track, TrackFrame +from edge.detectors.fall import FallDetector, FallConfig + +def _standing_kps(): + kps = [Keypoint(50, 150, 0.9) for _ in range(17)] + kps[5] = Keypoint(40, 100, 0.9); kps[6] = Keypoint(60, 100, 0.9) + kps[11] = Keypoint(40, 200, 0.9); kps[12] = Keypoint(60, 200, 0.9) + kps[0] = Keypoint(50, 60, 0.9) # nose above hips + return kps + +def _lying_kps(): + 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) # nose below hips + return kps + +def _track(): + return Track(1, deque(maxlen=90)) + +def _feed(det, track, kps, bbox, t): + track.frames.append(TrackFrame(t, bbox, kps)) + return det.update(track, now=t) + +def test_standing_produces_no_event(): + det = FallDetector(FallConfig()) + tr = _track() + sig, _ = _feed(det, tr, _standing_kps(), BBox(40, 60, 60, 210), t=0.0) + assert sig is None + +def test_lying_confirmed_after_prone_seconds_emits_fall(): + det = FallDetector(FallConfig(prone_confirm_s=1.0)) + tr = _track() + lying_box = BBox(90, 290, 210, 320) # 넓고 낮음 (aspect>1.25) + _feed(det, tr, _lying_kps(), lying_box, t=0.0) + sig, meta = _feed(det, tr, _lying_kps(), lying_box, t=1.5) # prone 1.5s > 1.0 + assert sig == "fall" + assert meta["torso_angle"] > 80 + +def test_front_fall_height_collapse_triggers_without_angle(): + # 정면 낙상: 각도/종횡비 안 변해도 bbox 높이 급붕괴 + det = FallDetector(FallConfig(prone_confirm_s=0.5)) + tr = _track() + tall = BBox(40, 60, 60, 210) # height 150 + _feed(det, tr, _standing_kps(), tall, t=0.0) + short = BBox(40, 180, 60, 210) # height 30 (붕괴 0.8) + _feed(det, tr, _standing_kps(), short, t=0.2) + sig, _ = _feed(det, tr, _standing_kps(), short, t=0.9) + assert sig == "fall" + +def test_no_motion_after_fall_escalates(): + det = FallDetector(FallConfig(prone_confirm_s=0.5, no_motion_s=1.0, motion_thr=0.05)) + tr = _track() + box = BBox(90, 290, 210, 320) + _feed(det, tr, _lying_kps(), box, t=0.0) + _feed(det, tr, _lying_kps(), box, t=0.6) # -> fall + _feed(det, tr, _lying_kps(), box, t=1.0) # 미동 지속 + sig, _ = _feed(det, tr, _lying_kps(), box, t=2.0) # 부동 >1.0s + assert sig == "no_motion" + +def test_recovery_clears_state(): + det = FallDetector(FallConfig(prone_confirm_s=0.5)) + tr = _track() + box = BBox(90, 290, 210, 320) + _feed(det, tr, _lying_kps(), box, t=0.0) + _feed(det, tr, _lying_kps(), box, t=0.6) # fall + sig, _ = _feed(det, tr, _standing_kps(), BBox(40, 60, 60, 210), t=1.0) + assert sig == "recover" + +def test_low_confidence_is_ignored(): + det = FallDetector(FallConfig()) + tr = _track() + kps = [Keypoint(0, 0, 0.1) for _ in range(17)] # 전부 저신뢰 + sig, _ = _feed(det, tr, kps, BBox(0, 0, 100, 30), t=0.0) + assert sig is None +``` + +- [ ] **Step 2: 실패 확인** + +Run: `pytest tests/test_fall.py -v` → Expected: FAIL — `ModuleNotFoundError: edge.detectors.fall` + +- [ ] **Step 3: 구현** — `edge/detectors/fall.py` + +```python +from dataclasses import dataclass + +from edge.tracking import Track +from edge.geometry import ( + torso_angle_deg, head_below_hips, core_visible, visible_count, normalize, +) + + +@dataclass +class FallConfig: + falling_angle: float = 45.0 + fallen_angle: float = 70.0 + aspect_ratio: float = 1.25 + height_collapse: float = 0.35 # baseline 대비 남은 높이 비율(이 값 미만이면 급붕괴) + prone_confirm_s: float = 3.0 + no_motion_s: float = 10.0 + motion_thr: float = 0.02 # 정규화 관절변위 임계 + min_conf: float = 0.5 + min_visible: int = 10 + + +@dataclass +class _State: + name: str = "STANDING" + baseline_height: float | None = None + prone_since: float | None = None + still_since: float | None = None + fall_emitted: bool = False + no_motion_emitted: bool = False + + +class FallDetector: + def __init__(self, cfg: FallConfig): + self.cfg = cfg + self._states: dict[int, _State] = {} + + def _prone(self, kps, bbox, st: _State) -> tuple[bool, dict]: + angle = torso_angle_deg(kps) + ar = bbox.aspect + remaining = 1.0 + if st.baseline_height: + remaining = bbox.height / st.baseline_height + prone = ( + angle > self.cfg.fallen_angle + or ar > self.cfg.aspect_ratio + or remaining < self.cfg.height_collapse # 높이 급붕괴(정면/후면 낙상) + or head_below_hips(kps) + ) + return prone, {"torso_angle": round(angle, 1), + "aspect_ratio": round(ar, 2), + "height_remaining": round(remaining, 2)} + + def _displacement(self, track: Track) -> float: + if len(track.frames) < 2: + return 1.0 + a, b = track.frames[-2], track.frames[-1] + h = b.bbox.height or 1.0 + d = sum(((p.x - q.x) ** 2 + (p.y - q.y) ** 2) ** 0.5 + for p, q in zip(a.keypoints, b.keypoints)) / len(b.keypoints) + return normalize(d, h) + + def update(self, track: Track, now: float) -> tuple[str | None, dict]: + st = self._states.setdefault(track.track_id, _State()) + f = track.latest + kps, bbox = f.keypoints, f.bbox + + # 신뢰도 게이트 + if not core_visible(kps, self.cfg.min_conf) or visible_count(kps) < self.cfg.min_visible: + return None, {} + + prone, meta = self._prone(kps, bbox, st) + + # incident 비활성일 때만 baseline(정상 키높이) 갱신 = incident 중 freeze + if st.name == "STANDING" and not prone: + st.baseline_height = bbox.height + + if not prone: + if st.name in ("FALLING", "FALLEN", "NO_MOTION"): + self._states[track.track_id] = _State(baseline_height=bbox.height) + return "recover", {} + st.prone_since = None + return None, {} + + # prone 상태 + if st.prone_since is None: + st.prone_since = now + st.name = "FALLING" + if now - st.prone_since >= self.cfg.prone_confirm_s and not st.fall_emitted: + st.name = "FALLEN" + st.fall_emitted = True + st.still_since = now + return "fall", meta + + # FALLEN 이후 미동 감시 + if st.fall_emitted: + if self._displacement(track) < self.cfg.motion_thr: + if st.still_since is None: + st.still_since = now + elif now - st.still_since >= self.cfg.no_motion_s and not st.no_motion_emitted: + st.name = "NO_MOTION" + st.no_motion_emitted = True + return "no_motion", {**meta, "immobility_s": round(now - st.still_since, 1)} + else: + st.still_since = None + return None, {} +``` + +- [ ] **Step 4: 통과 확인** + +Run: `pytest tests/test_fall.py -v` → Expected: PASS + +- [ ] **Step 5: 커밋** + +```bash +git add edge/detectors/__init__.py edge/detectors/fall.py tests/test_fall.py +git commit -m "feat(edge): FallDetector state machine (OR-logic, baseline freeze, post-fall no-motion)" +``` + +--- + +## Task 7: ImmobilityWatchdog (자세무관 부동) + +**Files:** +- Create: `edge/detectors/immobility.py`, `tests/test_immobility.py` + +**Interfaces:** +- Consumes: `Track` (Task 4); `normalize` (Task 2) +- Produces: + - `ImmobilityConfig(motion_thr=0.02, immobile_s=120.0, zone_immobile_s=30.0)` + - `ImmobilityWatchdog(cfg)` with `update(track: Track, now: float, in_danger_zone: bool = False) -> tuple[bool, dict]` — `(True, meta)` 최초 부동 임계 초과 시 1회, 이후 `(False, {})`; 움직이면 리셋 + +- [ ] **Step 1: 실패 테스트** — `tests/test_immobility.py` + +```python +from collections import deque +from edge.models import Keypoint, BBox +from edge.tracking import Track, TrackFrame +from edge.detectors.immobility import ImmobilityWatchdog, ImmobilityConfig + +def _kps(x, y): + return [Keypoint(x, y, 0.9) for _ in range(17)] + +def _push(track, x, y, t): + track.frames.append(TrackFrame(t, BBox(x, y, x + 20, y + 60), _kps(x, y))) + +def test_still_person_triggers_after_threshold(): + w = ImmobilityWatchdog(ImmobilityConfig(motion_thr=0.05, immobile_s=1.0)) + tr = Track(1, deque(maxlen=90)) + _push(tr, 100, 100, 0.0) + assert w.update(tr, now=0.0)[0] is False + _push(tr, 100, 100, 0.5) + assert w.update(tr, now=0.5)[0] is False + _push(tr, 100, 100, 1.5) # 1.5s 정지 + fired, meta = w.update(tr, now=1.5) + assert fired is True + assert meta["immobility_s"] >= 1.0 + +def test_fires_only_once_until_movement(): + w = ImmobilityWatchdog(ImmobilityConfig(motion_thr=0.05, immobile_s=1.0)) + tr = Track(1, deque(maxlen=90)) + _push(tr, 100, 100, 0.0) + _push(tr, 100, 100, 1.5); w.update(tr, now=1.5) + _push(tr, 100, 100, 2.0) + assert w.update(tr, now=2.0)[0] is False # 재발화 안 함 + +def test_movement_resets_timer(): + w = ImmobilityWatchdog(ImmobilityConfig(motion_thr=0.05, immobile_s=1.0)) + tr = Track(1, deque(maxlen=90)) + _push(tr, 100, 100, 0.0) + _push(tr, 100, 100, 0.9); w.update(tr, now=0.9) + _push(tr, 300, 300, 1.0) # 크게 이동 + fired, _ = w.update(tr, now=1.0) + assert fired is False + +def test_danger_zone_uses_shorter_window(): + w = ImmobilityWatchdog(ImmobilityConfig(motion_thr=0.05, immobile_s=100.0, zone_immobile_s=1.0)) + tr = Track(1, deque(maxlen=90)) + _push(tr, 100, 100, 0.0) + _push(tr, 100, 100, 1.5) + fired, _ = w.update(tr, now=1.5, in_danger_zone=True) + assert fired is True +``` + +- [ ] **Step 2: 실패 확인** + +Run: `pytest tests/test_immobility.py -v` → Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 구현** — `edge/detectors/immobility.py` + +```python +from dataclasses import dataclass + +from edge.tracking import Track +from edge.geometry import normalize + + +@dataclass +class ImmobilityConfig: + motion_thr: float = 0.02 + immobile_s: float = 120.0 + zone_immobile_s: float = 30.0 + + +class ImmobilityWatchdog: + def __init__(self, cfg: ImmobilityConfig): + self.cfg = cfg + self._still_since: dict[int, float] = {} + self._fired: set[int] = set() + + def _displacement(self, track: Track) -> float: + if len(track.frames) < 2: + return 1.0 + a, b = track.frames[-2], track.frames[-1] + h = b.bbox.height or 1.0 + d = sum(((p.x - q.x) ** 2 + (p.y - q.y) ** 2) ** 0.5 + for p, q in zip(a.keypoints, b.keypoints)) / len(b.keypoints) + return normalize(d, h) + + def update(self, track: Track, now: float, in_danger_zone: bool = False) -> tuple[bool, dict]: + tid = track.track_id + window = self.cfg.zone_immobile_s if in_danger_zone else self.cfg.immobile_s + if self._displacement(track) >= self.cfg.motion_thr: + self._still_since.pop(tid, None) + self._fired.discard(tid) + return False, {} + start = self._still_since.setdefault(tid, now) + if now - start >= window and tid not in self._fired: + self._fired.add(tid) + return True, {"immobility_s": round(now - start, 1)} + return False, {} +``` + +- [ ] **Step 4: 통과 확인** + +Run: `pytest tests/test_immobility.py -v` → Expected: PASS + +- [ ] **Step 5: 커밋** + +```bash +git add edge/detectors/immobility.py tests/test_immobility.py +git commit -m "feat(edge): ImmobilityWatchdog (posture-independent prolonged immobility)" +``` + +--- + +## Task 8: GhostIncidentDetector (트랙 소실 공간사고) + +**Files:** +- Create: `edge/detectors/ghost.py`, `tests/test_ghost.py` + +**Interfaces:** +- Consumes: `Track` (Task 4) +- Produces: + - `GhostConfig(reappear_s=10.0, reappear_dist=1.0)` (dist는 bbox 높이 배수) + - `GhostIncidentDetector(cfg)`: + - `note_state(track_id: int, state: str)` — FallDetector 상태를 공유(위험상태 트랙만 유령 후보) + - `on_lost(lost_tracks: list[Track], now: float)` — 위험상태(FALLING/FALLEN/NO_MOTION)로 사라진 트랙을 후보 등록 + - `update(active_tracks: list[Track], now: float) -> list[dict]` — 재출현 없이 reappear_s 경과한 후보를 유령사고로 방출(`{"spatial": (x,y), "last_state": ..., "waited_s": ...}`) + +- [ ] **Step 1: 실패 테스트** — `tests/test_ghost.py` + +```python +from collections import deque +from edge.models import Keypoint, BBox +from edge.tracking import Track, TrackFrame +from edge.detectors.ghost import GhostIncidentDetector, GhostConfig + +def _track(tid, x, y, t): + fr = TrackFrame(t, BBox(x, y, x + 20, y + 60), [Keypoint(x, y, 0.9) for _ in range(17)]) + return Track(tid, deque([fr], maxlen=90)) + +def test_lost_dangerous_track_becomes_ghost_after_wait(): + d = GhostIncidentDetector(GhostConfig(reappear_s=1.0)) + d.note_state(1, "FALLEN") + d.on_lost([_track(1, 100, 100, 0.0)], now=0.0) + assert d.update([], now=0.5) == [] # 아직 대기 + ghosts = d.update([], now=1.5) # 재출현 없음 > 1.0s + assert len(ghosts) == 1 + assert ghosts[0]["last_state"] == "FALLEN" + assert ghosts[0]["spatial"] == (110.0, 130.0) # bbox 중심 + +def test_non_dangerous_lost_track_is_ignored(): + d = GhostIncidentDetector(GhostConfig(reappear_s=1.0)) + d.note_state(1, "STANDING") + d.on_lost([_track(1, 100, 100, 0.0)], now=0.0) + assert d.update([], now=2.0) == [] + +def test_reappearance_near_location_cancels_ghost(): + d = GhostIncidentDetector(GhostConfig(reappear_s=1.0, reappear_dist=2.0)) + d.note_state(1, "FALLEN") + d.on_lost([_track(1, 100, 100, 0.0)], now=0.0) + # 인근에 새 트랙 재출현 + assert d.update([_track(2, 105, 105, 0.5)], now=0.5) == [] + assert d.update([], now=2.0) == [] # 취소되어 방출 없음 +``` + +- [ ] **Step 2: 실패 확인** + +Run: `pytest tests/test_ghost.py -v` → Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 구현** — `edge/detectors/ghost.py` + +```python +from dataclasses import dataclass + +from edge.tracking import Track + +_DANGEROUS = {"FALLING", "FALLEN", "NO_MOTION"} + + +@dataclass +class GhostConfig: + reappear_s: float = 10.0 + reappear_dist: float = 1.0 # bbox 높이 배수 + + +class GhostIncidentDetector: + def __init__(self, cfg: GhostConfig): + self.cfg = cfg + self._states: dict[int, str] = {} + # 후보: track_id -> (spatial(x,y), last_state, height, lost_at) + self._candidates: dict[int, tuple] = {} + + def note_state(self, track_id: int, state: str) -> None: + self._states[track_id] = state + + def on_lost(self, lost_tracks: list[Track], now: float) -> None: + for tr in lost_tracks: + state = self._states.get(tr.track_id, "STANDING") + if state in _DANGEROUS: + b = tr.latest.bbox + self._candidates[tr.track_id] = ((b.cx, b.cy), state, b.height, now) + self._states.pop(tr.track_id, None) + + def update(self, active_tracks: list[Track], now: float) -> list[dict]: + # 인근 재출현이면 후보 취소 + for tid, (spatial, _state, height, _lost_at) in list(self._candidates.items()): + for tr in active_tracks: + b = tr.latest.bbox + dist = ((b.cx - spatial[0]) ** 2 + (b.cy - spatial[1]) ** 2) ** 0.5 + if dist <= self.cfg.reappear_dist * (height or 1.0): + self._candidates.pop(tid, None) + break + # 대기 초과 방출 + out = [] + for tid, (spatial, state, _height, lost_at) in list(self._candidates.items()): + if now - lost_at >= self.cfg.reappear_s: + out.append({"spatial": spatial, "last_state": state, + "waited_s": round(now - lost_at, 1)}) + self._candidates.pop(tid, None) + return out +``` + +- [ ] **Step 4: 통과 확인** + +Run: `pytest tests/test_ghost.py -v` → Expected: PASS + +- [ ] **Step 5: 커밋** + +```bash +git add edge/detectors/ghost.py tests/test_ghost.py +git commit -m "feat(edge): GhostIncidentDetector (collapse-then-occluded spatial incident)" +``` + +--- + +## Task 9: ZoneIntrusionDetector (점-다각형 + 디바운스) + +**Files:** +- Create: `edge/detectors/zone.py`, `tests/test_zone.py` + +**Interfaces:** +- Consumes: `Track` (Task 4); `L_ANKLE, R_ANKLE` (Task 1) +- Produces: + - `point_in_polygon(x: float, y: float, poly: list[tuple[float,float]]) -> bool` (ray-casting) + - `Zone(zone_id: str, level: str, polygon: list[tuple[float,float]])` — level ∈ {"warning","restricted"} + - `ZoneConfig(debounce_s=0.75)` + - `ZoneIntrusionDetector(zones: list[Zone], cfg)` with `update(track: Track, now: float) -> list[dict]` — 발목점이 폴리곤 안에 debounce_s 이상 지속 시 `{"zone_id","level"}` 1회 방출; 이탈 시 리셋. `active_zone_ids(track_id) -> set[str]` (부동 워치독 연동용) + +- [ ] **Step 1: 실패 테스트** — `tests/test_zone.py` + +```python +from collections import deque +from edge.models import Keypoint, BBox +from edge.tracking import Track, TrackFrame +from edge.detectors.zone import point_in_polygon, Zone, ZoneIntrusionDetector, ZoneConfig + +SQUARE = [(0, 0), (100, 0), (100, 100), (0, 100)] + +def test_point_in_polygon_basic(): + assert point_in_polygon(50, 50, SQUARE) is True + assert point_in_polygon(150, 50, SQUARE) is False + +def _track_with_ankles(ax, ay, t): + kps = [Keypoint(ax, ay, 0.9) for _ in range(17)] + kps[15] = Keypoint(ax, ay, 0.9) # L_ANKLE + kps[16] = Keypoint(ax, ay, 0.9) # R_ANKLE + return Track(1, deque([TrackFrame(t, BBox(ax - 10, ay - 60, ax + 10, ay), kps)], maxlen=90)) + +def test_intrusion_fires_after_debounce(): + z = Zone("z1", "restricted", SQUARE) + det = ZoneIntrusionDetector([z], ZoneConfig(debounce_s=0.5)) + tr = _track_with_ankles(50, 50, 0.0) + assert det.update(tr, now=0.0) == [] + tr = _track_with_ankles(50, 50, 0.6) + out = det.update(tr, now=0.6) + assert out == [{"zone_id": "z1", "level": "restricted"}] + +def test_leaving_zone_resets(): + z = Zone("z1", "warning", SQUARE) + det = ZoneIntrusionDetector([z], ZoneConfig(debounce_s=0.5)) + det.update(_track_with_ankles(50, 50, 0.0), now=0.0) + det.update(_track_with_ankles(200, 200, 0.6), now=0.6) # 이탈 + out = det.update(_track_with_ankles(50, 50, 1.0), now=1.0) # 재진입, 디바운스 재시작 + assert out == [] +``` + +- [ ] **Step 2: 실패 확인** + +Run: `pytest tests/test_zone.py -v` → Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 구현** — `edge/detectors/zone.py` + +```python +from dataclasses import dataclass + +from edge.tracking import Track +from edge.models import L_ANKLE, R_ANKLE + + +def point_in_polygon(x: float, y: float, poly: list[tuple[float, float]]) -> bool: + inside = False + n = len(poly) + j = n - 1 + for i in range(n): + xi, yi = poly[i] + xj, yj = poly[j] + if ((yi > y) != (yj > y)) and (x < (xj - xi) * (y - yi) / ((yj - yi) or 1e-9) + xi): + inside = not inside + j = i + return inside + + +@dataclass(frozen=True) +class Zone: + zone_id: str + level: str # "warning" | "restricted" + polygon: list # list[tuple[float,float]] + + +@dataclass +class ZoneConfig: + debounce_s: float = 0.75 + + +class ZoneIntrusionDetector: + def __init__(self, zones: list[Zone], cfg: ZoneConfig): + self.zones = zones + self.cfg = cfg + self._inside_since: dict[tuple[int, str], float] = {} + self._fired: set[tuple[int, str]] = set() + + def _foot(self, track: Track) -> tuple[float, float]: + kps = track.latest.keypoints + a, b = kps[L_ANKLE], kps[R_ANKLE] + return ((a.x + b.x) / 2, (a.y + b.y) / 2) + + def active_zone_ids(self, track_id: int) -> set: + return {z for (tid, z) in self._inside_since if tid == track_id} + + def update(self, track: Track, now: float) -> list[dict]: + fx, fy = self._foot(track) + out = [] + for zone in self.zones: + key = (track.track_id, zone.zone_id) + if point_in_polygon(fx, fy, zone.polygon): + start = self._inside_since.setdefault(key, now) + if now - start >= self.cfg.debounce_s and key not in self._fired: + self._fired.add(key) + out.append({"zone_id": zone.zone_id, "level": zone.level}) + else: + self._inside_since.pop(key, None) + self._fired.discard(key) + return out +``` + +- [ ] **Step 4: 통과 확인** + +Run: `pytest tests/test_zone.py -v` → Expected: PASS + +- [ ] **Step 5: 커밋** + +```bash +git add edge/detectors/zone.py tests/test_zone.py +git commit -m "feat(edge): ZoneIntrusionDetector (foot-anchor point-in-polygon, debounce)" +``` + +--- + +## Task 10: EventEngine (오케스트레이션 + dedup + cooldown) + +**Files:** +- Create: `edge/engine.py`, `tests/test_engine.py` + +**Interfaces:** +- Consumes: `Track` (Task 4); `FallDetector` (Task 6); `ImmobilityWatchdog` (Task 7); `GhostIncidentDetector` (Task 8); `ZoneIntrusionDetector` (Task 9); `Event, EventType, severity_for` (Task 5) +- Produces: + - `EngineConfig(camera_id="cam", cooldown_s=30.0)` + - `EventEngine(fall, immobility, ghost, zone, cfg)` with `process(active_tracks: list[Track], lost_tracks: list[Track], now: float) -> list[Event]` + +- [ ] **Step 1: 실패 테스트** — `tests/test_engine.py` + +```python +from collections import deque +from edge.models import Keypoint, BBox +from edge.tracking import Track, TrackFrame +from edge.events import EventType +from edge.detectors.fall import FallDetector, FallConfig +from edge.detectors.immobility import ImmobilityWatchdog, ImmobilityConfig +from edge.detectors.ghost import GhostIncidentDetector, GhostConfig +from edge.detectors.zone import ZoneIntrusionDetector, ZoneConfig +from edge.engine import EventEngine, EngineConfig + +def _lying_kps(): + 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 kps + +def _lying_track(track, t): + track.frames.append(TrackFrame(t, BBox(90, 290, 210, 320), _lying_kps())) + return track + +def _engine(): + return EventEngine( + FallDetector(FallConfig(prone_confirm_s=0.5)), + ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)), + GhostIncidentDetector(GhostConfig()), + ZoneIntrusionDetector([], ZoneConfig()), + EngineConfig(camera_id="cam-01", cooldown_s=30.0), + ) + +def test_fall_produces_fall_event(): + eng = _engine() + tr = Track(1, deque(maxlen=90)) + _lying_track(tr, 0.0) + eng.process([tr], [], now=0.0) + _lying_track(tr, 0.6) + events = eng.process([tr], [], now=0.6) + types = [e.event_type for e in events] + assert EventType.FALL in types + assert events[0].camera_id == "cam-01" + +def test_cooldown_suppresses_duplicate(): + eng = _engine() + tr = Track(1, deque(maxlen=90)) + _lying_track(tr, 0.0); eng.process([tr], [], now=0.0) + _lying_track(tr, 0.6); eng.process([tr], [], now=0.6) # FALL 1회 + _lying_track(tr, 0.7) + again = eng.process([tr], [], now=0.7) + assert all(e.event_type != EventType.FALL for e in again) +``` + +- [ ] **Step 2: 실패 확인** + +Run: `pytest tests/test_engine.py -v` → Expected: FAIL — `ModuleNotFoundError: edge.engine` + +- [ ] **Step 3: 구현** — `edge/engine.py` + +```python +from dataclasses import dataclass + +from edge.tracking import Track +from edge.events import Event, EventType, severity_for +from edge.detectors.fall import FallDetector +from edge.detectors.immobility import ImmobilityWatchdog +from edge.detectors.ghost import GhostIncidentDetector +from edge.detectors.zone import ZoneIntrusionDetector + + +@dataclass +class EngineConfig: + camera_id: str = "cam" + cooldown_s: float = 30.0 + + +class EventEngine: + def __init__(self, fall: FallDetector, immobility: ImmobilityWatchdog, + ghost: GhostIncidentDetector, zone: ZoneIntrusionDetector, + cfg: EngineConfig): + self.fall = fall + self.immobility = immobility + self.ghost = ghost + self.zone = zone + self.cfg = cfg + self._last_emit: dict[tuple, float] = {} + + def _allow(self, key: tuple, now: float) -> bool: + last = self._last_emit.get(key) + if last is not None and now - last < self.cfg.cooldown_s: + return False + self._last_emit[key] = now + return True + + def _mk(self, etype: EventType, now: float, track_id=None, zone_id=None, + metrics=None, spatial=None) -> Event: + return Event(etype, severity_for(etype), self.cfg.camera_id, track_id, + zone_id, now, 0.0, metrics or {}, spatial) + + def process(self, active_tracks: list[Track], lost_tracks: list[Track], + now: float) -> list[Event]: + events: list[Event] = [] + self.ghost.on_lost(lost_tracks, now) + + for tr in active_tracks: + zone_hits = self.zone.update(tr, now) + in_zone = bool(self.zone.active_zone_ids(tr.track_id)) + + sig, meta = self.fall.update(tr, now) + state = self.fall._states.get(tr.track_id) + self.ghost.note_state(tr.track_id, state.name if state else "STANDING") + if sig == "fall" and self._allow((EventType.FALL, tr.track_id), now): + events.append(self._mk(EventType.FALL, now, tr.track_id, metrics=meta)) + elif sig == "no_motion" and self._allow((EventType.NO_MOTION, tr.track_id), now): + events.append(self._mk(EventType.NO_MOTION, now, tr.track_id, metrics=meta)) + + fired, imeta = self.immobility.update(tr, now, in_danger_zone=in_zone) + if fired and self._allow((EventType.IMMOBILITY, tr.track_id), now): + events.append(self._mk(EventType.IMMOBILITY, now, tr.track_id, metrics=imeta)) + + for hit in zone_hits: + etype = (EventType.ZONE_RESTRICTED if hit["level"] == "restricted" + else EventType.ZONE_WARNING) + if self._allow((etype, tr.track_id, hit["zone_id"]), now): + events.append(self._mk(etype, now, tr.track_id, zone_id=hit["zone_id"])) + + for g in self.ghost.update(active_tracks, now): + key = (EventType.GHOST, round(g["spatial"][0]), round(g["spatial"][1])) + if self._allow(key, now): + events.append(self._mk(EventType.GHOST, now, spatial=g["spatial"], + metrics={"last_state": g["last_state"], + "waited_s": g["waited_s"]})) + return events +``` + +- [ ] **Step 4: 통과 확인** + +Run: `pytest tests/test_engine.py -v` → Expected: PASS + +- [ ] **Step 5: 커밋** + +```bash +git add edge/engine.py tests/test_engine.py +git commit -m "feat(edge): EventEngine orchestration with dedup + cooldown" +``` + +--- + +## Task 11: RingBuffer + FaceBlurrer + Snapshotter + +**Files:** +- Create: `edge/snapshot.py`, `tests/test_snapshot.py` + +**Interfaces:** +- Consumes: `BBox` (Task 1) — 얼굴 블러 대상 박스 +- Produces: + - `RingBuffer(maxlen=90)` with `push(t: float, frame)` and `window(pre: int, post: int, index: int)` + - `FaceBlurrer()` with `blur(frame, head_boxes: list[BBox]) -> frame` (cv2 GaussianBlur) + - `Snapshotter(blurrer, quality=80)` with `capture(frames: list, head_boxes_per_frame: list) -> list[bytes]` — JPEG 최대 3장 인코딩 + +- [ ] **Step 1: 실패 테스트** — `tests/test_snapshot.py` + +```python +import numpy as np +from edge.models import BBox +from edge.snapshot import RingBuffer, FaceBlurrer, Snapshotter + +def _frame(val=128): + return np.full((64, 64, 3), val, dtype=np.uint8) + +def test_ringbuffer_keeps_recent(): + rb = RingBuffer(maxlen=3) + for i in range(5): + rb.push(float(i), _frame(i)) + assert len(rb.frames) == 3 + assert rb.frames[-1][0] == 4.0 + +def test_face_blur_changes_region(): + fb = FaceBlurrer() + f = _frame(0) + f[0:10, 0:10] = 255 + out = fb.blur(f.copy(), [BBox(0, 0, 10, 10)]) + assert out is not None + assert out.shape == f.shape + +def test_snapshotter_returns_jpeg_bytes(): + snap = Snapshotter(FaceBlurrer()) + imgs = snap.capture([_frame(100), _frame(120)], [[], []]) + assert 1 <= len(imgs) <= 3 + assert imgs[0][:2] == b"\xff\xd8" # JPEG magic +``` + +- [ ] **Step 2: 실패 확인** + +Run: `pytest tests/test_snapshot.py -v` → Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 구현** — `edge/snapshot.py` + +```python +from collections import deque + +import cv2 + +from edge.models import BBox + + +class RingBuffer: + def __init__(self, maxlen: int = 90): + self.frames: deque = deque(maxlen=maxlen) + + def push(self, t: float, frame) -> None: + self.frames.append((t, frame)) + + def window(self, pre: int, post: int, index: int) -> list: + lo = max(0, index - pre) + hi = min(len(self.frames), index + post + 1) + return [f for (_t, f) in list(self.frames)[lo:hi]] + + +class FaceBlurrer: + def blur(self, frame, head_boxes: list[BBox]): + for b in head_boxes: + x1, y1 = int(max(0, b.x1)), int(max(0, b.y1)) + x2, y2 = int(min(frame.shape[1], b.x2)), int(min(frame.shape[0], b.y2)) + if x2 > x1 and y2 > y1: + roi = frame[y1:y2, x1:x2] + frame[y1:y2, x1:x2] = cv2.GaussianBlur(roi, (0, 0), sigmaX=12) + return frame + + +class Snapshotter: + def __init__(self, blurrer: FaceBlurrer, quality: int = 80): + self.blurrer = blurrer + self.quality = quality + + def capture(self, frames: list, head_boxes_per_frame: list) -> list[bytes]: + out: list[bytes] = [] + for frame, heads in list(zip(frames, head_boxes_per_frame))[:3]: + blurred = self.blurrer.blur(frame.copy(), heads) + ok, buf = cv2.imencode(".jpg", blurred, + [cv2.IMWRITE_JPEG_QUALITY, self.quality]) + if ok: + out.append(buf.tobytes()) + return out +``` + +- [ ] **Step 4: 통과 확인** + +Run: `pytest tests/test_snapshot.py -v` → Expected: PASS + +- [ ] **Step 5: 커밋** + +```bash +git add edge/snapshot.py tests/test_snapshot.py +git commit -m "feat(edge): RingBuffer + face blur + JPEG snapshotter" +``` + +--- + +## Task 12: PostClient (HMAC + nonce 서명) + +**Files:** +- Create: `edge/alert.py`, `tests/test_alert.py` + +**Interfaces:** +- Consumes: `Event` (Task 5) +- Produces: + - `sign(secret: str, nonce: str, ts: str, body: bytes) -> str` (HMAC-SHA256 hex) + - `PostClient(url, device_id, secret, nonce_fn, now_fn, http_post=None, timeout=5.0)` with `send(event: Event, snapshots: list[bytes]) -> bool` + +- [ ] **Step 1: 실패 테스트** — `tests/test_alert.py` + +```python +import json +from edge.events import Event, EventType, Severity +from edge.alert import sign, PostClient + +def _event(): + return Event(EventType.FALL, Severity.HIGH, "cam-01", 7, None, 1000.0, 3.0, + {"torso_angle": 78.0}, None) + +class _FakeResp: + def __init__(self, code): self.status_code = code + +def test_sign_is_deterministic(): + s1 = sign("secret", "n1", "1000", b"body") + s2 = sign("secret", "n1", "1000", b"body") + assert s1 == s2 and len(s1) == 64 + +def test_send_posts_signed_payload(): + captured = {} + def fake_post(url, data=None, headers=None, timeout=None): + captured["url"] = url + captured["headers"] = headers + captured["body"] = data + return _FakeResp(200) + + client = PostClient("https://x/api/events", "pi-01", "secret", + nonce_fn=lambda: "n1", now_fn=lambda: 1000.0, + http_post=fake_post) + ok = client.send(_event(), [b"\xff\xd8jpeg"]) + assert ok is True + assert captured["headers"]["X-Nonce"] == "n1" + assert captured["headers"]["X-Timestamp"] == "1000.0" + payload = json.loads(captured["body"]) + assert payload["device_id"] == "pi-01" + assert payload["event_type"] == "fall" + assert len(payload["snapshots"]) == 1 + expected = sign("secret", "n1", "1000.0", captured["body"].encode()) + assert captured["headers"]["X-Signature"] == expected + +def test_send_returns_false_on_error_code(): + client = PostClient("https://x", "pi-01", "s", lambda: "n", lambda: 1.0, + http_post=lambda *a, **k: _FakeResp(500)) + assert client.send(_event(), []) is False +``` + +- [ ] **Step 2: 실패 확인** + +Run: `pytest tests/test_alert.py -v` → Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 구현** — `edge/alert.py` + +```python +import base64 +import hashlib +import hmac +import json + +from edge.events import Event + + +def sign(secret: str, nonce: str, ts: str, body: bytes) -> str: + msg = nonce.encode() + ts.encode() + body + return hmac.new(secret.encode(), msg, hashlib.sha256).hexdigest() + + +class PostClient: + def __init__(self, url: str, device_id: str, secret: str, + nonce_fn, now_fn, http_post=None, timeout: float = 5.0): + self.url = url + self.device_id = device_id + self.secret = secret + self.nonce_fn = nonce_fn + self.now_fn = now_fn + self.timeout = timeout + if http_post is None: + import requests + http_post = requests.post + self._post = http_post + + def send(self, event: Event, snapshots: list[bytes]) -> bool: + payload = event.to_payload(self.device_id) + payload["snapshots"] = [base64.b64encode(s).decode() for s in snapshots] + body = json.dumps(payload, sort_keys=True) + nonce = self.nonce_fn() + ts = str(self.now_fn()) + headers = { + "Content-Type": "application/json", + "X-Nonce": nonce, + "X-Timestamp": ts, + "X-Signature": sign(self.secret, nonce, ts, body.encode()), + } + try: + resp = self._post(self.url, data=body, headers=headers, timeout=self.timeout) + except Exception: + return False + return 200 <= resp.status_code < 300 +``` + +- [ ] **Step 4: 통과 확인** + +Run: `pytest tests/test_alert.py -v` → Expected: PASS + +- [ ] **Step 5: 커밋** + +```bash +git add edge/alert.py tests/test_alert.py +git commit -m "feat(edge): PostClient with HMAC+nonce signed event POST" +``` + +--- + +## Task 13: OfflineQueue (SQLite 지속 + 재전송) + +**Files:** +- Create: `edge/queue.py`, `tests/test_queue.py` + +**Interfaces:** +- Consumes: `PostClient` (Task 12); `Event, EventType, Severity` (Task 5) +- Produces: + - `OfflineQueue(db_path, client)`: + - `enqueue(event: Event, snapshots: list[bytes]) -> None` + - `try_send(event: Event, snapshots: list[bytes]) -> bool` — 즉시 전송, 실패 시 enqueue + - `flush() -> int` — 큐 재전송, 성공 개수 반환(실패 시 순서 보존 중단) + - `pending() -> int` + +- [ ] **Step 1: 실패 테스트** — `tests/test_queue.py` + +```python +from edge.events import Event, EventType, Severity +from edge.queue import OfflineQueue + +def _event(n=1): + return Event(EventType.FALL, Severity.HIGH, "cam-01", n, None, float(n), 3.0, {}, None) + +class _Client: + def __init__(self, fail=False): + self.fail = fail + self.sent = [] + def send(self, event, snapshots): + if self.fail: + return False + self.sent.append(event.track_id) + return True + +def test_try_send_success_not_queued(tmp_path): + q = OfflineQueue(str(tmp_path / "q.db"), _Client()) + assert q.try_send(_event(1), [b"x"]) is True + assert q.pending() == 0 + +def test_failure_is_queued(tmp_path): + q = OfflineQueue(str(tmp_path / "q.db"), _Client(fail=True)) + assert q.try_send(_event(1), [b"x"]) is False + assert q.pending() == 1 + +def test_flush_resends_when_recovered(tmp_path): + client = _Client(fail=True) + q = OfflineQueue(str(tmp_path / "q.db"), client) + q.try_send(_event(1), [b"x"]) + q.try_send(_event(2), [b"y"]) + assert q.pending() == 2 + client.fail = False + assert q.flush() == 2 + assert q.pending() == 0 + assert client.sent == [1, 2] +``` + +- [ ] **Step 2: 실패 확인** + +Run: `pytest tests/test_queue.py -v` → Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 구현** — `edge/queue.py` + +```python +import base64 +import json +import sqlite3 + +from edge.events import Event, EventType, Severity + + +class OfflineQueue: + def __init__(self, db_path: str, client): + self.client = client + self.db = sqlite3.connect(db_path) + self.db.execute( + "CREATE TABLE IF NOT EXISTS queue (" + "id INTEGER PRIMARY KEY AUTOINCREMENT, payload TEXT, snaps TEXT)" + ) + self.db.commit() + + def _rehydrate(self, payload: dict, snaps: list[bytes]) -> tuple[Event, list[bytes]]: + e = Event(EventType(payload["event_type"]), Severity(payload["severity"]), + payload["camera_id"], payload["track_id"], payload["zone_id"], + payload["occurred_at"], payload["duration_s"], payload["metrics"], + tuple(payload["spatial"]) if payload.get("spatial") else None) + return e, snaps + + def enqueue(self, event: Event, snapshots: list[bytes]) -> None: + payload = event.to_payload("") # device_id는 client가 다시 채움 + payload["event_type"] = event.event_type.value + payload["severity"] = event.severity.value + payload["spatial"] = list(event.spatial) if event.spatial else None + snaps = [base64.b64encode(s).decode() for s in snapshots] + self.db.execute("INSERT INTO queue (payload, snaps) VALUES (?, ?)", + (json.dumps(payload), json.dumps(snaps))) + self.db.commit() + + def try_send(self, event: Event, snapshots: list[bytes]) -> bool: + if self.client.send(event, snapshots): + return True + self.enqueue(event, snapshots) + return False + + def flush(self) -> int: + rows = self.db.execute("SELECT id, payload, snaps FROM queue ORDER BY id").fetchall() + sent = 0 + for rid, payload_json, snaps_json in rows: + payload = json.loads(payload_json) + snaps = [base64.b64decode(s) for s in json.loads(snaps_json)] + event, snapshots = self._rehydrate(payload, snaps) + if self.client.send(event, snapshots): + self.db.execute("DELETE FROM queue WHERE id = ?", (rid,)) + self.db.commit() + sent += 1 + else: + break # 여전히 오프라인 — 순서 보존 + return sent + + def pending(self) -> int: + return self.db.execute("SELECT COUNT(*) FROM queue").fetchone()[0] +``` + +- [ ] **Step 4: 통과 확인** + +Run: `pytest tests/test_queue.py -v` → Expected: PASS + +- [ ] **Step 5: 커밋** + +```bash +git add edge/queue.py tests/test_queue.py +git commit -m "feat(edge): OfflineQueue (SQLite persist + ordered resend)" +``` + +--- + +## Task 14: Config (env 로드) + +**Files:** +- Create: `edge/config.py`, `tests/test_config.py` + +**Interfaces:** +- Consumes: `FallConfig` (Task 6), `ImmobilityConfig` (Task 7), `GhostConfig` (Task 8), `ZoneConfig` (Task 9), `EngineConfig` (Task 10) +- Produces: + - `Config(device_id, camera_id, cafe24_url, secret, fall, immobility, ghost, zone, engine)` + - `load_config(env: dict) -> Config` + - env 키: `DEVICE_ID, CAMERA_ID, CAFE24_URL, DEVICE_SECRET, FALL_PRONE_S, FALL_NO_MOTION_S, IMMOBILE_S, GHOST_REAPPEAR_S, ZONE_DEBOUNCE_S, COOLDOWN_S` + +- [ ] **Step 1: 실패 테스트** — `tests/test_config.py` + +```python +from edge.config import load_config + +def test_defaults_when_env_empty(): + cfg = load_config({}) + assert cfg.device_id == "pi-unknown" + assert cfg.fall.prone_confirm_s == 3.0 + assert cfg.fall.no_motion_s == 10.0 + +def test_env_overrides(): + env = { + "DEVICE_ID": "pi-w1", "CAMERA_ID": "cam-9", + "CAFE24_URL": "https://x/api/events", "DEVICE_SECRET": "s", + "FALL_PRONE_S": "2.0", "FALL_NO_MOTION_S": "8", + "IMMOBILE_S": "90", "COOLDOWN_S": "45", + } + cfg = load_config(env) + assert cfg.device_id == "pi-w1" + assert cfg.camera_id == "cam-9" + assert cfg.fall.prone_confirm_s == 2.0 + assert cfg.fall.no_motion_s == 8.0 + assert cfg.immobility.immobile_s == 90.0 + assert cfg.engine.cooldown_s == 45.0 + assert cfg.engine.camera_id == "cam-9" +``` + +- [ ] **Step 2: 실패 확인** + +Run: `pytest tests/test_config.py -v` → Expected: FAIL — `ModuleNotFoundError` + +- [ ] **Step 3: 구현** — `edge/config.py` + +```python +from dataclasses import dataclass + +from edge.detectors.fall import FallConfig +from edge.detectors.immobility import ImmobilityConfig +from edge.detectors.ghost import GhostConfig +from edge.detectors.zone import ZoneConfig +from edge.engine import EngineConfig + + +@dataclass +class Config: + device_id: str + camera_id: str + cafe24_url: str + secret: str + fall: FallConfig + immobility: ImmobilityConfig + ghost: GhostConfig + zone: ZoneConfig + engine: EngineConfig + + +def _f(env: dict, key: str, default: float) -> float: + return float(env[key]) if key in env else default + + +def load_config(env: dict) -> Config: + camera_id = env.get("CAMERA_ID", "cam-01") + fall = FallConfig( + prone_confirm_s=_f(env, "FALL_PRONE_S", 3.0), + no_motion_s=_f(env, "FALL_NO_MOTION_S", 10.0), + ) + immobility = ImmobilityConfig(immobile_s=_f(env, "IMMOBILE_S", 120.0)) + ghost = GhostConfig(reappear_s=_f(env, "GHOST_REAPPEAR_S", 10.0)) + zone = ZoneConfig(debounce_s=_f(env, "ZONE_DEBOUNCE_S", 0.75)) + engine = EngineConfig(camera_id=camera_id, cooldown_s=_f(env, "COOLDOWN_S", 30.0)) + return Config( + device_id=env.get("DEVICE_ID", "pi-unknown"), + camera_id=camera_id, + cafe24_url=env.get("CAFE24_URL", ""), + secret=env.get("DEVICE_SECRET", ""), + fall=fall, immobility=immobility, ghost=ghost, zone=zone, engine=engine, + ) +``` + +- [ ] **Step 4: 통과 확인** + +Run: `pytest tests/test_config.py -v` → Expected: PASS + +- [ ] **Step 5: 커밋** + +```bash +git add edge/config.py tests/test_config.py +git commit -m "feat(edge): env-based Config for detectors and transport" +``` + +--- + +## Task 15: 메인 루프 배선 (app) + 통합 테스트 + +**Files:** +- Create: `edge/app.py`, `tests/test_app.py` + +**Interfaces:** +- Consumes: 전 태스크. `SourcePort, PosePort` (Task 4); `Person` (Task 1) +- Produces: + - `run_once(source, pose, tracker, smoother, engine, ring, snapshotter, sink, now_fn) -> list[Event]` — 한 프레임 처리 + - `build_pipeline(config: Config, source, pose) -> dict` — 실제 배선 팩토리 + +- [ ] **Step 1: 실패 테스트** — `tests/test_app.py` + +```python +import numpy as np +from edge.models import Person, BBox, Keypoint +from edge.tracking import IouTracker +from edge.smoothing import KeypointSmoother +from edge.detectors.fall import FallDetector, FallConfig +from edge.detectors.immobility import ImmobilityWatchdog, ImmobilityConfig +from edge.detectors.ghost import GhostIncidentDetector, GhostConfig +from edge.detectors.zone import ZoneIntrusionDetector, ZoneConfig +from edge.engine import EventEngine, EngineConfig +from edge.snapshot import RingBuffer, FaceBlurrer, Snapshotter +from edge.events import EventType +from edge.app import run_once + +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, event, snapshots): + self.events.append(event); return True + +def test_run_once_pipeline_emits_fall_after_prone(): + src, pose = _Source(), _Pose() + tracker = IouTracker() + smoother = KeypointSmoother() + engine = EventEngine( + FallDetector(FallConfig(prone_confirm_s=0.5)), + ImmobilityWatchdog(ImmobilityConfig(immobile_s=999)), + GhostIncidentDetector(GhostConfig()), + ZoneIntrusionDetector([], ZoneConfig()), + EngineConfig(camera_id="cam-01"), + ) + ring = RingBuffer(); snap = Snapshotter(FaceBlurrer()); sink = _Sink() + times = iter([0.0, 0.6]) + def now_fn(): return next(times) + run_once(src, pose, tracker, smoother, engine, ring, snap, sink, now_fn) + run_once(src, pose, tracker, smoother, engine, ring, snap, sink, now_fn) + assert any(e.event_type == EventType.FALL for e in sink.events) +``` + +- [ ] **Step 2: 실패 확인** + +Run: `pytest tests/test_app.py -v` → Expected: FAIL — `ModuleNotFoundError: edge.app` + +- [ ] **Step 3: 구현** — `edge/app.py` + +```python +from edge.config import Config +from edge.tracking import IouTracker +from edge.smoothing import KeypointSmoother +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 +from edge.models import Person, BBox + + +def _head_boxes(persons: list) -> list: + # 배경인물 머리영역 근사: 코 주변 작은 박스 (MVP; 실제는 얼굴 검출 포트로 교체) + boxes = [] + for p in persons: + nose = p.keypoints[0] + boxes.append(BBox(nose.x - 20, nose.y - 25, nose.x + 20, nose.y + 15)) + return boxes + + +def run_once(source, pose, tracker, smoother, engine, ring, snapshotter, sink, now_fn): + ok, frame = source.read() + if not ok: + return [] + now = now_fn() + persons = pose.infer(frame) + smoothed = [Person(p.bbox, smoother.smooth(p.keypoints, now)) for p in persons] + tracks = tracker.update(smoothed, now) + lost = tracker.lost() + ring.push(now, frame) + events = engine.process(tracks, lost, now) + if events: + snaps = snapshotter.capture([frame], [_head_boxes(persons)]) + for e in events: + sink.try_send(e, snaps) + return events + + +def build_pipeline(config: Config, source, pose) -> dict: + return { + "source": source, + "pose": pose, + "tracker": IouTracker(), + "smoother": KeypointSmoother(), + "engine": EventEngine( + FallDetector(config.fall), + ImmobilityWatchdog(config.immobility), + GhostIncidentDetector(config.ghost), + ZoneIntrusionDetector([], config.zone), + config.engine, + ), + "ring": RingBuffer(), + "snapshotter": Snapshotter(FaceBlurrer()), + } +``` + +- [ ] **Step 4: 통과 확인** + +Run: `pytest tests/test_app.py -v` → Expected: PASS + +- [ ] **Step 5: 전체 스위트 + 커밋** + +```bash +pytest -q +git add edge/app.py tests/test_app.py +git commit -m "feat(edge): main-loop wiring (run_once) + pipeline factory" +``` + +--- + +## Task 16: 하드웨어 어댑터 스텁 + 브링업 문서 (실장비 게이트) + +> ⚠️ **이 태스크는 실장비(Pi5+Hailo)에서 실행.** 개발 PC(Windows)에선 파일 생성·문서화까지만. fps 벤치·카메라 캡처는 Pi에서. + +**Files:** +- Create: `edge/adapters/__init__.py`(빈 파일), `edge/adapters/hailo_pose.py`, `edge/adapters/picamera_source.py`, `edge/adapters/uvc_source.py`, `docs/hardware-bringup.md` + +**Interfaces:** +- Consumes: `PosePort, SourcePort` (Task 4); `Person, BBox, Keypoint` (Task 1) +- Produces: `HailoPoseAdapter(hef_path)` implements PosePort; `PiCameraSource(index)`, `UvcSource(dev, mjpeg=True)` implement SourcePort. + +- [ ] **Step 1: 어댑터 스텁 작성** — `edge/adapters/hailo_pose.py` + +```python +"""Hailo yolov8s_pose HEF 어댑터 (실장비). PosePort 구현. +개발 PC에선 인스턴스화 불가(hailort 부재). Pi에서 HailoRT로 실제 추론 배선.""" + +from edge.models import Person, BBox, Keypoint + + +class HailoPoseAdapter: + def __init__(self, hef_path: str): + self.hef_path = hef_path + # 실장비 배선 지점: HailoRT VDevice/InferModel 로드, yolov8s_pose 후처리. + # 참고: hailo-rpi5-examples pose 파이프라인 (MIT). + raise NotImplementedError("실장비 전용: HailoRT 배선 필요") + + def infer(self, frame) -> list: + # frame(640x640 RGB) -> yolov8s_pose -> list[Person(bbox, 17 keypoints)] + raise NotImplementedError +``` + +- [ ] **Step 2: 소스 스텁 작성** — `edge/adapters/picamera_source.py`, `edge/adapters/uvc_source.py` + +```python +# picamera_source.py +"""CSI 카메라 소스 (실장비). 카메라당 별 프로세스로 picamera2 구동.""" + + +class PiCameraSource: + def __init__(self, index: int = 0): + self.index = index + raise NotImplementedError("실장비 전용: picamera2 배선 필요") + + def read(self): + raise NotImplementedError +``` + +```python +# uvc_source.py +"""USB UVC 소스 (실장비). v4l2 MJPEG. cv2.VideoCapture, 2개 USB3 컨트롤러 분산.""" + +import cv2 + + +class UvcSource: + def __init__(self, dev, mjpeg: bool = True): + self.cap = cv2.VideoCapture(dev) + if mjpeg: + self.cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc(*"MJPG")) + + def read(self): + return self.cap.read() +``` + +- [ ] **Step 3: 브링업 검증 문서** — `docs/hardware-bringup.md` + +다음 실장비 검증 절차 명시(스펙 §14): +1. PCIe Gen3 활성: `/boot/firmware/config.txt`에 `dtparam=pciex1_gen=3`, 재부팅 후 `lspci -vv | grep -i speed` 확인. +2. `yolov8s_pose` HEF 벤치: 단일·2·4 스트림 라운드로빈 fps·CPU·온도 측정. 목표 낙상 중요캠 ≥12~15fps. +3. fps 3티어(full/detection/motion) 각각 측정 → 캠 수·BOM 확정. +4. CSI 카메라당 별 프로세스 + USB MJPEG 2컨트롤러 분산, 수 시간 동시 캡처 안정성. +5. 실측 결과를 `thresholds`·카메라 `role`에 반영. + +- [ ] **Step 4: 개발 PC에서 UVC 어댑터 import 확인** + +Run: `python -c "import edge.adapters.uvc_source; print('ok')"` +Expected: `ok` (cv2만 필요; Hailo/picamera 스텁은 인스턴스화 시에만 NotImplementedError). + +- [ ] **Step 5: 커밋** + +```bash +git add edge/adapters docs/hardware-bringup.md +git commit -m "feat(edge): hardware adapter stubs + Pi bring-up/benchmark doc" +``` + +--- + +## Self-Review + +**1. Spec coverage (P1 범위 대비):** +- F1 넘어짐 → Task 6 ✓ · F2 낙상후 미동 위급 → Task 6(no_motion) ✓ · F3 자세복원 하향 → Task 6(recover) ✓ +- F3b 자세무관 부동 → Task 7 ✓ · F3c 유령사고 → Task 8 ✓ · F4 위험구역 2단계 → Task 9 ✓ +- F5 dedup+cooldown → Task 10 ✓ · F6 스냅샷 1~3장 → Task 11 ✓ · F8 오프라인 큐 → Task 13 ✓ +- 시점강건 낙상(OR) → Task 6 ✓ · 지터스무딩 → Task 3 ✓ · baseline freeze → Task 6 ✓ +- NF5 HMAC+nonce → Task 12 ✓ · 얼굴블러(F14 일부) → Task 11 ✓ · env 임계 → Task 14 ✓ +- fps 3티어 검증(§10.2) → Task 16 문서(실장비) ✓ +- **P1-A 범위 밖(다른 서브플랜):** 인증/RBAC·ingest 서버(P1-B), 라벨 평가셋·드릴게이트(P1-C), 컴플라이언스 화면(P1-B), 신뢰시간 RTC(P2). +- **알려진 갭(의도적):** F15 "현재프레임 캡처"는 카메라 소스 어댑터 후속(P1-B 구역설정과 함께). fps 3티어(T2/T3) 실제 배선은 Task 16 실장비 단계에서(코어 로직은 모델 무관). + +**2. Placeholder scan:** 코드 스텝 전부 실제 코드. Task 16만 실장비 스텁(의도적 하드웨어 게이트, 문서로 절차 명시). + +**3. Type consistency:** `Track.latest`, `TrackFrame(t,bbox,keypoints)`, `_State.name`(FallDetector↔EventEngine 참조), `Event.to_payload`, `PostClient.send(event,snapshots)`, `OfflineQueue.try_send`, detector `update(...)->tuple` 시그니처 Task 간 일치. `BBox` import는 app.py에 포함 ✓. `height_remaining` 메트릭 키 fall.py 내부 일관. + +--- + +## Execution Handoff + +**Plan complete. 저장: `docs/superpowers/plans/2026-07-11-p1a-edge-detection-core.md`.** + +두 실행 옵션: + +1. **Subagent-Driven (추천)** — 태스크마다 새 subagent 디스패치, 태스크 간 리뷰, 빠른 반복. `superpowers:subagent-driven-development`. +2. **Inline Execution** — 이 세션에서 태스크 순차 실행, 체크포인트 리뷰. `superpowers:executing-plans`. + +어느 방식으로? diff --git a/docs/superpowers/specs/2026-07-11-workplace-safety-ai-cctv-design.md b/docs/superpowers/specs/2026-07-11-workplace-safety-ai-cctv-design.md new file mode 100644 index 0000000..ca3ef1b --- /dev/null +++ b/docs/superpowers/specs/2026-07-11-workplace-safety-ai-cctv-design.md @@ -0,0 +1,456 @@ +# 작업장 안전관리 AI CCTV — 설계 문서 (Design Spec) + +- **문서 버전:** 2.0 (다관점 리뷰 반영 개정) +- **작성일:** 2026-07-11 +- **코드명:** `move_detect` +- **상태:** 설계 확정 대기(사용자 리뷰 전) + +> **v2.0 변경 요약:** 5개 렌즈(산업안전·엣지CV·운영UX·보안/개인정보·제품/배포) 다관점 리뷰 결과 반영. +> 핵심: (1) 기절 감지를 낙상 경로에서 **분리**(자세무관 부동 워치독 + 트랙소실 유령사고), (2) 시점/노이즈 강건 낙상(OR로직·공간바인딩·지터스무딩), (3) 대시보드 **인증/RBAC** 및 ingest/저장 하드닝, (4) **라벨 평가셋 + 드릴 커미셔닝 게이트**, (5) **fps 3티어 폴백**, (6) **PIPA·노사협의회 컴플라이언스 제품 내장**, (7) 증거 무결성(계층 보존·legal-hold·audit). +> **사용자 결정:** 호스팅 = Cafe24 유지(내부 하드닝) · v1+끼임 body-part(P2) · 라이브뷰 폐기(스냅샷 절대) · 컴플라이언스 제품 소유. + +--- + +## 1. 개념 (Concept) + +> Raspberry Pi 5 + AI HAT+ (Hailo-8, 26 TOPS)가 작업장 카메라 영상을 **실시간 자세추정(pose estimation)**으로 분석해, **넘어짐·미동/기절·위험구역 침입(+끼임)** 사고를 판단하고, 사고 순간 스냅샷과 함께 Cafe24 PHP 관제 대시보드로 즉시 경보하는 엣지 AI CCTV. 감지에 그치지 않고 **응답(에스컬레이션·119)·측정(정확도)·합법(컴플라이언스)·증거보존**까지 책임지는 안전관리 제품. + +핵심 철학: + +- **무거운 AI 판단은 전부 엣지(Pi)에서.** 클라우드는 얇게 — 수신·저장·표시·발송·대응관리. +- **영상 스트림/라이브뷰를 절대 내보내지 않는다.** 개인정보·대역폭 이유로 사고 순간 **정지 스냅샷만** (라이브뷰·클립 없음, 사용자 결정). +- **오탐(FP)은 안전 시스템에서 치명적.** 자세추정으로 앉음/숙임/의도적 눕기와 실제 낙상을 구분한다. +- **미탐(FN)은 더 치명적.** 가장 위험한 케이스(좌위 실신·가림 붕괴)에서 경보가 나가도록 설계한다. +- **감지≠안전.** 응답자에게 닿고, 정확도를 측정·튜닝하고, 법적 증거로 남아야 안전이 완성된다. + +--- + +## 2. 범위 (Scope) + +### 2.1 목표 (In scope) + +- 사고 감지: 넘어짐, 넘어짐 후 장시간 미동(기절 의심), **자세무관 장시간 부동**, 위험구역 침입(2단계), **끼임 body-part 침입(P2)**. +- 1대의 Pi가 2~4대 카메라(CSI + USB UVC 혼합) 동시 처리, **fps 3티어 폴백**으로 우아한 저하. +- 사고 시 스냅샷 1~3장(사고 前/中/後, 링버퍼) + 메타데이터를 Cafe24 PHP로 HTTPS POST(HMAC+nonce). +- PHP 관제 대시보드: **인증/RBAC**, 실시간 사고 목록·상세, 대응 에스컬레이션, 위험구역 설정, 감도 다이얼, 장비/플릿 상태, 이력·통계, 오탐 피드백. +- 대응 계층: 확인 SLA·에스컬레이션 사다리·온콜 라우팅·119 안내 카드. +- 컴플라이언스 모듈(**제품 소유**): 노사협의회 협의 기록, PIPA 안내판/처리방침 메타, DPIA 체크리스트 게이트, 데이터주체 조회/삭제 도구, **배경인물 얼굴블러 default-ON**. +- 증거: 계층형 보존 + legal-hold(실사고 3~5년), 위변조방지 audit 로그, 서버 신뢰 타임스탬프. +- (선택) 위급 시 SMS(기존 `leak_sms` 게이트 재사용), 현장 GPIO 경보(경광등/부저). + +### 2.2 비목표 (Out of scope, YAGNI) + +- **라이브뷰·영상클립·상시 녹화·클라우드 스트리밍** — 절대 미포함(사용자 결정, no-streaming 원칙 고수). +- 얼굴 **인식**/개인 식별 — 미포함. (단 배경인물 **탐지-후-블러**는 최소화 요건으로 포함; 식별 아님) +- 실외·야간·IR 환경 — 대상은 **실내 공장/창고, 주간 조명**. (조명꺼짐은 `monitoring_degraded`로 처리) +- 떨어짐(추락)·부딪힘(차량)·밀폐공간 질식·화재/가스 — v1 미포함, 외부센서/별도클래스 필요 → P4 범위 결정(§16). +- 기계 비상정지(E-stop) **능동 개입** — 안전PLC 인증·책임 이슈로 v1 미포함, 훅만 설계(§16 미결정). +- 시간축 딥러닝 행동인식(ST-GCN) — v1은 자세 휴리스틱. 오탐 과다 시 P4 재검토. +- 대규모 온사이트 다중 Pi 집계 서버 — P4. (단 **경량 멀티사이트 플릿 레지스트리는 P2**로 포함 — 클라우드가 이미 device_id 키 기반) + +### 2.3 컴플라이언스 (제품 소유 — v1 In scope) + +v1.0에서 "운영정책, 범위밖"으로 뺐던 항목을 **제품 기능으로 내장**한다(사용자 결정, 차별화). 이는 배포의 법적 전제: + +- **근로자참여법 §20-14:** 감시장비 설치는 노사협의회 **필수 협의 사항** → 협의 기록(일자·안건·근로자대표 서명) 관리 기능. +- **PIPA §25:** 영상정보처리기기 안내판, 운영·관리방침, 목적 제한(근태감시 전용(轉用) 차단), 데이터주체 조회/삭제 권리 도구. +- 낙상/실신 영상은 **민감정보 소지** → 최소수집·접근통제·접근로그. +- 배포 전 **DPIA 체크리스트 게이트**. + +--- + +## 3. 요구사항 (Requirements) + +### 3.1 기능 요구사항 + +| ID | 요구사항 | 단계 | +|----|----------|------| +| F1 | 사람이 넘어져 완전히 넘어진 자세(prone)가 > 3초 지속되면 넘어짐 이벤트를 발생시킨다. | P1 | +| F2 | 넘어짐 상태에서 관절 움직임이 임계 미만으로 ≥10초 지속되면 "위급(기절 의심)" 최고등급 경보를 발생시킨다. | P1 | +| F3 | 스스로 일어나면(자세 복원) 넘어짐/미동 경보를 취소·하향한다. | P1 | +| **F3b** | **낙상 전이 없이도** 추적 중인 사람이 자세와 무관하게 관절변위 임계 미만으로 N초(좌위 기본 90~180초, 위험구역 더 짧게) 지속되면 "이상 정지" 저·중등급 확인 경보를 발생시킨다. | **P1** | +| **F3c** | FALLING/FALLEN/NO_MOTION 트랙이 복원 확인 없이 **사라지면**(가림) 마지막 위치에 공간 "유령사고"를 만들고, T초 내 인근에 트랙 재출현 없으면 상향한다. | **P1** | +| F4 | 카메라별 위험구역 폴리곤(2단계: 경고/절대금지)에 사람 발끝이 진입하면 침입 이벤트를 발생시킨다. | P1 | +| **F4b** | 기계 위험점 구역에 대해 **임의 신체 관절점** 진입(발목만 아님)으로 끼임 위험을 판정한다. | **P2** | +| F5 | 모든 이벤트는 (트랙ID 또는 공간영역) 기준 1회 발생 + 쿨다운으로 반복 폭탄을 방지한다. | P1 | +| F6 | 이벤트 시 스냅샷 1~3장(前/中/後, 링버퍼) + 메타데이터를 Cafe24로 전송한다. | P1 | +| F7 | 관리자는 **인증 후** 실시간 사고 목록·상세를 보고, 위험구역을 그리고, 감도를 조정하고, 이력·통계를 조회한다. | P1 | +| F8 | 전송 실패 시 로컬 큐잉 후 재전송하여 사고를 유실하지 않는다. | P1 | +| **F9** | 미확인 위급 경보는 확인 SLA 초과 시 2·3차 온콜로 자동 에스컬레이션한다. | P3(위급 최소본 P1) | +| **F10** | 이벤트 처리결과(실제/오탐/현장처리/드릴/검토요)를 라벨링하고, 온디바이스 키포인트 시계열로 재생·튜닝한다. | P2 | +| **F11** | 임계값을 (device, camera, zone)별로 오버라이드하고, 관리자용 감도 프리셋(Conservative↔Aggressive)으로 노출한다. | P2 | +| **F12** | 카메라 가림/초점흐림/조명꺼짐/탬퍼를 self-check해 `monitoring_degraded`를 발생시킨다. | P2 | +| **F13** | 실사고로 분류된 건은 legal-hold로 자동만료에서 제외하고 증거묶음을 내보낸다. | P3 | +| **F14** | 컴플라이언스: 노사협의회 기록·안내판메타·DPIA·데이터주체 도구·얼굴블러를 제공한다. | P1 | +| **F15** | "현재 프레임 캡처" 명령으로 이벤트 전에도 구역 그리기·커미셔닝·탬퍼 기준 스냅샷을 얻는다. | P1 | + +### 3.2 비기능 요구사항 + +| ID | 요구사항 | +|----|----------| +| NF1 | 낙상 중요 카메라 유효 **≥12~15 fps**, 저위험 8~10 fps. fps 티어별 보장. | +| NF2 | 사고 발생→대시보드 표시 지연 **≤ 5초**(정상 네트워크). 위급 확인 SLA 별도(§8). | +| NF3 | Pi 소프트웨어 **오프라인 설치**(모든 `.hef`·의존성 사전 번들; 폐쇄망). 업데이트는 **서명 검증 오프라인 번들**. | +| NF4 | 파이프라인 멈춤/카메라 끊김/가림 자동 복구·감지(워치독·재연결·self-check). | +| NF5 | POST는 HTTPS + **HMAC + nonce/timestamp(replay 방지) + per-device 키**. | +| NF6 | 보관은 **계층형** + **env 설정**(하드코딩 금지): 일상/오탐 이벤트 보존일·실사고 legal-hold 기간 모두 환경변수. 기본값 TBD(법률 확정 시 조정). | +| NF7 | 대시보드 **인증 필수**(argon2id·세션 하드닝·RBAC·brute-force 잠금). 무인증 접근 0. | +| NF8 | 사고 **정확도 목표**(낙상 recall, 오탐/카메라/일, 에스컬레이션 지연)를 정의하고 커미셔닝에서 검증. | +| NF9 | 심각도는 **색상 단독 금지**(텍스트+아이콘+WCAG-AA+색맹안전). 데스크 오디오·모바일 push. | +| NF10 | 사고 시각은 **신뢰 시간원**(HW RTC + 로컬 NTP) + 서버 received_at 이중기록. | + +--- + +## 4. 아키텍처 (Architecture) + +2계층: **엣지(Pi)** 감지·판단·대응트리거, **클라우드(Cafe24)** 수신·인증·표시·대응관리·컴플라이언스. + +``` +┌──────────── 엣지: Raspberry Pi 5 + AI HAT+ (Hailo-8 26TOPS, PCIe Gen3, HW RTC) ────────────┐ +│ │ +│ [CSI캠]─(별프로세스)─┐ ┌─▶ EventEngine (상태머신 × 트랙/공간) │ +│ [USB캠 MJPEG]────────┼─▶ Capture ─▶ roundrobin ─▶ │ ├ FallDetector (OR로직, 시점강건) │ +│ [USB캠 MJPEG]────────┘ (정규화640) hailonet │ ├ ImmobilityWatchdog (자세무관 부동) │ +│ (공유 pose, └streamrouter│ ├ GhostIncident (트랙소실 공간사고) │ +│ batch≥N) → Tracker │ └ ZoneIntrusion (발끝/body-part) │ +│ (ByteTrack └────────────┬──────────────────────────┤ +│ +공간바인딩, │ │ +│ 지터스무딩) RingBuffer(前/中/後 프레임) │ +│ │ ▼ │ │ +│ Snapshotter+FaceBlur Alerter(POST+GPIO) │ +│ │ │ │ │ +│ LocalQueue◀─재시도─┘ HealthAgent │ +│ (암호화) (하트비트/워치독/ │ +│ │ 카메라 self-check)│ +│ Config(구역·임계값 per-cam/zone, 동기) │ +│ Secrets(TPM/keyring, per-device 키) │ +└────────────────────────────────┼───────────────────────────────────────┬───────────────────┘ + HTTPS POST(이벤트+JPEG, HMAC+nonce) GPIO │ + 하트비트(fps·온도·스로틀·큐·config버전) ▼ +┌──────────────── 클라우드: Cafe24 웹호스팅 (PHP + MySQL) ─────────────────┐ [경광등/부저] +│ [인증/RBAC] ── 세션하드닝 ── 접근로그 │ +│ Ingest API(HMAC+nonce검증, 이미지 재인코딩, rate limit) │ +│ │ │ +│ MySQL ── 스냅샷은 웹루트 밖 저장 → 인증 PHP 프록시로만 서빙 │ +│ ├─ 실시간 관제(사고 목록·상세, 심각도 텍스트+아이콘, 데스크오디오) │ +│ ├─ 대응(에스컬레이션 사다리·온콜 로스터·119 카드·NEW→CLAIMED→RESOLVED) │ +│ ├─ 위험구역 설정(폴리곤 드로잉) + 감도 다이얼(per-cam/zone) │ +│ ├─ 장비/플릿(사이트별, config버전, 온도/스로틀, 카메라 degraded) │ +│ ├─ 오탐 피드백(처리결과 라벨·재생·튜닝 제안) │ +│ ├─ 이력·통계·KPI + 증거묶음 내보내기(legal-hold, 산업재해조사표) │ +│ └─ 컴플라이언스(노사협의회 기록·안내판메타·DPIA·데이터주체·블러정책) │ +│ audit 로그(append-only 해시체인: 임계값·조회·내보내기·확인) │ +│ SMS 발송(위급, leak_sms 재사용) · Config/구역 동기 API(버전확인) │ +│ 모바일 PWA + Web Push (high/warning) · SMS(critical 폴백) │ +└────────────────────────────────────────────────────────────────────────────┘ +``` + +### 4.1 단일 공유모델 원칙 (유지) + +모든 카메라가 단일 `yolov8s_pose` HEF를 공유하고, 사람 박스·발끝/관절(구역·끼임)·관절 기하(낙상)·관절 변위(미동)를 **같은 pose 출력에서 파생**. 2모델 동시는 컨텍스트스위치 손해라 회피. + +--- + +## 5. 엣지 컴포넌트 + +| 컴포넌트 | 책임 | 신규/변경 | +|---|---|---| +| **Capture** | CSI=picamera2(카메라당 별 프로세스), USB=v4l2 MJPEG. 640 정규화. "현재프레임 캡처" 명령 지원. | +F15 | +| **Inference** | 공유 `yolov8s_pose` HEF. roundrobin→hailonet(batch≥N)→streamrouter. **fps 3티어**(§10.2). | +티어 | +| **Tracker** | ByteTrack/OC-SORT. **공간(IoU) 바인딩**으로 ID스위치 시 상태·타이머 승계. **per-keypoint One-Euro/Kalman 스무딩**. | 변경 | +| **EventEngine** | 트랙/공간 상태머신 4서브: FallDetector·ImmobilityWatchdog·GhostIncident·ZoneIntrusion(+body-part). | 대폭 | +| **RingBuffer** | 사전 프레임 링버퍼 → 사고 前/中/後 스냅샷 공급. 온디바이스 키포인트 시계열 기록. | 신규 | +| **Snapshotter+FaceBlur** | 링버퍼에서 1~3장 JPEG. **배경인물 얼굴 블러 default-ON**(피재자 제외 옵션). | +블러 | +| **Alerter** | 이벤트+스냅샷 POST(HMAC+nonce). 위급 시 GPIO + SMS 트리거. | +nonce | +| **LocalQueue** | 실패 시 **암호화(LUKS)** SQLite 큐, 백오프 재전송. | +암호화 | +| **Config** | 구역·임계값 **per-(device,camera,zone)** 로컬 관리, 클라우드 동기 + 적용버전 보고. | 변경 | +| **Secrets** | HMAC/SMS 키를 **TPM/keyring**, per-device. 평문파일 금지. | 신규 | +| **HealthAgent** | 하트비트(fps·온도·스로틀·큐깊이·config버전). 프레임정체·카메라끊김 복구. **카메라 가림/초점/조명/탬퍼 self-check → `monitoring_degraded`**. | 대폭 | + +--- + +## 6. 이벤트 판정 로직 + +모든 픽셀 임계는 **사람 bbox 높이로 정규화**. 단 이는 스케일만 보정 — 렌즈왜곡·마운트유형(탑다운 vs 사각)은 **per-camera 캘리브레이션 프로파일**로 보완(§16 캘리브 도구). 임계는 실장비 튜닝 전 **시작값**, per-cam/zone 조정 가능. + +### 6.1 트랙/공간 상태머신 (기절 경로 분리) + +``` + ┌──────────── ImmobilityWatchdog (자세 무관, 독립) ───────────┐ + │ 임의 트랙이 관절변위<임계 N초 지속 → "이상정지" 저·중경보 │ + └──────────────────────────────────────────────────────────────┘ + +STANDING/SITTING ──(급격 수직하강 or 낙상 전이)──▶ FALLING ──(prone >3s)──▶ FALLEN + ▲ │ │ + │(자세복원, F3) │(복원) │(관절변위<임계 ≥10s) + └──────────────────────────────────────────────┴─────────────────────────┤ + ▼ + NO_MOTION (위급/기절) + [임의 상태에서 트랙 소실(복원확인 없이)] ──▶ GhostIncident(마지막 위치) ──(T초 내 미재출현)──▶ 상향 +``` + +- **핵심 변경:** 위급(기절) 경보는 이제 **FALLEN 경로 전용이 아니다.** ImmobilityWatchdog가 자세·트랙연속성과 무관하게 병렬로 돈다. 좌위 실신·기대 슬럼프도 잡힘. +- **incident 활성 중 baseline freeze:** §6.5 재보정이 넘어진 자세를 "정상"으로 재앵커하지 않도록, 임의 incident 상태 동안 정상자세 기준을 동결. + +### 6.2 넘어짐 (Fall) — F1, 시점 강건 (OR 로직) + +각도 = 몸통(어깨중심→엉덩이중심) 벡터가 수직축과 이루는 각. 서있음≈0°, 누움≈90°. + +**FALLING 진입(전이):** 몸통각도 ~45° 상향 돌파 **AND** COM 하강속도·낙차 유의미(10프레임 윈도우, 정규화). + +**FALLEN 확정 — 직교 단서 OR 결합(정면/후면 낙상 대응):** + +- 다음 중 **하나 이상** 확신 + prone > 3초(20프레임 다수결): + - 몸통각도 수직대비 **> 70°**(측면 낙상), **또는** + - bbox **수직크기(높이) 급붕괴**(정면/후면 낙상 — 각도·종횡비 안 변해도 키가 확 줄음), **또는** + - COM 하강속도 큼 + 바닥권 정지, **또는** + - 머리 y > 엉덩이 y **또는** 종횡비 W/H > 1.25. +- **AND 게이트 폐기 이유:** 카메라 향해/등지고 넘어지면 몸통벡터 단축·종횡비 불변 → v1.0의 AND 조합은 한 방향 통째 미탐이었음. +- 관절 신뢰도 게이트: conf<0.5 폐기, 핵심 4관절 필수, 가시 ≥10. + +### 6.3 미동/기절 (No-motion) — F2, F3, F3b + +- **낙상 후 미동:** FALLEN 진입 시 타이머, 관절변위<임계 ≥10초 → NO_MOTION 위급(대시보드+SMS+부저). +- **자세무관 부동(F3b):** ImmobilityWatchdog가 임의 트랙에 대해 변위<임계 지속(좌위 90~180초, 위험구역 더 짧게) → 저·중 확인경보 → 미해소 상향. +- **지터 대응:** INT8 pose 관절은 FP보다 노이즈 큼 → per-keypoint 스무딩, 미동 임계 = **측정된 정지지터 + 마진**(무한 미발화 방지). +- 움직임 재개 → 취소·하향. 위급 후 쿨다운. + +### 6.4 위험구역 침입 — F4 / 끼임 body-part — F4b(P2) + +- **일반 침입(F4):** 발목 관절점 point-in-polygon. 경고구역(노랑)/절대금지(주황+옵션 부저). 디바운스 0.5~1초. +- **끼임 body-part(F4b, P2):** 기계 위험점 구역은 **임의 관절점**(손목·팔꿈치 등) 진입으로 판정. 손/팔이 프레스·롤러 위험점에 들어가는 상황 커버. point-in-polygon 소확장. +- 구역은 대시보드에서 스냅샷 위 폴리곤 드로잉. (E-stop 능동개입은 훅만, §16) + +### 6.5 공통 억제 규칙 + +- (트랙ID 또는 공간영역)당 1회 + 유형별 쿨다운. GhostIncident는 공간영역 기준. +- 저신뢰 검출 판정 제외. **incident 활성 중 baseline 동결**(§6.1). + +--- + +## 7. 클라우드 (Cafe24 PHP + MySQL) — 하드닝 + +> Cafe24 공유호스팅 유지(사용자 결정). 공유호스팅 한계를 인정하되 **내부 하드닝 최대**로. 스냅샷 웹루트 밖 + 프록시, 인증/RBAC, replay 방지가 필수. + +### 7.1 인증 / 접근제어 (신규, P1) + +- **로그인 필수**(argon2id, HttpOnly+Secure+SameSite 쿠키, CSRF 토큰, brute-force 잠금, HTTPS/HSTS). +- **RBAC:** admin(설정·임계값·컴플라이언스) / operator(확인·대응) / viewer(조회). 사이트 스코프. +- `acknowledged_by`는 인증 주체에 바인딩. 모든 조회/내보내기 접근로그(PIPA). + +### 7.2 화면 (Dashboard) + +| 화면 | 기능 | +|---|---| +| **실시간 관제** | 이벤트 실시간 목록, 미확인 상단고정, 심각도 = **텍스트(위급/높음/경고)+아이콘+색(색맹안전)**, **데스크 Web Audio**(위급 반복·무음불가). | +| **사고 상세** | 스냅샷 1~3장(前/中/後), 카메라·시각·구역·트랙ID·지속, 처리결과 라벨(실제/오탐/…), 확인처리(인증). | +| **대응** | 확인 SLA 타이머, 에스컬레이션 사다리(2/3차 온콜·음성콜), 온콜 로스터/교대, **119 카드**(위치도·주소·게이트코드·AED·응급처치), NEW→CLAIMED→RESOLVED. | +| **위험구역 설정** | 최신/현재캡처 스냅샷 위 폴리곤 드래그, 경고/금지/끼임 지정, Pi 동기. | +| **감도** | (device/camera/zone)별 프리셋 Conservative↔Balanced↔Aggressive + 고급 raw 슬라이더. | +| **장비/플릿** | 사이트별 Pi·카메라 상태, config 적용버전(drift 표시), 온도/스로틀/fps, `monitoring_degraded`. | +| **오탐 피드백** | 처리결과 라벨 집계, 키포인트 시계열 재생, 임계 튜닝 제안. | +| **이력·통계** | 날짜·구역·유형 이력, KPI(오탐률·MTTA), 히트맵, **증거묶음 내보내기**(legal-hold, 산업재해조사표 프리필). | +| **컴플라이언스** | 노사협의회 협의 기록, 안내판/처리방침 메타, DPIA 체크리스트, 데이터주체 조회/삭제, 얼굴블러 정책. | +| **모바일** | PWA 반응형 + Web Push(high/warning), SMS(critical 폴백), 딥링크. | + +### 7.3 데이터 계약 (Ingest API) — 하드닝 + +`POST /api/events` (JSON+base64 또는 multipart), 헤더 `X-Signature: HMAC-SHA256(nonce+ts+body, per_device_secret)`, `X-Nonce`, `X-Timestamp`. + +- **replay 방지:** nonce 1회성 + timestamp 윈도우 검증. +- **이미지:** 서버가 GD/Imagick로 **재인코딩**(페이로드 제거, 강제 .jpg, 랜덤 UUID 파일명), **웹루트 밖** 저장, **인증 PHP 프록시로만** 서빙(IDOR 차단). +- **SQL:** prepared statement + event_type/severity **화이트리스트** enum. 출력 인코딩 + CSP(저장 XSS). +- **남용:** per-device rate limit + 크기/inode 쿼터(위조 홍수로 스토리지·대시보드 다운 방지). + +```json +{ + "device_id": "pi-warehouse-01", + "camera_id": "cam-02", + "event_type": "fall | no_motion | immobility | ghost | zone_warning | zone_restricted | pinch | monitoring_degraded | device_fault", + "severity": "critical | high | warning | info", + "track_id": 42, + "zone_id": "zone-loading-dock", + "occurred_at": "2026-07-11T09:15:32+09:00", + "received_at": "(서버 기록)", + "duration_s": 12.4, + "disposition": "unset | real_emergency | false_alarm | handled_onsite | drill | needs_review", + "is_test": false, + "metrics": { "torso_angle": 78.2, "height_collapse": 0.55, "immobility_s": 10.1 }, + "snapshots": ["", "", ""] +} +``` + +`POST /api/heartbeat` — device_id, 카메라별 상태·fps·온도·스로틀·큐깊이·config버전. +`GET /api/config?device_id=` — 구역·임계값(per-cam/zone) 최신본; Pi가 적용버전 회신. + +### 7.4 DB 스키마(개략) + +- `users(user_id, role, site_scope, password_hash, ...)`, `sites(site_id, name, address)` +- `devices(device_id, site_id, name, last_heartbeat, temp, throttle, config_version, status)` +- `cameras(camera_id, device_id, name, role[fall_critical|zone|low], last_frame_at)` +- `zones(zone_id, camera_id, level[warning|restricted|pinch], polygon_json, updated_at)` +- `events(event_id, device_id, camera_id, event_type, severity, track_id, zone_id, occurred_at, received_at, duration_s, disposition, is_test, legal_hold, metrics_json, acknowledged_by, acknowledged_at, lifecycle_state)` +- `event_snapshots(snapshot_id, event_id, file_uuid, content_hash, created_at)` — 웹루트 밖. +- `thresholds(device_id, camera_id, zone_id, key, value)` — per-cam/zone. +- `escalations(event_id, level, recipient, sent_at, delivered, responded_at)`, `oncall(site_id, shift, recipient, priority)` +- `audit_log(id, actor, action, target, before, after, ts, prev_hash, hash)` — append-only 해시체인. +- `compliance(site_id, nosa_record_json, dpia_json, signage_meta, subject_requests_json)` +- `retention_policy(data_class, days, legal_hold_years)` + +--- + +## 8. 신뢰성 / 대응 / 에러 처리 + +- **오프라인 큐:** 실패 시 암호화 SQLite 저장, 백오프 재전송. 유실 0. +- **하트비트:** 30초 주기(fps·온도·스로틀·큐·config버전). 90초 무신호 → "감시 중단" 경보. +- **카메라 재연결 + self-check:** 끊김 자동 재오픈; 가림/초점흐림/조명꺼짐/탬퍼 → `monitoring_degraded`(다운과 구분). active-hours 스케줄로 off-shift 블라인드는 정상 처리. +- **워치독:** 프레임 정체 → systemd watchdog 재시작. +- **대응 에스컬레이션(F9):** 위급 미확인 SLA 초과 → 2/3차 온콜·음성콜 폴백 → 119 카드 → 지방노동관서 보고 초안. SMS 전송영수증 추적·미전달 재라우팅. +- **경보 이중화:** 위급 = 대시보드+SMS+부저+데스크오디오 동시. +- **신뢰 시간:** HW RTC + 로컬 NTP; 서버 received_at 병기; skew 플래그. + +--- + +## 9. 개인정보 / 보안 + +### 9.1 데이터 / 프라이버시 +- 클라우드엔 **사고 스냅샷만**(라이브뷰·영상 없음). 배경인물 **얼굴블러 default-ON**. +- **계층형 보존(NF6):** 일상/오탐 30~90일, 실사고 **legal-hold 3~5년**(산안법 §164 대응; 정확 값 법률 확정 §16). 자동삭제가 중대재해 증거 파기하지 않도록 legal-hold 우선. +- 목적 제한(근태감시 전용 차단), 데이터주체 조회/삭제, 모든 조회/내보내기 접근로그. + +### 9.2 애플리케이션 보안 +- 대시보드 인증/RBAC/세션하드닝(§7.1). ingest HMAC+nonce+per-device 키(§7.3). +- 스냅샷 웹루트 밖 + 프록시, 이미지 재인코딩, prepared statement, CSP, rate limit. +- **위변조방지 audit**: 임계값·구역·확인·조회·내보내기 전부 append-only 해시체인(누가·언제·전·후). 감도 몰래 낮춰 사고 은폐 탐지. + +### 9.3 엣지 / 공급망 +- 큐/스냅샷 파티션 **LUKS 암호화**, 시크릿 **TPM/keyring**(평문 금지), 함체 탬퍼 GPIO → 보안이벤트. +- SSH 키전용·root 비활성·egress 방화벽(Cafe24만). TLS 엄격검증 + **인증서 핀닝 fail-closed**. +- 모든 vendored 의존성·`.hef` **SHA256/GPG 검증**(빌드+로드시) + SBOM. (참고 fall repo의 Azure 문자열 유출 사례 = 공급망 리스크 근거.) + +### 9.4 시크릿 관리 +- HMAC·SMS·DB 키 환경변수/시크릿 매니저. 소스·아티팩트 하드코딩 금지. per-device 키 회전/폐기. + +--- + +## 10. 하드웨어 · 모델 · 런타임 (조사 근거) + +### 10.1 모델 +- **주 모델 `yolov8s_pose`**(640×640, 17 관절). Hailo-8 pose HEF 라이트. 세 이벤트 전부 여기서 파생. +- Hailo HEF 배포 → 런타임 **ultralytics(AGPL) 비의존**. + +### 10.2 fps 3티어 폴백 (신규, P1 설계 의무) + +미검증 pose fps 리스크에 **제품적 답**. 카메라 `role`이 NPU 스케줄링 구동: + +| 티어 | 처리 | 용도 | 부하 | +|---|---|---|---| +| **T1 full-pose** | 관절17 → 낙상+미동+구역+끼임 | fall_critical 카메라 | 최대 | +| **T2 detection-only** | 사람박스+발끝앵커 → 구역+거친 낙상(종횡비)만 | zone 카메라 | 중 | +| **T3 motion-occupancy** | 움직임/점유만 → 부동·구역 진입만 | low-priority | 최소 | + +- **PCIe Gen3 강제**(`dtparam=pciex1_gen=3`), `batch-size ≥ N`, 640 유지. +- **실병목 = Pi CPU**(파이프라인당 ~1코어). 후처리 C/GStreamer, Python 콜백 최소. +- 3티어 전부 **P1에서 실장비 검증** → 우아한 저하 제품 확보 후 고객약속. 벤치 결과 → cameras-per-Pi·Pis-per-site·BOM·가격 산정. +- **스로틀 텔레메트리:** 뜨거운 밀폐 함체 스로틀이 fps 저하 → 온도·스로틀 하트비트로 티어 신뢰. + +### 10.3 카메라 캡처 +- CSI: picamera2, **카메라당 별 프로세스**. USB: v4l2 **MJPEG 필수**, **2개 USB3 컨트롤러 분산** + 전원 허브. + +### 10.4 런타임 스택 +- **hailo-rpi5-examples(MIT) + hailo-apps-infra + GStreamer.** `N소스 → hailoroundrobin → 단일 hailonet(공유 pose, batch≥N) → hailostreamrouter → 캠별 추적/룰`. + +### 10.5 참조 BOM / 사이트 준비 (P2) +- 카메라 SKU·렌즈/FoV, **IP5x+ 방열·필터 함체**(동작온도 명시), 마운트/조준 가이드, 전원 허브, **소형 UPS**(정전 라이드), 네트워크 업링크/페일오버. "현재프레임 캡처"(F15)로 신규캠 구역 그리기 chicken-egg 해소. + +--- + +## 11. 재사용 OSS 및 라이선스 + +| 용도 | 프로젝트 | 라이선스 | 방식 | +|---|---|---|---| +| 플러밍 뼈대 | hailo-ai/hailo-rpi5-examples | MIT | fork | +| 추적+체류 참조 | Seeed YOLOv11-Hailo-Tracker | MIT | ByteTrack+dwell → 구역/미동 타이머 | +| 구역/끼임 | roboflow/supervision | MIT | PolygonZone 앵커(발끝/임의관절) | +| 구역 드로잉 UI | roboflow/polygonzone | MIT | 참조 | +| 추적기 | ByteTrack / OC-SORT | MIT | 기본/대안 | +| 키포인트 스무딩 | One-Euro filter | 공개 알고리즘 | 자체 구현 | +| 낙상 휴리스틱 | InvictusRex, Y-B-Class 등 | 무라이선스 | 알고리즘만 참고·재구현 | + +**라이선스 함정:** `ultralytics`(AGPL) 런타임 배제 → Hailo HEF + 자체 로직. 무라이선스 repo = 알고리즘 참고만. (`16dina/fall-detection` Azure 문자열 유출 블록 복사 금지.) + +--- + +## 12. 테스트 / 검증 전략 + +- **유닛:** 상태머신(전이·부동 워치독·유령사고·디바운스·쿨다운·baseline freeze) 합성 시퀀스, 80%+ 코드커버리지. +- **탐지 정확도(NF8, 신규):** **라벨 평가셋** 구축 — 공개 낙상 코퍼스 + 매트 연출낙상 + **강한 하드네거티브**(앉음·깊은 숙임·의도적 눕기·무거운 물건 낙하·2인 겹침) + 정면/후면·가림·기어감·프레임가장자리 케이스. 목표: 낙상 recall, 오탐/카메라/일, 에스컬레이션 지연. +- **커미셔닝 게이트(신규):** **드릴 모드**로 실장착 카메라별 연출낙상 → pass/fail 리포트. `is_test=true`로 실제 119/SMS 억제. **go-live 전제 + 중대재해처벌법 due-diligence 증거.** +- **부하:** 4캠 라운드로빈 fps·CPU·온도·지연 실측(티어별). +- **E2E:** 모의사고 → 스냅샷 → Cafe24 수신(인증/HMAC) → 대시보드 → 에스컬레이션 → (위급) SMS. +- **오프라인:** 네트워크 차단 설치·구동(NF3), POST 실패→큐→복구 재전송. +- **보안:** 인증/RBAC·replay·IDOR·업로드·SQLi·rate limit 침투 테스트. + +--- + +## 13. 단계별 로드맵 (리뷰 반영 재배치) + +- **P1 (MVP + 안전 필수):** 1캠 낙상 + **자세무관 부동(F3b)** + **유령사고(F3c)** + 스냅샷 POST. **시점강건 낙상(OR)·지터스무딩·baseline freeze.** **대시보드 인증/RBAC(F7,NF7)** + **ingest 하드닝(NF5).** **라벨 평가셋 + 드릴 커미셔닝 게이트(NF8).** **fps 3티어 검증(§10.2).** **컴플라이언스 최소본(F14: 노사협의회·안내판·블러·DPIA).** 색맹안전 심각도(NF9). "현재프레임 캡처"(F15). 위급 최소 에스컬레이션. +- **P2:** 멀티캠(2~4) + **끼임 body-part(F4b)** + 위험구역 2단계·감도 다이얼(F11) + 오탐 피드백·링버퍼(F10) + 카메라 self-check(F12) + 신뢰시간(NF10) + 경량 멀티사이트 플릿 + 플릿 관측성·물리내구 + BOM/함체/UPS. +- **P3:** 대응 에스컬레이션 풀(F9) + 증거 무결성·legal-hold·audit(F13) + 엣지 하드닝(LUKS/TPM/탬퍼) + 서명 오프라인 OTA + SMS/GPIO 이중화 + 이력·통계·KPI. +- **P4 (선택/범위결정):** 떨어짐·부딪힘·밀폐공간(외부센서) · 끼임 E-stop 능동개입 · 행동인식(ST-GCN) · 대규모 다중 Pi 집계 · RTSP 레트로핏 수신. + +--- + +## 14. 미해결 리스크 / 실장비 검증 체크리스트 + +1. **[최우선] Pi `yolov8s_pose` 엔드투엔드 fps 미검증.** 3티어 전부 실장비 벤치 → 캠 수·BOM 확정. +2. **PCIe Gen3 안정성** — 지속부하·발열 링크 안정. +3. **멀티 CSI 불안정** — 카메라당 별프로세스, 장시간 CSI+USB 동시 검증. +4. **USB UVC 대역폭** — MJPEG, 2 USB3 분산. +5. **마운팅 각도별 휴리스틱 오탐** — per-camera 캘리브 + 현장영상 튜닝. +6. **저fps 낙상 누락** — 낙상 중요캠 ≥12~15fps, 속도 휴리스틱 발화 확인. +7. **INT8 pose 정확도/지터** — recall·미동 임계에 영향, 평가셋으로 측정. +8. **가림/ID스위치** — 공간바인딩·유령사고 효과 실측. +9. **Cafe24 공유호스팅 하드닝 한계** — 웹루트밖·프록시·rate limit로 완화, 침투 테스트로 검증. +10. **폐쇄망 패키징** — `.hef`·휠·PHP 의존성 번들, 차단 설치 검증, HailoRT↔Compiler 버전 정렬. +11. **열/스로틀** — 함체 방열 검증, 스로틀 텔레메트리. +12. **SD 마모** — SSD/NVMe 부팅 + WAL로 "유실0" 보전. + +--- + +## 15. 설정 파라미터 레퍼런스 (시작값, per-cam/zone 조정) + +| 파라미터 | 시작값 | 비고 | +|---|---|---| +| 몸통각도(FALLING 전이) | ~45° 상향돌파 + 하강속도 | 서있음0° 누움90° | +| FALLEN 확정 (OR) | 각도>70° **또는** 높이급붕괴 **또는** COM하강 **또는** 머리<엉덩이/AR>1.25 | 시점강건 | +| bbox 종횡비 | > 1.25 (Δ>0.35 급변) | | +| 관절 신뢰도 | conf ≥ 0.5, 핵심4·가시≥10 | | +| 낙상 확정 지속 | prone > 3초 (20프레임 다수결) | | +| 낙상후 미동 | 관절변위<임계 ≥10초 | 5~15초 | +| 자세무관 부동(F3b) | 좌위 90~180초 / 위험구역 더 짧게 | | +| 미동 임계 | 측정 정지지터 + 마진 | 지터 무한미발화 방지 | +| 유령사고 재출현 대기(T) | 5~15초 | 미재출현 시 상향 | +| 구역 침입 디바운스 | 0.5~1초 | | +| 하트비트/다운 | 30초 / 90초 무신호 | | +| 보관 | 일상 30~90일 / 실사고 3~5년 legal-hold | 법률 확정 §16 | +| 낙상 중요캠 fps | ≥ 12~15 / 저위험 8~10 | | +| PCIe | Gen3 강제 | `dtparam=pciex1_gen=3` | +| 확인 SLA (위급) | 예: 60~120초 → 2차 에스컬레이션 | 온콜 구조 확정 §16 | + +--- + +## 16. 남은 미결정 (사용자/법률 확인 필요) + +- **보관기간:** env 설정으로 진행(정확값 TBD, 법률 확정 시 조정). 일상/실사고 legal-hold 모두 환경변수. 낙상/실신=민감정보 확정은 법률 확인. +- **끼임 E-stop 능동개입:** 안전PLC 인증·책임 감수하고 기계정지 출력? 아니면 detect-and-alert 한정? (v1은 훅만) +- **온콜/에스컬레이션 현실:** 각 사이트 24/7 온콜 구조? 119/노동관서 보고 연동 형태? +- **RTC BOM 비용:** 유닛마다 HW RTC + 로컬 NTP 추가 수용? +- **업데이트 물류:** 서명 번들을 사이트 업링크 vs USB 스니커넷? 서명·배포 권한자? +- **카메라 소싱:** 그린필드 CSI/USB 한정 vs 기존 IP/RTSP/ONVIF 수신 + 고객 VMS 이벤트 내보내기(레트로핏 세일즈)? +- **떨어짐/부딪힘/밀폐공간:** P4 착수 여부 및 우선순위(외부센서 필요). + +--- + +_문서 끝. 다음 단계: 사용자 리뷰 → writing-plans 스킬로 P1 구현 계획 수립._