Added CPR Refactored

CPR/CPRAnalyzer.py

@@ -0,0 +1,242 @@
+# main.py
+import cv2
+import time
+import tkinter as tk  # For screen size detection
+from datetime import datetime
+
+from pose_estimation import PoseEstimator
+from role_classifier import RoleClassifier
+from chest_initializer import ChestInitializer
+from metrics_calculator import MetricsCalculator
+from posture_analyzer import PostureAnalyzer
+
+class CPRAnalyzer:
+    """Main CPR analysis pipeline with execution tracing"""
+    
+    def __init__(self, video_path):
+        print(f"\n[INIT] Initializing CPR Analyzer for: {video_path}")
+        self.cap = cv2.VideoCapture(video_path)
+        if not self.cap.isOpened():
+            print("[ERROR] Failed to open video file")
+            return
+
+        self.frame_count = int(self.cap.get(cv2.CAP_PROP_FRAME_COUNT))
+        print(f"[INFO] Video properties: {self.frame_count} frames, "
+              f"{self.cap.get(cv2.CAP_PROP_FPS):.1f} FPS")
+
+        # Get screen dimensions
+        root = tk.Tk()
+        self.screen_width = root.winfo_screenwidth()
+        self.screen_height = root.winfo_screenheight()
+        root.destroy()
+        
+        print(f"[DISPLAY] Detected screen resolution: {self.screen_width}x{self.screen_height}")
+        
+        # Initialize components
+        self.pose_estimator = PoseEstimator()
+        self.role_classifier = RoleClassifier()
+        self.chest_initializer = ChestInitializer()
+        self.metrics_calculator = MetricsCalculator(shoulder_width_cm=45)
+        self.posture_analyzer = PostureAnalyzer()
+        
+        # Window configuration
+        self.window_name = "CPR Analysis"
+        cv2.namedWindow(self.window_name, cv2.WINDOW_NORMAL)
+        print("[INIT] System components initialized successfully")
+
+    def run_analysis(self):
+        """Main processing loop with execution tracing"""
+
+        print("\n[PHASE] Starting initialization phase")
+        start_time = time.time()
+        
+        if not self._initialize_system():
+            return
+        
+        # Get the first frame to determine orientation
+        ret, frame = self.cap.read()
+        if ret:
+            self._handle_chest_point_rotation(frame)
+            self.cap.set(cv2.CAP_PROP_POS_FRAMES, 0)  # Reset to beginning
+        
+        print(f"[INIT] Completed in {time.time()-start_time:.2f}s\n")
+        print("[PHASE] Starting main processing loop")
+        
+        frame_counter = 0
+        while self.cap.isOpened():
+            ret, frame = self.cap.read()
+            if not ret:
+                print("\n[INFO] End of video stream reached")
+                break
+
+            frame = self._handle_frame_rotation(frame)
+            print(f"\n[FRAME {int(self.cap.get(cv2.CAP_PROP_POS_FRAMES))}/{self.frame_count}] Processing")
+            
+            frame = self._process_frame(frame)
+            
+            self._display_frame(frame)
+
+            if cv2.waitKey(1) == ord('q'):
+                print("\n[USER] Analysis interrupted by user")
+                break
+                
+        self._finalize_analysis()
+
+    def _initialize_system(self):
+        """Initialize system components with status tracking"""
+        print("[INIT] Starting chest point initialization")
+        init_start = time.time()
+        
+        if not self.chest_initializer.initialize(self.cap, self.pose_estimator, self.role_classifier):
+            print("[ERROR] Chest initialization failed - check video content")
+            return False
+        
+        self.cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
+        print(f"[INIT] Chest point initialized at {self.chest_initializer.chest_point}")
+        print(f"[INIT] Sampled {len(self.chest_initializer.shoulder_samples)} valid shoulder positions")
+        print(f"[INIT] Initialization completed in {time.time()-init_start:.2f}s")
+        return True
+
+    def _handle_chest_point_rotation(self, frame):
+        """Handle chest point rotation once during initialization"""
+        if frame.shape[1] > frame.shape[0]:  # Landscape orientation
+            print(f"[ROTATE] Detected landscape video")
+            
+            if self.chest_initializer.chest_point:
+                print(f"[ROTATE] Adjusting chest point coordinates")
+                cx, cy = self.chest_initializer.chest_point
+                # For 90° clockwise rotation:
+                # New x = original height - original y
+                # New y = original x
+                new_cx = frame.shape[0] - cy
+                new_cy = cx
+                self.chest_initializer.chest_point = (int(new_cx), int(new_cy))
+                print(f"[ROTATE] New chest point: {self.chest_initializer.chest_point}")
+
+    def _handle_frame_rotation(self, frame):
+        """Handle frame rotation without adjusting chest point"""
+        if frame.shape[1] > frame.shape[0]:  # Width > Height
+            frame = cv2.rotate(frame, cv2.ROTATE_90_CLOCKWISE)
+        return frame
+
+    def _process_frame(self, frame):
+        """Process frame with execution tracing"""
+        processing_start = time.time()
+        
+        # Pose estimation
+        pose_results = self.pose_estimator.detect_poses(frame)
+        if not pose_results:
+            print("[PROCESS] No poses detected in frame")
+            return frame
+            
+        print(f"[POSE] Detected {len(pose_results.boxes)} people")
+        frame = self.pose_estimator.draw_keypoints(frame, pose_results)
+        frame = self._analyze_rescuer(frame, pose_results)
+        
+        # Draw chest point
+        frame = self.chest_initializer.draw_chest_marker(frame)
+        print(f"[PROCESS] Frame processed in {(time.time()-processing_start)*1000:.1f}ms")
+        return frame
+
+    def _analyze_rescuer(self, frame, pose_results):
+        """Analyze rescuer with detailed logging"""
+        rescuer_id = self.role_classifier.find_rescuer(pose_results, frame.shape[:2])
+        if rescuer_id is None:
+            print("[RESCUER] No rescuer identified in frame")
+            return frame
+            
+        print(f"[RESCUER] Identified at index {rescuer_id}")
+        keypoints = self.pose_estimator.get_keypoints(pose_results, rescuer_id)
+        
+        # Draw rescuer bounding box
+        frame = self._draw_rescuer_box(frame, pose_results, rescuer_id)
+        
+        # Posture analysis
+        warnings = self.posture_analyzer.validate_posture(keypoints, self.chest_initializer.chest_point)
+        frame = self.posture_analyzer.display_warnings(frame, warnings)
+
+        if warnings:
+            print(f"[WARNING] Posture issues: {', '.join(warnings)}")
+        else:
+            # Track midpoints
+            frame = self.role_classifier.track_rescuer_midpoints(keypoints, frame)
+            print(f"[TRACKING] Midpoint added at {self.role_classifier.midpoints[-1]}")
+            self.role_classifier.update_shoulder_distance()
+            
+        return frame
+
+    def _draw_rescuer_box(self, frame, results, rescuer_id):
+        """Draw rescuer box with dimension logging"""
+        boxes = results.boxes.xyxy.cpu().numpy()
+        if rescuer_id < len(boxes):
+            x1, y1, x2, y2 = boxes[rescuer_id].astype(int)
+            w, h = x2-x1, y2-y1
+            print(f"[BOX] Rescuer bounding box: {w}x{h} pixels")
+            cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 3)
+            cv2.putText(frame, "RESCUER", (x1, y1-10),
+                        cv2.FONT_HERSHEY_SIMPLEX, 0.9, (0, 255, 0), 2)
+        return frame
+
+    def _display_frame(self, frame):
+        """Adaptive window sizing with aspect ratio preservation"""
+        display_start = time.time()
+        
+        # Get original frame dimensions
+        h, w = frame.shape[:2]
+        if w == 0 or h == 0:
+            return
+
+        # Calculate maximum possible scale while maintaining aspect ratio
+        scale_w = self.screen_width / w
+        scale_h = self.screen_height / h
+        scale = min(scale_w, scale_h) * 0.9  # 90% of max to leave some margin
+
+        # Calculate new dimensions
+        new_w = int(w * scale)
+        new_h = int(h * scale)
+
+        # Resize and display
+        resized = cv2.resize(frame, (new_w, new_h), interpolation=cv2.INTER_AREA)
+        
+        # Center window
+        pos_x = (self.screen_width - new_w) // 2
+        pos_y = (self.screen_height - new_h) // 2
+        cv2.moveWindow(self.window_name, pos_x, pos_y)
+        
+        cv2.imshow(self.window_name, resized)
+        print(f"[DISPLAY] Resized to {new_w}x{new_h} (scale: {scale:.2f}) in {(time.time()-display_start)*1000:.1f}ms")
+
+    def _finalize_analysis(self):
+        """Final analysis with detailed reporting"""
+        print("\n[PHASE] Starting final analysis")
+        start_time = time.time()
+        
+        try:
+            print("[METRICS] Smoothing midpoint data...")
+            self.metrics_calculator.smooth_midpoints(self.role_classifier.midpoints)
+            
+            print("[METRICS] Detecting compression peaks...")
+            self.metrics_calculator.detect_peaks()
+            
+            print("[METRICS] Calculating depth and rate...")
+            depth, rate = self.metrics_calculator.calculate_metrics(
+                self.role_classifier.shoulder_distances,
+                self.cap.get(cv2.CAP_PROP_FPS))
+
+            print(f"[RESULTS] Compression Depth: {depth:.1f} cm")
+            print(f"[RESULTS] Compression Rate: {rate:.1f} cpm")
+            
+            print("[VISUAL] Generating motion curve plot...")
+            self.metrics_calculator.plot_motion_curve()
+            
+        except Exception as e:
+            print(f"[ERROR] Metric calculation failed: {str(e)}")
+            
+        finally:
+            self.cap.release()
+            cv2.destroyAllWindows()
+            print(f"\n[ANALYSIS] Completed in {time.time()-start_time:.1f}s")
+            print("[CLEANUP] Resources released")
+
+
+

