from __future__ import annotations import math from dataclasses import dataclass from datetime import datetime, timezone from typing import Callable, Iterable, Sequence from uuid import uuid4 @dataclass(frozen=True) class Point: x: float y: float def __post_init__(self) -> None: if not all(math.isfinite(value) for value in (self.x, self.y)) or not ( 0.0 <= self.x <= 1.0 and 0.0 <= self.y <= 1.0 ): raise ValueError("point coordinates must be within [0, 1]") @dataclass(frozen=True) class Box: x1: float y1: float x2: float y2: float def __post_init__(self) -> None: values = (self.x1, self.y1, self.x2, self.y2) if not all(math.isfinite(value) for value in values) or any(value < 0.0 or value > 1.0 for value in values): raise ValueError("box coordinates must be within [0, 1]") if self.x1 >= self.x2 or self.y1 >= self.y2: raise ValueError("box must have positive area") @property def center(self) -> Point: return Point((self.x1 + self.x2) / 2.0, (self.y1 + self.y2) / 2.0) @dataclass(frozen=True) class Detection: track_id: str class_name: str box: Box detector_score: float | None = None def __post_init__(self) -> None: if not self.track_id or len(self.track_id) > 128: raise ValueError("track_id must contain 1 to 128 characters") if self.class_name != "person": raise ValueError("T-017 only supports anonymous person detections") if self.detector_score is not None and not math.isfinite(self.detector_score): raise ValueError("detector_score must be finite") @dataclass(frozen=True) class Zone: zone_id: str name: str version: int points: tuple[Point, ...] def __post_init__(self) -> None: if not self.zone_id or len(self.zone_id) > 128: raise ValueError("zone_id must contain 1 to 128 characters") if not self.name.strip() or len(self.name) > 80: raise ValueError("zone name must contain 1 to 80 characters") if self.version < 1: raise ValueError("zone version must be positive") if not 3 <= len(self.points) <= 32: raise ValueError("zone must contain 3 to 32 points") @dataclass(frozen=True) class EventCandidate: source_event_id: str kind: str source_ref: str track_id: str zone_id: str zone_version: int occurred_at: str confidence: None fixture: bool def as_dict(self) -> dict[str, object]: # Bell owns the platform `id`; it is deliberately absent here. return { "candidate_version": "brain-demo-v1", "source_event_id": self.source_event_id, "kind": self.kind, "source_ref": self.source_ref, "track_id": self.track_id, "zone_id": self.zone_id, "zone_version": self.zone_version, "occurred_at": self.occurred_at, "confidence": self.confidence, "fixture": self.fixture, } def _on_segment(point: Point, start: Point, end: Point, epsilon: float = 1e-9) -> bool: cross = (point.y - start.y) * (end.x - start.x) - (point.x - start.x) * (end.y - start.y) if abs(cross) > epsilon: return False return ( min(start.x, end.x) - epsilon <= point.x <= max(start.x, end.x) + epsilon and min(start.y, end.y) - epsilon <= point.y <= max(start.y, end.y) + epsilon ) def point_in_polygon(point: Point, polygon: Sequence[Point]) -> bool: if len(polygon) < 3: return False inside = False previous = polygon[-1] for current in polygon: if _on_segment(point, previous, current): return True crosses = (current.y > point.y) != (previous.y > point.y) if crosses: boundary_x = (previous.x - current.x) * (point.y - current.y) / (previous.y - current.y) + current.x if point.x < boundary_x: inside = not inside previous = current return inside class ZoneEntryEvaluator: def __init__( self, source_ref: str, fixture: bool, track_ttl_frames: int = 8, event_id_factory: Callable[[], str] | None = None, ) -> None: if track_ttl_frames < 1: raise ValueError("track_ttl_frames must be positive") self._source_ref = source_ref self._fixture = fixture self._track_ttl_frames = track_ttl_frames self._event_id_factory = event_id_factory or (lambda: f"BRN-{uuid4().hex}") self._inside: dict[str, bool] = {} self._last_seen: dict[str, int] = {} def reset(self) -> None: self._inside.clear() self._last_seen.clear() def evaluate( self, sequence: int, occurred_at: datetime, zone: Zone, detections: Iterable[Detection], ) -> tuple[list[EventCandidate], dict[str, bool]]: if sequence < 0: raise ValueError("sequence cannot be negative") if occurred_at.tzinfo is None: raise ValueError("occurred_at must be timezone-aware") expired = [ track_id for track_id, last_seen in self._last_seen.items() if sequence - last_seen > self._track_ttl_frames ] for track_id in expired: self._inside.pop(track_id, None) self._last_seen.pop(track_id, None) events: list[EventCandidate] = [] states: dict[str, bool] = {} for detection in detections: inside = point_in_polygon(detection.box.center, zone.points) previous = self._inside.get(detection.track_id) if previous is False and inside: events.append( EventCandidate( source_event_id=self._event_id_factory(), kind="zone_entry", source_ref=self._source_ref, track_id=detection.track_id, zone_id=zone.zone_id, zone_version=zone.version, occurred_at=occurred_at.astimezone(timezone.utc).isoformat().replace("+00:00", "Z"), confidence=None, fixture=self._fixture, ) ) self._inside[detection.track_id] = inside self._last_seen[detection.track_id] = sequence states[detection.track_id] = inside return events, states def zone_from_payload(payload: object, current: Zone) -> Zone: if not isinstance(payload, dict): raise ValueError("request body must be an object") if set(payload) - {"name", "points"}: raise ValueError("unknown zone fields are not allowed") name = payload.get("name", current.name) raw_points = payload.get("points") if not isinstance(name, str): raise ValueError("zone name must be a string") if not isinstance(raw_points, list): raise ValueError("zone points must be an array") points: list[Point] = [] for raw_point in raw_points: if not isinstance(raw_point, dict) or set(raw_point) != {"x", "y"}: raise ValueError("each point must contain only x and y") x = raw_point["x"] y = raw_point["y"] if isinstance(x, bool) or isinstance(y, bool) or not isinstance(x, (int, float)) or not isinstance(y, (int, float)): raise ValueError("point coordinates must be numbers") points.append(Point(float(x), float(y))) return Zone(current.zone_id, name.strip(), current.version + 1, tuple(points))