""" Profile Engine v2 — Multi-Modal Biometric Re-Identification ============================================================= 6-Signal Enrollment & Inference Pipeline: 1. Gait (DeepGaitV2) → 256-d gait_vector (weight: 0.35) 2. Biomechanics (Pose) → 64-d biomech_vector (weight: 0.25) 3. Appearance (OSNet) → 512-d appearance_vector (weight: 0.25) 4. Height (Pinhole) → float height_cm (weight: 0.15) 5. Face (ArcFace) → 512-d face_vector (optional: 0.30 blend) 6. FAISS Vector DB → fast similarity search Anti-Spoofing Gates: - Signal agreement: ≥2 of (gait, pose, appearance) must independently exceed 0.70 - Pose quality: reject if avg keypoint confidence < 0.6 - Temporal smoothing: 15 consecutive frames above threshold before alert - Height pre-filter: reject |detected - target| > 12cm """ import os import cv2 import json import time import uuid import threading import numpy as np from ultralytics import YOLO from gait_models.gait_engine import GaitEngine from biomech_engine import BiomechEngine from appearance_engine import AppearanceEngine from height_estimator import HeightEstimator from vector_db import ProfileVectorDB class TargetVerifier: """ Temporal evidence accumulator for target person verification. Instead of making an instant decision per-frame, this collects evidence across multiple frames and only declares a verified match after sufficient consistent evidence has been gathered. States: scanning → No candidate found yet verifying → A candidate person is being tracked; accumulating evidence confirmed → Target identity verified with high confidence lost → Target was confirmed but lost from view """ # Temporal smoothing requirements REQUIRED_CONSECUTIVE_FRAMES = 15 # frames above threshold before alerting MIN_VERIFY_SECONDS = 2.0 # minimum wallclock time before confirming GAIT_CYCLE_FRAMES = 30 # frames for gait analysis # Thresholds HIGH_CONFIDENCE_THRESHOLD = 0.85 # GREEN box — "MATCH: XX%" POSSIBLE_MATCH_THRESHOLD = 0.65 # YELLOW box — "POSSIBLE: XX%" CONFIRM_SCORE_THRESHOLD = 0.55 # fused score to confirm identity FACE_MATCH_THRESHOLD = 0.35 # per-frame face cosine sim # Anti-spoofing MIN_POSE_CONFIDENCE = 0.6 # reject frames below this SIGNAL_AGREEMENT_THRESHOLD = 0.70 # per-signal threshold for agreement gate MIN_AGREEING_SIGNALS = 2 # at least 2 must agree # Timeouts CANDIDATE_LOST_TIMEOUT = 3.0 CONFIRMED_LOST_TIMEOUT = 5.0 def __init__(self): self.reset() def reset(self): """Reset all verification state.""" self.state = "scanning" self.candidate_track_id = None self.candidate_bbox = None # Per-modality evidence buffers self.face_scores = [] self.face_embeddings_live = [] self.gait_silhouettes = [] self.biomech_vectors = [] self.appearance_embeddings = [] self.height_estimates = [] self.clothing_scores = [] # Anti-spoofing counters self.consecutive_match_frames = 0 self.pose_confidence_history = [] # Timing self.verify_start_time = 0.0 self.last_seen_time = 0.0 # Per-modality live scores (updated each frame) self.live_scores = { "gait": 0.0, "biomech": 0.0, "appearance": 0.0, "height": 0.0, "face": None, "ensemble": 0.0 } # Final result self.confirmed_score = 0.0 self.confirmed_bbox = None self.verification_progress = 0.0 self.status_message = "Scanning for target..." def get_status(self): return { "state": self.state, "progress": self.verification_progress, "message": self.status_message, "score": self.confirmed_score, "face_hits": len(self.face_scores), "gait_frames": len(self.gait_silhouettes), "consecutive_frames": self.consecutive_match_frames, "live_scores": self.live_scores.copy(), } class ProfileEngine: """ Multi-modal biometric profile engine. Enrollment: video → 6-signal feature extraction → FAISS storage Inference: live frame → 6-signal extraction → weighted ensemble → alert """ # Ensemble weights (per spec) W_GAIT = 0.35 W_BIOMECH = 0.25 W_APPEARANCE = 0.25 W_HEIGHT = 0.15 W_FACE_BLEND = 0.30 # face blends into the 4-signal score def __init__(self, face_app): print("[ProfileEngine] Initializing multi-modal engine...") self.face_app = face_app # YOLOv8 models self.yolo_pose = YOLO("yolov8n-pose.pt") self.yolo_det = YOLO("yolov8n.pt") # Sub-engines self.gait_engine = GaitEngine() self.biomech_engine = BiomechEngine() self.appearance_engine = AppearanceEngine() self.height_estimator = HeightEstimator() # FAISS vector database self.vector_db = ProfileVectorDB(db_dir="saved_profiles") # Profile storage self.PROFILES_DIR = "saved_profiles" self.PROFILES_JSON = "profiles.json" os.makedirs(self.PROFILES_DIR, exist_ok=True) if not os.path.exists(self.PROFILES_JSON): with open(self.PROFILES_JSON, "w") as f: json.dump([], f) self.active_jobs = {} # Active verifier self.verifier = TargetVerifier() # Cache for loaded target data self._cached_target_id = None self._cached_face_embs = None self._cached_gait_embs = None self._cached_biomech_vec = None self._cached_appearance_vec = None self._cached_profile = None # Match event log self.match_log = [] print("[ProfileEngine] Multi-modal engine ready.") # ================================================================ # Camera Calibration # ================================================================ def update_camera_calibration(self, calibration): """Update height estimator with camera calibration params.""" self.height_estimator.update_calibration(calibration) # ================================================================ # Color Extraction (kept for backward compat) # ================================================================ def get_dominant_colors(self, image_crop, k=3): """Extract dominant colors from a bounding box using K-Means.""" if image_crop.size == 0: return [] img = cv2.resize(image_crop, (50, 50)) hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV) pixels = hsv.reshape((-1, 3)).astype(np.float32) criteria = (cv2.TERM_CRITERIA_EPS + cv2.TERM_CRITERIA_MAX_ITER, 10, 1.0) _, labels, centers = cv2.kmeans(pixels, k, None, criteria, 10, cv2.KMEANS_RANDOM_CENTERS) label_counts = np.bincount(labels.flatten()) percentages = label_counts / float(len(pixels)) colors = [] for i, center in enumerate(centers): colors.append({ "h": float(center[0]), "s": float(center[1]), "v": float(center[2]), "weight": float(percentages[i]) }) colors.sort(key=lambda x: x["weight"], reverse=True) return colors def compare_colors(self, hsv1, hsv2): """Compare two HSV colors (0-1 score).""" h_diff = min(abs(hsv1['h'] - hsv2['h']), 180 - abs(hsv1['h'] - hsv2['h'])) / 90.0 s_diff = abs(hsv1['s'] - hsv2['s']) / 255.0 v_diff = abs(hsv1['v'] - hsv2['v']) / 255.0 score = 1.0 - (0.6 * h_diff + 0.2 * s_diff + 0.2 * v_diff) return max(0.0, score) # ================================================================ # PHASE 1: ENROLLMENT — Video Processing # ================================================================ def process_video_bg(self, video_path, person_name, job_id): """Background worker to analyze video and build full multi-modal profile.""" try: self.active_jobs[job_id] = { "status": "processing", "progress": 0, "message": "Reading video...", "name": person_name } cap = cv2.VideoCapture(video_path) total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) fps = cap.get(cv2.CAP_PROP_FPS) if fps <= 0: fps = 30 sample_rate = max(1, int(total_frames / 60)) # Accumulators face_embeddings = [] clothing_colors = [] appearance_embeddings = [] biomech_accum = BiomechEngine() height_estimates = [] gait_bbox_track = [] frame_idx = 0 while True: ret, frame = cap.read() if not ret: break if frame_idx % sample_rate == 0: self.active_jobs[job_id]["progress"] = int((frame_idx / total_frames) * 60) self.active_jobs[job_id]["message"] = f"Analyzing frame {frame_idx}/{total_frames}..." # 1. Detect person with pose pose_results = self.yolo_pose(frame, conf=0.5, verbose=False) best_person_box = None best_area = 0 best_keypoints = None best_kp_conf = None for r in pose_results: if r.keypoints is None or len(r.keypoints.data) == 0: continue for idx_p in range(len(r.boxes)): cls = int(r.boxes.cls[idx_p]) if hasattr(r.boxes, 'cls') else 0 if cls != 0: continue box = r.boxes.xyxy[idx_p].cpu().numpy().astype(int) area = (box[2] - box[0]) * (box[3] - box[1]) if area > best_area: best_area = area best_person_box = box kp_data = r.keypoints.data[idx_p].cpu().numpy() best_keypoints = kp_data[:, :2] # (17, 2) best_kp_conf = kp_data[:, 2] # (17,) if best_person_box is not None: px1, py1, px2, py2 = best_person_box person_crop = frame[py1:py2, px1:px2] gait_bbox_track.append((frame_idx, best_person_box)) if person_crop.size > 0: # 2. Face (InsightFace/ArcFace) faces = self.face_app.get(person_crop) if faces: face_embeddings.append(faces[0].normed_embedding) # 3. Appearance (OSNet) app_emb = self.appearance_engine.extract_embedding(person_crop) if app_emb is not None: appearance_embeddings.append(app_emb) # 4. Biomechanics (from keypoints) if best_keypoints is not None and best_kp_conf is not None: bvec = biomech_accum.extract_features( best_keypoints, best_kp_conf, best_person_box ) if bvec is not None: biomech_accum.accumulate_template(bvec) # 5. Height h_est = self.height_estimator.estimate_height( best_keypoints, best_kp_conf, best_person_box, frame_shape=frame.shape[:2] ) height_estimates.append(h_est) # 6. Clothing colors h, w = person_crop.shape[:2] torso = person_crop[int(h*0.2):int(h*0.6), int(w*0.2):int(w*0.8)] colors = self.get_dominant_colors(torso) if colors: clothing_colors.append(colors) else: # Track bbox for gait on non-sampled frames too det_results = self.yolo_det(frame, conf=0.5, verbose=False) for r in det_results: for idx_d, cls in enumerate(r.boxes.cls): if int(cls) == 0: box = r.boxes.xyxy[idx_d].cpu().numpy().astype(int) gait_bbox_track.append((frame_idx, box)) break break frame_idx += 1 cap.release() # === GAIT EXTRACTION === self.active_jobs[job_id]["progress"] = 65 self.active_jobs[job_id]["message"] = "Extracting gait features (walking style)..." gait_results = self.gait_engine.process_walking_video(video_path, gait_bbox_track) # === AGGREGATE RESULTS === self.active_jobs[job_id]["progress"] = 85 self.active_jobs[job_id]["message"] = "Building multi-modal profile..." profile_id = str(uuid.uuid4())[:8] # Face embeddings face_emb_path = "" if face_embeddings: emb_array = np.array(face_embeddings) face_emb_path = os.path.join(self.PROFILES_DIR, f"{person_name}_{profile_id}_faces.npy") np.save(face_emb_path, emb_array) # Appearance vector (averaged) appearance_vec_path = "" avg_appearance = None if appearance_embeddings: avg_appearance = AppearanceEngine.average_embeddings(appearance_embeddings) appearance_vec_path = os.path.join(self.PROFILES_DIR, f"{person_name}_{profile_id}_appearance.npy") np.save(appearance_vec_path, avg_appearance) # Biomech vector (averaged template) biomech_vec_path = "" avg_biomech = biomech_accum.get_averaged_template() if avg_biomech is not None: biomech_vec_path = os.path.join(self.PROFILES_DIR, f"{person_name}_{profile_id}_biomech.npy") np.save(biomech_vec_path, avg_biomech) # Gait vectors gait_v2_path = "" gait_gl_path = "" gait_former_path = "" gait_gei_path = "" if gait_results["deepgaitv2_embedding"] is not None: gait_v2_path = os.path.join(self.PROFILES_DIR, f"{person_name}_{profile_id}_gait_v2.npy") np.save(gait_v2_path, gait_results["deepgaitv2_embedding"]) if gait_results.get("gaitgl_embedding") is not None: gait_gl_path = os.path.join(self.PROFILES_DIR, f"{person_name}_{profile_id}_gait_gl.npy") np.save(gait_gl_path, gait_results["gaitgl_embedding"]) if gait_results.get("gaitformer_embedding") is not None: gait_former_path = os.path.join(self.PROFILES_DIR, f"{person_name}_{profile_id}_gait_former.npy") np.save(gait_former_path, gait_results["gaitformer_embedding"]) if gait_results.get("gei_embedding") is not None: gait_gei_path = os.path.join(self.PROFILES_DIR, f"{person_name}_{profile_id}_gait_gei.npy") np.save(gait_gei_path, gait_results["gei_embedding"]) # GEI visualization gei_img_path = "" if gait_results.get("gei_image") is not None: gei_img_path = os.path.join(self.PROFILES_DIR, f"{person_name}_{profile_id}_gei.png") cv2.imwrite(gei_img_path, (gait_results["gei_image"] * 255).astype(np.uint8)) # Height height_cm = None height_ratio = 0.8 if height_estimates: cm_vals = [h["height_cm"] for h in height_estimates if h.get("height_cm")] ratio_vals = [h["height_ratio"] for h in height_estimates if h.get("height_ratio", 0) > 0] if cm_vals: height_cm = round(float(np.median(cm_vals)), 1) if ratio_vals: height_ratio = float(np.mean(ratio_vals)) # Clothing final_colors = [] if clothing_colors: avg_h = np.mean([c[0]["h"] for c in clothing_colors if c]) avg_s = np.mean([c[0]["s"] for c in clothing_colors if c]) avg_v = np.mean([c[0]["v"] for c in clothing_colors if c]) final_colors = [{"h": avg_h, "s": avg_s, "v": avg_v, "weight": 1.0}] # Build profile entry profile_entry = { "id": profile_id, "name": person_name, # Face "face_count": len(face_embeddings), "embeddings_file": face_emb_path, # Gait "gait_v2_file": gait_v2_path, "gait_gl_file": gait_gl_path, "gait_former_file": gait_former_path, "gait_gei_file": gait_gei_path, "gait_gei_image": gei_img_path, "gait_silhouette_count": gait_results["silhouette_count"], # Biomech "biomech_file": biomech_vec_path, # Appearance "appearance_file": appearance_vec_path, # Height "height_cm": height_cm, "height_ratio": float(height_ratio), # Clothing (backward compat) "clothing_colors": final_colors, } # Save to JSON with open(self.PROFILES_JSON, "r") as f: db = json.load(f) db.append(profile_entry) with open(self.PROFILES_JSON, "w") as f: json.dump(db, f, indent=4) # Add to FAISS faiss_vectors = {} if gait_results["deepgaitv2_embedding"] is not None: faiss_vectors["gait"] = gait_results["deepgaitv2_embedding"] if avg_biomech is not None: faiss_vectors["biomech"] = avg_biomech if avg_appearance is not None: faiss_vectors["appearance"] = avg_appearance if face_embeddings: avg_face = np.mean(np.array(face_embeddings), axis=0) n = np.linalg.norm(avg_face) if n > 0: avg_face = avg_face / n faiss_vectors["face"] = avg_face if faiss_vectors: self.vector_db.add_profile(profile_id, faiss_vectors) # Build summary modalities = [] if face_embeddings: modalities.append(f"{len(face_embeddings)} faces") if gait_results["silhouette_count"] >= 30: modalities.append(f"{gait_results['silhouette_count']} gait frames") if avg_biomech is not None: modalities.append("biomechanics") if avg_appearance is not None: modalities.append("appearance") if height_cm: modalities.append(f"height: {height_cm}cm") summary = ", ".join(modalities) if modalities else "limited data" self.active_jobs[job_id] = { "status": "done", "progress": 100, "message": f"Complete! ({summary})", "profile": profile_entry } # Sync profiles to cloud try: import hf_db hf_db.sync_single_file(self.PROFILES_JSON) hf_db.sync_single_folder(self.PROFILES_DIR) except Exception as sync_err: print(f"[ProfileEngine] Cloud sync warning: {sync_err}") except Exception as e: print(f"[ProfileEngine] Error processing video: {e}") import traceback traceback.print_exc() self.active_jobs[job_id] = {"status": "error", "message": str(e), "progress": 0} # Clean up video if os.path.exists(video_path): try: os.remove(video_path) except Exception: pass def start_video_processing(self, video_path, person_name): job_id = str(uuid.uuid4()) threading.Thread(target=self.process_video_bg, args=(video_path, person_name, job_id), daemon=True).start() return job_id def get_job_status(self, job_id): return self.active_jobs.get(job_id, {"status": "not_found"}) # ================================================================ # Target Cache Loading # ================================================================ def _load_npy_safe(self, path): """Safely load a numpy file, returning None on any error.""" if path and os.path.exists(path): try: return np.load(path) except Exception: pass return None def _load_target_cache(self, target_profile): """Load and cache target profile data for fast per-frame access.""" pid = target_profile.get("id", "") if self._cached_target_id == pid and self._cached_profile is not None: return self._cached_target_id = pid self._cached_profile = target_profile # Face embeddings self._cached_face_embs = self._load_npy_safe(target_profile.get("embeddings_file", "")) # Gait embeddings self._cached_gait_embs = { "deepgaitv2_embedding": self._load_npy_safe(target_profile.get("gait_v2_file", "")), "gei_embedding": self._load_npy_safe(target_profile.get("gait_gei_file", "")), "gaitgl_embedding": self._load_npy_safe(target_profile.get("gait_gl_file", "")), "gaitformer_embedding": self._load_npy_safe(target_profile.get("gait_former_file", "")), } # Biomech vector self._cached_biomech_vec = self._load_npy_safe(target_profile.get("biomech_file", "")) # Appearance vector self._cached_appearance_vec = self._load_npy_safe(target_profile.get("appearance_file", "")) modalities = [] if self._cached_face_embs is not None: modalities.append(f"faces={self._cached_face_embs.shape[0]}") if any(v is not None for v in self._cached_gait_embs.values()): modalities.append("gait") if self._cached_biomech_vec is not None: modalities.append("biomech") if self._cached_appearance_vec is not None: modalities.append("appearance") print(f"[ProfileEngine] Cached target '{target_profile.get('name', '?')}': {', '.join(modalities)}") # ================================================================ # SEARCH SESSION MANAGEMENT # ================================================================ def start_search(self, target_profile): """Initialize a new search session for a target profile.""" self._load_target_cache(target_profile) self.verifier = TargetVerifier() self.verifier.status_message = "Scanning for target..." self.biomech_engine.reset() print(f"[ProfileEngine] Search started for '{target_profile.get('name', '?')}'") def stop_search(self): """Stop the current search session.""" self.verifier.reset() self._cached_target_id = None self._cached_face_embs = None self._cached_gait_embs = None self._cached_biomech_vec = None self._cached_appearance_vec = None self._cached_profile = None # ================================================================ # PHASE 2: INFERENCE — Per-Frame Evidence Accumulation # ================================================================ def _compute_face_score(self, person_crop): """Run face detection and compare to cached target faces.""" if self._cached_face_embs is None or len(self._cached_face_embs) == 0: return 0.0, None, False faces = self.face_app.get(person_crop) if not faces: return 0.0, None, False emb = faces[0].normed_embedding norm = np.linalg.norm(emb) if norm == 0: return 0.0, None, True emb_normed = emb / norm sims = np.dot(self._cached_face_embs, emb_normed) top_k = min(5, len(sims)) top_k_sims = np.sort(sims)[-top_k:] score = float(np.mean(top_k_sims)) return score, emb_normed, True def _compute_appearance_score(self, person_crop): """Compute appearance similarity using OSNet.""" if self._cached_appearance_vec is None: return 0.0, None emb = self.appearance_engine.extract_embedding(person_crop) if emb is None: return 0.0, None score = AppearanceEngine.compute_similarity(emb, self._cached_appearance_vec) return score, emb def _compute_biomech_score(self, keypoints_xy, confidence, bbox): """Compute biomechanical similarity.""" if self._cached_biomech_vec is None: return 0.0, None vec = self.biomech_engine.extract_features(keypoints_xy, confidence, bbox) if vec is None: return 0.0, None score = BiomechEngine.compute_similarity(vec, self._cached_biomech_vec) return score, vec def _compute_height_score(self, keypoints_xy, confidence, bbox, frame_shape): """Compute height match score.""" target_cm = self._cached_profile.get("height_cm") target_ratio = self._cached_profile.get("height_ratio", 0) h_est = self.height_estimator.estimate_height( keypoints_xy, confidence, bbox, frame_shape ) score = HeightEstimator.compute_match_score(h_est, target_cm, target_ratio) return score, h_est def _compute_live_gait_score(self): """Compute gait similarity from accumulated live silhouettes.""" v = self.verifier if len(v.gait_silhouettes) < self.gait_engine.MIN_FRAMES_FOR_GAIT: return 0.0 if self._cached_gait_embs is None or not any( val is not None for val in self._cached_gait_embs.values() ): return 0.0 # Build live gait embeddings gei = self.gait_engine.compute_gei(v.gait_silhouettes) if gei is None: return 0.0 live_gait = { "deepgaitv2_embedding": self.gait_engine.compute_deepgaitv2_embedding(v.gait_silhouettes), "gei_embedding": self.gait_engine.compute_gei_embedding(gei), "gaitgl_embedding": self.gait_engine.compute_gaitgl_embedding(v.gait_silhouettes), "gaitformer_embedding": self.gait_engine.compute_gaitformer_embedding(v.gait_silhouettes), } return self.gait_engine.compute_gait_similarity(live_gait, self._cached_gait_embs) def _compute_clothing_score(self, person_crop): """Compare clothing colors of crop against cached target.""" target_colors = self._cached_profile.get("clothing_colors", []) if not target_colors: return 0.0 h, w = person_crop.shape[:2] torso_crop = person_crop[int(h*0.2):int(h*0.6), int(w*0.2):int(w*0.8)] if torso_crop.size == 0: return 0.0 current_colors = self.get_dominant_colors(torso_crop, k=1) if not current_colors: return 0.0 return self.compare_colors(current_colors[0], target_colors[0]) def _compute_weighted_ensemble(self, gait_sim, biomech_sim, appearance_sim, height_sim, face_sim=None): """Compute weighted ensemble score per spec.""" score = (self.W_GAIT * gait_sim + self.W_BIOMECH * biomech_sim + self.W_APPEARANCE * appearance_sim + self.W_HEIGHT * height_sim) if face_sim is not None and face_sim > 0: score = (1.0 - self.W_FACE_BLEND) * score + self.W_FACE_BLEND * face_sim return score def _check_signal_agreement(self, gait_sim, biomech_sim, appearance_sim): """Anti-spoofing: at least 2 of 3 must independently exceed threshold.""" thresh = self.verifier.SIGNAL_AGREEMENT_THRESHOLD agreeing = sum(1 for s in [gait_sim, biomech_sim, appearance_sim] if s >= thresh) return agreeing >= self.verifier.MIN_AGREEING_SIGNALS def _bb_iou(self, boxA, boxB): """IoU between two [x1,y1,x2,y2] boxes.""" xA = max(boxA[0], boxB[0]); yA = max(boxA[1], boxB[1]) xB = min(boxA[2], boxB[2]); yB = min(boxA[3], boxB[3]) inter = max(0, xB - xA) * max(0, yB - yA) aA = (boxA[2]-boxA[0]) * (boxA[3]-boxA[1]) aB = (boxB[2]-boxB[0]) * (boxB[3]-boxB[1]) union = aA + aB - inter return inter / float(union) if union > 0 else 0.0 def _log_match_event(self, score, bbox): """Log a match event for the dashboard.""" event = { "timestamp": time.time(), "name": self._cached_profile.get("name", "Unknown"), "score": round(score, 3), "bbox": [int(c) for c in bbox], "scores": self.verifier.live_scores.copy(), } self.match_log.append(event) # Keep last 50 events if len(self.match_log) > 50: self.match_log = self.match_log[-50:] def _extract_keypoints_for_bbox(self, frame, bbox): """Run YOLOv8-pose on a person crop and return keypoints.""" x1, y1, x2, y2 = [int(c) for c in bbox] crop = frame[y1:y2, x1:x2] if crop.size == 0: return None, None try: results = self.yolo_pose(crop, verbose=False, conf=0.3) for r in results: if r.keypoints is not None and len(r.keypoints.data) > 0: kp_data = r.keypoints.data[0].cpu().numpy() kp_xy = kp_data[:, :2] kp_conf = kp_data[:, 2] # Offset keypoints to full-frame coordinates kp_xy[:, 0] += x1 kp_xy[:, 1] += y1 return kp_xy, kp_conf except Exception: pass return None, None # ================================================================ # MAIN PROCESS_FRAME — Temporal State Machine # ================================================================ def process_frame(self, frame, person_detections): """ Process one frame during active target search. Args: frame: Full BGR frame person_detections: List of [x1,y1,x2,y2] bounding boxes Returns: list of (label, score, bbox) results to draw. """ v = self.verifier now = time.time() results = [] if not person_detections or len(person_detections) == 0: if v.state == "verifying": if now - v.last_seen_time > v.CANDIDATE_LOST_TIMEOUT: v.status_message = "Candidate lost. Rescanning..." v.reset() elif v.state == "confirmed": if now - v.last_seen_time > v.CONFIRMED_LOST_TIMEOUT: v.state = "lost" v.status_message = "Target lost from view" return results # ---- STATE: SCANNING ---- if v.state == "scanning": best_candidate = None best_score = 0.0 for bbox in person_detections: x1, y1, x2, y2 = [int(c) for c in bbox] crop = frame[y1:y2, x1:x2] if crop.size == 0: continue # Height pre-filter kp_xy, kp_conf = self._extract_keypoints_for_bbox(frame, bbox) if kp_xy is not None and kp_conf is not None: h_est = self.height_estimator.estimate_height( kp_xy, kp_conf, [x1,y1,x2,y2], frame.shape[:2] ) if not HeightEstimator.passes_prefilter( h_est, self._cached_profile.get("height_cm"), self._cached_profile.get("height_ratio") ): results.append(("Unknown", 0.0, np.array([x1,y1,x2,y2]))) continue # Quick multi-signal screening face_score, face_emb, face_detected = self._compute_face_score(crop) app_score, app_emb = self._compute_appearance_score(crop) clothing_score = self._compute_clothing_score(crop) # Screening score if face_detected and face_score >= v.FACE_MATCH_THRESHOLD: screen_score = face_score * 0.5 + app_score * 0.3 + clothing_score * 0.2 elif not face_detected: screen_score = app_score * 0.5 + clothing_score * 0.3 screen_score *= 0.6 else: screen_score = 0.0 if screen_score > best_score: best_score = screen_score best_candidate = { "bbox": [x1, y1, x2, y2], "face_score": face_score, "face_emb": face_emb, "face_detected": face_detected, "app_score": app_score, "app_emb": app_emb, "clothing_score": clothing_score, } results.append(("Unknown", 0.0, np.array([x1, y1, x2, y2]))) # Transition to VERIFYING if best_candidate and best_score >= 0.15: v.state = "verifying" v.candidate_bbox = best_candidate["bbox"] v.verify_start_time = now v.last_seen_time = now v.consecutive_match_frames = 0 if best_candidate["face_detected"] and best_candidate["face_score"] >= v.FACE_MATCH_THRESHOLD: v.face_scores.append(best_candidate["face_score"]) if best_candidate["app_emb"] is not None: v.appearance_embeddings.append(best_candidate["app_emb"]) v.clothing_scores.append(best_candidate["clothing_score"]) sil = self.gait_engine.extract_silhouette(frame, best_candidate["bbox"]) if sil is not None: v.gait_silhouettes.append(sil) v.status_message = "Potential target found. Verifying..." v.verification_progress = 0.05 return results # ---- STATE: VERIFYING ---- elif v.state == "verifying": best_iou = 0.0 matched_bbox = None for bbox in person_detections: iou = self._bb_iou(v.candidate_bbox, [int(c) for c in bbox]) if iou > best_iou: best_iou = iou matched_bbox = [int(c) for c in bbox] if matched_bbox is None or best_iou < 0.15: if now - v.last_seen_time > v.CANDIDATE_LOST_TIMEOUT: v.status_message = "Candidate lost. Rescanning..." v.reset() for bbox in person_detections: x1, y1, x2, y2 = [int(c) for c in bbox] results.append(("Unknown", 0.0, np.array([x1,y1,x2,y2]))) return results # Smooth bbox tracking v.candidate_bbox = [ int(0.7 * v.candidate_bbox[i] + 0.3 * matched_bbox[i]) for i in range(4) ] v.last_seen_time = now # Collect evidence x1, y1, x2, y2 = matched_bbox crop = frame[y1:y2, x1:x2] if crop.size > 0: # Keypoints kp_xy, kp_conf = self._extract_keypoints_for_bbox(frame, matched_bbox) # Pose quality gate if kp_xy is not None and kp_conf is not None: avg_conf = BiomechEngine.avg_keypoint_confidence(kp_conf) v.pose_confidence_history.append(avg_conf) if avg_conf >= v.MIN_POSE_CONFIDENCE: # Biomech b_score, b_vec = self._compute_biomech_score(kp_xy, kp_conf, matched_bbox) if b_vec is not None: v.biomech_vectors.append(b_vec) # Height h_score, h_est = self._compute_height_score(kp_xy, kp_conf, matched_bbox, frame.shape[:2]) v.height_estimates.append(h_est) # Face face_score, face_emb, face_detected = self._compute_face_score(crop) if face_detected and face_score >= v.FACE_MATCH_THRESHOLD: v.face_scores.append(face_score) elif face_detected and face_score < 0.20: v.status_message = "Face mismatch detected. Rescanning..." v.reset() for bbox in person_detections: bx = [int(c) for c in bbox] results.append(("Unknown", 0.0, np.array(bx))) return results # Appearance app_score, app_emb = self._compute_appearance_score(crop) if app_emb is not None: v.appearance_embeddings.append(app_emb) # Clothing clth_score = self._compute_clothing_score(crop) v.clothing_scores.append(clth_score) # Gait silhouette sil = self.gait_engine.extract_silhouette(frame, matched_bbox) if sil is not None: v.gait_silhouettes.append(sil) # Compute verification progress elapsed = now - v.verify_start_time time_factor = min(1.0, elapsed / v.MIN_VERIFY_SECONDS) gait_factor = min(1.0, len(v.gait_silhouettes) / v.GAIT_CYCLE_FRAMES) face_factor = min(1.0, len(v.face_scores) / 5) if self._cached_face_embs is not None else 0.5 app_factor = min(1.0, len(v.appearance_embeddings) / 5) v.verification_progress = (time_factor * 0.2 + face_factor * 0.3 + gait_factor * 0.25 + app_factor * 0.25) # Compute live scores gait_sim = self._compute_live_gait_score() biomech_sim = BiomechEngine.compute_similarity( BiomechEngine.average_embeddings(v.biomech_vectors) if v.biomech_vectors else None, self._cached_biomech_vec ) if v.biomech_vectors and self._cached_biomech_vec is not None else 0.0 appearance_sim = AppearanceEngine.compute_similarity( AppearanceEngine.average_embeddings(v.appearance_embeddings), self._cached_appearance_vec ) if v.appearance_embeddings and self._cached_appearance_vec is not None else 0.0 height_sim = float(np.mean([ HeightEstimator.compute_match_score(h, self._cached_profile.get("height_cm"), self._cached_profile.get("height_ratio")) for h in v.height_estimates[-10:] ])) if v.height_estimates else 0.5 face_sim_avg = float(np.mean(v.face_scores)) if v.face_scores else None v.live_scores = { "gait": round(gait_sim, 3), "biomech": round(biomech_sim, 3), "appearance": round(appearance_sim, 3), "height": round(height_sim, 3), "face": round(face_sim_avg, 3) if face_sim_avg else None, "ensemble": 0.0, } ensemble = self._compute_weighted_ensemble( gait_sim, biomech_sim, appearance_sim, height_sim, face_sim_avg ) v.live_scores["ensemble"] = round(ensemble, 3) v.status_message = (f"Verifying... gait:{len(v.gait_silhouettes)}f " f"face:{len(v.face_scores)} " f"app:{len(v.appearance_embeddings)} " f"({v.verification_progress:.0%})") # Check if we can confirm can_confirm = False if elapsed >= v.MIN_VERIFY_SECONDS: # Signal agreement gate if self._check_signal_agreement(gait_sim, biomech_sim, appearance_sim): can_confirm = ensemble >= v.CONFIRM_SCORE_THRESHOLD elif face_sim_avg and face_sim_avg >= 0.5 and len(v.face_scores) >= 3: # Strong face can override signal agreement can_confirm = ensemble >= v.CONFIRM_SCORE_THRESHOLD else: # Weaker case: need higher threshold can_confirm = ensemble >= v.CONFIRM_SCORE_THRESHOLD + 0.10 # Temporal smoothing check if ensemble >= v.POSSIBLE_MATCH_THRESHOLD: v.consecutive_match_frames += 1 else: v.consecutive_match_frames = max(0, v.consecutive_match_frames - 2) if can_confirm and v.consecutive_match_frames < v.REQUIRED_CONSECUTIVE_FRAMES: can_confirm = False if can_confirm: v.state = "confirmed" v.confirmed_score = ensemble v.confirmed_bbox = matched_bbox v.verification_progress = 1.0 match_pct = ensemble * 100 v.status_message = f"TARGET CONFIRMED — {match_pct:.1f}% confidence" self._log_match_event(ensemble, matched_bbox) print(f"[ProfileEngine] TARGET CONFIRMED: {self._cached_profile.get('name', '?')} " f"score={ensemble:.3f} gait={gait_sim:.2f} biomech={biomech_sim:.2f} " f"app={appearance_sim:.2f} height={height_sim:.2f} face={face_sim_avg}") # Draw results for bbox in person_detections: bx1, by1, bx2, by2 = [int(c) for c in bbox] iou = self._bb_iou(matched_bbox, [bx1, by1, bx2, by2]) if iou > 0.3: label = f"[VERIFYING] {v.verification_progress:.0%}" results.append((label, v.verification_progress, np.array([bx1,by1,bx2,by2]))) else: results.append(("Unknown", 0.0, np.array([bx1,by1,bx2,by2]))) return results # ---- STATE: CONFIRMED ---- elif v.state == "confirmed": best_iou = 0.0 matched_bbox = None for bbox in person_detections: iou = self._bb_iou(v.candidate_bbox, [int(c) for c in bbox]) if iou > best_iou: best_iou = iou matched_bbox = [int(c) for c in bbox] if matched_bbox and best_iou >= 0.15: v.candidate_bbox = [ int(0.7 * v.candidate_bbox[i] + 0.3 * matched_bbox[i]) for i in range(4) ] v.last_seen_time = now v.confirmed_bbox = matched_bbox # Continue updating evidence x1, y1, x2, y2 = matched_bbox crop = frame[y1:y2, x1:x2] if crop.size > 0: # Gait sil = self.gait_engine.extract_silhouette(frame, matched_bbox) if sil is not None: v.gait_silhouettes.append(sil) if len(v.gait_silhouettes) > 120: v.gait_silhouettes = v.gait_silhouettes[-60:] # Appearance app_score, app_emb = self._compute_appearance_score(crop) # Recompute ensemble gait_sim = self._compute_live_gait_score() biomech_sim = v.live_scores.get("biomech", 0.0) appearance_sim = app_score if app_score > 0 else v.live_scores.get("appearance", 0.0) height_sim = v.live_scores.get("height", 0.5) face_sim_avg = v.live_scores.get("face") ensemble = self._compute_weighted_ensemble( gait_sim, biomech_sim, appearance_sim, height_sim, face_sim_avg ) v.confirmed_score = 0.9 * v.confirmed_score + 0.1 * ensemble v.live_scores["ensemble"] = round(v.confirmed_score, 3) match_pct = v.confirmed_score * 100 if match_pct >= 85: v.status_message = f"TARGET LOCKED — {match_pct:.1f}% confidence" elif match_pct >= 65: v.status_message = f"POSSIBLE MATCH — {match_pct:.1f}%" else: if now - v.last_seen_time > v.CONFIRMED_LOST_TIMEOUT: v.state = "lost" v.status_message = "Target lost from view. Rescanning..." # Draw results for bbox in person_detections: bx1, by1, bx2, by2 = [int(c) for c in bbox] if matched_bbox and self._bb_iou(matched_bbox, [bx1,by1,bx2,by2]) > 0.3: name = self._cached_profile.get("name", "Target") results.append((name, v.confirmed_score, np.array([bx1,by1,bx2,by2]))) else: results.append(("Unknown", 0.0, np.array([bx1,by1,bx2,by2]))) return results # ---- STATE: LOST ---- elif v.state == "lost": v.status_message = "Target lost. Attempting re-acquisition..." best_reacq_score = 0.0 best_reacq_bbox = None for bbox in person_detections: x1, y1, x2, y2 = [int(c) for c in bbox] crop = frame[y1:y2, x1:x2] if crop.size == 0: continue face_score, _, face_detected = self._compute_face_score(crop) app_score, _ = self._compute_appearance_score(crop) clothing_score = self._compute_clothing_score(crop) reacq_score = 0.0 if face_detected and face_score >= v.FACE_MATCH_THRESHOLD: reacq_score = face_score * 0.5 + app_score * 0.3 + clothing_score * 0.2 elif not face_detected: reacq_score = app_score * 0.4 + clothing_score * 0.3 if reacq_score > best_reacq_score: best_reacq_score = reacq_score best_reacq_bbox = [x1, y1, x2, y2] if best_reacq_bbox and best_reacq_score >= 0.35: v.state = "confirmed" v.candidate_bbox = best_reacq_bbox v.confirmed_bbox = best_reacq_bbox v.last_seen_time = now v.confirmed_score = best_reacq_score v.status_message = f"Target re-acquired — {best_reacq_score:.0%}" elif now - v.last_seen_time > v.CONFIRMED_LOST_TIMEOUT * 3: v.reset() v.status_message = "Target lost too long. Rescanning..." for bbox in person_detections: bx1, by1, bx2, by2 = [int(c) for c in bbox] if (best_reacq_bbox and v.state == "confirmed" and self._bb_iou(best_reacq_bbox, [bx1,by1,bx2,by2]) > 0.3): name = self._cached_profile.get("name", "Target") results.append((name, v.confirmed_score, np.array([bx1,by1,bx2,by2]))) else: results.append(("Unknown", 0.0, np.array([bx1,by1,bx2,by2]))) return results # Fallback for bbox in person_detections: x1, y1, x2, y2 = [int(c) for c in bbox] results.append(("Unknown", 0.0, np.array([x1,y1,x2,y2]))) return results # Legacy single-frame matcher (kept for backward compat) def match_person(self, person_crop, target_profile): """ Legacy single-frame matcher. Kept for API compatibility. New code should use start_search() + process_frame() flow. """ self._load_target_cache(target_profile) face_score, _, face_detected = self._compute_face_score(person_crop) app_score, _ = self._compute_appearance_score(person_crop) clothing_score = self._compute_clothing_score(person_crop) has_face_data = self._cached_face_embs is not None and len(self._cached_face_embs) > 0 if face_detected and face_score >= 0.35 and has_face_data: final = face_score * 0.50 + app_score * 0.30 + clothing_score * 0.20 return final elif face_detected and face_score < 0.25: return 0.0 elif not face_detected: return min(0.40, app_score * 0.5 + clothing_score * 0.5) else: return 0.0