from dataclasses import dataclass from edge.tracking import Track from edge.models import L_ANKLE, R_ANKLE def point_in_polygon(x: float, y: float, poly: list[tuple[float, float]]) -> bool: inside = False n = len(poly) j = n - 1 for i in range(n): xi, yi = poly[i] xj, yj = poly[j] if ((yi > y) != (yj > y)) and (x < (xj - xi) * (y - yi) / ((yj - yi) or 1e-9) + xi): inside = not inside j = i return inside @dataclass(frozen=True) class Zone: zone_id: str level: str # "warning" | "restricted" polygon: list # list[tuple[float,float]] @dataclass class ZoneConfig: debounce_s: float = 0.75 class ZoneIntrusionDetector: def __init__(self, zones: list[Zone], cfg: ZoneConfig): self.zones = zones self.cfg = cfg self._inside_since: dict[tuple[int, str], float] = {} self._fired: set[tuple[int, str]] = set() def forget(self, track_id: int) -> None: self._inside_since = {k: v for k, v in self._inside_since.items() if k[0] != track_id} self._fired = {k for k in self._fired if k[0] != track_id} def _foot(self, track: Track, frame_wh=None) -> tuple[float, float]: kps = track.latest.keypoints a, b = kps[L_ANKLE], kps[R_ANKLE] fx, fy = (a.x + b.x) / 2, (a.y + b.y) / 2 # Zones are stored NORMALIZED (0..1 of the image) by the dashboard, which stretches the # camera snapshot to a fixed canvas. The foot is camera-native pixels, so normalize it # by the frame size to compare in the same space — otherwise every zone is misplaced # (a zone drawn at image-centre would be checked against a raw pixel like 480, matching # nothing on a 1280-wide frame). frame_wh=None keeps raw coords for unit tests that # supply their own matching space. if frame_wh: w, h = frame_wh if w > 0 and h > 0: fx, fy = fx / w, fy / h return (fx, fy) def active_zone_ids(self, track_id: int) -> set: return {z for (tid, z) in self._inside_since if tid == track_id} def update(self, track: Track, now: float, frame_wh=None) -> list[dict]: fx, fy = self._foot(track, frame_wh) out = [] for zone in self.zones: key = (track.track_id, zone.zone_id) if point_in_polygon(fx, fy, zone.polygon): start = self._inside_since.setdefault(key, now) if now - start >= self.cfg.debounce_s and key not in self._fired: self._fired.add(key) out.append({"zone_id": zone.zone_id, "level": zone.level}) else: self._inside_since.pop(key, None) self._fired.discard(key) return out