218 lines
7.5 KiB
Python
218 lines
7.5 KiB
Python
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))
|