CPR/chest_initializer.py

@@ -0,0 +1,153 @@
+# chest_initializer.py
+import cv2
+import numpy as np
+from keypoints import CocoKeypoints
+
+class ChestInitializer:
+    """Handles chest point detection with separated debug visualization"""
+    
+    def __init__(self, num_init_frames=15, min_confidence=0.2, aspect_ratio_thresh=1.2):
+        self.num_init_frames = num_init_frames
+        self.min_confidence = min_confidence
+        self.aspect_ratio_thresh = aspect_ratio_thresh
+        self.chest_point = None
+        self.shoulder_samples = []
+        self.hip_samples = []
+        self.debug_window = "Chest Detection Debug"
+
+    def initialize(self, cap, pose_estimator, role_classifier):
+        """Main initialization routine"""
+        success = self._sample_initial_frames(cap, pose_estimator)
+        self._calculate_chest_point(cap)
+        cv2.destroyWindow(self.debug_window)
+        return success
+
+    def _sample_initial_frames(self, cap, pose_estimator):
+        """Collect valid shoulder positions"""
+        cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
+        valid_samples = 0
+
+        for i in range(self.num_init_frames):
+            ret, frame = cap.read()
+            if not ret:
+                break
+
+            results = pose_estimator.detect_poses(frame) #!could remove this to prevent double detection
+            if not results:
+                continue
+
+            debug_frame = results.plot()
+            shoulders = self._detect_valid_shoulders(results, frame.shape)
+
+            if shoulders:
+                left_shoulder, right_shoulder, left_hip, right_hip = shoulders
+                self.shoulder_samples.append((left_shoulder, right_shoulder))
+                self.hip_samples.append((left_hip, right_hip))
+                valid_samples += 1
+                debug_frame = self._draw_debug_info(
+                    debug_frame, 
+                    left_shoulder, 
+                    right_shoulder, 
+                    frame.shape,
+                    i+1
+                )
+
+            cv2.imshow(self.debug_window, debug_frame)
+            if cv2.waitKey(400) == ord('q'):
+                break
+
+        return valid_samples > 0
+
+    def _draw_debug_info(self, frame, left_shoulder, right_shoulder, frame_shape, frame_num):
+        """Helper function for drawing debug information"""
+        # Convert normalized coordinates to pixel values
+        lx, ly = (left_shoulder * np.array([frame_shape[1], frame_shape[0]])).astype(int)
+        rx, ry = (right_shoulder * np.array([frame_shape[1], frame_shape[0]])).astype(int)
+        
+        # Draw shoulder points
+        cv2.circle(frame, (lx, ly), 5, (0, 0, 255), -1)
+        cv2.circle(frame, (rx, ry), 5, (0, 0, 255), -1)
+        
+        # Add informational text
+        cv2.putText(frame, f"CHEST pts - Frame {frame_num}", (lx, ly - 10),
+                   cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2)
+        
+        return frame
+
+    def _detect_valid_shoulders(self, results, frame_shape):
+        """Validate detected shoulders using aspect ratio and confidence"""
+        boxes = results.boxes.xywh.cpu().numpy()
+        keypoints = results.keypoints.xyn.cpu().numpy()
+        confs = results.keypoints.conf.cpu().numpy()
+
+        for i, (box, kp) in enumerate(zip(boxes, keypoints)):
+            x, y, w, h = box
+            # Aspect ratio validation
+            if h < w * self.aspect_ratio_thresh:
+                continue
+
+            # Confidence check
+            if (confs[i][CocoKeypoints.LEFT_SHOULDER.value] < self.min_confidence or 
+                confs[i][CocoKeypoints.RIGHT_SHOULDER.value] < self.min_confidence):
+                continue
+
+            return (kp[CocoKeypoints.LEFT_SHOULDER.value], 
+                    kp[CocoKeypoints.RIGHT_SHOULDER.value],
+                    kp[CocoKeypoints.LEFT_HIP.value],
+                    kp[CocoKeypoints.RIGHT_HIP.value])
+
+        return None
+
+    def _calculate_chest_point(self, cap):
+        """Calculate final chest point from valid samples"""
+        if not self.shoulder_samples:
+            return
+
+        avg_left = np.median([s[0] for s in self.shoulder_samples], axis=0)
+        avg_right = np.median([s[1] for s in self.shoulder_samples], axis=0)
+        avg_left_hip = np.median([h[0] for h in self.hip_samples], axis=0)
+        avg_right_hip = np.median([h[1] for h in self.hip_samples], axis=0)
+
+        frame_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+        frame_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+        
+        avg_left_px = avg_left * np.array([frame_width, frame_height])
+        avg_right_px = avg_right * np.array([frame_width, frame_height])
+        avg_left_hip_px = avg_left_hip * np.array([frame_width, frame_height])
+        avg_right_hip_px = avg_right_hip * np.array([frame_width, frame_height])
+
+        #midpoint = (avg_left_px + avg_right_px) / 2
+        #shoulder_dist = np.linalg.norm(avg_left_px - avg_right_px)
+        #downward_offset = 0.4 * shoulder_dist
+        #self.chest_point = (int(midpoint[0]), int(midpoint[1] + downward_offset))
+
+        if avg_left_px[1] < avg_right_px[1]:
+            shoulder = np.array(avg_left_px)
+            hip = np.array(avg_left_hip_px)
+        else:
+            shoulder = np.array(avg_right_px)
+            hip = np.array(avg_right_hip_px)
+
+        alpha = 0.412  # Relative chest position between shoulder and hip
+        offset = 10  # move 10 pixels upward into the body
+        self.chest_point = (
+            int(shoulder[0] + alpha * (hip[0] - shoulder[0])),
+            int(shoulder[1] + alpha * (hip[1] - shoulder[1])) - offset
+        )
+
+        # Visualize the chest point in the debug window for 2 seconds
+        cap.set(cv2.CAP_PROP_POS_FRAMES, 0)  # Reset to the first frame
+        ret, frame = cap.read()
+        if ret:
+            frame_with_marker = self.draw_chest_marker(frame)
+            cv2.imshow(self.debug_window, frame_with_marker)
+            cv2.waitKey(2000)  # Wait for 2 seconds
+
+    def draw_chest_marker(self, frame):
+        """Draw chest point with visualization"""
+        print(f"Chest point: {self.chest_point}")
+        if self.chest_point:
+            cv2.circle(frame, self.chest_point, 8, (0, 55, 120), -1)
+            cv2.putText(frame, "Chest", (self.chest_point[0] + 5, self.chest_point[1] - 10),
+                       cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2)
+        return frame

