enhanced logs around missing frames

backend/gradio_labanmovementanalysis/json_generator.py

@@ -8,6 +8,7 @@ from datetime import datetime
 
 from .pose_estimation import PoseResult, Keypoint
 from .notation_engine import MovementMetrics, Direction, Intensity, Speed
+import numpy as np
 
 
 def generate_json(

backend/gradio_labanmovementanalysis/labanmovementanalysis.py

@@ -51,6 +51,7 @@ class LabanMovementAnalysis(Component):
                  elem_classes: Optional[List[str]] = None,
                  render: bool = True,
                  **kwargs):
+        print("[TRACE] LabanMovementAnalysis.__init__ called")
         """
         Initialize the Laban Movement Analysis component.
         
@@ -89,6 +90,7 @@ class LabanMovementAnalysis(Component):
         # SkateFormer features reserved for Version 2
     
     def preprocess(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        print("[TRACE] LabanMovementAnalysis.preprocess called")
         """
         Preprocess input from the frontend.
         
@@ -126,6 +128,7 @@ class LabanMovementAnalysis(Component):
         return options
     
     def postprocess(self, value: Any) -> Dict[str, Any]:
+        print("[TRACE] LabanMovementAnalysis.postprocess called")
         """
         Postprocess analysis results for the frontend.
         
@@ -153,6 +156,7 @@ class LabanMovementAnalysis(Component):
     def process_video(self, video_input: Union[str, os.PathLike], model: str = DEFAULT_MODEL,
                      enable_visualization: bool = True,
                      include_keypoints: bool = False) -> Tuple[Dict[str, Any], Optional[str]]:
+        print(f"[TRACE] LabanMovementAnalysis.process_video called with model={model}, enable_visualization={enable_visualization}, include_keypoints={include_keypoints}")
         """
         Main processing function that performs pose analysis on a video.
         
@@ -266,6 +270,7 @@ class LabanMovementAnalysis(Component):
         return json_output, visualization_path
     
     def __call__(self, video_path: str, **kwargs) -> Tuple[Dict[str, Any], Optional[str]]:
+        print(f"[TRACE] LabanMovementAnalysis.__call__ called with video_path={video_path}")
         """
         Make the component callable for easy use.
         
@@ -285,12 +290,14 @@ class LabanMovementAnalysis(Component):
     # will be available in the next major release
     
     def cleanup(self):
+        print("[TRACE] LabanMovementAnalysis.cleanup called")
         """Clean up temporary files and resources."""
         # Clean up video input handler
         if hasattr(self, 'video_input'):
             self.video_input.cleanup()
     
     def example_payload(self) -> Dict[str, Any]:
+        print("[TRACE] LabanMovementAnalysis.example_payload called")
         """Example input payload for documentation."""
         return {
             "video": {"path": "/path/to/video.mp4"},
@@ -300,6 +307,7 @@ class LabanMovementAnalysis(Component):
         }
     
     def example_value(self) -> Dict[str, Any]:
+        print("[TRACE] LabanMovementAnalysis.example_value called")
         """Example output value for documentation."""
         return {
             "json_output": {
@@ -348,6 +356,7 @@ class LabanMovementAnalysis(Component):
         }
     
     def api_info(self) -> Dict[str, Any]:
+        print("[TRACE] LabanMovementAnalysis.api_info called")
         """API information for the component."""
         return {
             "type": "composite",

backend/gradio_labanmovementanalysis/pose_estimation.py

@@ -7,6 +7,7 @@ from abc import ABC, abstractmethod
 import numpy as np
 from typing import List, Dict, Optional, Tuple, Any
 from dataclasses import dataclass
+import math
 
 
 @dataclass
@@ -136,14 +137,19 @@ class MoveNetPoseEstimator(PoseEstimator):
         # Convert to our format
         keypoints = []
         for i, (y, x, score) in enumerate(keypoints_with_scores):
+            # Sanitize NaN values
+            if any(map(math.isnan, [x, y, score])):
+                continue
             keypoints.append(Keypoint(
                 x=float(x),
                 y=float(y),
                 confidence=float(score),
                 name=self.KEYPOINT_NAMES[i]
             ))
-        
-        return [PoseResult(keypoints=keypoints, frame_index=0)]
+        if keypoints:
+            return [PoseResult(keypoints=keypoints, frame_index=0)]
+        else:
+            return []
     
     def get_keypoint_names(self) -> List[str]:
         return self.KEYPOINT_NAMES.copy()
@@ -239,7 +245,7 @@ class YOLOPoseEstimator(PoseEstimator):
         "left_knee", "right_knee", "left_ankle", "right_ankle"
     ]
     
-    def __init__(self, model_version: str = "v11", model_size: str = "n", confidence_threshold: float = 0.25):
+    def __init__(self, model_version: str = "v11", model_size: str = "n", confidence_threshold: float = 0.15):
         """
         Initialize YOLO pose model.
         
