""" Vehicle Counting Module Dua metode counting: 1. VirtualLineCounter - hitung kendaraan yang melewati garis virtual 2. PolygonRegionCounter - hitung kendaraan yang masuk ke area polygon Kedua counter melacak ID unik supaya satu kendaraan tidak dihitung dua kali. """ import numpy as np from collections import defaultdict class VirtualLineCounter: """ Menghitung kendaraan yang melewati garis virtual horizontal. Cara kerja: - Definisikan garis horizontal pada posisi y tertentu - Untuk setiap tracked object, cek apakah centroid-nya berpindah dari atas ke bawah (atau sebaliknya) melewati garis - Setiap kendaraan hanya dihitung sekali berdasarkan track_id """ def __init__(self, line_position_ratio=0.5, frame_height=720): """ Args: line_position_ratio: posisi garis sebagai rasio dari tinggi frame (0.0 - 1.0) frame_height: tinggi frame video """ self.line_position_ratio = line_position_ratio self.frame_height = frame_height self.line_y = int(frame_height * line_position_ratio) # set untuk menyimpan ID yang sudah dihitung self.counted_ids = set() # counter per kelas self.class_counts = defaultdict(int) self.total_count = 0 def update_line(self, line_position_ratio, frame_height): """Update posisi garis jika berubah.""" self.line_position_ratio = line_position_ratio self.frame_height = frame_height self.line_y = int(frame_height * line_position_ratio) def update(self, tracked_objects): """ Cek setiap tracked object apakah melewati garis. Args: tracked_objects: list of dict dari ByteTracker.update() harus punya: track_id, class_name, center_history """ for obj in tracked_objects: track_id = obj["track_id"] # skip kalau sudah pernah dihitung if track_id in self.counted_ids: continue history = obj.get("center_history", []) if len(history) < 2: continue # ambil posisi y sekarang dan sebelumnya prev_y = history[-2][1] curr_y = history[-1][1] # cek crossing: dari atas ke bawah ATAU bawah ke atas crossed = False if prev_y < self.line_y and curr_y >= self.line_y: crossed = True # atas ke bawah elif prev_y > self.line_y and curr_y <= self.line_y: crossed = True # bawah ke atas if crossed: self.counted_ids.add(track_id) self.total_count += 1 self.class_counts[obj["class_name"]] += 1 def get_counts(self): """ Return hasil counting. Returns: dict dengan keys: - total: int - per_class: dict {class_name: count} """ return { "total": self.total_count, "per_class": dict(self.class_counts) } def get_line_coordinates(self, frame_width): """ Return koordinat garis untuk drawing. Returns: tuple: ((x1, y1), (x2, y2)) """ return ((0, self.line_y), (frame_width, self.line_y)) def reset(self): """Reset semua counter.""" self.counted_ids = set() self.class_counts = defaultdict(int) self.total_count = 0 class PolygonRegionCounter: """ Menghitung kendaraan yang masuk ke area polygon. Cara kerja: - Definisikan polygon region (list of points) - Cek apakah centroid kendaraan berada di dalam polygon - Setiap kendaraan hanya dihitung sekali berdasarkan track_id """ def __init__(self, polygon_points=None, frame_width=1280, frame_height=720): """ Args: polygon_points: list of (x, y) tuples, definisikan vertices polygon Jika None, akan dibuat default rectangle di tengah frame frame_width: lebar frame frame_height: tinggi frame """ self.frame_width = frame_width self.frame_height = frame_height if polygon_points is None: # default: rectangle di area tengah-bawah frame margin_x = int(frame_width * 0.15) margin_top = int(frame_height * 0.4) margin_bottom = int(frame_height * 0.1) self.polygon = [ (margin_x, margin_top), (frame_width - margin_x, margin_top), (frame_width - margin_x, frame_height - margin_bottom), (margin_x, frame_height - margin_bottom) ] else: self.polygon = polygon_points self.counted_ids = set() self.class_counts = defaultdict(int) self.total_count = 0 def _point_in_polygon(self, x, y): """ Cek apakah point (x, y) berada di dalam polygon. Menggunakan ray casting algorithm. """ n = len(self.polygon) inside = False j = n - 1 for i in range(n): xi, yi = self.polygon[i] xj, yj = self.polygon[j] if ((yi > y) != (yj > y)) and (x < (xj - xi) * (y - yi) / (yj - yi) + xi): inside = not inside j = i return inside def update(self, tracked_objects): """ Cek setiap tracked object apakah centroid-nya di dalam polygon. Args: tracked_objects: list of dict dari ByteTracker.update() """ for obj in tracked_objects: track_id = obj["track_id"] if track_id in self.counted_ids: continue cx, cy = obj["center"] if self._point_in_polygon(cx, cy): self.counted_ids.add(track_id) self.total_count += 1 self.class_counts[obj["class_name"]] += 1 def get_counts(self): """Return hasil counting.""" return { "total": self.total_count, "per_class": dict(self.class_counts) } def get_polygon_points(self): """Return polygon points untuk drawing.""" return self.polygon def reset(self): """Reset semua counter.""" self.counted_ids = set() self.class_counts = defaultdict(int) self.total_count = 0