Lag reduction

app.py

@@ -7,11 +7,37 @@ from torchvision.models.detection import FasterRCNN_ResNet50_FPN_Weights
 from PIL import Image
 import mediapipe as mp
 from fer import FER  # Facial emotion recognition
+from concurrent.futures import ThreadPoolExecutor
 
 # -----------------------------
-# Constants
+# Global Asynchronous Executor & Caches
 # -----------------------------
-SKIP_RATE = 5  # Run heavy detection every 5 frames
+executor = ThreadPoolExecutor(max_workers=4)
+latest_results = {
+    "posture": None,
+    "emotion": None,
+    "objects": None,
+    "faces": None
+}
+futures = {
+    "posture": None,
+    "emotion": None,
+    "objects": None,
+    "faces": None
+}
+
+def async_analyze(key, func, image):
+    """
+    Runs the heavy detection function 'func' in a background thread.
+    Returns the last computed result (if available) so that the output
+    FPS remains high even if the detection lags.
+    """
+    if futures[key] is None or futures[key].done():
+        futures[key] = executor.submit(func, image)
+    if futures[key].done():
+        latest_results[key] = futures[key].result()
+    # Return latest result if available; otherwise, compute synchronously
+    return latest_results.get(key, func(image))
 
 # -----------------------------
 # Initialize Models and Helpers
@@ -37,141 +63,91 @@ obj_transform = transforms.Compose([transforms.ToTensor()])
 emotion_detector = FER(mtcnn=True)
 
 # -----------------------------
-# Define Analysis Functions with Frame Skipping
+# Heavy (Synchronous) Analysis Functions
 # -----------------------------
 
-def analyze_posture(image):
-    """
-    Processes an image from the webcam with MediaPipe Pose.
-    Runs heavy detection every SKIP_RATE frames; otherwise, returns last result.
-    """
-    if not hasattr(analyze_posture, "counter"):
-        analyze_posture.counter = 0
-        analyze_posture.last_output = None
-    analyze_posture.counter += 1
-
-    # If first frame or time to run detection:
-    if analyze_posture.counter % SKIP_RATE == 0 or analyze_posture.last_output is None:
-        # Convert from PIL (RGB) to OpenCV BGR format
-        frame = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
-        output_frame = frame.copy()
-        frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-
-        posture_result = "No posture detected"
-        pose_results = pose.process(frame_rgb)
-        if pose_results.pose_landmarks:
-            posture_result = "Posture detected"
-            mp_drawing.draw_landmarks(
-                output_frame, pose_results.pose_landmarks, mp_pose.POSE_CONNECTIONS,
-                mp_drawing.DrawingSpec(color=(0, 255, 0), thickness=2, circle_radius=2),
-                mp_drawing.DrawingSpec(color=(0, 0, 255), thickness=2)
-            )
-
-        annotated_image = cv2.cvtColor(output_frame, cv2.COLOR_BGR2RGB)
-        result = (annotated_image, f"Posture Analysis: {posture_result}")
-        analyze_posture.last_output = result
-        return result
+def _analyze_posture(image):
+    # Convert from PIL (RGB) to OpenCV BGR format
+    frame = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
+    output_frame = frame.copy()
+    frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+    posture_result = "No posture detected"
+    pose_results = pose.process(frame_rgb)
+    if pose_results.pose_landmarks:
+        posture_result = "Posture detected"
+        mp_drawing.draw_landmarks(
+            output_frame, pose_results.pose_landmarks, mp_pose.POSE_CONNECTIONS,
+            mp_drawing.DrawingSpec(color=(0, 255, 0), thickness=2, circle_radius=2),
+            mp_drawing.DrawingSpec(color=(0, 0, 255), thickness=2)
+        )
+    annotated_image = cv2.cvtColor(output_frame, cv2.COLOR_BGR2RGB)
+    return annotated_image, f"Posture Analysis: {posture_result}"
+
+def _analyze_emotion(image):
+    frame = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
+    frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+    emotions = emotion_detector.detect_emotions(frame_rgb)
+    if emotions:
+        top_emotion, score = max(emotions[0]["emotions"].items(), key=lambda x: x[1])
+        emotion_text = f"{top_emotion} ({score:.2f})"
     else:
-        # For frames in between, return last result
-        return analyze_posture.last_output
+        emotion_text = "No face detected for emotion analysis"
+    annotated_image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+    return annotated_image, f"Emotion Analysis: {emotion_text}"
+
+def _analyze_objects(image):
+    frame = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
+    output_frame = frame.copy()
+    frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+    image_pil = Image.fromarray(frame_rgb)
+    img_tensor = obj_transform(image_pil)
+    with torch.no_grad():
+        detections = object_detection_model([img_tensor])[0]
+    threshold = 0.8
+    detected_boxes = detections["boxes"][detections["scores"] > threshold]
+    for box in detected_boxes:
+        box = box.int().cpu().numpy()
+        cv2.rectangle(output_frame, (box[0], box[1]), (box[2], box[3]), (255, 255, 0), 2)
+    object_result = f"Detected {len(detected_boxes)} object(s)" if len(detected_boxes) else "No objects detected"
+    annotated_image = cv2.cvtColor(output_frame, cv2.COLOR_BGR2RGB)
+    return annotated_image, f"Object Detection: {object_result}"
+
+def _analyze_faces(image):
+    frame = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
+    output_frame = frame.copy()
+    frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+    face_results = face_detection.process(frame_rgb)
+    face_result = "No faces detected"
+    if face_results.detections:
+        face_result = f"Detected {len(face_results.detections)} face(s)"
+        h, w, _ = output_frame.shape
+        for detection in face_results.detections:
+            bbox = detection.location_data.relative_bounding_box
+            x = int(bbox.xmin * w)
+            y = int(bbox.ymin * h)
+            box_w = int(bbox.width * w)
+            box_h = int(bbox.height * h)
+            cv2.rectangle(output_frame, (x, y), (x + box_w, y + box_h), (0, 0, 255), 2)
+    annotated_image = cv2.cvtColor(output_frame, cv2.COLOR_BGR2RGB)
+    return annotated_image, f"Face Detection: {face_result}"
 