CPR/keypoints.py

@@ -0,0 +1,22 @@
+# keypoints.py
+from enum import Enum
+
+class CocoKeypoints(Enum):
+    """Enum for COCO keypoints (17 points)"""
+    NOSE = 0
+    LEFT_EYE = 1
+    RIGHT_EYE = 2
+    LEFT_EAR = 3
+    RIGHT_EAR = 4
+    LEFT_SHOULDER = 5
+    RIGHT_SHOULDER = 6
+    LEFT_ELBOW = 7
+    RIGHT_ELBOW = 8
+    LEFT_WRIST = 9
+    RIGHT_WRIST = 10
+    LEFT_HIP = 11
+    RIGHT_HIP = 12
+    LEFT_KNEE = 13
+    RIGHT_KNEE = 14
+    LEFT_ANKLE = 15
+    RIGHT_ANKLE = 16

CPR/metrics_calculator.py

@@ -0,0 +1,120 @@
+# metrics_calculator.py
+import numpy as np
+from scipy.signal import savgol_filter, find_peaks
+import matplotlib.pyplot as plt
+
+class MetricsCalculator:
+    """Rate and depth calculation from motion data with improved peak detection"""
+    
+    def __init__(self, shoulder_width_cm=45):
+        self.shoulder_width_cm = shoulder_width_cm
+        self.peaks = np.array([])
+        self.peaks_max = np.array([])
+        self.peaks_min = np.array([])
+        self.y_smoothed = np.array([])
+        self.cm_px_ratio = None
+        self.midpoints_list = np.array([])
+        self.shoulder_distances = []
+
+    def smooth_midpoints(self, midpoints):
+        """Apply Savitzky-Golay filter to smooth motion data"""
+        self.midpoints_list = np.array(midpoints)
+        
+        if len(self.midpoints_list) > 5:  # Ensure enough data points
+            try:
+                self.y_smoothed = savgol_filter(
+                    self.midpoints_list[:, 1], 
+                    window_length=10, 
+                    polyorder=2, 
+                    mode='nearest'
+                )
+                return True
+            except Exception as e:
+                print(f"Smoothing error: {e}")
+                self.y_smoothed = self.midpoints_list[:, 1]  # Fallback to original
+                return False
+        else:
+            self.y_smoothed = self.midpoints_list[:, 1]  # Not enough points
+            return False
+
+    def detect_peaks(self):
+        """Improved peak detection with separate max/min peaks"""
+        if self.y_smoothed.size == 0:
+            print("No smoothed values found for peak detection")
+            return False
+            
+        try:
+            self.peaks_max, _ = find_peaks(self.y_smoothed, distance=10)
+            self.peaks_min, _ = find_peaks(-self.y_smoothed, distance=10)
+            self.peaks = np.sort(np.concatenate((self.peaks_max, self.peaks_min)))
+            return len(self.peaks) > 0
+        except Exception as e:
+            print(f"Peak detection error: {e}")
+            return False
+
+    def calculate_metrics(self, shoulder_distances, fps):
+        """Calculate compression metrics with improved calculations"""
+        self.shoulder_distances = shoulder_distances
+        
+        try:
+            # Calculate pixel to cm ratio
+            if len(self.shoulder_distances) > 0:
+                avg_dist = np.mean(self.shoulder_distances)
+                self.cm_px_ratio = self.shoulder_width_cm / avg_dist
+            else:
+                print("No shoulder distances available")
+                return None, None
+
+            # Depth calculation using all peaks
+            depth = None
+            if len(self.peaks) > 1:
+                depth = np.mean(np.abs(np.diff(self.y_smoothed[self.peaks]))) * self.cm_px_ratio
+
+            # Rate calculation using only compression peaks (peaks_max)
+            rate = None
+            if len(self.peaks_max) > 1:
+                rate = 1 / (np.mean(np.diff(self.peaks_max)) / fps) * 60  # Convert to CPM
+
+            return depth, rate
+            
+        except Exception as e:
+            print(f"Metric calculation error: {e}")
+            return None, None
+
+    def plot_motion_curve(self):
+        """Enhanced visualization with original and smoothed data"""
+        if self.midpoints_list.size == 0:
+            print("No midpoint data to plot")
+            return
+
+        plt.figure(figsize=(12, 6))
+        
+        # Plot original and smoothed data
+        plt.plot(self.midpoints_list[:, 1], 
+                label="Original Motion", 
+                color="red", 
+                linestyle="dashed", 
+                alpha=0.6)
+                
+        plt.plot(self.y_smoothed, 
+                label="Smoothed Motion", 
+                color="blue", 
+                linewidth=2)
+
+        # Plot peaks if detected
+        if self.peaks.size > 0:
+            plt.plot(self.peaks, 
+                    self.y_smoothed[self.peaks], 
+                    "x", 
+                    color="green", 
+                    markersize=10, 
+                    label="Peaks")
+        else:
+            print("No peaks to plot")
+
+        plt.xlabel("Frame Number")
+        plt.ylabel("Vertical Position (px)")
+        plt.title("Compression Motion Analysis")
+        plt.grid(True)
+        plt.legend()
+        plt.show()

