algoryn/validation
0
1"""2Gesture validation service for identity verification.3 4This module provides gesture validation functionality by leveraging the existing5gesture detection system in src/gesturedetection/. It processes user videos to6detect specific gestures and validates them against a list of required gestures.7"""8 9import os10import logging11import tempfile12from typing import List, Dict, Any, Optional, Tuple13from datetime import datetime, timezone14 15from .models import ValidationResult, ValidationStatus, GestureRequirement16 17logger = logging.getLogger(__name__)18 19 20class GestureValidator:21 """22 Gesture validation service for identity verification.23 24 This class processes user videos to detect and validate specific gestures25 against a list of required gestures. It uses the existing gesture detection26 pipeline from src/gesturedetection/ and provides configurable validation27 parameters including error margins and minimum requirements.28 """29 30 def __init__(31 self,32 detector_path: str = "models/hand_detector.onnx",33 classifier_path: str = "models/crops_classifier.onnx",34 frame_skip: int = 1,35 min_gesture_duration: int = 5,36 confidence_threshold: float = 0.737 ):38 """39 Initialize the gesture validator.40 41 Parameters42 ----------43 detector_path : str, optional44 Path to the hand detection ONNX model, by default "models/hand_detector.onnx"45 classifier_path : str, optional46 Path to the gesture classification ONNX model, by default "models/crops_classifier.onnx"47 frame_skip : int, optional48 Number of frames to skip between processing, by default 149 min_gesture_duration : int, optional50 Minimum duration for gesture detection, by default 551 confidence_threshold : float, optional52 Minimum confidence threshold for gesture detection, by default 0.753 """54 self.detector_path = detector_path55 self.classifier_path = classifier_path56 self.frame_skip = frame_skip57 self.min_gesture_duration = min_gesture_duration58 self.confidence_threshold = confidence_threshold59 60 # Import here to avoid circular imports and handle missing dependencies gracefully61 try:62 from ..gesturedetection.main_controller import MainController63 from ..gesturedetection.models import PRODUCTION_GESTURE_MAPPING64 self._main_controller_class = MainController65 self._gesture_mapping = PRODUCTION_GESTURE_MAPPING66 self._initialized = True67 logger.info("GestureValidator initialized successfully with PRODUCTION_GESTURE_MAPPING")68 except ImportError as e:69 logger.warning(f"Could not import gesture detection components: {e}")70 self._initialized = False71 72 def validate_gestures(73 self,74 video_path: str,75 required_gestures: List[str],76 error_margin: float = 0.33,77 require_all: bool = True78 ) -> ValidationResult:79 """80 Validate that required gestures are present in the video.81 82 Parameters83 ----------84 video_path : str85 Path to the video file to analyze86 required_gestures : List[str]87 List of gesture names that must be detected88 error_margin : float, optional89 Fraction of gestures that can be missed (0.0-1.0), by default 0.3390 require_all : bool, optional91 Whether all gestures must be present, by default True92 93 Returns94 -------95 ValidationResult96 Validation result with success status and detailed metrics97 """98 if not self._initialized:99 error_msg = "GestureValidator not properly initialized - missing gesture detection components"100 logger.error(error_msg)101 return ValidationResult(102 status=ValidationStatus.FAILED,103 success=False,104 confidence=0.0,105 error_message=error_msg106 )107 108 logger.info(f"Starting gesture validation for video: {video_path}")109 logger.info(f"Required gestures: {required_gestures}, error_margin: {error_margin}")110 111 # Validate input file112 if not os.path.exists(video_path):113 error_msg = f"Video file not found: {video_path}"114 logger.error(error_msg)115 return ValidationResult(116 status=ValidationStatus.FAILED,117 success=False,118 confidence=0.0,119 error_message=error_msg120 )121 122 # Validate required gestures123 if not required_gestures:124 error_msg = "No gestures specified for validation"125 logger.error(error_msg)126 return ValidationResult(127 status=ValidationStatus.FAILED,128 success=False,129 confidence=0.0,130 error_message=error_msg131 )132 133 try:134 # Process video using existing gesture detection pipeline135 detected_gestures = self._process_video_for_gestures(video_path)136 137 # Analyze detected gestures against requirements138 validation_metrics = self._analyze_gesture_requirements(139 detected_gestures, required_gestures, error_margin, require_all140 )141 142 # Determine overall success143 if require_all:144 success = validation_metrics["required_gestures_met"] >= len(required_gestures)145 else:146 # Allow for error margin147 min_required = max(1, int(len(required_gestures) * (1.0 - error_margin)))148 success = validation_metrics["required_gestures_met"] >= min_required149 150 # Calculate confidence based on detection quality151 confidence = self._calculate_confidence(detected_gestures, validation_metrics)152 153 status = ValidationStatus.SUCCESS if success else ValidationStatus.PARTIAL154 155 result = ValidationResult(156 status=status,157 success=success,158 confidence=confidence,159 details={160 "detected_gestures": [161 {162 "gesture": g["gesture"],163 "duration": g["duration"],164 "confidence": g["confidence"]165 }166 for g in detected_gestures167 ],168 "validation_metrics": validation_metrics,169 "required_gestures": required_gestures,170 "error_margin": error_margin,171 "require_all": require_all,172 "processing_timestamp": datetime.now(timezone.utc).isoformat()173 }174 )175 176 logger.info(f"Gesture validation completed: success={success}, confidence={confidence}")177 return result178 179 except Exception as e:180 error_msg = f"Error during gesture validation: {str(e)}"181 logger.error(error_msg, exc_info=True)182 return ValidationResult(183 status=ValidationStatus.FAILED,184 success=False,185 confidence=0.0,186 error_message=error_msg187 )188 189 def _process_video_for_gestures(self, video_path: str) -> List[Dict[str, Any]]:190 """191 Process video file to detect gestures using existing pipeline.192 193 Parameters194 ----------195 video_path : str196 Path to the video file197 198 Returns199 -------200 List[Dict[str, Any]]201 List of detected gestures with metadata202 """203 logger.debug(f"Processing video for gestures: {video_path}")204 205 # Initialize the main controller206 controller = self._main_controller_class(self.detector_path, self.classifier_path)207 208 # Import video processing function from existing API209 try:210 from ..gesturedetection.api import process_video_for_gestures211 gestures = process_video_for_gestures(212 video_path,213 detector_path=self.detector_path,214 classifier_path=self.classifier_path,215 frame_skip=self.frame_skip216 )217 except ImportError:218 # Fallback: use controller directly if import fails219 logger.warning("Using fallback gesture processing method")220 gestures = self._process_video_with_controller(controller, video_path)221 222 # Convert to our internal format223 detected_gestures = []224 for gesture in gestures:225 # Map gesture names to standardized format226 gesture_name = self._normalize_gesture_name(gesture.gesture)227 228 detected_gestures.append({229 "gesture": gesture_name,230 "duration": gesture.duration,231 "confidence": gesture.confidence,232 "raw_gesture": gesture.gesture233 })234 235 logger.debug(f"Detected {len(detected_gestures)} gestures")236 return detected_gestures237 238 def _process_video_with_controller(self, controller, video_path: str) -> List[Dict[str, Any]]:239 """240 Fallback method to process video using controller directly.241 242 This is used if the import from api.py fails for any reason.243 """244 import cv2245 from collections import defaultdict246 247 logger.debug("Processing video with controller fallback method")248 249 # Open video file250 cap = cv2.VideoCapture(video_path)251 if not cap.isOpened():252 raise ValueError(f"Could not open video file: {video_path}")253 254 gesture_tracks = defaultdict(list)255 frame_count = 0256 257 try:258 while True:259 ret, frame = cap.read()260 if not ret:261 break262 263 # Skip frames based on frame_skip parameter264 if frame_count % self.frame_skip == 0:265 # Process frame through the controller266 bboxes, ids, labels = controller(frame)267 268 if bboxes is not None and ids is not None and labels is not None:269 # Track gestures for each detected hand270 for i in range(len(bboxes)):271 hand_id = int(ids[i])272 gesture_id = labels[i]273 274 if gesture_id is not None:275 confidence = 0.8 # Default confidence276 gesture_tracks[hand_id].append((gesture_id, confidence))277 278 frame_count += 1279 280 finally:281 cap.release()282 283 # Process gesture tracks to find continuous gestures284 detected_gestures = []285 286 for hand_id, gesture_sequence in gesture_tracks.items():287 if not gesture_sequence:288 continue289 290 # Group consecutive identical gestures291 current_gesture = None292 current_duration = 0293 current_confidence = 0.0294 295 for gesture_id, confidence in gesture_sequence:296 if current_gesture is None or current_gesture != gesture_id:297 # Save previous gesture if it was significant298 if current_gesture is not None and current_duration >= self.min_gesture_duration:299 gesture_name = self._gesture_mapping.get(current_gesture, f"unknown_{current_gesture}")300 avg_confidence = current_confidence / current_duration if current_duration > 0 else 0.0301 scaled_duration = current_duration * self.frame_skip302 303 detected_gestures.append({304 "gesture": gesture_name,305 "duration": scaled_duration,306 "confidence": avg_confidence307 })308 309 # Start new gesture310 current_gesture = gesture_id311 current_duration = 1312 current_confidence = confidence313 else:314 # Continue current gesture315 current_duration += 1316 current_confidence += confidence317 318 # Don't forget the last gesture319 if current_gesture is not None and current_duration >= self.min_gesture_duration:320 gesture_name = self._gesture_mapping.get(current_gesture, f"unknown_{current_gesture}")321 avg_confidence = current_confidence / current_duration if current_duration > 0 else 0.0322 scaled_duration = current_duration * self.frame_skip323 324 detected_gestures.append({325 "gesture": gesture_name,326 "duration": scaled_duration,327 "confidence": avg_confidence328 })329 330 return detected_gestures331 332 def _analyze_gesture_requirements(333 self,334 detected_gestures: List[Dict[str, Any]],335 required_gestures: List[str],336 error_margin: float,337 require_all: bool338 ) -> Dict[str, Any]:339 """340 Analyze detected gestures against requirements.341 342 Parameters343 ----------344 detected_gestures : List[Dict[str, Any]]345 List of detected gestures346 required_gestures : List[str]347 List of required gesture names348 error_margin : float349 Error margin for validation350 require_all : bool351 Whether all gestures are required352 353 Returns354 -------355 Dict[str, Any]356 Validation metrics and analysis357 """358 logger.debug("Analyzing gesture requirements")359 360 # Create lookup for detected gestures361 detected_gesture_counts = {}362 for gesture in detected_gestures:363 gesture_name = gesture["gesture"]364 if gesture_name not in detected_gesture_counts:365 detected_gesture_counts[gesture_name] = []366 detected_gesture_counts[gesture_name].append(gesture)367 368 # Analyze each required gesture369 required_gestures_met = 0370 gesture_analysis = {}371 372 for required_gesture in required_gestures:373 detected_instances = detected_gesture_counts.get(required_gesture, [])374 375 # Filter by minimum duration and confidence if specified376 valid_instances = [377 g for g in detected_instances378 if g["duration"] >= self.min_gesture_duration and379 g["confidence"] >= self.confidence_threshold380 ]381 382 met_requirement = len(valid_instances) > 0383 384 gesture_analysis[required_gesture] = {385 "required": True,386 "detected": len(detected_instances),387 "valid_instances": len(valid_instances),388 "met_requirement": met_requirement,389 "best_confidence": max([g["confidence"] for g in detected_instances], default=0.0),390 "best_duration": max([g["duration"] for g in detected_instances], default=0)391 }392 393 if met_requirement:394 required_gestures_met += 1395 396 # Calculate success rate397 total_required = len(required_gestures)398 success_rate = required_gestures_met / total_required if total_required > 0 else 0.0399 400 # Determine if validation passes based on error margin401 if require_all:402 passes_validation = required_gestures_met >= total_required403 else:404 min_required = max(1, int(total_required * (1.0 - error_margin)))405 passes_validation = required_gestures_met >= min_required406 407 metrics = {408 "total_required_gestures": total_required,409 "required_gestures_met": required_gestures_met,410 "success_rate": success_rate,411 "passes_validation": passes_validation,412 "error_margin": error_margin,413 "require_all": require_all,414 "gesture_analysis": gesture_analysis415 }416 417 logger.debug(f"Gesture analysis completed: {required_gestures_met}/{total_required} gestures met requirement")418 return metrics419 420 def _calculate_confidence(421 self,422 detected_gestures: List[Dict[str, Any]],423 validation_metrics: Dict[str, Any]424 ) -> float:425 """426 Calculate overall confidence score for gesture validation.427 428 Parameters429 ----------430 detected_gestures : List[Dict[str, Any]]431 List of detected gestures432 validation_metrics : Dict[str, Any]433 Validation metrics from analysis434 435 Returns436 -------437 float438 Overall confidence score (0.0-1.0)439 """440 if not detected_gestures:441 return 0.0442 443 # Base confidence on success rate444 success_rate = validation_metrics.get("success_rate", 0.0)445 446 # Boost confidence based on average gesture quality447 if detected_gestures:448 avg_confidence = sum(g["confidence"] for g in detected_gestures) / len(detected_gestures)449 avg_duration = sum(g["duration"] for g in detected_gestures) / len(detected_gestures)450 451 # Normalize duration to confidence boost (longer, more confident gestures = higher score)452 duration_boost = min(0.2, avg_duration / 100.0) # Cap at 0.2 boost453 confidence_boost = min(0.1, avg_confidence * 0.1) # Cap at 0.1 boost454 455 success_rate = min(1.0, success_rate + duration_boost + confidence_boost)456 457 return success_rate458 459 def _normalize_gesture_name(self, gesture_name: str) -> str:460 """461 Normalize gesture names to production-standard format.462 463 Handles legacy naming and variations to ensure consistent gesture names464 across different parts of the system. Maps old names like "like" to 465 "thumbs_up", and handles hand-agnostic counting variations.466 467 Parameters468 ----------469 gesture_name : str470 Raw gesture name from detection471 472 Returns473 -------474 str475 Normalized gesture name matching PRODUCTION_GESTURE_MAPPING476 """477 # Convert to lowercase and remove common variations478 normalized = gesture_name.lower().strip()479 480 # Handle common variations and legacy names481 variations = {482 "thumbs_up": ["thumbsup", "thumb_up", "like"], # "like" is legacy name483 "one": ["one_finger", "one_left", "one_right", "one_down"], # Hand-agnostic484 "two": ["peace_sign", "victory", "two_fingers", "two_up", "two_left", "two_right", "two_down"], # Hand-agnostic485 "three": ["three_fingers", "three2", "three3"], # Hand-agnostic486 "four": ["four_fingers"],487 "five": ["palm", "open_palm", "five_fingers"], # "palm" is alias for "five"488 "peace_inverted": ["peace_inverted_sign"],489 "ok": ["okay", "ok_sign"],490 "call": ["call_me", "phone"],491 "fist": ["closed_fist"],492 "point": ["pointing"],493 "stop": ["stop_sign"],494 "middle_finger": ["middle"],495 }496 497 for standard_name, variant_list in variations.items():498 if normalized in variant_list or normalized == standard_name:499 return standard_name500 501 return normalized502