-def analyze_emotion(image):
-    """
-    Uses FER to detect facial emotions from the webcam image.
-    Runs heavy detection every SKIP_RATE frames.
-    """
-    if not hasattr(analyze_emotion, "counter"):
-        analyze_emotion.counter = 0
-        analyze_emotion.last_output = None
-    analyze_emotion.counter += 1
+# -----------------------------
+# Asynchronous (Fast) Analysis Functions
+# -----------------------------
+def analyze_posture(image):
+    return async_analyze("posture", _analyze_posture, image)
 
-    if analyze_emotion.counter % SKIP_RATE == 0 or analyze_emotion.last_output is None:
-        frame = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
-        frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-        emotions = emotion_detector.detect_emotions(frame_rgb)
-        if emotions:
-            top_emotion, score = max(emotions[0]["emotions"].items(), key=lambda x: x[1])
-            emotion_text = f"{top_emotion} ({score:.2f})"
-        else:
-            emotion_text = "No face detected for emotion analysis"
-        annotated_image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-        result = (annotated_image, f"Emotion Analysis: {emotion_text}")
-        analyze_emotion.last_output = result
-        return result
-    else:
-        return analyze_emotion.last_output
+def analyze_emotion(image):
+    return async_analyze("emotion", _analyze_emotion, image)
 
 def analyze_objects(image):
-    """
-    Uses Faster R-CNN to detect objects in the webcam image.
-    Heavy detection is run every SKIP_RATE frames.
-    """
-    if not hasattr(analyze_objects, "counter"):
-        analyze_objects.counter = 0
-        analyze_objects.last_output = None
-    analyze_objects.counter += 1
-
-    if analyze_objects.counter % SKIP_RATE == 0 or analyze_objects.last_output is None:
-        frame = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
-        output_frame = frame.copy()
-        frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-        image_pil = Image.fromarray(frame_rgb)
-        img_tensor = obj_transform(image_pil)
-
-        with torch.no_grad():
-            detections = object_detection_model([img_tensor])[0]
-
-        threshold = 0.8
-        detected_boxes = detections["boxes"][detections["scores"] > threshold]
-        for box in detected_boxes:
-            box = box.int().cpu().numpy()
-            cv2.rectangle(output_frame, (box[0], box[1]), (box[2], box[3]), (255, 255, 0), 2)
-
-        object_result = f"Detected {len(detected_boxes)} object(s)" if len(detected_boxes) else "No objects detected"
-        annotated_image = cv2.cvtColor(output_frame, cv2.COLOR_BGR2RGB)
-        result = (annotated_image, f"Object Detection: {object_result}")
-        analyze_objects.last_output = result
-        return result
-    else:
-        return analyze_objects.last_output
+    return async_analyze("objects", _analyze_objects, image)
 
 def analyze_faces(image):
-    """
-    Uses MediaPipe to detect faces in the webcam image.
-    Runs heavy detection every SKIP_RATE frames.
-    """
-    if not hasattr(analyze_faces, "counter"):
-        analyze_faces.counter = 0
-        analyze_faces.last_output = None
-    analyze_faces.counter += 1
-
-    if analyze_faces.counter % SKIP_RATE == 0 or analyze_faces.last_output is None:
-        frame = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
-        output_frame = frame.copy()
-        frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
-        face_results = face_detection.process(frame_rgb)
-
-        face_result = "No faces detected"
-        if face_results.detections:
-            face_result = f"Detected {len(face_results.detections)} face(s)"
-            h, w, _ = output_frame.shape
-            for detection in face_results.detections:
-                bbox = detection.location_data.relative_bounding_box
-                x = int(bbox.xmin * w)
-                y = int(bbox.ymin * h)
-                box_w = int(bbox.width * w)
-                box_h = int(bbox.height * h)
-                cv2.rectangle(output_frame, (x, y), (x + box_w, y + box_h), (0, 0, 255), 2)
-
-        annotated_image = cv2.cvtColor(output_frame, cv2.COLOR_BGR2RGB)
-        result = (annotated_image, f"Face Detection: {face_result}")
-        analyze_faces.last_output = result
-        return result
-    else:
-        return analyze_faces.last_output
+    return async_analyze("faces", _analyze_faces, image)
 
 # -----------------------------
-# Custom CSS for a High-Tech Look (with white fonts)
+# Custom CSS for a High-Tech Look (White Font)
 # -----------------------------
 custom_css = """
 @import url('https://fonts.googleapis.com/css2?family=Orbitron:wght@400;700&display=swap');
@@ -204,7 +180,7 @@ body {
 """
 
 # -----------------------------
-# Create Individual Interfaces for Each Analysis (using real-time webcam input)
+# Create Individual Interfaces for Each Analysis
 # -----------------------------
 posture_interface = gr.Interface(
     fn=analyze_posture,