CPR/pose_estimation.py

@@ -0,0 +1,41 @@
+# pose_estimation.py
+import cv2
+import numpy as np
+from ultralytics import YOLO
+from keypoints import CocoKeypoints
+
+class PoseEstimator:
+    """Human pose estimation using YOLO"""
+    
+    def __init__(self, model_path="yolo11n-pose.pt", min_confidence=0.2):
+        self.model = YOLO(model_path)
+        self.min_confidence = min_confidence
+
+    def detect_poses(self, frame):
+        """Detect human poses in a frame"""
+        try:
+            results = self.model(frame, verbose=False)
+            if not results or len(results[0].keypoints.xy) == 0:
+                return None
+            return results[0]
+        except Exception as e:
+            print(f"Pose detection error: {e}")
+            return None
+
+    def get_keypoints(self, results, person_idx=0):
+        """Extract keypoints for a detected person"""
+        try:
+            if not results or len(results.keypoints.xy) <= person_idx:
+                return None
+            return results.keypoints.xy[person_idx].cpu().numpy()
+        except Exception as e:
+            print(f"Keypoint extraction error: {e}")
+            return None
+
+    def draw_keypoints(self, frame, results):
+        """Draw detected keypoints on frame"""
+        try:
+            return results.plot()
+        except Exception as e:
+            print(f"Keypoint drawing error: {e}")
+            return frame

