Spaces:
Running
Running
| import numpy as np | |
| class InteractionDetector: | |
| def __init__(self): | |
| # Store historical hand positions for tracking velocity | |
| # Format: {person_id: {"left_hand": [positions...], "right_hand": [positions...]}} | |
| self.hand_histories = {} | |
| # Simple tracker state: {person_id: last_bounding_box} | |
| self.tracked_people = {} | |
| self.next_id = 1 | |
| self.max_history_len = 5 # Number of frames to calculate velocity | |
| def update_tracker(self, detections): | |
| """ | |
| Simple centroid-based tracking to keep IDs consistent between frames. | |
| detections: list of bounding boxes [x_min, y_min, x_max, y_max] | |
| returns: list of dicts {"id": person_id, "bbox": bbox} | |
| """ | |
| current_tracked = {} | |
| updated_detections = [] | |
| for bbox in detections: | |
| x_min, y_min, x_max, y_max = bbox | |
| cx, cy = (x_min + x_max) / 2, (y_min + y_max) / 2 | |
| # Find matching person from previous frame | |
| matched_id = None | |
| min_dist = float('inf') | |
| for pid, prev_bbox in self.tracked_people.items(): | |
| px_min, py_min, px_max, py_max = prev_bbox | |
| pcx, pcy = (px_min + px_max) / 2, (py_min + py_max) / 2 | |
| dist = np.sqrt((cx - pcx)**2 + (cy - pcy)**2) | |
| # Threshold for distance (relative to box size) | |
| box_width = x_max - x_min | |
| if dist < box_width * 0.8 and dist < min_dist: | |
| min_dist = dist | |
| matched_id = pid | |
| if matched_id is None: | |
| matched_id = self.next_id | |
| self.next_id += 1 | |
| current_tracked[matched_id] = bbox | |
| updated_detections.append({"id": matched_id, "bbox": bbox}) | |
| # Cleanup lost histories | |
| for pid in list(self.hand_histories.keys()): | |
| if pid not in current_tracked: | |
| del self.hand_histories[pid] | |
| self.tracked_people = current_tracked | |
| return updated_detections | |
| def check_aggression(self, people_landmarks, frame_w, frame_h): | |
| """ | |
| Analyzes landmarks of all tracked people to detect contact and aggressive behavior. | |
| people_landmarks: dict {person_id: pose_landmarks_object} | |
| returns: list of alerts (strings) | |
| """ | |
| alerts = [] | |
| # 1. Update histories | |
| for pid, landmarks in people_landmarks.items(): | |
| if landmarks is None: | |
| continue | |
| # MediaPipe Pose landmarks: | |
| # 15: left_wrist, 16: right_wrist | |
| # 19: left_index, 20: right_index | |
| lw = landmarks[15] | |
| rw = landmarks[16] | |
| lw_pos = np.array([lw.x * frame_w, lw.y * frame_h]) | |
| rw_pos = np.array([rw.x * frame_w, rw.y * frame_h]) | |
| if pid not in self.hand_histories: | |
| self.hand_histories[pid] = {"left": [], "right": []} | |
| self.hand_histories[pid]["left"].append(lw_pos) | |
| self.hand_histories[pid]["right"].append(rw_pos) | |
| # Keep history short | |
| if len(self.hand_histories[pid]["left"]) > self.max_history_len: | |
| self.hand_histories[pid]["left"].pop(0) | |
| if len(self.hand_histories[pid]["right"]) > self.max_history_len: | |
| self.hand_histories[pid]["right"].pop(0) | |
| # 2. Check for interactions between every pair of people | |
| pids = list(people_landmarks.keys()) | |
| for i in range(len(pids)): | |
| for j in range(i + 1, len(pids)): | |
| pid_a, pid_b = pids[i], pids[j] | |
| lm_a = people_landmarks[pid_a] | |
| lm_b = people_landmarks[pid_b] | |
| if lm_a is None or lm_b is None: | |
| continue | |
| # Get coordinates | |
| # Person B head (nose: 0, left shoulder: 11, right shoulder: 12) | |
| head_b = np.array([lm_b[0].x * frame_w, lm_b[0].y * frame_h]) | |
| shoulder_b = np.array([ | |
| (lm_b[11].x + lm_b[12].x) / 2 * frame_w, | |
| (lm_b[11].y + lm_b[12].y) / 2 * frame_h | |
| ]) | |
| # Person A head and shoulder | |
| head_a = np.array([lm_a[0].x * frame_w, lm_a[0].y * frame_h]) | |
| shoulder_a = np.array([ | |
| (lm_a[11].x + lm_a[12].x) / 2 * frame_w, | |
| (lm_a[11].y + lm_a[12].y) / 2 * frame_h | |
| ]) | |
| # Get Person A's hands | |
| left_hand_a = np.array([lm_a[15].x * frame_w, lm_a[15].y * frame_h]) | |
| right_hand_a = np.array([lm_a[16].x * frame_w, lm_a[16].y * frame_h]) | |
| # Get Person B's hands | |
| left_hand_b = np.array([lm_b[15].x * frame_w, lm_b[15].y * frame_h]) | |
| right_hand_b = np.array([lm_b[16].x * frame_w, lm_b[16].y * frame_h]) | |
| # Estimate head-shoulder span as bounding distance threshold | |
| dist_threshold = np.linalg.norm(head_b - shoulder_b) * 1.5 | |
| if dist_threshold < 20: # Fallback minimum pixel distance | |
| dist_threshold = 40 | |
| # Check Person A hitting Person B | |
| for hand_side, hand_pos in [("Left", left_hand_a), ("Right", right_hand_a)]: | |
| # Distance from A's hand to B's face/neck | |
| dist_to_head = np.linalg.norm(hand_pos - head_b) | |
| dist_to_shoulder = np.linalg.norm(hand_pos - shoulder_b) | |
| if dist_to_head < dist_threshold or dist_to_shoulder < dist_threshold: | |
| # Contact detected! Now check velocity of the hand | |
| velocity = self._calculate_hand_velocity(pid_a, hand_side.lower()) | |
| if velocity > 25.0: # Threshold for strike/hit velocity | |
| alerts.append({ | |
| "type": "aggression", | |
| "message": f"Person {pid_a} hit Person {pid_b}! (Sudden velocity: {velocity:.1f}px/fr)", | |
| "parties": [pid_a, pid_b] | |
| }) | |
| else: | |
| alerts.append({ | |
| "type": "contact", | |
| "message": f"Physical contact: Person {pid_a} touching Person {pid_b}", | |
| "parties": [pid_a, pid_b] | |
| }) | |
| # Check Person B hitting Person A | |
| for hand_side, hand_pos in [("Left", left_hand_b), ("Right", right_hand_b)]: | |
| dist_to_head = np.linalg.norm(hand_pos - head_a) | |
| dist_to_shoulder = np.linalg.norm(hand_pos - shoulder_a) | |
| if dist_to_head < dist_threshold or dist_to_shoulder < dist_threshold: | |
| velocity = self._calculate_hand_velocity(pid_b, hand_side.lower()) | |
| if velocity > 25.0: | |
| alerts.append({ | |
| "type": "aggression", | |
| "message": f"Person {pid_b} hit Person {pid_a}! (Sudden velocity: {velocity:.1f}px/fr)", | |
| "parties": [pid_b, pid_a] | |
| }) | |
| else: | |
| alerts.append({ | |
| "type": "contact", | |
| "message": f"Physical contact: Person {pid_b} touching Person {pid_a}", | |
| "parties": [pid_b, pid_a] | |
| }) | |
| # Remove duplicate alerts (just in case) | |
| unique_alerts = [] | |
| seen = set() | |
| for alert in alerts: | |
| key = (alert["type"], alert["message"]) | |
| if key not in seen: | |
| seen.add(key) | |
| unique_alerts.append(alert) | |
| return unique_alerts | |
| def _calculate_hand_velocity(self, pid, hand_side): | |
| """ | |
| Calculates the instantaneous pixel velocity of a hand. | |
| """ | |
| if pid not in self.hand_histories: | |
| return 0.0 | |
| history = self.hand_histories[pid][hand_side] | |
| if len(history) < 2: | |
| return 0.0 | |
| # Velocity is distance between last frame and current frame | |
| diffs = [] | |
| for i in range(1, len(history)): | |
| diffs.append(np.linalg.norm(history[i] - history[i-1])) | |
| # Return max diff in history to catch sudden movements | |
| return float(np.max(diffs)) | |