""" Gesture validation service for identity verification. This module provides gesture validation functionality by leveraging the existing gesture detection system in src/gesturedetection/. It processes user videos to detect specific gestures and validates them against a list of required gestures. """ import os import logging import tempfile from typing import List, Dict, Any, Optional, Tuple from datetime import datetime, timezone from .models import ValidationResult, ValidationStatus, GestureRequirement logger = logging.getLogger(__name__) class GestureValidator: """ Gesture validation service for identity verification. This class processes user videos to detect and validate specific gestures against a list of required gestures. It uses the existing gesture detection pipeline from src/gesturedetection/ and provides configurable validation parameters including error margins and minimum requirements. """ def __init__( self, detector_path: str = "models/hand_detector.onnx", classifier_path: str = "models/crops_classifier.onnx", frame_skip: int = 1, min_gesture_duration: int = 5, confidence_threshold: float = 0.7 ): """ Initialize the gesture validator. Parameters ---------- detector_path : str, optional Path to the hand detection ONNX model, by default "models/hand_detector.onnx" classifier_path : str, optional Path to the gesture classification ONNX model, by default "models/crops_classifier.onnx" frame_skip : int, optional Number of frames to skip between processing, by default 1 min_gesture_duration : int, optional Minimum duration for gesture detection, by default 5 confidence_threshold : float, optional Minimum confidence threshold for gesture detection, by default 0.7 """ self.detector_path = detector_path self.classifier_path = classifier_path self.frame_skip = frame_skip self.min_gesture_duration = min_gesture_duration self.confidence_threshold = confidence_threshold # Import here to avoid circular imports and handle missing dependencies gracefully try: from ..gesturedetection.main_controller import MainController from ..gesturedetection.models import PRODUCTION_GESTURE_MAPPING self._main_controller_class = MainController self._gesture_mapping = PRODUCTION_GESTURE_MAPPING self._initialized = True logger.info("GestureValidator initialized successfully with PRODUCTION_GESTURE_MAPPING") except ImportError as e: logger.warning(f"Could not import gesture detection components: {e}") self._initialized = False def validate_gestures( self, video_path: str, required_gestures: List[str], error_margin: float = 0.33, require_all: bool = True ) -> ValidationResult: """ Validate that required gestures are present in the video. Parameters ---------- video_path : str Path to the video file to analyze required_gestures : List[str] List of gesture names that must be detected error_margin : float, optional Fraction of gestures that can be missed (0.0-1.0), by default 0.33 require_all : bool, optional Whether all gestures must be present, by default True Returns ------- ValidationResult Validation result with success status and detailed metrics """ if not self._initialized: error_msg = "GestureValidator not properly initialized - missing gesture detection components" logger.error(error_msg) return ValidationResult( status=ValidationStatus.FAILED, success=False, confidence=0.0, error_message=error_msg ) logger.info(f"Starting gesture validation for video: {video_path}") logger.info(f"Required gestures: {required_gestures}, error_margin: {error_margin}") # Validate input file if not os.path.exists(video_path): error_msg = f"Video file not found: {video_path}" logger.error(error_msg) return ValidationResult( status=ValidationStatus.FAILED, success=False, confidence=0.0, error_message=error_msg ) # Validate required gestures if not required_gestures: error_msg = "No gestures specified for validation" logger.error(error_msg) return ValidationResult( status=ValidationStatus.FAILED, success=False, confidence=0.0, error_message=error_msg ) try: # Process video using existing gesture detection pipeline detected_gestures = self._process_video_for_gestures(video_path) # Analyze detected gestures against requirements validation_metrics = self._analyze_gesture_requirements( detected_gestures, required_gestures, error_margin, require_all ) # Determine overall success if require_all: success = validation_metrics["required_gestures_met"] >= len(required_gestures) else: # Allow for error margin min_required = max(1, int(len(required_gestures) * (1.0 - error_margin))) success = validation_metrics["required_gestures_met"] >= min_required # Calculate confidence based on detection quality confidence = self._calculate_confidence(detected_gestures, validation_metrics) status = ValidationStatus.SUCCESS if success else ValidationStatus.PARTIAL result = ValidationResult( status=status, success=success, confidence=confidence, details={ "detected_gestures": [ { "gesture": g["gesture"], "duration": g["duration"], "confidence": g["confidence"] } for g in detected_gestures ], "validation_metrics": validation_metrics, "required_gestures": required_gestures, "error_margin": error_margin, "require_all": require_all, "processing_timestamp": datetime.now(timezone.utc).isoformat() } ) logger.info(f"Gesture validation completed: success={success}, confidence={confidence}") return result except Exception as e: error_msg = f"Error during gesture validation: {str(e)}" logger.error(error_msg, exc_info=True) return ValidationResult( status=ValidationStatus.FAILED, success=False, confidence=0.0, error_message=error_msg ) def _process_video_for_gestures(self, video_path: str) -> List[Dict[str, Any]]: """ Process video file to detect gestures using existing pipeline. Parameters ---------- video_path : str Path to the video file Returns ------- List[Dict[str, Any]] List of detected gestures with metadata """ logger.debug(f"Processing video for gestures: {video_path}") # Initialize the main controller controller = self._main_controller_class(self.detector_path, self.classifier_path) # Import video processing function from existing API try: from ..gesturedetection.api import process_video_for_gestures gestures = process_video_for_gestures( video_path, detector_path=self.detector_path, classifier_path=self.classifier_path, frame_skip=self.frame_skip ) except ImportError: # Fallback: use controller directly if import fails logger.warning("Using fallback gesture processing method") gestures = self._process_video_with_controller(controller, video_path) # Convert to our internal format detected_gestures = [] for gesture in gestures: # Map gesture names to standardized format gesture_name = self._normalize_gesture_name(gesture.gesture) detected_gestures.append({ "gesture": gesture_name, "duration": gesture.duration, "confidence": gesture.confidence, "raw_gesture": gesture.gesture }) logger.debug(f"Detected {len(detected_gestures)} gestures") return detected_gestures def _process_video_with_controller(self, controller, video_path: str) -> List[Dict[str, Any]]: """ Fallback method to process video using controller directly. This is used if the import from api.py fails for any reason. """ import cv2 from collections import defaultdict logger.debug("Processing video with controller fallback method") # Open video file cap = cv2.VideoCapture(video_path) if not cap.isOpened(): raise ValueError(f"Could not open video file: {video_path}") gesture_tracks = defaultdict(list) frame_count = 0 try: while True: ret, frame = cap.read() if not ret: break # Skip frames based on frame_skip parameter if frame_count % self.frame_skip == 0: # Process frame through the controller bboxes, ids, labels = controller(frame) if bboxes is not None and ids is not None and labels is not None: # Track gestures for each detected hand for i in range(len(bboxes)): hand_id = int(ids[i]) gesture_id = labels[i] if gesture_id is not None: confidence = 0.8 # Default confidence gesture_tracks[hand_id].append((gesture_id, confidence)) frame_count += 1 finally: cap.release() # Process gesture tracks to find continuous gestures detected_gestures = [] for hand_id, gesture_sequence in gesture_tracks.items(): if not gesture_sequence: continue # Group consecutive identical gestures current_gesture = None current_duration = 0 current_confidence = 0.0 for gesture_id, confidence in gesture_sequence: if current_gesture is None or current_gesture != gesture_id: # Save previous gesture if it was significant if current_gesture is not None and current_duration >= self.min_gesture_duration: gesture_name = self._gesture_mapping.get(current_gesture, f"unknown_{current_gesture}") avg_confidence = current_confidence / current_duration if current_duration > 0 else 0.0 scaled_duration = current_duration * self.frame_skip detected_gestures.append({ "gesture": gesture_name, "duration": scaled_duration, "confidence": avg_confidence }) # Start new gesture current_gesture = gesture_id current_duration = 1 current_confidence = confidence else: # Continue current gesture current_duration += 1 current_confidence += confidence # Don't forget the last gesture if current_gesture is not None and current_duration >= self.min_gesture_duration: gesture_name = self._gesture_mapping.get(current_gesture, f"unknown_{current_gesture}") avg_confidence = current_confidence / current_duration if current_duration > 0 else 0.0 scaled_duration = current_duration * self.frame_skip detected_gestures.append({ "gesture": gesture_name, "duration": scaled_duration, "confidence": avg_confidence }) return detected_gestures def _analyze_gesture_requirements( self, detected_gestures: List[Dict[str, Any]], required_gestures: List[str], error_margin: float, require_all: bool ) -> Dict[str, Any]: """ Analyze detected gestures against requirements. Parameters ---------- detected_gestures : List[Dict[str, Any]] List of detected gestures required_gestures : List[str] List of required gesture names error_margin : float Error margin for validation require_all : bool Whether all gestures are required Returns ------- Dict[str, Any] Validation metrics and analysis """ logger.debug("Analyzing gesture requirements") # Create lookup for detected gestures detected_gesture_counts = {} for gesture in detected_gestures: gesture_name = gesture["gesture"] if gesture_name not in detected_gesture_counts: detected_gesture_counts[gesture_name] = [] detected_gesture_counts[gesture_name].append(gesture) # Analyze each required gesture required_gestures_met = 0 gesture_analysis = {} for required_gesture in required_gestures: detected_instances = detected_gesture_counts.get(required_gesture, []) # Filter by minimum duration and confidence if specified valid_instances = [ g for g in detected_instances if g["duration"] >= self.min_gesture_duration and g["confidence"] >= self.confidence_threshold ] met_requirement = len(valid_instances) > 0 gesture_analysis[required_gesture] = { "required": True, "detected": len(detected_instances), "valid_instances": len(valid_instances), "met_requirement": met_requirement, "best_confidence": max([g["confidence"] for g in detected_instances], default=0.0), "best_duration": max([g["duration"] for g in detected_instances], default=0) } if met_requirement: required_gestures_met += 1 # Calculate success rate total_required = len(required_gestures) success_rate = required_gestures_met / total_required if total_required > 0 else 0.0 # Determine if validation passes based on error margin if require_all: passes_validation = required_gestures_met >= total_required else: min_required = max(1, int(total_required * (1.0 - error_margin))) passes_validation = required_gestures_met >= min_required metrics = { "total_required_gestures": total_required, "required_gestures_met": required_gestures_met, "success_rate": success_rate, "passes_validation": passes_validation, "error_margin": error_margin, "require_all": require_all, "gesture_analysis": gesture_analysis } logger.debug(f"Gesture analysis completed: {required_gestures_met}/{total_required} gestures met requirement") return metrics def _calculate_confidence( self, detected_gestures: List[Dict[str, Any]], validation_metrics: Dict[str, Any] ) -> float: """ Calculate overall confidence score for gesture validation. Parameters ---------- detected_gestures : List[Dict[str, Any]] List of detected gestures validation_metrics : Dict[str, Any] Validation metrics from analysis Returns ------- float Overall confidence score (0.0-1.0) """ if not detected_gestures: return 0.0 # Base confidence on success rate success_rate = validation_metrics.get("success_rate", 0.0) # Boost confidence based on average gesture quality if detected_gestures: avg_confidence = sum(g["confidence"] for g in detected_gestures) / len(detected_gestures) avg_duration = sum(g["duration"] for g in detected_gestures) / len(detected_gestures) # Normalize duration to confidence boost (longer, more confident gestures = higher score) duration_boost = min(0.2, avg_duration / 100.0) # Cap at 0.2 boost confidence_boost = min(0.1, avg_confidence * 0.1) # Cap at 0.1 boost success_rate = min(1.0, success_rate + duration_boost + confidence_boost) return success_rate def _normalize_gesture_name(self, gesture_name: str) -> str: """ Normalize gesture names to production-standard format. Handles legacy naming and variations to ensure consistent gesture names across different parts of the system. Maps old names like "like" to "thumbs_up", and handles hand-agnostic counting variations. Parameters ---------- gesture_name : str Raw gesture name from detection Returns ------- str Normalized gesture name matching PRODUCTION_GESTURE_MAPPING """ # Convert to lowercase and remove common variations normalized = gesture_name.lower().strip() # Handle common variations and legacy names variations = { "thumbs_up": ["thumbsup", "thumb_up", "like"], # "like" is legacy name "one": ["one_finger", "one_left", "one_right", "one_down"], # Hand-agnostic "two": ["peace_sign", "victory", "two_fingers", "two_up", "two_left", "two_right", "two_down"], # Hand-agnostic "three": ["three_fingers", "three2", "three3"], # Hand-agnostic "four": ["four_fingers"], "five": ["palm", "open_palm", "five_fingers"], # "palm" is alias for "five" "peace_inverted": ["peace_inverted_sign"], "ok": ["okay", "ok_sign"], "call": ["call_me", "phone"], "fist": ["closed_fist"], "point": ["pointing"], "stop": ["stop_sign"], "middle_finger": ["middle"], } for standard_name, variant_list in variations.items(): if normalized in variant_list or normalized == standard_name: return standard_name return normalized