CPR/posture_analyzer.py

@@ -0,0 +1,124 @@
+# posture_analyzer.py
+import math
+import cv2
+import numpy as np
+from keypoints import CocoKeypoints
+
+class PostureAnalyzer:
+    """Posture analysis and visualization with comprehensive validation"""
+    
+    def __init__(self, right_arm_angle_threshold=210, left_arm_angle_threshold=150, wrist_distance_threshold=170):
+        self.right_arm_angles = []
+        self.left_arm_angles = []
+        self.wrist_distances = []
+        self.right_arm_angle_threshold = right_arm_angle_threshold
+        self.left_arm_angle_threshold = left_arm_angle_threshold
+        self.wrist_distance_threshold = wrist_distance_threshold
+        self.warning_positions = {
+            'arm_angles': (50, 50),
+            'one_handed': (50, 100)
+        }
+
+    def calculate_angle(self, a, b, c):
+        """Calculate angle between three points"""
+        try:
+            ang = math.degrees(math.atan2(c[1]-b[1], c[0]-b[0]) - 
+                         math.atan2(a[1]-b[1], a[0]-b[0]))
+            return ang + 360 if ang < 0 else ang
+        except Exception as e:
+            print(f"Angle calculation error: {e}")
+            return 0
+
+    def check_bended_arms(self, keypoints):
+        """Check for proper arm positioning (returns warnings)"""
+        warnings = []
+        try:
+            # Right arm analysis
+            shoulder = keypoints[CocoKeypoints.RIGHT_SHOULDER.value]
+            elbow = keypoints[CocoKeypoints.RIGHT_ELBOW.value]
+            wrist = keypoints[CocoKeypoints.RIGHT_WRIST.value]
+            right_angle = self.calculate_angle(wrist, elbow, shoulder)
+            self.right_arm_angles.append(right_angle)
+            
+            # Left arm analysis
+            shoulder = keypoints[CocoKeypoints.LEFT_SHOULDER.value]
+            elbow = keypoints[CocoKeypoints.LEFT_ELBOW.value]
+            wrist = keypoints[CocoKeypoints.LEFT_WRIST.value]
+            left_angle = self.calculate_angle(wrist, elbow, shoulder)
+            self.left_arm_angles.append(left_angle)
+            
+            # Analyze angles with moving average
+            avg_right = np.mean(self.right_arm_angles[-10:] if self.right_arm_angles else 0)
+            avg_left = np.mean(self.left_arm_angles[-10:] if self.left_arm_angles else 0)
+            
+            if avg_right > self.right_arm_angle_threshold:
+                warnings.append("Right arm bent")
+            if avg_left < self.left_arm_angle_threshold:
+                warnings.append("Left arm bent")
+                
+        except Exception as e:
+            print(f"Arm angle check error: {e}")
+        
+        return warnings
+
+    def check_missing_arms(self, keypoints):
+        """Check for one-handed CPR pattern (returns warning)"""
+        try:
+            left_wrist = keypoints[CocoKeypoints.LEFT_WRIST.value]
+            right_wrist = keypoints[CocoKeypoints.RIGHT_WRIST.value]
+            
+            wrist_distance = np.linalg.norm(left_wrist - right_wrist)
+            self.wrist_distances.append(wrist_distance)
+            
+            avg_distance = np.mean(self.wrist_distances[-10:] if self.wrist_distances else 0)
+            
+            if avg_distance > self.wrist_distance_threshold:
+                return ["One-handed CPR detected!"]
+        except Exception as e:
+            print(f"One-handed check error: {e}")
+        
+        return []
+
+    def check_correct_hand_position(self, keypoints, chest_point):
+        """Check if hands are in correct position (returns warning)"""
+        try:
+            left_wrist = keypoints[CocoKeypoints.LEFT_WRIST.value]
+            right_wrist = keypoints[CocoKeypoints.RIGHT_WRIST.value]
+            
+            # Calculate distance from chest to wrists
+            left_distance = np.linalg.norm(left_wrist - chest_point)
+            right_distance = np.linalg.norm(right_wrist - chest_point)
+
+            print(f"Left wrist distance: {left_distance}, Right wrist distance: {right_distance}")
+            
+            if left_distance > 100 and right_distance > 100:
+                return ["Hands not in correct position"]
+        except Exception as e:
+            print(f"Hand position check error: {e}")
+        
+        return []
+
+    def validate_posture(self, keypoints, chest_point):
+        """Run all posture validations (returns aggregated warnings)"""
+        warnings = []
+        warnings += self.check_bended_arms(keypoints)
+        warnings += self.check_missing_arms(keypoints)
+        warnings += self.check_correct_hand_position(keypoints, chest_point)
+        return warnings
+
+    def display_warnings(self, frame, warnings):
+
+        """Display all posture warnings at the top of the frame"""
+
+        try:
+            if warnings:
+                # Join all warnings with commas
+                warning_text = ", ".join(warnings)
+                # Display the combined warning text at the top
+                cv2.putText(frame, warning_text,
+                            (10, 30),  # Top-left corner
+                            cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 2)
+        except Exception as e:
+            print(f"Warning display error: {e}")
+
+        return frame

