"""Line-crossing counter — hybrid zone-based state tracking. Counting logic (Low-FPS robust): Uses y1 (top edge) of the stabilized sack bounding box. Each track_id goes through states: UNKNOWN → ABOVE → COUNTED (when seen below line) UNKNOWN → BELOW (ghost/appeared below line first → never counted) Loading: track had state ABOVE, now detected BELOW the zone Unloading: track had state BELOW, now detected ABOVE the zone (if needed) 4-Layer deduplication: Layer 1: State guard — must have been ABOVE before counting Layer 2: Entry circle — must have exited the radius from first appearance Layer 3: Spatial dedup radius — same position can't trigger twice Layer 4: Track ID — one track_id can only be counted once per direction This approach is immune to low FPS because it doesn't require detecting the exact frame of crossing. It only needs the track to have been seen ABOVE the line at ANY point in its lifetime. """ from __future__ import annotations import time from src.interfaces import Detection class LineCrossCounter: """Counts sacks crossing a horizontal zone using y1 (top edge). The zone is a band [line_y - margin, line_y + margin]. A sack is "above" if y1 < line_y - margin, "below" if y1 > line_y + margin. While y1 is inside the band, state is held (no trigger). Loading = track was ever "above", now "below" (entered truck) Unloading = track was ever "below", now "above" (left truck) """ def __init__( self, line_y: int, line_x_start: int, line_x_end: int, margin: int = 20, dedup_radius: float = 60.0, ) -> None: self._line_y = line_y self._line_x_start = line_x_start self._line_x_end = line_x_end self._margin = margin self._dedup_radius = dedup_radius self._loading_count = 0 self._unloading_count = 0 # track_id -> zone state for y1: "above" | "below" | None self._state: dict[int, str | None] = {} # track_id -> whether this track has EVER been in each zone self._has_been_above: dict[int, bool] = {} self._has_been_below: dict[int, bool] = {} # track_id -> set of directions already counted self._counted: dict[int, set[str]] = {} # track_id -> initial coordinates (cx, y1) when first tracked self._entry_points: dict[int, tuple[float, float]] = {} # track_id -> whether this track has exited its entry circle # A track must move beyond dedup_radius from its first appearance # before it can be counted. This prevents ghost bboxes that suddenly # appear near the counting line from being counted immediately. self._has_exited_entry_circle: dict[int, bool] = {} # list of active deduplication circles self._dedup_circles: list[dict] = [] @property def entry_points(self) -> dict[int, tuple[float, float]]: return self._entry_points @property def counted_tracks(self) -> dict[int, set[str]]: return self._counted @property def line_y(self) -> int: return self._line_y @line_y.setter def line_y(self, value: int) -> None: self._line_y = value @property def line_x_start(self) -> int: return self._line_x_start @line_x_start.setter def line_x_start(self, value: int) -> None: self._line_x_start = value @property def line_x_end(self) -> int: return self._line_x_end @line_x_end.setter def line_x_end(self, value: int) -> None: self._line_x_end = value def update(self, detections: list[Detection]) -> list[dict]: """Process detections, return list of crossing events. Hybrid approach: - Tracks zone state per frame (above/below/in-band) - BUT uses accumulated history (has_been_above) for counting decision - A track counts as "loading" when: 1. It has been seen ABOVE the line at any previous point 2. It has exited its entry circle (moved beyond dedup_radius from first appearance) 3. Its current y1 is now BELOW the line 4. It hasn't been counted for loading yet 5. It passes spatial dedup check """ now_t = time.time() events: list[dict] = [] upper = self._line_y - self._margin lower = self._line_y + self._margin # Clean up expired dedup circles (older than 3.0 seconds) self._dedup_circles = [c for c in self._dedup_circles if (now_t - c["time"]) <= 3.0] for det in detections: if det.track_id is None: continue x1, y1, x2, y2 = det.bbox cx = (x1 + x2) / 2.0 tid = det.track_id if tid not in self._entry_points: self._entry_points[tid] = (cx, y1) self._has_exited_entry_circle[tid] = False # Check if sack has exited its entry circle (moved beyond dedup_radius # from where it first appeared). Once exited, stays exited permanently. if not self._has_exited_entry_circle.get(tid, False): entry_cx, entry_y1 = self._entry_points[tid] dist_from_entry = ((cx - entry_cx) ** 2 + (y1 - entry_y1) ** 2) ** 0.5 if dist_from_entry > self._dedup_radius: self._has_exited_entry_circle[tid] = True # Skip if centroid X outside counting bounds if cx < self._line_x_start or cx > self._line_x_end: continue counted_dirs = self._counted.setdefault(tid, set()) # Determine y1 zone state (top edge of sack bbox) if y1 < upper: new_state = "above" elif y1 > lower: new_state = "below" else: new_state = self._state.get(tid) # in band: hold prev_state = self._state.get(tid) self._state[tid] = new_state # Track zone history — CRITICAL for low-FPS robustness # Once a track has been seen above/below, it stays recorded forever if new_state == "above": self._has_been_above[tid] = True elif new_state == "below": self._has_been_below[tid] = True # --- HYBRID COUNTING LOGIC --- # Loading: track was EVER above, NOW below (entered truck from top) # AND has exited its entry circle (proven real movement, not ghost bbox) # This works even if the track jumped over the line between frames has_exited = self._has_exited_entry_circle.get(tid, False) is_loading = ( new_state == "below" and self._has_been_above.get(tid, False) and has_exited and "loading" not in counted_dirs ) # Unloading: track was EVER below, NOW above (left truck) is_unloading = ( new_state == "above" and self._has_been_below.get(tid, False) and has_exited and "unloading" not in counted_dirs ) if is_loading or is_unloading: # Check spatial distance against all active dedup circles is_duplicate = False for circle in self._dedup_circles: dist = ((cx - circle["x"]) ** 2 + (y1 - circle["y"]) ** 2) ** 0.5 if dist <= self._dedup_radius: is_duplicate = True break if is_duplicate: continue # Add this coordinate to the active dedup circles self._dedup_circles.append({ "x": cx, "y": y1, "time": now_t, "track_id": tid }) if is_loading: self._loading_count += 1 counted_dirs.add("loading") events.append({ "track_id": tid, "direction": "loading", "cx": cx, "cy": y1 }) elif is_unloading: self._unloading_count += 1 counted_dirs.add("unloading") events.append({ "track_id": tid, "direction": "unloading", "cx": cx, "cy": y1 }) return events @property def loading_count(self) -> int: return self._loading_count @property def unloading_count(self) -> int: return self._unloading_count @property def net_count(self) -> int: return self._loading_count - self._unloading_count def reset(self) -> None: """Reset all counters (new batch).""" self._loading_count = 0 self._unloading_count = 0 self._state.clear() self._has_been_above.clear() self._has_been_below.clear() self._counted.clear() self._entry_points.clear() self._has_exited_entry_circle.clear() self._dedup_circles.clear()