File size: 2,243 Bytes
e13bc33
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
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