move_detect/edge/zones.py

import json
import logging

from edge.detectors.zone import Zone

log = logging.getLogger(__name__)

# Zones are stored NORMALIZED (0..1 of the image) by the current dashboard. Polygons drawn by the
# pre-normalization dashboard were saved in canvas PIXELS (values in the hundreds); after the edge
# normalizes the foot point, a pixel polygon can never match — silently killing that zone's
# intrusion detection with no error. Flag such rows loudly so the operator redraws them, instead
# of failing dark. A normalized coordinate may round slightly past 1.0, so allow a small margin.
_NORMALIZED_MAX = 1.5


def _looks_like_pixels(polygon: list[tuple[float, float]]) -> bool:
    return any(abs(x) > _NORMALIZED_MAX or abs(y) > _NORMALIZED_MAX for x, y in polygon)


_RULE_INT_KEYS = ("min_workers", "max_workers", "max_dwell_s")

# 키마다 하한이 다르다. max_workers 만 0 이 유효하다 — "아무도 있으면 안 되는 구역"
# (크레인으로 든 짐 아래, 굴착기 선회 반경). 전 키에 v <= 0 을 걸면 그 규칙을 저장한 구역이
# 통째로 '규칙 없음'이 되고, 같이 넣은 2인1조까지 함께 사라진다.
_RULE_MIN = {"min_workers": 1, "max_workers": 0, "max_dwell_s": 1}

# 정수가 아니라 참/거짓인 규칙. 이 목록에 없으면 파서가 조용히 버린다 — elevated 를 여기
# 넣지 않아 고소작업 프리셋이 통째로 무동작이었다.
_RULE_FLAG_KEYS = ("elevated",)


def _rules_of(row: dict) -> dict | None:
    """Optional occupancy rules from the zone row's rules_json. A malformed rules blob must
    not kill the zone's INTRUSION detection (the polygon is still valid) — raise here and the
    caller keeps the zone but logs the rules loudly as inactive."""
    raw = row.get("rules_json")
    if raw in (None, "", "null"):
        return None
    data = json.loads(raw) if isinstance(raw, str) else raw
    if not isinstance(data, dict):
        raise ValueError(f"rules_json is not an object: {data!r}")
    rules = {}
    for key in _RULE_INT_KEYS:
        if data.get(key) is None:
            continue
        v = int(data[key])
        lo = _RULE_MIN[key]
        if v < lo:
            raise ValueError(f"{key} must be >= {lo}, got {v}")
        rules[key] = v
    for key in _RULE_FLAG_KEYS:
        if data.get(key) is None:
            continue
        rules[key] = bool(data[key])
    return rules or None


def zones_from_config(config: dict, camera_id: str | None = None) -> list[Zone]:
    """Parse server config zone rows into Zone objects. When camera_id is given, only
    rows for that camera are kept (multi-camera routing); None keeps all rows."""
    out: list[Zone] = []
    rows = config.get("zones", []) or []
    if not isinstance(rows, list):
        return []
    for row in rows:
        try:
            if camera_id is not None and str(row["camera_id"]) != str(camera_id):
                continue
            raw = row["polygon_json"]
            pts = json.loads(raw) if isinstance(raw, str) else raw
            polygon = [(float(p[0]), float(p[1])) for p in pts]
            if len(polygon) < 3:
                continue
            if _looks_like_pixels(polygon):
                log.warning(
                    "zone %s on camera %s has un-normalized (pixel-space) coordinates %r — it "
                    "would match nothing after foot normalization. Skipping; redraw this zone in "
                    "the dashboard so it is stored normalized (0..1).",
                    row.get("zone_id"), row.get("camera_id"), polygon[:2])
                continue
            try:
                rules = _rules_of(row)
            except (ValueError, TypeError, json.JSONDecodeError) as e:
                # Same "never fail dark" principle: the polygon is fine, so intrusion detection
                # stays live — but the occupancy rules the operator THINKS are active are not.
                log.warning("zone %s on camera %s has malformed rules_json (%s: %s) — its "
                            "occupancy rules are INACTIVE until fixed in the dashboard.",
                            row.get("zone_id"), row.get("camera_id"), type(e).__name__, e)
                rules = None
            out.append(Zone(str(row["zone_id"]), str(row["level"]), polygon, rules))
        except (KeyError, ValueError, TypeError, IndexError, AttributeError, json.JSONDecodeError) as e:
            # Same principle as the pixel-space branch above: never fail dark. A row missing a key
            # (level/zone_id), with unparseable polygon_json, or degenerate points would otherwise be
            # dropped silently — intrusion detection for that zone dead with no log, no dashboard
            # signal. Flag it so it can be fixed instead of discovered after a worker enters the zone.
            zid = row.get("zone_id") if isinstance(row, dict) else None
            log.warning("skipping malformed zone row %r (%s: %s) — its intrusion detection is "
                        "inactive until the row is fixed in the dashboard.", zid, type(e).__name__, e)
            continue
    return out


def thresholds_from_config(config: dict) -> dict:
    t = config.get("thresholds")
    return t if isinstance(t, dict) else {}