@@ -276,7 +282,7 @@ class YOLOPoseEstimator(PoseEstimator):
             self._load_model()
         
         # Run inference
-        results = self.model(frame, conf=self.confidence_threshold)
+        results = self.model(frame, conf=self.confidence_threshold, iou=0.7)
         
         pose_results = []
         
@@ -289,6 +295,9 @@ class YOLOPoseEstimator(PoseEstimator):
                     # YOLO returns keypoints as [x, y, conf]
                     height, width = frame.shape[:2]
                     for i, (x, y, conf) in enumerate(keypoints_data):
+                        # Sanitize NaN values
+                        if any(map(math.isnan, [x, y, conf])):
+                            continue
                         keypoints.append(Keypoint(
                             x=float(x) / width,  # Normalize to 0-1
                             y=float(y) / height,  # Normalize to 0-1
@@ -296,11 +305,12 @@ class YOLOPoseEstimator(PoseEstimator):
                             name=self.KEYPOINT_NAMES[i] if i < len(self.KEYPOINT_NAMES) else f"joint_{i}"
                         ))
                     
-                    pose_results.append(PoseResult(
-                        keypoints=keypoints,
-                        frame_index=0,
-                        person_id=person_idx
-                    ))
+                    if keypoints:
+                        pose_results.append(PoseResult(
+                            keypoints=keypoints,
+                            frame_index=0,
+                            person_id=person_idx
+                        ))
         
         return pose_results
     
@@ -377,4 +387,38 @@ def get_pose_estimator(model_spec: str) -> PoseEstimator:
     
     # Legacy format support
     else:
-        return create_pose_estimator(model_spec) 
+        return create_pose_estimator(model_spec)
+
+
+def _safe_pose_from_dets(dets: List[PoseResult], frame_idx: int) -> List[PoseResult]:
+    """
+    Helper function to fill missing poses with the previous pose and keep a missing_mask.
+    After the loop, interpolate missing poses in pose_seq before running metrics.
+    Add debug prints when a pose is missing and when interpolation is performed.
+    """
+    safe_poses = []
+    missing_mask = []
+    prev_pose = None
+
+    for det in dets:
+        if det.frame_index == frame_idx:
+            if prev_pose is None:
+                print(f"Warning: No previous pose found for frame {frame_idx}")
+                safe_poses.append(PoseResult(keypoints=[], frame_index=frame_idx))
+                missing_mask.append(True)
+            else:
+                safe_poses.append(PoseResult(keypoints=prev_pose.keypoints, frame_index=frame_idx))
+                missing_mask.append(False)
+            prev_pose = det
+        elif det.frame_index > frame_idx:
+            break
+
+    if prev_pose is None:
+        print(f"Warning: No poses found for frame {frame_idx}")
+        safe_poses.append(PoseResult(keypoints=[], frame_index=frame_idx))
+        missing_mask.append(True)
+    else:
+        safe_poses.append(PoseResult(keypoints=prev_pose.keypoints, frame_index=frame_idx))
+        missing_mask.append(False)
+
+    return safe_poses, missing_mask 

backend/gradio_labanmovementanalysis/video_utils.py

@@ -52,6 +52,9 @@ def get_video_info(video_path: str) -> Tuple[int, float, Tuple[int, int]]:
         fps = cap.get(cv2.CAP_PROP_FPS)
         width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
         height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+        print("DEBUG: fps =", fps, "frames =", frame_count)
+        if np.isnan(fps) or fps <= 0:
+            raise ValueError("FPS came back NaN/0 – did the capture fail?")
         return frame_count, fps, (width, height)
     finally:
         cap.release()

backend/gradio_labanmovementanalysis/visualizer.py

@@ -7,6 +7,8 @@ import numpy as np
 from typing import List, Tuple, Optional, Dict, Any
 from collections import deque
 import colorsys
