move_detect/edge/beacon.py

"""현장 경광등·부저 — 엣지 노드가 스스로 울리는 유일한 통보 경로.

WHY LOCAL
    폐쇄망 현장에서는 FCM 도 Slack 도 쓸 수 없고, 사내 SMTP·웹훅조차 없는 곳이 있다. 게다가
    서버 경유 통보는 업링크가 살아 있을 때만 동작한다 — 정전이나 스위치 장애로 회선이 끊긴
    순간은 사고 위험이 오히려 높아지는 때다. GPIO 출력은 그 모든 것과 무관하게 동작한다:
    감지기가 이벤트를 만든 그 자리에서, 네트워크 없이, 수십 밀리초 안에.

    그래서 이것이 어떤 사이트든 **가장 먼저 설정해야 할 통보 경로**다. 서버측 채널
    (App\\Alert\\Channels)은 이 위에 얹는 보조 수단이다.

설정 전달
    서버 /api/config 의 `beacons` 배열로 내려온다 (핀 번호, latch/pulse, 유지 시간,
    active_high). 관리 화면에서 켜고 끄므로 현장에서 edge.env 를 고칠 필요가 없다.

하드웨어 부재
    gpiozero / RPi.GPIO 가 없거나(개발 PC, CI) 핀 열기에 실패하면 NullBeacon 으로 조용히
    떨어진다. 경광등이 없다고 감지가 멈추면 안 된다 — 통보는 감지의 부가 기능이지 전제가
    아니다. 다만 그 사실은 로그로 남긴다: 아무도 모르게 경광등이 죽어 있는 것이 최악이다.

실장비 검증
    핀이 실제로 떨어지는지는 이 파일의 단위 테스트로 증명할 수 없다. tests/hardware/ 의
    OFF-HW 항목으로 남긴다 (docs/hardware-bringup.md).
"""

import time

_SEVERITY_RANK = {"info": 0, "warning": 1, "high": 2, "critical": 3}


class NullBeacon:
    """하드웨어가 없을 때의 무동작 구현. 인터페이스는 동일하다."""

    available = False

    def fire(self, severity: str, now: float) -> bool:
        return False

    def tick(self, now: float) -> None:
        pass

    def release(self) -> None:
        pass


class GpioBeacon:
    """핀 하나를 latch(고정) 또는 pulse(일정 시간 후 해제) 로 구동한다.

    `latch` 는 사람이 확인할 때까지 계속 울리는 대신 hold_s 뒤 자동 해제한다. 무한 latch 는
    현장에 사람이 갈 수 없는 시간대(야간·주말)에 아무도 끌 수 없는 소음이 되고, 그러면 다음
    주에 배선이 뽑혀 있다.
    """

    available = True

    def __init__(self, pin: int, *, mode: str = "latch", hold_s: float = 30.0,
                 active_high: bool = False, min_severity: str = "high",
                 backend=None, log=None):
        self.pin = int(pin)
        self.mode = mode if mode in ("latch", "pulse") else "latch"
        self.hold_s = max(0.5, float(hold_s))
        self.active_high = bool(active_high)
        self.min_rank = _SEVERITY_RANK.get(min_severity, 2)
        self._on_until: float | None = None
        self._log = log or (lambda m: None)
        self._backend = backend if backend is not None else _open_pin(self.pin, self.active_high)

    def _write(self, on: bool) -> None:
        # active_high=False 는 릴레이 모듈의 통상 배선(LOW 에서 도통)이다. 반대로 물린
        # 현장에서 경광등이 상시 켜져 있는 것을 설정으로 뒤집을 수 있어야 한다.
        self._backend.set(on if self.active_high else not on)

    def fire(self, severity: str, now: float) -> bool:
        if _SEVERITY_RANK.get(severity, 0) < self.min_rank:
            return False
        self._write(True)
        # 이미 울리는 중에 새 이벤트가 오면 유지 시간을 연장한다 — 사고가 이어지는 동안
        # 경광등이 중간에 꺼졌다 켜지면 사람이 "끝났다"로 읽는다.
        self._on_until = now + self.hold_s
        return True

    def tick(self, now: float) -> None:
        """루프가 매 틱 호출한다. 유지 시간이 지나면 해제."""
        if self._on_until is not None and now >= self._on_until:
            self._write(False)
            self._on_until = None

    def release(self) -> None:
        try:
            self._write(False)
        finally:
            close = getattr(self._backend, "close", None)
            if close:
                close()


class _PinBackend:
    """gpiozero / RPi.GPIO 중 있는 쪽에 얇게 붙는다."""

    def __init__(self, device=None, rpi=None, pin: int = 0):
        self._device = device
        self._rpi = rpi
        self._pin = pin

    def set(self, high: bool) -> None:
        if self._device is not None:
            if high:
                self._device.on()
            else:
                self._device.off()
        elif self._rpi is not None:
            self._rpi.output(self._pin, self._rpi.HIGH if high else self._rpi.LOW)

    def close(self) -> None:
        if self._device is not None:
            self._device.close()
        elif self._rpi is not None:
            self._rpi.cleanup(self._pin)


def _open_pin(pin: int, active_high: bool):  # pragma: no cover (hardware)
    try:
        from gpiozero import DigitalOutputDevice
        # initial_value=False 는 "비활성" 을 뜻한다. active_high 해석은 _write 가 담당하므로
        # 여기서는 항상 raw 출력으로 열고, 부팅 순간 경광등이 번쩍이지 않도록 초기값을 준다.
        dev = DigitalOutputDevice(pin, active_high=True, initial_value=not active_high)
        return _PinBackend(device=dev, pin=pin)
    except Exception:
        pass
    try:
        import RPi.GPIO as GPIO
        GPIO.setmode(GPIO.BCM)
        GPIO.setup(pin, GPIO.OUT, initial=GPIO.LOW if active_high else GPIO.HIGH)
        return _PinBackend(rpi=GPIO, pin=pin)
    except Exception:
        raise RuntimeError("no usable GPIO backend")


def build_beacons(config: dict, log=None) -> list:
    """/api/config 의 `beacons` 배열 -> 구동 가능한 비콘 목록.

    핀 하나가 열리지 않아도 나머지는 살린다. 전부 실패하면 빈 목록이고, 그때도 감지·전송은
    그대로 돈다.
    """
    log = log or (lambda m: None)
    out = []
    for b in (config or {}).get("beacons") or []:
        if not isinstance(b, dict):
            continue
        pin = int(b.get("gpio_pin") or 0)
        if pin <= 0:
            continue                      # 0 = 미배선 (관리 화면에서 핀을 아직 안 정한 상태)
        try:
            out.append(GpioBeacon(
                pin,
                mode=str(b.get("mode") or "latch"),
                hold_s=float(b.get("hold_s") or 30.0),
                active_high=bool(b.get("active_high")),
                min_severity=str(b.get("min_severity") or "high"),
                log=log,
            ))
        except Exception as e:
            # 조용히 넘어가면 아무도 경광등이 죽은 줄 모른다.
            log(f"beacon pin {pin} unavailable ({type(e).__name__}: {e}); alerts on this node "
                f"will not sound locally")
    return out


def fire_all(beacons, events, now: float | None = None) -> int:
    """이벤트 목록 중 가장 높은 심각도로 비콘을 울린다. @return 실제로 울린 비콘 수."""
    if not beacons or not events:
        return 0
    now = time.monotonic() if now is None else now
    worst = max((getattr(e.severity, "value", e.severity) for e in events),
                key=lambda s: _SEVERITY_RANK.get(s, 0))
    return sum(1 for b in beacons if b.fire(worst, now))