move_detect/edge/detectors/height.py

"""추락(높이에서 떨어짐) 판정의 1차 신호 — 급격한 하강.

WHY THIS EXISTS
    이 제품이 잡고 있던 것은 같은 높이에서의 **넘어짐**이다. 그런데 산업재해 사망 1위 유형은
    **떨어짐**이고, 둘은 다른 사건이다. GhostIncidentDetector 는 트랙이 이미 위험 자세일 때만
    무장하므로(_DANGEROUS), 발판 위에 서 있다가 떨어지는 사람은 소실 순간 STANDING 이라
    아무것도 걸리지 않는다. 화각 밖으로 떨어지면 넘어진 자세를 볼 기회조차 없다.

WHY THE FOOT LINE (y2) AND NOT THE CENTRE
    이 파일에서 가장 중요한 판단이다.

        쭈그려 앉기 → 중심(cy)은 내려가지만 **발 밑선(y2)은 그 자리**다. 높이만 줄어든다.
        계단 내려가기 → 둘 다 천천히 내려간다.
        카메라 쪽 접근 → 둘 다 내려가지만 **사람이 커진다**.
        추락        → 발 밑선이 통째로, 빠르게 내려간다.

    중심으로 판정하면 쭈그려 앉아 작업하는 사람이 전부 추락이 된다. 원근 접근은 y2 도
    내려가므로 높이 증가 가드로 따로 막는다.

WHAT THIS DOES NOT DO
    이 감시자는 이벤트를 만들지 않는다. "이 트랙이 방금 떨어졌다"는 사실만 들고 있다가
    EventEngine 이 물어보면 답한다. 실제 발보는 두 갈래로 갈린다:
      · 프레임 안에서 넘어짐이 확정되면 -> 유형을 fall 대신 fall_from_height 로 승격
      · 화각 밖으로 사라지면          -> GhostIncidentDetector 가 무장해 발보
    한 번의 추락이 두 건으로 올라가지 않게 하려는 분리다.

    Track.frames 가 90프레임(약 7.5초) 이력을 들고 있어 별도 버퍼가 필요 없다.
"""
from dataclasses import dataclass

from edge.tracking import Track


@dataclass
class DescentConfig:
    # 평소 키의 몇 배를 내려가야 추락으로 보는가. 1.2배 ≈ 2m — 계단 한 층은 못 미친다.
    drop_ratio: float = 1.2
    # 그 하강이 이 시간 안에 일어나야 한다. 3m 자유낙하가 약 0.78초이므로 1.2초면 담긴다.
    drop_window_s: float = 1.2
    # 구간 동안 사람이 이 배수 이상 커졌으면 카메라 쪽으로 걸어온 것이다.
    grow_guard: float = 1.35
    # 하강과 착지를 같은 사건으로 묶는 허용 오차.
    #
    # 이 값을 "하강 후 now 까지"로 재면 안 된다. 넘어짐은 prone 이 prone_confirm_s(기본 3초)
    # 지속돼야 확정되므로, 확정 시점에는 하강으로부터 항상 3초 넘게 지나 있다 — 기본값끼리
    # 부딪혀 승격이 영원히 일어나지 않았다. 착지(prone 시작) 시각과 비교해야 한다.
    link_s: float = 3.0
    # 무장 유효 기간. 하강 한 번이 트랙의 남은 평생을 무장 상태로 두면, 몇 분 뒤 그냥
    # 걸어 나간 사람이 최상위 추락 경보가 된다.
    arm_ttl_s: float = 20.0


class DescentWatch:
    def __init__(self, cfg: DescentConfig):
        self.cfg = cfg
        self._seen: dict[int, tuple[float, dict]] = {}   # track_id -> (감지 시각, 상세)

    def forget(self, track_id: int) -> None:
        self._seen.pop(track_id, None)

    def armed_at(self, track_id: int) -> float | None:
        """이 트랙의 하강이 확정된 시각. 없으면 None."""
        hit = self._seen.get(track_id)
        return hit[0] if hit is not None else None

    def landed_from_drop(self, track_id: int, prone_since: float | None) -> bool:
        """이 트랙의 prone 시작이 하강 직후인가 — 즉 떨어져서 누운 것인가.

        now 가 아니라 **착지 시각**과 비교하는 것이 핵심이다. 확정은 prone_confirm_s 만큼
        늦게 오므로 now 로 재면 기본값에서 영원히 거짓이 된다.
        """
        hit = self._seen.get(track_id)
        if hit is None or prone_since is None:
            return False
        return abs(prone_since - hit[0]) <= self.cfg.link_s

    def armed(self, track_id: int, now: float) -> bool:
        """소실 시 추락으로 볼 만큼 최근에 떨어졌는가."""
        hit = self._seen.get(track_id)
        return hit is not None and (now - hit[0]) <= self.cfg.arm_ttl_s

    def descending(self, track_id: int, now: float) -> bool:
        """하위 호환용 별칭. 새 코드는 landed_from_drop / armed 를 쓴다."""
        return self.armed(track_id, now)

    def detail(self, track_id: int) -> dict | None:
        hit = self._seen.get(track_id)
        return dict(hit[1]) if hit is not None else None

    def update(self, track: Track, now: float) -> dict | None:
        """이번 틱에 하강이 확정되면 상세를 돌려준다. 같은 하강은 한 번만 확정한다."""
        frames = list(track.frames)
        if len(frames) < 2:
            return None
        if self.armed(track.track_id, now):
            return None   # 이미 무장돼 있다 — 같은 추락을 매 틱 다시 확정하지 않는다

        last = frames[-1]
        # 창 안에서 가장 오래된 프레임을 기준으로 삼는다. 그보다 앞은 다른 사건이다.
        start = None
        for f in frames:
            if last.t - f.t <= self.cfg.drop_window_s:
                start = f
                break
        if start is None or start is last:
            return None

        baseline = start.bbox.height
        if baseline <= 0:
            return None

        drop = (last.bbox.y2 - start.bbox.y2) / baseline
        if drop < self.cfg.drop_ratio:
            return None

        # 원근 접근 기각: 떨어지는 사람의 겉보기 크기는 그렇게 커지지 않는다.
        if last.bbox.height > baseline * self.cfg.grow_guard:
            return None

        info = {
            "drop_ratio": round(drop, 2),
            "drop_s": round(last.t - start.t, 2),
            "from_y2": round(start.bbox.y2, 3),
            "to_y2": round(last.bbox.y2, 3),
        }
        self._seen[track.track_id] = (now, info)
        return info