+import math
+import traceback
 
 from .pose_estimation import PoseResult, Keypoint
 from .notation_engine import MovementMetrics, Direction, Intensity, Speed
@@ -66,6 +68,7 @@ class PoseVisualizer:
             show_direction_arrows: Whether to show movement direction arrows
             show_metrics: Whether to display text metrics on frame
         """
+        print("[TRACE] PoseVisualizer.__init__ called")
         self.trail_length = trail_length
         self.show_skeleton = show_skeleton
         self.show_trails = show_trails
@@ -99,11 +102,18 @@ class PoseVisualizer:
         Returns:
             Annotated frame
         """
-        # Create a copy to avoid modifying original
+        print(f"[TRACE] visualize_frame(frame_index={frame_index}, num_poses={len(pose_results)})")
         vis_frame = frame.copy()
+        if not pose_results:
+            print(f"[DEBUG] Frame {frame_index}: No pose results (no detections)")
+            return vis_frame  # No people detected, return original frame
         
         # Draw for each detected person
         for person_idx, pose in enumerate(pose_results):
+            if not pose.keypoints or all(math.isnan(kp.x) or math.isnan(kp.y) for kp in pose.keypoints):
+                print(f"[DEBUG] Frame {frame_index}, Person {person_idx}: All keypoints are NaN or missing, skipping drawing for this person.")
+                continue
+            
             # Update trails
             if self.show_trails:
                 self._update_trails(pose, person_idx)
