"""Load N named counting zones from environment variables. Env: ZONE_COUNT (default 1) ZONE_i_NAME ZONE_i_X1..Y2 (optional absolute pixels) ZONE_i_X1_FRAC..Y2_FRAC ZONE_i_COOLDOWN_SEC, ZONE_i_DWELL_SEC """ from __future__ import annotations import os from dataclasses import dataclass from typing import Mapping, Optional, Sequence @dataclass(frozen=True) class ZoneDef: id: str name: str x1: Optional[int] y1: Optional[int] x2: Optional[int] y2: Optional[int] x1_frac: float y1_frac: float x2_frac: float y2_frac: float cooldown_sec: float dwell_sec: float def resolve_rect(self, frame_width: int, frame_height: int): def coord(abs_val, frac, span): if abs_val is not None: return int(abs_val) return int(span * frac) x1 = coord(self.x1, self.x1_frac, frame_width) y1 = coord(self.y1, self.y1_frac, frame_height) x2 = coord(self.x2, self.x2_frac, frame_width) y2 = coord(self.y2, self.y2_frac, frame_height) return (min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2)) def _env_int(environ: Mapping[str, str], key: str) -> Optional[int]: raw = environ.get(key) if raw is None or str(raw).strip() == "": return None return int(raw) def _env_float(environ: Mapping[str, str], key: str, default: float) -> float: raw = environ.get(key) if raw is None or str(raw).strip() == "": return float(default) return float(raw) def load_zones( environ: Optional[Mapping[str, str]] = None, camera_name: Optional[str] = None, ) -> list[ZoneDef]: """Parse ZONE_COUNT + ZONE_i_* into ZoneDef list. Fail fast on invalid config.""" env = environ if environ is not None else os.environ cam = camera_name if camera_name is not None else env.get("CAMERA_NAME", "camera") raw_count = env.get("ZONE_COUNT", "1") try: zone_count = int(raw_count) except (TypeError, ValueError) as exc: raise ValueError(f"ZONE_COUNT must be an integer >= 1, got {raw_count!r}") from exc if zone_count < 1: raise ValueError(f"ZONE_COUNT must be >= 1, got {zone_count}") # Default geometry: single zone covers center-ish; multi-zone splits horizontally. defaults_by_count = { 1: { 1: (0.10, 0.10, 0.90, 1.00), }, } zones: list[ZoneDef] = [] for i in range(1, zone_count + 1): prefix = f"ZONE_{i}_" zone_id = f"zone_{i}" name = env.get(f"{prefix}NAME") or f"{cam}_zone_{i}" if zone_count == 1: dx1, dy1, dx2, dy2 = defaults_by_count[1][1] else: # Even horizontal slices across the frame. slice_w = 1.0 / zone_count gap = 0.02 dx1 = (i - 1) * slice_w + (gap if i > 1 else 0.0) dx2 = i * slice_w - (gap if i < zone_count else 0.0) dy1, dy2 = 0.10, 1.00 # Require at least frac defaults present logically — always have defaults. x1_frac = _env_float(env, f"{prefix}X1_FRAC", dx1) y1_frac = _env_float(env, f"{prefix}Y1_FRAC", dy1) x2_frac = _env_float(env, f"{prefix}X2_FRAC", dx2) y2_frac = _env_float(env, f"{prefix}Y2_FRAC", dy2) if not (0.0 <= x1_frac <= 1.0 and 0.0 <= x2_frac <= 1.0): raise ValueError( f"{prefix}X*_FRAC must be in [0,1], got x1={x1_frac} x2={x2_frac}" ) if not (0.0 <= y1_frac <= 1.0 and 0.0 <= y2_frac <= 1.0): raise ValueError( f"{prefix}Y*_FRAC must be in [0,1], got y1={y1_frac} y2={y2_frac}" ) if x1_frac == x2_frac and _env_int(env, f"{prefix}X1") is None: raise ValueError(f"{prefix} geometry has zero width (X1_FRAC == X2_FRAC)") if y1_frac == y2_frac and _env_int(env, f"{prefix}Y1") is None: raise ValueError(f"{prefix} geometry has zero height (Y1_FRAC == Y2_FRAC)") zones.append( ZoneDef( id=zone_id, name=name, x1=_env_int(env, f"{prefix}X1"), y1=_env_int(env, f"{prefix}Y1"), x2=_env_int(env, f"{prefix}X2"), y2=_env_int(env, f"{prefix}Y2"), x1_frac=x1_frac, y1_frac=y1_frac, x2_frac=x2_frac, y2_frac=y2_frac, cooldown_sec=_env_float(env, f"{prefix}COOLDOWN_SEC", 0.0), dwell_sec=_env_float(env, f"{prefix}DWELL_SEC", 2.0), ) ) return zones def zone_ids(zones: Sequence[ZoneDef]) -> list[str]: return [z.id for z in zones] def zone_meta(zones: Sequence[ZoneDef]) -> list[dict]: """Dashboard/API-friendly [{id, name}, ...].""" return [{"id": z.id, "name": z.name} for z in zones] def zones_by_id(zones: Sequence[ZoneDef]) -> dict[str, ZoneDef]: return {z.id: z for z in zones}