CPR/quick_test.ipynb

@@ -0,0 +1,37 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "c819609e",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Example notebook usage\n",
+    "from pose_estimation import PoseEstimator\n",
+    "from role_classifier import RoleClassifier\n",
+    "from metrics_calculator import MetricsCalculator\n",
+    "\n",
+    "# Test pose estimation\n",
+    "estimator = PoseEstimator()\n",
+    "frame = cv2.imread(\"test_frame.jpg\")\n",
+    "results = estimator.detect_poses(frame)\n",
+    "annotated = estimator.draw_keypoints(frame, results)\n",
+    "cv2.imshow(\"Pose Test\", annotated)\n",
+    "\n",
+    "# Test metrics calculation\n",
+    "calculator = MetricsCalculator()\n",
+    "calculator.smooth_midpoints(midpoints_data)\n",
+    "calculator.detect_peaks()\n",
+    "calculator.plot_motion_curve()"
+   ]
+  }
+ ],
+ "metadata": {
+  "language_info": {
+   "name": "python"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

CPR/role_classifier.py

@@ -0,0 +1,81 @@
+# role_classifier.py
+import cv2
+import numpy as np
+from keypoints import CocoKeypoints
+
+class RoleClassifier:
+    """Role classification and tracking using image processing"""
+    
+    def __init__(self, proximity_thresh=0.3):
+        self.proximity_thresh = proximity_thresh
+        self.rescuer_id = None
+        self.midpoints = []
+        self.shoulder_distances = []
+        self.rescuer_keypoints = None
+
+    def find_rescuer(self, pose_results, frame_shape):
+        """Identify rescuer based on proximity to horizontal objects"""
+        try:
+            boxes = pose_results.boxes.xywh.cpu().numpy()
+            horizontal_objects = [b for b in boxes if b[2] > b[3]*1.5]
+            
+            if not horizontal_objects:
+                return None
+
+            people = []
+            for i, box in enumerate(boxes):
+                if box[3] > box[2]*1.2 and len(pose_results.keypoints) > i:
+                    people.append((i, box[0], box[1]))
+
+            min_distance = float('inf')
+            for (i, px, py) in people:
+                for (hx, hy, hw, hh) in horizontal_objects:
+                    distance = np.sqrt(((px-hx)/frame_shape[1])**2 + ((py-hy)/frame_shape[0])**2)
+                    if distance < min_distance and distance < self.proximity_thresh:
+                        min_distance = distance
+                        self.rescuer_id = i
+            return self.rescuer_id
+        except Exception as e:
+            print(f"Rescuer finding error: {e}")
+            return None
+        
+    def track_rescuer_midpoints(self, keypoints, frame):
+        """Track and draw rescuer's wrist midpoints (using absolute pixel coordinates)"""
+        try:
+            # Store keypoints for shoulder distance calculation
+            self.rescuer_keypoints = keypoints
+            
+            # Get wrist coordinates directly in pixels
+            lw = keypoints[CocoKeypoints.LEFT_WRIST.value]
+            rw = keypoints[CocoKeypoints.RIGHT_WRIST.value]
+            
+            # Calculate midpoint in pixel space
+            midpoint = (
+                int((lw[0] + rw[0]) / 2),
+                int((lw[1] + rw[1]) / 2)
+            )
+            self.midpoints.append(midpoint)
+            
+            # Draw tracking
+            cv2.circle(frame, midpoint, 8, (0,255,0), -1)
+            cv2.putText(frame, "MIDPOINT", (midpoint[0]+5, midpoint[1]-10),
+                    cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0,255,0), 2)
+            return frame
+        except Exception as e:
+            print(f"Midpoint tracking error: {e}")
+            return frame
+
+    def update_shoulder_distance(self):
+        """Helper function to calculate and store shoulder distance"""
+        if self.rescuer_keypoints is None:
+            return
+
+        try:
+            left_shoulder = self.rescuer_keypoints[CocoKeypoints.LEFT_SHOULDER.value]
+            right_shoulder = self.rescuer_keypoints[CocoKeypoints.RIGHT_SHOULDER.value]
+            
+            shoulder_dist = np.linalg.norm(left_shoulder - right_shoulder)
+            self.shoulder_distances.append(shoulder_dist)
+            print(f"Shoulder distance: {shoulder_dist:.2f} pixels")
+        except Exception as e:
+            print(f"Shoulder distance calculation error: {e}")

CPR/yolo11n-pose.pt

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

app.py

@@ -21,6 +21,8 @@ from fastapi import HTTPException
 from fastapi import WebSocket, WebSocketDisconnect
 import base64
 import cv2
+import time
+from CPR import CPRAnalyzer
 
 
 
@@ -199,21 +201,15 @@ async def process_video(file: UploadFile = File(...)):
     with open(video_path, "wb") as buffer:
         shutil.copyfileobj(file.file, buffer)
 
-    model = YOLO("yolo11n-pose_float16.tflite")  # fixed variable name
+    start_time = time.time()
+    print("[START] CPR Analysis started")
 
-    print("Model loaded successfully")
+    analyzer = CPRAnalyzer(video_path)
+    analyzer.run_analysis()
+
+    end_time = time.time()
+    print(f"[END] Total execution time: {end_time - start_time:.2f}s")
 
-    # Run YOLO detection 
-    try:
-        results = model(
-            source=video_path,
-            show=True,     # Set True if running locally and want to view
-            save=True,
-            project="runs/detect",
-            name="testResult"
-        )
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=f"YOLO processing error: {str(e)}")
 
     return JSONResponse(content={"message": "Video processed successfully", "result_dir": "runs/detect/testResult"})
 

requirements.txt

@@ -102,4 +102,5 @@ wfdb==4.3.0
 wrapt==1.17.2
 yarl==1.20.0
 websockets
-xgboost 
+xgboost
+time