148 lines
4.8 KiB
Python
148 lines
4.8 KiB
Python
"""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}
|