cw2007/move_detect
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
move_detect/docs/superpowers/plans/2026-07-11-p1a-edge-detection-core.md
유창욱 a4f2ead211 docs: workplace safety AI CCTV design spec v2.0 + P1-A edge core plan 
- 설계 스펙 v2.0 (다관점 리뷰 반영: 기절경로 분리, 시점강건 낙상, 인증/RBAC, 평가셋, fps 3티어, 컴플라이언스, 증거 무결성)
- P1-A 엣지 감지 코어 구현계획 (16 태스크 TDD)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ne4rU4FToTLHcBUMUvUnT1
2026-07-11 20:08:01 +09:00

76 KiB
Raw Blame History

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 작성

[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
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
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

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

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
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: 커밋
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

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
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: 커밋
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

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
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:

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: 커밋
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

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
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: 커밋
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

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
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: 커밋
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

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
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: 커밋
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

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
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: 커밋
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

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
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: 커밋
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

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
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: 커밋
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

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
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: 커밋
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

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
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: 커밋
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

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
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: 커밋
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

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
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: 커밋
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

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
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: 전체 스위트 + 커밋
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

"""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
# picamera_source.py
"""CSI 카메라 소스 (실장비). 카메라당 별 프로세스로 picamera2 구동."""


class PiCameraSource:
    def __init__(self, index: int = 0):
        self.index = index
        raise NotImplementedError("실장비 전용: picamera2 배선 필요")

    def read(self):
        raise NotImplementedError

# 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: 커밋
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.

어느 방식으로?