""" Forensic Touch Tracker - Gradio Space ===================================== Tracks humans in video, estimates body/hand keypoints, detects surface proximity, and logs potential fingerprint touch events for forensic evidence collection. Optimized design: - Single SAM3 pass: tracks persons + segments surfaces simultaneously (saves ~50% compute) - Temporal contact grouping: groups consecutive touch frames into discrete events - Persistent person IDs across entire video via SAM3 object IDs - Precise fingertip localization via MediaPipe 21-landmark hands Components: - SAM3 Video (facebook/sam3) – promptable concept segmentation + tracking - RT-DETR (PekingU/rtdetr_r50vd_coco_o365) – person detection box pre-filter - ViTPose (usyd-community/vitpose-base-simple) – 17 COCO body keypoints - MediaPipe Hands – 21 hand landmarks per hand (5 fingertips) """ import json import os from dataclasses import dataclass, field from pathlib import Path from typing import Dict, List, Optional, Tuple import cv2 import numpy as np import torch from PIL import Image import gradio as gr # --------------------------------------------------------------------------- # Optional dependencies # --------------------------------------------------------------------------- try: from transformers import ( Sam3VideoModel, Sam3VideoProcessor, RTDetrForObjectDetection, AutoProcessor, VitPoseForPoseEstimation, ) HAS_TRANSFORMERS = True except Exception as e: HAS_TRANSFORMERS = False TRANSFORMERS_ERR = str(e) print("WARNING: transformers not available:", TRANSFORMERS_ERR) try: import mediapipe as mp HAS_MEDIAPIPE = True except Exception as e: HAS_MEDIAPIPE = False MEDIAPIPE_ERR = str(e) print("WARNING: mediapipe not available:", MEDIAPIPE_ERR) # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- @dataclass class ForensicConfig: sam3_model_id: str = "facebook/sam3" rtdetr_model_id: str = "PekingU/rtdetr_r50vd_coco_o365" vitpose_model_id: str = "usyd-community/vitpose-base-simple" device: str = "cuda" if torch.cuda.is_available() else "cpu" max_frames: int = 0 contact_threshold_px: int = 25 min_contact_frames: int = 3 confidence_threshold: float = 0.5 save_debug_video: bool = True output_dir: str = "./forensic_output" surface_prompts: List[str] = field(default_factory=lambda: [ "door handle", "countertop", "table", "desk", "wall", "railing", "keyboard", "mouse", "phone", "cup", "bottle", "drawer handle" ]) # --------------------------------------------------------------------------- # Data structures # --------------------------------------------------------------------------- @dataclass class TouchEvent: timestamp_seconds_start: float timestamp_seconds_end: float frame_index_start: int frame_index_end: int person_id: int body_part: str touch_point: Tuple[int, int] # median point across contact window surface: Optional[str] confidence: float bbox: List[int] video_resolution: Tuple[int, int] num_frames: int def to_dict(self) -> dict: return { "timestamp_seconds_start": round(self.timestamp_seconds_start, 3), "timestamp_seconds_end": round(self.timestamp_seconds_end, 3), "frame_index_start": self.frame_index_start, "frame_index_end": self.frame_index_end, "person_id": self.person_id, "body_part": self.body_part, "touch_point": {"x": self.touch_point[0], "y": self.touch_point[1]}, "surface": self.surface, "confidence": round(self.confidence, 4), "bbox": self.bbox, "video_resolution": list(self.video_resolution), "num_frames": self.num_frames, } # --------------------------------------------------------------------------- # Utilities # --------------------------------------------------------------------------- def extract_frames(video_path: str, max_frames: int = 0): cap = cv2.VideoCapture(video_path) fps = cap.get(cv2.CAP_PROP_FPS) or 30.0 w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) frames = [] while True: ret, frame = cap.read() if not ret: break frames.append(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)) if max_frames and len(frames) >= max_frames: break cap.release() return frames, fps, (w, h) def distance_point_to_mask(point: Tuple[int, int], mask: np.ndarray) -> float: ys, xs = np.where(mask) if len(xs) == 0: return float("inf") dx = xs - point[0] dy = ys - point[1] return float(np.sqrt((dx ** 2 + dy ** 2)).min()) def mask_center(mask: np.ndarray) -> Tuple[int, int]: ys, xs = np.where(mask) return int(xs.mean()), int(ys.mean()) def box_iou_xyxy(a, b): x1 = max(a[0], b[0]); y1 = max(a[1], b[1]) x2 = min(a[2], b[2]); y2 = min(a[3], b[3]) inter = max(0, x2 - x1) * max(0, y2 - y1) area_a = (a[2] - a[0]) * (a[3] - a[1]) area_b = (b[2] - b[0]) * (b[3] - b[1]) union = area_a + area_b - inter return inter / union if union > 0 else 0.0 # --------------------------------------------------------------------------- # Hand landmarks # --------------------------------------------------------------------------- FINGERTIP_INDICES = [4, 8, 12, 16, 20] FINGERTIP_NAMES = ["thumb_tip", "index_tip", "middle_tip", "ring_tip", "pinky_tip"] def get_mediapipe_hands(): if not HAS_MEDIAPIPE: return None mp_hands = mp.solutions.hands return mp_hands.Hands( static_image_mode=False, max_num_hands=2, min_detection_confidence=0.5, min_tracking_confidence=0.5, ) def extract_hand_fingertips(image: np.ndarray, hands_detector) -> List[dict]: results = [] if hands_detector is None: return results h, w = image.shape[:2] mp_results = hands_detector.process(image) if not mp_results or not mp_results.multi_hand_landmarks: return results for hand_idx, hand_landmarks in enumerate(mp_results.multi_hand_landmarks): fingertips = {} for idx, name in zip(FINGERTIP_INDICES, FINGERTIP_NAMES): lm = hand_landmarks.landmark[idx] fingertips[name] = (int(lm.x * w), int(lm.y * h)) results.append({"hand_index": hand_idx, "fingertips": fingertips}) return results # --------------------------------------------------------------------------- # Core tracker # --------------------------------------------------------------------------- class ForensicTouchTracker: def __init__(self, config: Optional[ForensicConfig] = None): self.cfg = config or ForensicConfig() self.device = self.cfg.device self.output_dir = Path(self.cfg.output_dir) self.output_dir.mkdir(parents=True, exist_ok=True) self._sam3_model = None self._sam3_processor = None self._rtdetr_model = None self._rtdetr_processor = None self._vitpose_model = None self._vitpose_processor = None self._hands_detector = None self.events: List[TouchEvent] = [] def _load_sam3(self): if self._sam3_model is not None: return if not HAS_TRANSFORMERS: raise RuntimeError(f"transformers not available: {TRANSFORMERS_ERR}") print("Loading SAM3 Video...") self._sam3_model = Sam3VideoModel.from_pretrained( self.cfg.sam3_model_id, device_map=self.device ) self._sam3_processor = Sam3VideoProcessor.from_pretrained(self.cfg.sam3_model_id) def _load_rtdetr(self): if self._rtdetr_model is not None: return if not HAS_TRANSFORMERS: raise RuntimeError(f"transformers not available: {TRANSFORMERS_ERR}") print("Loading RT-DETR...") self._rtdetr_model = RTDetrForObjectDetection.from_pretrained( self.cfg.rtdetr_model_id, device_map=self.device ) self._rtdetr_processor = AutoProcessor.from_pretrained(self.cfg.rtdetr_model_id) def _load_vitpose(self): if self._vitpose_model is not None: return if not HAS_TRANSFORMERS: raise RuntimeError(f"transformers not available: {TRANSFORMERS_ERR}") print("Loading ViTPose...") self._vitpose_model = VitPoseForPoseEstimation.from_pretrained( self.cfg.vitpose_model_id, device_map=self.device ) self._vitpose_processor = AutoProcessor.from_pretrained(self.cfg.vitpose_model_id) def _load_hands(self): if self._hands_detector is not None: return if not HAS_MEDIAPIPE: raise RuntimeError(f"mediapipe not available: {MEDIAPIPE_ERR}") self._hands_detector = get_mediapipe_hands() def _detect_person_boxes(self, image: Image.Image) -> np.ndarray: self._load_rtdetr() inputs = self._rtdetr_processor(images=image, return_tensors="pt").to(self._rtdetr_model.device) with torch.no_grad(): outputs = self._rtdetr_model(**inputs) target_sizes = torch.tensor([(image.height, image.width)]) results = self._rtdetr_processor.post_process_object_detection( outputs, target_sizes=target_sizes, threshold=self.cfg.confidence_threshold )[0] person_mask = results["labels"] == 0 return results["boxes"][person_mask].cpu().numpy() def _estimate_poses(self, image: Image.Image, boxes_xyxy: np.ndarray): if len(boxes_xyxy) == 0: return [] self._load_vitpose() boxes_xywh = boxes_xyxy.copy() boxes_xywh[:, 2] -= boxes_xywh[:, 0] boxes_xywh[:, 3] -= boxes_xywh[:, 1] inputs = self._vitpose_processor(image, boxes=[boxes_xywh], return_tensors="pt").to(self._vitpose_model.device) with torch.no_grad(): outputs = self._vitpose_model(**inputs) poses = self._vitpose_processor.post_process_pose_estimation(outputs, boxes=[boxes_xywh]) return poses[0] if isinstance(poses, list) and len(poses) > 0 else [] def _sam3_single_pass(self, video_frames: List[np.ndarray]): """ One SAM3 propagation with all prompts (person + surfaces). Returns per-frame: persons dict and surfaces list. """ self._load_sam3() pil_frames = [Image.fromarray(f) for f in video_frames] inference_session = self._sam3_processor.init_video_session( video=pil_frames, inference_device=self.device, processing_device="cpu", video_storage_device="cpu", ) all_prompts = ["person"] + self.cfg.surface_prompts inference_session = self._sam3_processor.add_text_prompt( inference_session=inference_session, text=all_prompts ) per_frame_persons = [] per_frame_surfaces = [] max_track = self.cfg.max_frames if self.cfg.max_frames > 0 else len(pil_frames) for model_outputs in self._sam3_model.propagate_in_video_iterator( inference_session=inference_session, max_frame_num_to_track=max_track ): processed = self._sam3_processor.postprocess_outputs(inference_session, model_outputs) prompt_to_obj_ids = processed.get("prompt_to_obj_ids", {}) obj_ids = processed.get("object_ids", []) boxes = processed.get("boxes", []) masks = processed.get("masks", []) scores = processed.get("scores", []) persons = {} surfaces = [] for prompt, ids in prompt_to_obj_ids.items(): for oid in ids: idx = list(obj_ids).index(oid) if oid in list(obj_ids) else -1 if idx < 0: continue if scores is not None and len(scores) > idx and scores[idx] < self.cfg.confidence_threshold: continue mask = masks[idx].cpu().numpy() if hasattr(masks[idx], "cpu") else np.array(masks[idx]) bbox = boxes[idx].cpu().numpy().tolist() if hasattr(boxes[idx], "cpu") else list(boxes[idx]) entry = {"bbox": bbox, "mask": mask, "score": float(scores[idx]) if scores is not None and len(scores) > idx else 1.0} if prompt == "person": persons[int(oid)] = entry else: entry["name"] = prompt entry["center"] = mask_center(mask) surfaces.append(entry) per_frame_persons.append(persons) per_frame_surfaces.append(surfaces) return per_frame_persons, per_frame_surfaces def _get_touch_candidates(self, image: np.ndarray, person_boxes: np.ndarray) -> List[dict]: self._load_hands() candidates = [] pil = Image.fromarray(image) poses = self._estimate_poses(pil, person_boxes) for i, box in enumerate(person_boxes): if i < len(poses): kp = poses[i].get("keypoints", []) for wrist_name, wrist_idx in [("left_wrist", 9), ("right_wrist", 10)]: if wrist_idx < len(kp): x, y = float(kp[wrist_idx][0]), float(kp[wrist_idx][1]) candidates.append({ "person_idx": i, "body_part": wrist_name, "point": (int(x), int(y)), "source": "body_pose", }) x1, y1, x2, y2 = map(int, box) x1, y1 = max(0, x1), max(0, y1) x2, y2 = min(image.shape[1], x2), min(image.shape[0], y2) crop = image[y1:y2, x1:x2] if crop.size == 0: continue hands = extract_hand_fingertips(crop, self._hands_detector) for hand in hands: for tip_name, tip_coords in hand["fingertips"].items(): candidates.append({ "person_idx": i, "body_part": f"hand_{hand['hand_index']}_{tip_name}", "point": (x1 + tip_coords[0], y1 + tip_coords[1]), "source": "hand_landmark", }) return candidates def process_video(self, video_path: str, progress=None): print(f"[ForensicTouchTracker] Processing {video_path}") frames, fps, (W, H) = extract_frames(video_path, max_frames=self.cfg.max_frames) total = len(frames) print(f" -> {total} frames @ {fps:.2f} fps, {W}x{H}") if total == 0: return [], str(self.output_dir) if progress: progress(0.05, desc="Extracting frames...") # Single SAM3 pass for persons + surfaces print("[SAM3] Single-pass tracking & surface segmentation...") person_tracks, surface_masks = self._sam3_single_pass(frames) print(f" -> got persons in {len(person_tracks)} frames, surfaces in {len(surface_masks)} frames") if progress: progress(0.45, desc="Tracking & segmentation done") # Stage 3: per-frame pose + hands + contact inference with temporal grouping print("[Stage 3] Pose, hands, and temporal contact grouping...") active_contacts: Dict[Tuple[int, str], dict] = {} annotated_frames = [] def emit_event(key, rec): points = rec["points"] x = int(np.median([p[0] for p in points])) y = int(np.median([p[1] for p in points])) dists = rec["dists"] med_dist = float(np.median(dists)) conf = max(0.0, 1.0 - (med_dist / (self.cfg.contact_threshold_px * 2))) self.events.append(TouchEvent( timestamp_seconds_start=rec["time_start"], timestamp_seconds_end=rec["time_end"], frame_index_start=rec["frame_start"], frame_index_end=rec["frame_end"], person_id=key[0], body_part=key[1], touch_point=(x, y), surface=rec["surface"], confidence=round(conf, 4), bbox=rec["bbox"], video_resolution=(W, H), num_frames=rec["count"], )) for frame_idx, frame in enumerate(frames): if progress and frame_idx % 10 == 0: progress(0.45 + 0.50 * (frame_idx / total), desc=f"Frame {frame_idx}/{total}") t_sec = frame_idx / fps persons = person_tracks[frame_idx] surfaces = surface_masks[frame_idx] if len(persons) == 0: # flush all active contacts (no persons => no contact possible) for key, rec in list(active_contacts.items()): if rec["count"] >= self.cfg.min_contact_frames: emit_event(key, rec) del active_contacts[key] annotated_frames.append(frame) continue pil = Image.fromarray(frame) person_boxes = self._detect_person_boxes(pil) if len(person_boxes) == 0: for key, rec in list(active_contacts.items()): if rec["count"] >= self.cfg.min_contact_frames: emit_event(key, rec) del active_contacts[key] annotated_frames.append(frame) continue candidates = self._get_touch_candidates(frame, person_boxes) # Map each candidate to SAM3 person ID via IoU matched_ids = {} for cand in candidates: i = cand["person_idx"] best_iou = 0.0 best_oid = None for oid, pdata in persons.items(): iou = box_iou_xyxy(pdata["bbox"], person_boxes[i]) if iou > best_iou: best_iou = iou best_oid = oid matched_ids[i] = best_oid # Determine which keys are still in contact this frame touched_keys = set() for cand in candidates: pid = matched_ids.get(cand["person_idx"]) if pid is None: continue point = cand["point"] best_surface = None best_dist = float("inf") for surf in surfaces: d = distance_point_to_mask(point, surf["mask"]) if d < best_dist: best_dist = d best_surface = surf if best_surface is not None and best_dist <= self.cfg.contact_threshold_px: key = (pid, cand["body_part"]) touched_keys.add(key) if key not in active_contacts: active_contacts[key] = { "frame_start": frame_idx, "time_start": t_sec, "points": [], "dists": [], "surface": best_surface["name"], "bbox": person_boxes[cand["person_idx"]].astype(int).tolist(), "count": 0, } rec = active_contacts[key] rec["frame_end"] = frame_idx rec["time_end"] = t_sec rec["points"].append(point) rec["dists"].append(best_dist) rec["count"] += 1 # Emit events for keys that lost contact this frame for key, rec in list(active_contacts.items()): if key not in touched_keys: if rec["count"] >= self.cfg.min_contact_frames: emit_event(key, rec) del active_contacts[key] annotated = self._annotate_frame(frame, persons, surfaces, candidates) annotated_frames.append(annotated) # Flush remaining active contacts at EOF for key, rec in list(active_contacts.items()): if rec["count"] >= self.cfg.min_contact_frames: emit_event(key, rec) del active_contacts[key] self._write_outputs(annotated_frames, fps, (W, H)) print(f"[Done] {len(self.events)} events. Output: {self.output_dir}") return self.events, str(self.output_dir) def _annotate_frame(self, frame, persons, surfaces, candidates): out = frame.copy() for surf in surfaces: m = surf["mask"] if m.shape[:2] != out.shape[:2]: continue overlay = np.zeros_like(out) overlay[m > 0] = np.array([0, 255, 255], dtype=np.uint8) out = cv2.addWeighted(out, 1.0, overlay, 0.3, 0) cx, cy = surf["center"] cv2.putText(out, surf["name"], (cx, cy), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 120, 120), 2) for oid, pdata in persons.items(): x1, y1, x2, y2 = map(int, pdata["bbox"]) cv2.rectangle(out, (x1, y1), (x2, y2), (0, 255, 0), 2) cv2.putText(out, f"Person {oid}", (x1, y1 - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2) for cand in candidates: pt = cand["point"] color = (255, 0, 0) if "hand" in cand["body_part"] else (0, 0, 255) cv2.circle(out, pt, 4, color, -1) return out def _write_outputs(self, annotated_frames, fps, resolution): log_path = self.output_dir / "forensic_touch_log.jsonl" with open(log_path, "w") as f: for ev in self.events: f.write(json.dumps(ev.to_dict(), default=str) + "\n") summary = { "total_events": len(self.events), "unique_persons": sorted({e.person_id for e in self.events}), "surfaces_touched": sorted({e.surface for e in self.events if e.surface}), "body_parts": sorted({e.body_part for e in self.events}), "resolution": list(resolution), "fps": fps, } with open(self.output_dir / "forensic_summary.json", "w") as f: json.dump(summary, f, indent=2) if self.cfg.save_debug_video and annotated_frames: vid_path = self.output_dir / "annotated_video.mp4" fourcc = cv2.VideoWriter_fourcc(*"mp4v") writer = cv2.VideoWriter(str(vid_path), fourcc, fps, resolution) for af in annotated_frames: writer.write(cv2.cvtColor(af, cv2.COLOR_RGB2BGR)) writer.release() # --------------------------------------------------------------------------- # Gradio UI # --------------------------------------------------------------------------- def run_tracker(video_file, max_frames, contact_threshold, min_contact_frames): if video_file is None: return "No video uploaded.", None, None, None cfg = ForensicConfig( max_frames=int(max_frames), contact_threshold_px=int(contact_threshold), min_contact_frames=int(min_contact_frames), output_dir="./forensic_output", ) tracker = ForensicTouchTracker(cfg) events, out_dir = tracker.process_video(video_file, progress=gr.Progress()) if not events: md = "## No touch events detected.\n\nTry lowering the contact threshold or min-contact-frames." else: md = f"## Detected {len(events)} Forensic Touch Event(s)\n\n" md += "| Start | End | Person | Body Part | Surface | Conf | Frames | Touch Point |\n" md += "|-------|-----|--------|-----------|---------|------|--------|-------------|\n" for ev in events: md += f"| {ev.timestamp_seconds_start:.2f}s | {ev.timestamp_seconds_end:.2f}s | {ev.person_id} | {ev.body_part} | {ev.surface or 'unknown'} | {ev.confidence:.2f} | {ev.num_frames} | ({ev.touch_point[0]},{ev.touch_point[1]}) |\n" log_path = os.path.join(out_dir, "forensic_touch_log.jsonl") summary_path = os.path.join(out_dir, "forensic_summary.json") video_path_out = os.path.join(out_dir, "annotated_video.mp4") files_out = [p for p in [log_path, summary_path, video_path_out] if os.path.exists(p)] return md, files_out[0] if len(files_out) > 0 else None, files_out[1] if len(files_out) > 1 else None, files_out[2] if len(files_out) > 2 else None with gr.Blocks(title="Forensic Touch Tracker") as demo: gr.Markdown( """ # 🕵️ Forensic Touch Tracker **Track human movement and log touch points for fingerprint forensics.** Upload a surveillance or body-cam video. The pipeline: 1. **Track persons** with persistent IDs across the entire video (SAM3 Video) 2. **Segment forensic surfaces** in a single pass — door handles, countertops, walls, etc. (SAM3 Video) 3. **Estimate body poses** (ViTPose) and **hand fingertips** (MediaPipe) 4. **Log discrete touch events** when fingertips come near a surface for ≥N consecutive frames Outputs: structured JSONL forensic log, summary JSON, and an annotated debug video. """ ) with gr.Row(): with gr.Column(scale=1): video_input = gr.Video(label="Upload Video", format="mp4") max_frames = gr.Number(value=0, label="Max Frames (0 = all)", precision=0) contact_threshold = gr.Number(value=25, label="Contact Threshold (px)", precision=0) min_contact_frames = gr.Number(value=3, label="Min Contact Frames", precision=0) run_btn = gr.Button("Run Analysis", variant="primary") with gr.Column(scale=2): results_md = gr.Markdown("## Results will appear here") with gr.Row(): log_file = gr.File(label="Forensic Touch Log (JSONL)") summary_file = gr.File(label="Forensic Summary (JSON)") debug_video = gr.Video(label="Annotated Debug Video") run_btn.click( fn=run_tracker, inputs=[video_input, max_frames, contact_threshold, min_contact_frames], outputs=[results_md, log_file, summary_file, debug_video], ) if __name__ == "__main__": demo.launch()