move_detect/edge/detectors/zone.py

from dataclasses import dataclass

from edge.tracking import Track
from edge.models import L_ANKLE, R_ANKLE

# Below this an ankle keypoint is treated as unusable (typically hallucinated at conf~0).
MIN_ANKLE_CONF = 0.2


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]]
    # 선택적 존 규칙 (occupancy.py가 소비): {"min_workers": 2, "max_workers": 4,
    # "max_dwell_s": 600} — 키가 없으면 그 규칙은 꺼진 것. None = 규칙 없음.
    rules: dict | None = None


@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 forget(self, track_id: int) -> None:
        self._inside_since = {k: v for k, v in self._inside_since.items() if k[0] != track_id}
        self._fired = {k for k in self._fired if k[0] != track_id}

    def _foot(self, track: Track, frame_wh=None) -> tuple[float, float]:
        p = track.latest
        a, b = p.keypoints[L_ANKLE], p.keypoints[R_ANKLE]
        # Average only the ankles the model is confident about. A hallucinated ankle (conf~0,
        # often pinned to (0,0) or a frame corner) would otherwise drag the foot point into or
        # out of a zone. With neither ankle trustworthy, fall back to the bbox bottom-centre — a
        # robust standing-foot proxy (same bbox-fallback discipline as fall.py's cues).
        good = [k for k in (a, b) if k.conf >= MIN_ANKLE_CONF]
        if good:
            fx = sum(k.x for k in good) / len(good)
            fy = sum(k.y for k in good) / len(good)
        else:
            fx, fy = p.bbox.cx, p.bbox.y2
        # Zones are stored NORMALIZED (0..1 of the image) by the dashboard, which stretches the
        # camera snapshot to a fixed canvas. The foot is camera-native pixels, so normalize it
        # by the frame size to compare in the same space — otherwise every zone is misplaced
        # (a zone drawn at image-centre would be checked against a raw pixel like 480, matching
        # nothing on a 1280-wide frame). frame_wh=None keeps raw coords for unit tests that
        # supply their own matching space.
        if frame_wh:
            w, h = frame_wh
            if w > 0 and h > 0:
                fx, fy = fx / w, fy / h
        return (fx, fy)

    def in_elevated_zone(self, track_id: int) -> bool:
        """이 트랙이 '고소' 표시된 구역 안에 있는가.

        rules_json 의 `elevated: true` 한 키로 붙는다 — 스키마 변경이 없다. 운영자가 발판·
        비계·개구부를 그려두면 추락 판정이 훨씬 정확해지지만, 구역이 없어도 하강 신호만으로
        동작한다. 위험은 운영자가 선언하되, 선언이 없어도 최소한은 본다.
        """
        active = self.active_zone_ids(track_id)
        if not active:
            return False
        for z in self.zones:
            if z.zone_id in active and (z.rules or {}).get("elevated"):
                return True
        return False

    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, frame_wh=None) -> list[dict]:
        fx, fy = self._foot(track, frame_wh)
        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