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 {}