#!/usr/bin/env python3 """ Main entry point for the unified gesture detection and identity validation API. Provides a flat API structure with all endpoints at the root level. """ import uvicorn import os import sys import tempfile import time import json import logging from typing import Optional from datetime import datetime, timezone # Add the project root to Python path project_root = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, project_root) from fastapi import FastAPI, UploadFile, File, Form, HTTPException, Depends from fastapi.responses import ORJSONResponse # Import gesture detection functionality from src.gesturedetection.api import process_video_for_gestures from src.gesturedetection.models import GestureResponse # Import validation functionality from src.validate.models import ValidationRequest, ValidationResponse, ValidationStatus from src.validate.facial_validator import FacialValidator from src.validate.gesture_validator import GestureValidator from src.validate.api import get_validation_request from src.validate.config import config # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Create main FastAPI application app = FastAPI( title="Gesture Detection & Identity Validation API", description="Unified API for gesture detection and identity validation services", version="1.0.0", docs_url="/docs", redoc_url="/redoc", default_response_class=ORJSONResponse ) # Initialize validators for validation endpoint facial_validator = FacialValidator() gesture_validator = GestureValidator() @app.get("/") async def root(): """ Root endpoint providing API information. Returns ------- dict API information and available endpoints """ return { "name": "Gesture Detection & Identity Validation API", "version": "1.0.0", "description": "Unified API providing gesture detection and identity validation services", "endpoints": { "GET /": "API information", "GET /health": "Health check", "POST /validate": "Validate identity using facial recognition and gestures", "POST /gestures": "Detect gestures in video", "GET /docs": "Interactive API documentation" } } @app.get("/health") async def health(): """ Health check endpoint for the unified API. Returns ------- dict Health status of all service components """ return { "status": "healthy", "service": "unified-api", "version": "1.0.0", "timestamp": datetime.now(timezone.utc).isoformat(), "components": { "gesture_detection": "available", "identity_validation": "available", "facial_validator": "initialized", "gesture_validator": "initialized" } } @app.post("/gestures", response_model=GestureResponse) async def detect_gestures(video: UploadFile = File(...), frame_skip: int = Form(1)): """ Detect gestures in an uploaded video file. Parameters ---------- video : UploadFile The video file to process frame_skip : int Number of frames to skip between processing (1 = process every frame, 3 = process every 3rd frame) Returns ------- GestureResponse Response containing detected gestures with duration and confidence """ logger.info(f"Gesture detection request received: {video.filename}") # Validate file type if not video.content_type or not video.content_type.startswith('video/'): raise HTTPException(status_code=400, detail="File must be a video") # Create temporary file to save uploaded video with tempfile.NamedTemporaryFile(delete=False, suffix='.mp4') as temp_file: try: # Write uploaded content to temporary file content = await video.read() temp_file.write(content) temp_file.flush() logger.info(f"Processing video: {temp_file.name} ({len(content)} bytes)") # Process the video with frame skip parameter gestures = process_video_for_gestures(temp_file.name, frame_skip=frame_skip) logger.info(f"Gesture detection completed: {len(gestures)} gestures detected") return GestureResponse(gestures=gestures) except Exception as e: logger.error(f"Error processing video: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail=f"Error processing video: {str(e)}") finally: # Clean up temporary file if os.path.exists(temp_file.name): os.unlink(temp_file.name) logger.debug(f"Cleaned up temporary file: {temp_file.name}") @app.post("/validate", response_model=ValidationResponse) async def validate_identity( photo: UploadFile = File(...), video: UploadFile = File(...), request: ValidationRequest = Depends(get_validation_request) ): """ Validate user identity using facial recognition and gesture validation. This endpoint accepts an ID document photo, a user video containing the person's face and required gestures, and a list of gestures that must be performed. It returns validation results for both facial recognition and gesture compliance. Parameters ---------- photo : UploadFile ID document photo file (image format) video : UploadFile User video file containing face and gestures (video format) request : ValidationRequest Validation configuration and gesture requirements Returns ------- ValidationResponse Validation results with success indicators and optional details Raises ------ HTTPException If validation fails or processing errors occur """ start_time = time.time() logger.info(f"Identity validation request received for {request.asked_gestures}") # Validate file types if not photo.content_type or not photo.content_type.startswith(('image/', 'application/')): raise HTTPException( status_code=400, detail="Photo file must be an image" ) if not video.content_type or not video.content_type.startswith('video/'): raise HTTPException( status_code=400, detail="Video file must be a video" ) # Validate file sizes (basic check) MAX_FILE_SIZE = 100 * 1024 * 1024 # 100MB if photo.size and photo.size > MAX_FILE_SIZE: raise HTTPException( status_code=413, detail="Photo file too large (max 100MB)" ) if video.size and video.size > MAX_FILE_SIZE: raise HTTPException( status_code=413, detail="Video file too large (max 100MB)" ) # Create temporary files for processing temp_photo = None temp_video = None try: # Save uploaded files to temporary location with tempfile.NamedTemporaryFile(delete=False, suffix=f"_photo.{photo.filename.split('.')[-1] if '.' in photo.filename else 'jpg'}") as temp_photo_file: temp_photo = temp_photo_file.name photo_content = await photo.read() temp_photo_file.write(photo_content) with tempfile.NamedTemporaryFile(delete=False, suffix=f"_video.{video.filename.split('.')[-1] if '.' in video.filename else 'mp4'}") as temp_video_file: temp_video = temp_video_file.name video_content = await video.read() temp_video_file.write(video_content) logger.info(f"Files saved: photo={temp_photo}, video={temp_video}") # Perform facial validation logger.info("Starting facial validation") # Update facial validator with request-specific parameters if provided if request.similarity_threshold is not None: facial_validator.similarity_threshold = request.similarity_threshold if request.frame_sample_rate is not None: facial_validator.frame_sample_rate = request.frame_sample_rate face_result = facial_validator.validate_facial_match(temp_photo, temp_video) # Perform gesture validation logger.info("Starting gesture validation") # Update gesture validator with request-specific parameters if provided if request.confidence_threshold is not None: gesture_validator.confidence_threshold = request.confidence_threshold if request.min_gesture_duration is not None: gesture_validator.min_gesture_duration = request.min_gesture_duration gesture_result = gesture_validator.validate_gestures( temp_video, request.asked_gestures, error_margin=request.error_margin, require_all=request.require_all_gestures ) # Determine overall result overall_success = face_result.success and gesture_result.success overall_status = ValidationStatus.SUCCESS if overall_success else ValidationStatus.PARTIAL # Calculate processing time processing_time_ms = int((time.time() - start_time) * 1000) # Build response response = ValidationResponse( face=face_result.success, gestures=gesture_result.success, overall=overall_success, status=overall_status, face_result=face_result if request.include_details else None, gesture_result=gesture_result if request.include_details else None, processing_time_ms=processing_time_ms, timestamp=datetime.now(timezone.utc).isoformat() ) # Log results logger.info( "Identity validation completed", extra={ "face_success": face_result.success, "gesture_success": gesture_result.success, "overall_success": overall_success, "processing_time_ms": processing_time_ms, "requested_gestures": request.asked_gestures } ) return response except Exception as e: logger.error(f"Error during identity validation: {str(e)}", exc_info=True) raise HTTPException( status_code=500, detail=f"Internal server error during validation: {str(e)}" ) finally: # Clean up temporary files for temp_file in [temp_photo, temp_video]: if temp_file and os.path.exists(temp_file): try: os.unlink(temp_file) logger.debug(f"Cleaned up temporary file: {temp_file}") except Exception as e: logger.warning(f"Failed to clean up temporary file {temp_file}: {e}") def main(): """Start the unified API server.""" # Get port from environment variable, default to 7860 for HF Spaces compatibility port = int(os.getenv("PORT", 7860)) print("🚀 Starting Unified Gesture Detection & Identity Validation API") print(f"📍 API will be available at: http://localhost:{port}") print(f"📚 API documentation at: http://localhost:{port}/docs") print(f"❤️ Health check at: http://localhost:{port}/health") print(f"🔐 Identity validation at: POST http://localhost:{port}/validate") print(f"👋 Gesture detection at: POST http://localhost:{port}/gestures") print("\nPress Ctrl+C to stop the server") uvicorn.run( app, host="0.0.0.0", port=port, reload=False, # Disable reload in production/Docker log_level="info" ) if __name__ == "__main__": main()