Spaces:
Sleeping
Sleeping
File size: 14,138 Bytes
95db528 afecd35 95db528 afecd35 95db528 afecd35 95db528 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 |
import cv2
import numpy as np
import tempfile
import os
from collections import defaultdict
from typing import Dict, List, Tuple, Optional
from fastapi import FastAPI, UploadFile, File, HTTPException, Form
from fastapi.responses import ORJSONResponse
from fastapi.encoders import jsonable_encoder
from .models import Gesture, GestureResponse, GESTURE_MAPPING, FULL_GESTURE_MAPPING, PRODUCTION_GESTURE_MAPPING
from .config import get_logfire_token, is_monitoring_enabled
# Import the gesture detection components
from .main_controller import MainController
# Configure logfire monitoring if token is available
logfire = None
if is_monitoring_enabled():
try:
import logfire
logfire.configure(token=get_logfire_token())
logfire.instrument_fastapi = logfire.instrument_fastapi
except ImportError:
logfire = None
app = FastAPI(default_response_class=ORJSONResponse)
# Instrument FastAPI with logfire if monitoring is enabled
if logfire is not None:
logfire.instrument_fastapi(app, capture_headers=True)
def process_video_for_gestures(video_path: str, detector_path: str = "models/hand_detector.onnx",
classifier_path: str = "models/crops_classifier.onnx",
frame_skip: int = 1) -> List[Gesture]:
"""
Process a video file to detect gestures using the MainController.
Parameters
----------
video_path : str
Path to the video file to process
detector_path : str
Path to the hand detection ONNX model
classifier_path : str
Path to the gesture classification ONNX model
frame_skip : int
Number of frames to skip between processing (1 = process every frame, 3 = process every 3rd frame)
Returns
-------
List[Gesture]
List of detected gestures with duration and confidence
"""
# Create monitoring span for video processing
span_context = None
if logfire is not None:
span_context = logfire.span('process_video_for_gestures',
video_path=video_path,
detector_path=detector_path,
classifier_path=classifier_path)
span_context.__enter__()
try:
# Initialize the main controller
if logfire is not None:
with logfire.span('initialize_controller'):
controller = MainController(detector_path, classifier_path)
else:
controller = MainController(detector_path, classifier_path)
# Open video file
cap = cv2.VideoCapture(video_path)
if not cap.isOpened():
raise ValueError(f"Could not open video file: {video_path}")
# Get video properties for monitoring
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
fps = cap.get(cv2.CAP_PROP_FPS)
if logfire is not None:
logfire.info('Video properties',
total_frames=total_frames,
fps=fps,
duration_seconds=total_frames/fps if fps > 0 else 0)
# Track gestures per hand ID
gesture_tracks: Dict[int, List[Tuple[int, float]]] = defaultdict(list) # {hand_id: [(gesture_id, confidence), ...]}
frame_count = 0
processed_frames = 0
detection_stats = {
'frames_with_detections': 0,
'total_detections': 0,
'gesture_counts': defaultdict(int)
}
try:
while True:
ret, frame = cap.read()
if not ret:
break
# Skip frames based on frame_skip parameter
if frame_count % frame_skip == 0:
# Process frame through the controller
bboxes, ids, labels = controller(frame)
processed_frames += 1
if bboxes is not None and ids is not None and labels is not None:
detection_stats['frames_with_detections'] += 1
detection_stats['total_detections'] += len(bboxes)
# 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:
# Get confidence from bbox (assuming it's the last element)
confidence = 0.8 # Default confidence, could be extracted from bbox if available
gesture_tracks[hand_id].append((gesture_id, confidence))
detection_stats['gesture_counts'][gesture_id] += 1
# Log individual detections for debugging
if logfire is not None:
gesture_name = FULL_GESTURE_MAPPING.get(gesture_id, f"unknown_{gesture_id}")
logfire.debug('Hand detection',
frame=frame_count,
hand_id=hand_id,
gesture_id=gesture_id,
gesture_name=gesture_name,
confidence=confidence,
bbox=bboxes[i].tolist() if len(bboxes[i]) >= 4 else None)
else:
# Advance tracker on skipped frames to keep state consistent
controller.update(np.empty((0, 5)), None)
frame_count += 1
# Log progress every 100 frames
if frame_count % 100 == 0 and logfire is not None:
progress = (frame_count / total_frames) * 100 if total_frames > 0 else 0
logfire.info('Processing progress',
frame=frame_count,
total_frames=total_frames,
progress_percent=round(progress, 2))
finally:
cap.release()
# Log final detection statistics
if logfire is not None:
logfire.info('Detection statistics',
total_frames=frame_count,
processed_frames=processed_frames,
frame_skip=frame_skip,
frames_with_detections=detection_stats['frames_with_detections'],
total_detections=detection_stats['total_detections'],
detection_rate=detection_stats['frames_with_detections']/processed_frames if processed_frames > 0 else 0,
gesture_counts=dict(detection_stats['gesture_counts']))
# 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
# Adjust minimum duration based on frame skip
min_duration = max(5, frame_skip * 2) # At least 2 processed frames
if current_gesture is not None and current_duration >= min_duration:
gesture_name = PRODUCTION_GESTURE_MAPPING.get(current_gesture, f"unknown_{current_gesture}")
avg_confidence = current_confidence / current_duration if current_duration > 0 else 0.0
# Scale duration back to original frame count
scaled_duration = current_duration * frame_skip
detected_gestures.append(Gesture(
gesture=gesture_name,
duration=scaled_duration,
confidence=avg_confidence
))
# Log significant gesture detection
if logfire is not None:
logfire.info('Significant gesture detected',
hand_id=hand_id,
gesture=gesture_name,
duration_frames=current_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
min_duration = max(5, frame_skip * 2) # At least 2 processed frames
if current_gesture is not None and current_duration >= min_duration:
gesture_name = PRODUCTION_GESTURE_MAPPING.get(current_gesture, f"unknown_{current_gesture}")
avg_confidence = current_confidence / current_duration if current_duration > 0 else 0.0
# Scale duration back to original frame count
scaled_duration = current_duration * frame_skip
detected_gestures.append(Gesture(
gesture=gesture_name,
duration=scaled_duration,
confidence=avg_confidence
))
# Log final gesture detection
if logfire is not None:
logfire.info('Final gesture detected',
hand_id=hand_id,
gesture=gesture_name,
duration_frames=current_duration,
confidence=avg_confidence)
# Log final results
if logfire is not None:
logfire.info('Video processing completed',
total_gestures_detected=len(detected_gestures),
unique_hands=len(gesture_tracks),
gestures=[{'gesture': g.gesture, 'duration': g.duration, 'confidence': g.confidence} for g in detected_gestures])
return detected_gestures
finally:
if span_context is not None:
span_context.__exit__(None, None, None)
@app.get("/health")
async def health():
"""Health check endpoint."""
if logfire is not None:
logfire.info('Health check requested')
return {"message": "OK"}
@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
"""
# Log request details
if logfire is not None:
logfire.info('Gesture detection request received',
filename=video.filename,
content_type=video.content_type,
content_length=video.size if hasattr(video, 'size') else 'unknown')
# Validate file type
if not video.content_type.startswith('video/'):
if logfire is not None:
logfire.warning('Invalid file type received', content_type=video.content_type)
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()
if logfire is not None:
logfire.info('Video file saved for processing',
temp_file=temp_file.name,
file_size_bytes=len(content))
# Process the video with frame skip parameter
gestures = process_video_for_gestures(temp_file.name, frame_skip=frame_skip)
if logfire is not None:
logfire.info('Gesture detection completed successfully',
total_gestures=len(gestures),
gestures=[g.gesture for g in gestures])
return GestureResponse(gestures=gestures)
except Exception as e:
if logfire is not None:
logfire.error('Error processing video',
error=str(e),
error_type=type(e).__name__,
temp_file=temp_file.name)
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)
if logfire is not None:
logfire.debug('Temporary file cleaned up', temp_file=temp_file.name)
|