| from __future__ import annotations |
|
|
| from dataclasses import dataclass |
|
|
|
|
| @dataclass |
| class ZoneEvent: |
| track_id: int |
| label: int |
| kind: str |
| frame_index: int |
| position: tuple[float, float] |
|
|
|
|
| def point_in_polygon(point: tuple[float, float], polygon: list[tuple[float, float]]) -> bool: |
| x, y = point |
| inside = False |
| n = len(polygon) |
| j = n - 1 |
| for i in range(n): |
| xi, yi = polygon[i] |
| xj, yj = polygon[j] |
| if (yi > y) != (yj > y) and x < (xj - xi) * (y - yi) / (yj - yi) + xi: |
| inside = not inside |
| j = i |
| return inside |
|
|
|
|
| class RestrictedZoneMonitor: |
| """Fires an event when a tracked object's ground position crosses a zone boundary. |
| |
| Consumes `Track` objects from `SortTracker.update()`. The zone is a polygon in |
| the same pixel coordinates as the tracked boxes (a rectangle is just a |
| four-point polygon). Uses each box's bottom-center point rather than its |
| centroid, since a zone drawn on a floor plane should key off where someone is |
| standing, not their torso. Only reports the entered/exited transition, not |
| every frame a track spends inside the zone. |
| """ |
|
|
| def __init__(self, polygon: list[tuple[float, float]]) -> None: |
| self.polygon = polygon |
| self._inside_state: dict[int, bool] = {} |
|
|
| def update(self, tracks, frame_index: int = 0) -> list[ZoneEvent]: |
| events: list[ZoneEvent] = [] |
| live_ids = set() |
| for track in tracks: |
| live_ids.add(track.id) |
| x0, y0, x1, y1 = track.box |
| ground_point = ((x0 + x1) / 2.0, y1) |
| is_inside = point_in_polygon(ground_point, self.polygon) |
| was_inside = self._inside_state.get(track.id, False) |
| if is_inside and not was_inside: |
| events.append(ZoneEvent(track.id, track.label, "entered", frame_index, ground_point)) |
| elif was_inside and not is_inside: |
| events.append(ZoneEvent(track.id, track.label, "exited", frame_index, ground_point)) |
| self._inside_state[track.id] = is_inside |
|
|
| for track_id in list(self._inside_state): |
| if track_id not in live_ids: |
| del self._inside_state[track_id] |
|
|
| return events |
|
|