@@ -146,26 +156,30 @@ class PoseVisualizer:
         Returns:
             Path to created video
         """
-        if len(frames) != len(all_pose_results) or len(frames) != len(all_movement_metrics):
-            raise ValueError("Mismatched lengths between frames, poses, and metrics")
-        
-        # Reset trails
-        self.trails = {}
-        
-        # Process each frame
-        annotated_frames = []
-        for i, (frame, poses, metrics) in enumerate(
-            zip(frames, all_pose_results, all_movement_metrics)
-        ):
-            annotated_frame = self.visualize_frame(frame, poses, metrics, i)
-            annotated_frames.append(annotated_frame)
-        
-        # Import video_utils locally to avoid circular import
-        from . import video_utils
-        return video_utils.assemble_video(annotated_frames, output_path, fps)
+        print(f"[TRACE] generate_overlay_video(num_frames={len(frames)})")
+        try:
+            if len(frames) != len(all_pose_results) or len(frames) != len(all_movement_metrics):
+                raise ValueError("Mismatched lengths between frames, poses, and metrics")
+            self.trails = {}
+            annotated_frames = []
+            for i, (frame, poses, metrics) in enumerate(
+                zip(frames, all_pose_results, all_movement_metrics)
+            ):
+                annotated_frame = self.visualize_frame(frame, poses, metrics, i)
+                annotated_frames.append(annotated_frame)
+            print(f"[DEBUG] About to assemble video with {len(annotated_frames)} frames.")
+            for idx, f in enumerate(annotated_frames):
+                print(f"[DEBUG] Frame {idx} shape: {f.shape if hasattr(f, 'shape') else type(f)}")
+            from . import video_utils
+            return video_utils.assemble_video(annotated_frames, output_path, fps)
+        except Exception as e:
+            print(f'[FATAL ERROR] Exception during visualization: {e}')
+            print(traceback.format_exc())
+            raise
     
     def _update_trails(self, pose: PoseResult, person_id: int):
         """Update motion trails for a person."""
+        print(f"[TRACE] _update_trails(person_id={person_id})")
         if person_id not in self.trails:
             self.trails[person_id] = {}
         
@@ -182,6 +196,7 @@ class PoseVisualizer:
     
     def _draw_trails(self, frame: np.ndarray, person_id: int):
         """Draw motion trails for a person."""
+        print(f"[TRACE] _draw_trails(person_id={person_id})")
         if person_id not in self.trails:
             return
         
@@ -193,19 +208,25 @@ class PoseVisualizer:
             
             # Draw trail with fading effect
             for i in range(1, len(trail)):
-                # Calculate opacity based on position in trail
+                x1, y1 = trail[i-1]
+                x2, y2 = trail[i]
+                if math.isnan(x1) or math.isnan(y1) or math.isnan(x2) or math.isnan(y2):
+                    print(f"[DEBUG] _draw_trails: NaN detected in trail for person {person_id}, joint {joint_name}, skipping segment.")
+                    continue
+                print(f"[DEBUG] _draw_trails: Converting to int: ({x1}*{w}, {y1}*{h}) and ({x2}*{w}, {y2}*{h})")
                 alpha = i / len(trail)
                 color = tuple(int(c * alpha) for c in (255, 255, 255))
                 
                 # Convert normalized to pixel coordinates
-                pt1 = (int(trail[i-1][0] * w), int(trail[i-1][1] * h))
-                pt2 = (int(trail[i][0] * w), int(trail[i][1] * h))
+                pt1 = (int(x1 * w), int(y1 * h))
+                pt2 = (int(x2 * w), int(y2 * h))
                 
                 # Draw trail segment
                 cv2.line(frame, pt1, pt2, color, thickness=max(1, int(3 * alpha)))
     
     def _draw_skeleton(self, frame: np.ndarray, pose: PoseResult, color: Tuple[int, int, int]):
         """Draw pose skeleton."""
+        print(f"[TRACE] _draw_skeleton()")
         h, w = frame.shape[:2]
         
         # Create keypoint lookup
@@ -229,6 +250,11 @@ class PoseVisualizer:
                     kp1 = kp_dict[name1]
                     kp2 = kp_dict[name2]
                     
+                    if math.isnan(kp1.x) or math.isnan(kp1.y) or math.isnan(kp2.x) or math.isnan(kp2.y):
+                        print(f"[DEBUG] _draw_skeleton: NaN detected in skeleton for keypoints {name1}, {name2}, skipping line.")
+                        continue
+                    
+                    print(f"[DEBUG] _draw_skeleton: Converting to int: ({kp1.x}*{w}, {kp1.y}*{h}) and ({kp2.x}*{w}, {kp2.y}*{h})")
                     # Convert to pixel coordinates
                     pt1 = (int(kp1.x * w), int(kp1.y * h))
                     pt2 = (int(kp2.x * w), int(kp2.y * h))
@@ -239,12 +265,18 @@ class PoseVisualizer:
     def _draw_keypoints(self, frame: np.ndarray, pose: PoseResult, 
                        metrics: Optional[MovementMetrics] = None):
         """Draw individual keypoints."""
+        print(f"[TRACE] _draw_keypoints()")
         h, w = frame.shape[:2]
         
         for kp in pose.keypoints:
             if kp.confidence < 0.3:
                 continue
             
+            if math.isnan(kp.x) or math.isnan(kp.y):
+                print(f"[DEBUG] _draw_keypoints: NaN detected in keypoint {kp.name}, skipping.")
+                continue
+            
+            print(f"[DEBUG] _draw_keypoints: Converting to int: ({kp.x}*{w}, {kp.y}*{h})")
             # Convert to pixel coordinates
             pt = (int(kp.x * w), int(kp.y * h))
             
@@ -258,16 +290,24 @@ class PoseVisualizer:
     def _draw_direction_arrow(self, frame: np.ndarray, pose: PoseResult,
                              metrics: MovementMetrics):
         """Draw arrow indicating movement direction."""
+        print(f"[TRACE] _draw_direction_arrow()")
         if metrics.direction == Direction.STATIONARY:
             return
         
         h, w = frame.shape[:2]
         
         # Get body center
-        center_x = np.mean([kp.x for kp in pose.keypoints if kp.confidence > 0.3])
-        center_y = np.mean([kp.y for kp in pose.keypoints if kp.confidence > 0.3])
-        
-        # Convert to pixel coordinates
+        xs = [kp.x for kp in pose.keypoints if kp.confidence > 0.3 and not math.isnan(kp.x)]
+        ys = [kp.y for kp in pose.keypoints if kp.confidence > 0.3 and not math.isnan(kp.y)]
+        if not xs or not ys:
+            print(f"[DEBUG] _draw_direction_arrow: No valid keypoints for direction arrow, skipping.")
+            return
+        print(f"[DEBUG] _draw_direction_arrow: Converting to int: ({np.mean(xs)}*{w}, {np.mean(ys)}*{h})")
+        center_x = np.mean(xs)
+        center_y = np.mean(ys)
+        if math.isnan(center_x) or math.isnan(center_y):
+            print(f"[DEBUG] _draw_direction_arrow: NaN detected in center, skipping arrow.")
+            return
         center = (int(center_x * w), int(center_y * h))
         
         # Calculate arrow endpoint based on direction
@@ -294,18 +334,29 @@ class PoseVisualizer:
     
     def _draw_metrics_overlay(self, frame: np.ndarray, metrics: MovementMetrics):
         """Draw text overlay with movement metrics."""
+        print(f"[TRACE] _draw_metrics_overlay()")
         # Define text properties
         font = cv2.FONT_HERSHEY_SIMPLEX
         font_scale = 0.6
         thickness = 2
         
-        # Create text lines
+        # Sanitize metrics
+        def safe_metric(val, name):
+            if isinstance(val, float) and math.isnan(val):
+                print(f"[DEBUG] _draw_metrics_overlay: {name} is NaN, replacing with 0.0")
+                return 0.0
+            return val
+
+        velocity = safe_metric(metrics.velocity, "velocity")
+        fluidity = safe_metric(metrics.fluidity, "fluidity")
+        expansion = safe_metric(metrics.expansion, "expansion")
+
         lines = [
             f"Direction: {metrics.direction.value}",
-            f"Speed: {metrics.speed.value} ({metrics.velocity:.2f})",
+            f"Speed: {metrics.speed.value} ({velocity:.2f})",
             f"Intensity: {metrics.intensity.value}",
-            f"Fluidity: {metrics.fluidity:.2f}",
-            f"Expansion: {metrics.expansion:.2f}"
+            f"Fluidity: {fluidity:.2f}",
+            f"Expansion: {expansion:.2f}"
         ]
         
         # Draw background rectangle
@@ -330,6 +381,7 @@ class PoseVisualizer:
     
     def _get_color_for_metrics(self, metrics: Optional[MovementMetrics]) -> Tuple[int, int, int]:
         """Get color based on movement metrics."""
+        print(f"[TRACE] _get_color_for_metrics()")
         if metrics is None:
             return (255, 255, 255)  # White default
         
@@ -337,6 +389,7 @@ class PoseVisualizer:
     
     def _confidence_to_color(self, confidence: float) -> Tuple[int, int, int]:
         """Convert confidence score to color (green=high, red=low)."""
+        print(f"[TRACE] _confidence_to_color(confidence={confidence})")
         # Use HSV color space for smooth gradient
         hue = confidence * 120  # 0=red, 120=green
         rgb = colorsys.hsv_to_rgb(hue / 360, 1.0, 1.0)
@@ -344,6 +397,7 @@ class PoseVisualizer:
     
     def _get_skeleton_for_model(self, keypoints: List[Keypoint]) -> List[Tuple[int, int]]:
         """Determine which skeleton definition to use based on keypoints."""
+        print(f"[TRACE] _get_skeleton_for_model(num_keypoints={len(keypoints)})")
         # Simple heuristic: if we have more than 20 keypoints, use MediaPipe skeleton
         if len(keypoints) > 20:
             return self.MEDIAPIPE_SKELETON
@@ -351,6 +405,7 @@ class PoseVisualizer:
     
     def _get_keypoint_names_for_model(self, keypoints: List[Keypoint]) -> List[str]:
         """Get ordered list of keypoint names for the model."""
+        print(f"[TRACE] _get_keypoint_names_for_model(num_keypoints={len(keypoints)})")
         # If keypoints have names, use them
         if keypoints and keypoints[0].name:
             return [kp.name for kp in keypoints]

demo/app.py

@@ -6,7 +6,8 @@ Author: Csaba (BladeSzaSza)
 
 import gradio as gr
 import os
-from backend.gradio_labanmovementanalysis import LabanMovementAnalysis
+# from backend.gradio_labanmovementanalysis import LabanMovementAnalysis
+from gradio_labanmovementanalysis import LabanMovementAnalysis
 
 # Import agent API if available
 # Initialize agent API if available
@@ -144,6 +145,7 @@ def create_demo() -> gr.Blocks:
                             label="Examples"
                         )
 
+
                     # Output column
                     with gr.Column(scale=2, min_width=320):
                         viz_out = gr.Video(label="Annotated Video", scale=1, height=400)

yolo11n-pose.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:869e83fcdffdc7371fa4e34cd8e51c838cc729571d1635e5141e3075e9319dc0
+size 6255593

yolov8n-pose.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:c6fa93dd1ee4a2c18c900a45c1d864a1c6f7aba75d84f91648a30b7fb641d212
+size 6832633