randomshit11/frrf
0
1# import streamlit as st2# import cv23# import mediapipe as mp4# import math5# from PIL import Image6# import numpy as np7 8# ## Build and Load Model9# def attention_block(inputs, time_steps):10# """11# Attention layer for deep neural network12 13# """14# # Attention weights15# a = Permute((2, 1))(inputs)16# a = Dense(time_steps, activation='softmax')(a)17 18# # Attention vector19# a_probs = Permute((2, 1), name='attention_vec')(a)20 21# # Luong's multiplicative score22# output_attention_mul = multiply([inputs, a_probs], name='attention_mul') 23 24# return output_attention_mul25 26# @st.cache(allow_output_mutation=True)27# def build_model(HIDDEN_UNITS=256, sequence_length=30, num_input_values=33*4, num_classes=3):28 29# # Input30# inputs = Input(shape=(sequence_length, num_input_values))31# # Bi-LSTM32# lstm_out = Bidirectional(LSTM(HIDDEN_UNITS, return_sequences=True))(inputs)33# # Attention34# attention_mul = attention_block(lstm_out, sequence_length)35# attention_mul = Flatten()(attention_mul)36# # Fully Connected Layer37# x = Dense(2*HIDDEN_UNITS, activation='relu')(attention_mul)38# x = Dropout(0.5)(x)39# # Output40# x = Dense(num_classes, activation='softmax')(x)41# # Bring it all together42# model = Model(inputs=[inputs], outputs=x)43 44# ## Load Model Weights45# load_dir = "./models/LSTM_Attention.h5" 46# model.load_weights(load_dir)47 48# return model49# threshold1 = st.slider("Minimum Keypoint Detection Confidence", 0.00, 1.00, 0.50)50# threshold2 = st.slider("Minimum Tracking Confidence", 0.00, 1.00, 0.50)51# threshold3 = st.slider("Minimum Activity Classification Confidence", 0.00, 1.00, 0.50)52# ## Real Time Machine Learning and Computer Vision Processes53# class VideoProcessor:54# def __init__(self):55# # Parameters56# self.actions = np.array(['curl', 'press', 'squat'])57# self.sequence_length = 3058# self.colors = [(245,117,16), (117,245,16), (16,117,245)]59# self.threshold = 0.50 # Default threshold for activity classification confidence60 61# # Detection variables62# self.sequence = []63# self.current_action = ''64 65# # Initialize pose model66# self.mp_pose = mp.solutions.pose67# self.mp_drawing = mp.solutions.drawing_utils68# self.pose = self.mp_pose.Pose(min_detection_confidence=0.5, min_tracking_confidence=0.5)69 70# @st.cache()71# def draw_landmarks(self, image, results):72# """73# This function draws keypoints and landmarks detected by the human pose estimation model74 75# """76# self.mp_drawing.draw_landmarks(image, results.pose_landmarks, self.mp_pose.POSE_CONNECTIONS,77# self.mp_drawing.DrawingSpec(color=(245,117,66), thickness=2, circle_radius=2), 78# self.mp_drawing.DrawingSpec(color=(245,66,230), thickness=2, circle_radius=2) 79# )80# return image81 82# @st.cache()83# def extract_keypoints(self, results):84# """85# Processes and organizes the keypoints detected from the pose estimation model 86# to be used as inputs for the exercise decoder models87 88# """89# pose = np.array([[res.x, res.y, res.z, res.visibility] for res in results.pose_landmarks.landmark]).flatten() if results.pose_landmarks else np.zeros(33*4)90# return pose91 92# @st.cache()93# def calculate_angle(self, a, b, c):94# """95# Computes 3D joint angle inferred by 3 keypoints and their relative positions to one another96 97# """98# a = np.array(a) # First99# b = np.array(b) # Mid100# c = np.array(c) # End101 102# radians = np.arctan2(c[1]-b[1], c[0]-b[0]) - np.arctan2(a[1]-b[1], a[0]-b[0])103# angle = np.abs(radians*180.0/np.pi)104 105# if angle > 180.0:106# angle = 360-angle107 108# return angle 109 110# @st.cache()111# def get_coordinates(self, landmarks, side, joint):112# """113# Retrieves x and y coordinates of a particular keypoint from the pose estimation model114 115# Args:116# landmarks: processed keypoints from the pose estimation model117# side: 'left' or 'right'. Denotes the side of the body of the landmark of interest.118# joint: 'shoulder', 'elbow', 'wrist', 'hip', 'knee', or 'ankle'. Denotes which body joint is associated with the landmark of interest.119 120# """121# coord = getattr(self.mp_pose.PoseLandmark, side.upper() + "_" + joint.upper())122# x_coord_val = landmarks[coord.value].x123# y_coord_val = landmarks[coord.value].y124# return [x_coord_val, y_coord_val] 125 126# @st.cache()127# def viz_joint_angle(self, image, angle, joint):128# """129# Displays the joint angle value near the joint within the image frame130 131# """132# cv2.putText(image, str(int(angle)), 133# tuple(np.multiply(joint, [640, 480]).astype(int)), 134# cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 255), 2, cv2.LINE_AA135# )136# return137 138# @st.cache()139# def process_video_input(self, threshold1, threshold2, threshold3):140# """141# Processes the video input and performs real-time action recognition and rep counting.142 143# """144# video_file = st.file_uploader("Upload Video", type=["mp4", "avi"])145# if video_file is None:146# st.warning("Please upload a video file.")147# return148 149# cap = cv2.VideoCapture(video_file)150# if not cap.isOpened():151# st.error("Error opening video stream or file.")152# return153 154# while cap.isOpened():155# ret, frame = cap.read()156# if not ret:157# break158 159# # Convert frame to RGB (Mediapipe requires RGB input)160# frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)161 162# # Pose estimation163# results = self.pose.process(frame_rgb)164 165# # Draw landmarks166# self.draw_landmarks(frame, results)167 168# # Extract keypoints169# keypoints = self.extract_keypoints(results)170 171# # Visualize probabilities172# if len(self.sequence) == self.sequence_length:173# sequence = np.array([self.sequence])174# res = model.predict(sequence)175# frame = self.prob_viz(res[0], frame)176 177# # Append frame to output frames178# out_frames.append(frame)179 180# # Release video capture181# cap.release()182# # # Create an instance of VideoProcessor183# # video_processor = VideoProcessor()184 185# # # Call the process_video_input method186# # video_processor.process_video_input(threshold1, threshold2, threshold3)187 188# # Define Streamlit app189# def main():190# st.title("Real-time Exercise Detection")191# video_file = st.file_uploader("Upload a video file", type=["mp4", "avi"])192# if video_file is not None:193# st.video(video_file)194# video_processor = VideoProcessor()195# frames = video_processor.process_video(video_file)196# for frame in frames:197# st.image(frame, channels="BGR")198 199# if __name__ == "__main__":200# main()201 202 203 204import streamlit as st205import cv2206import mediapipe as mp207import numpy as np208import math209from tensorflow.keras.models import Model210from tensorflow.keras.layers import (LSTM, Dense, Dropout, Input, Flatten, 211 Bidirectional, Permute, multiply)212 213# Load the pose estimation model from Mediapipe214mp_pose = mp.solutions.pose 215mp_drawing = mp.solutions.drawing_utils 216pose = mp_pose.Pose(min_detection_confidence=0.5, min_tracking_confidence=0.5) 217 218# Define the attention block for the LSTM model219def attention_block(inputs, time_steps):220 a = Permute((2, 1))(inputs)221 a = Dense(time_steps, activation='softmax')(a)222 a_probs = Permute((2, 1), name='attention_vec')(a)223 output_attention_mul = multiply([inputs, a_probs], name='attention_mul') 224 return output_attention_mul225 226# Build and load the LSTM model227@st.cache(allow_output_mutation=True)228def build_model(HIDDEN_UNITS=256, sequence_length=30, num_input_values=33*4, num_classes=3):229 inputs = Input(shape=(sequence_length, num_input_values))230 lstm_out = Bidirectional(LSTM(HIDDEN_UNITS, return_sequences=True))(inputs)231 attention_mul = attention_block(lstm_out, sequence_length)232 attention_mul = Flatten()(attention_mul)233 x = Dense(2*HIDDEN_UNITS, activation='relu')(attention_mul)234 x = Dropout(0.5)(x)235 x = Dense(num_classes, activation='softmax')(x)236 model = Model(inputs=[inputs], outputs=x)237 load_dir = "./models/LSTM_Attention.h5" 238 model.load_weights(load_dir)239 return model240 241# Define the VideoProcessor class for real-time video processing242class VideoProcessor:243 def __init__(self):244 self.actions = np.array(['curl', 'press', 'squat'])245 self.sequence_length = 30246 self.colors = [(245,117,16), (117,245,16), (16,117,245)]247 self.pose = mp_pose.Pose(min_detection_confidence=0.5, min_tracking_confidence=0.5)248 self.model = build_model()249 250 def process_video(self, video_file):251 cap = cv2.VideoCapture(video_file)252 out_frames = []253 while cap.isOpened():254 ret, frame = cap.read()255 if not ret:256 break257 frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)258 results = self.pose.process(frame_rgb)259 frame = self.draw_landmarks(frame, results)260 out_frames.append(frame)261 cap.release()262 return out_frames263 264 def draw_landmarks(self, image, results):265 mp_drawing.draw_landmarks(image, results.pose_landmarks, mp_pose.POSE_CONNECTIONS,266 mp_drawing.DrawingSpec(color=(245,117,66), thickness=2, circle_radius=2), 267 mp_drawing.DrawingSpec(color=(245,66,230), thickness=2, circle_radius=2))268 return image269 270# Define Streamlit app271def main():272 st.title("Real-time Exercise Detection")273 video_file = st.file_uploader("Upload a video file", type=["mp4", "avi"])274 if video_file is not None:275 st.video(video_file)276 video_processor = VideoProcessor()277 frames = video_processor.process_video(video_file)278 for frame in frames:279 st.image(frame, channels="BGR")280 281if __name__ == "__main__":282 main()283 284 285 286 287 288# import streamlit as st289# import cv2290 291# from tensorflow.keras.models import Model292# from tensorflow.keras.layers import (LSTM, Dense, Dropout, Input, Flatten, 293# Bidirectional, Permute, multiply)294 295# import numpy as np296# import mediapipe as mp297# import math298# import streamlit as st299# import cv2300# import mediapipe as mp301# import math302 303# # from streamlit_webrtc import webrtc_streamer, WebRtcMode, RTCConfiguration304# import av305# from io import BytesIO306# import av307# from PIL import Image308 309# ## Build and Load Model310# def attention_block(inputs, time_steps):311# """312# Attention layer for deep neural network313 314# """315# # Attention weights316# a = Permute((2, 1))(inputs)317# a = Dense(time_steps, activation='softmax')(a)318 319# # Attention vector320# a_probs = Permute((2, 1), name='attention_vec')(a)321 322# # Luong's multiplicative score323# output_attention_mul = multiply([inputs, a_probs], name='attention_mul') 324 325# return output_attention_mul326 327# @st.cache(allow_output_mutation=True)328# def build_model(HIDDEN_UNITS=256, sequence_length=30, num_input_values=33*4, num_classes=3):329 330# # Input331# inputs = Input(shape=(sequence_length, num_input_values))332# # Bi-LSTM333# lstm_out = Bidirectional(LSTM(HIDDEN_UNITS, return_sequences=True))(inputs)334# # Attention335# attention_mul = attention_block(lstm_out, sequence_length)336# attention_mul = Flatten()(attention_mul)337# # Fully Connected Layer338# x = Dense(2*HIDDEN_UNITS, activation='relu')(attention_mul)339# x = Dropout(0.5)(x)340# # Output341# x = Dense(num_classes, activation='softmax')(x)342# # Bring it all together343# model = Model(inputs=[inputs], outputs=x)344 345# ## Load Model Weights346# load_dir = "./models/LSTM_Attention.h5" 347# model.load_weights(load_dir)348 349# return model350 351# HIDDEN_UNITS = 256352# model = build_model(HIDDEN_UNITS)353# threshold1 = st.slider("Minimum Keypoint Detection Confidence", 0.00, 1.00, 0.50)354# threshold2 = st.slider("Minimum Tracking Confidence", 0.00, 1.00, 0.50)355# threshold3 = st.slider("Minimum Activity Classification Confidence", 0.00, 1.00, 0.50)356 357# ## Mediapipe358# mp_pose = mp.solutions.pose # Pre-trained pose estimation model from Google Mediapipe359# mp_drawing = mp.solutions.drawing_utils # Supported Mediapipe visualization tools360# pose = mp_pose.Pose(min_detection_confidence=threshold1, min_tracking_confidence=threshold2) # mediapipe pose model361 362# ## Real Time Machine Learning and Computer Vision Processes363# class VideoProcessor:364# def __init__(self):365# # Parameters366# self.actions = np.array(['curl', 'press', 'squat'])367# self.sequence_length = 30368# self.colors = [(245,117,16), (117,245,16), (16,117,245)]369# self.threshold = threshold3370 371# # Detection variables372# self.sequence = []373# self.current_action = ''374 375# # Rep counter logic variables376# self.curl_counter = 0377# self.press_counter = 0378# self.squat_counter = 0379# self.curl_stage = None380# self.press_stage = None381# self.squat_stage = None382 383# @st.cache() 384# def draw_landmarks(self, image, results):385# """386# This function draws keypoints and landmarks detected by the human pose estimation model387 388# """389# mp_drawing.draw_landmarks(image, results.pose_landmarks, mp_pose.POSE_CONNECTIONS,390# mp_drawing.DrawingSpec(color=(245,117,66), thickness=2, circle_radius=2), 391# mp_drawing.DrawingSpec(color=(245,66,230), thickness=2, circle_radius=2) 392# )393# return394 395# @st.cache()396# def extract_keypoints(self, results):397# """398# Processes and organizes the keypoints detected from the pose estimation model 399# to be used as inputs for the exercise decoder models400 401# """402# pose = np.array([[res.x, res.y, res.z, res.visibility] for res in results.pose_landmarks.landmark]).flatten() if results.pose_landmarks else np.zeros(33*4)403# return pose404 405# @st.cache()406# def calculate_angle(self, a,b,c):407# """408# Computes 3D joint angle inferred by 3 keypoints and their relative positions to one another409 410# """411# a = np.array(a) # First412# b = np.array(b) # Mid413# c = np.array(c) # End414 415# radians = np.arctan2(c[1]-b[1], c[0]-b[0]) - np.arctan2(a[1]-b[1], a[0]-b[0])416# angle = np.abs(radians*180.0/np.pi)417 418# if angle > 180.0:419# angle = 360-angle420 421# return angle 422 423# @st.cache()424# def get_coordinates(self, landmarks, mp_pose, side, joint):425# """426# Retrieves x and y coordinates of a particular keypoint from the pose estimation model427 428# Args:429# landmarks: processed keypoints from the pose estimation model430# mp_pose: Mediapipe pose estimation model431# side: 'left' or 'right'. Denotes the side of the body of the landmark of interest.432# joint: 'shoulder', 'elbow', 'wrist', 'hip', 'knee', or 'ankle'. Denotes which body joint is associated with the landmark of interest.433 434# """435# coord = getattr(mp_pose.PoseLandmark,side.upper()+"_"+joint.upper())436# x_coord_val = landmarks[coord.value].x437# y_coord_val = landmarks[coord.value].y438# return [x_coord_val, y_coord_val] 439 440# @st.cache()441# def viz_joint_angle(self, image, angle, joint):442# """443# Displays the joint angle value near the joint within the image frame444 445# """446# cv2.putText(image, str(int(angle)), 447# tuple(np.multiply(joint, [640, 480]).astype(int)), 448# cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 255), 2, cv2.LINE_AA449# )450# return451# @st.cache()452# def process_video(self, video_file):453# """454# Processes each frame of the input video, performs pose estimation, 455# and counts repetitions of each exercise.456 457# Args:458# video_file (BytesIO): Input video file.459 460# Returns:461# tuple: A tuple containing the processed video frames with annotations462# and the final count of repetitions for each exercise.463# """464# cap = cv2.VideoCapture(video_file)465# out_frames = []466# # Initialize repetition counters467# self.curl_counter = 0468# self.press_counter = 0469# self.squat_counter = 0470 471# while cap.isOpened():472# ret, frame = cap.read()473# if not ret:474# break475 476# # Convert frame to RGB (Mediapipe requires RGB input)477# frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)478 479# # Pose estimation480# results = pose.process(frame_rgb)481 482# # Draw landmarks483# self.draw_landmarks(frame, results)484 485# # Extract keypoints486# keypoints = self.extract_keypoints(results)487 488# # Count repetitions489# self.count_reps(frame, results.pose_landmarks, mp_pose)490 491# # Visualize probabilities492# if len(self.sequence) == self.sequence_length:493# sequence = np.array([self.sequence])494# res = model.predict(sequence)495# frame = self.prob_viz(res[0], frame)496 497# # Append frame to output frames498# out_frames.append(frame)499 500# # Release video capture501# cap.release()502 503# # Return annotated frames and repetition counts504# return out_frames, {'curl': self.curl_counter, 'press': self.press_counter, 'squat': self.squat_counter}505# @st.cache()506 507# def count_reps(self, image, landmarks, mp_pose):508# """509# Counts repetitions of each exercise. Global count and stage (i.e., state) variables are updated within this function.510 511# """512 513# if self.current_action == 'curl':514# # Get coords515# shoulder = self.get_coordinates(landmarks, mp_pose, 'left', 'shoulder')516# elbow = self.get_coordinates(landmarks, mp_pose, 'left', 'elbow')517# wrist = self.get_coordinates(landmarks, mp_pose, 'left', 'wrist')518 519# # calculate elbow angle520# angle = self.calculate_angle(shoulder, elbow, wrist)521 522# # curl counter logic523# if angle < 30:524# self.curl_stage = "up" 525# if angle > 140 and self.curl_stage =='up':526# self.curl_stage="down" 527# self.curl_counter +=1528# self.press_stage = None529# self.squat_stage = None530 531# # Viz joint angle532# self.viz_joint_angle(image, angle, elbow)533 534# elif self.current_action == 'press': 535# # Get coords536# shoulder = self.get_coordinates(landmarks, mp_pose, 'left', 'shoulder')537# elbow = self.get_coordinates(landmarks, mp_pose, 'left', 'elbow')538# wrist = self.get_coordinates(landmarks, mp_pose, 'left', 'wrist')539 540# # Calculate elbow angle541# elbow_angle = self.calculate_angle(shoulder, elbow, wrist)542 543# # Compute distances between joints544# shoulder2elbow_dist = abs(math.dist(shoulder,elbow))545# shoulder2wrist_dist = abs(math.dist(shoulder,wrist))546 547# # Press counter logic548# if (elbow_angle > 130) and (shoulder2elbow_dist < shoulder2wrist_dist):549# self.press_stage = "up"550# if (elbow_angle < 50) and (shoulder2elbow_dist > shoulder2wrist_dist) and (self.press_stage =='up'):551# self.press_stage='down'552# self.press_counter += 1553# self.curl_stage = None554# self.squat_stage = None555 556# # Viz joint angle557# self.viz_joint_angle(image, elbow_angle, elbow)558 559# elif self.current_action == 'squat':560# # Get coords561# # left side562# left_shoulder = self.get_coordinates(landmarks, mp_pose, 'left', 'shoulder')563# left_hip = self.get_coordinates(landmarks, mp_pose, 'left', 'hip')564# left_knee = self.get_coordinates(landmarks, mp_pose, 'left', 'knee')565# left_ankle = self.get_coordinates(landmarks, mp_pose, 'left', 'ankle')566# # right side567# right_shoulder = self.get_coordinates(landmarks, mp_pose, 'right', 'shoulder')568# right_hip = self.get_coordinates(landmarks, mp_pose, 'right', 'hip')569# right_knee = self.get_coordinates(landmarks, mp_pose, 'right', 'knee')570# right_ankle = self.get_coordinates(landmarks, mp_pose, 'right', 'ankle')571 572# # Calculate knee angles573# left_knee_angle = self.calculate_angle(left_hip, left_knee, left_ankle)574# right_knee_angle = self.calculate_angle(right_hip, right_knee, right_ankle)575 576# # Calculate hip angles577# left_hip_angle = self.calculate_angle(left_shoulder, left_hip, left_knee)578# right_hip_angle = self.calculate_angle(right_shoulder, right_hip, right_knee)579 580# # Squat counter logic581# thr = 165582# if (left_knee_angle < thr) and (right_knee_angle < thr) and (left_hip_angle < thr) and (right_hip_angle < thr):583# self.squat_stage = "down"584# if (left_knee_angle > thr) and (right_knee_angle > thr) and (left_hip_angle > thr) and (right_hip_angle > thr) and (self.squat_stage =='down'):585# self.squat_stage='up'586# self.squat_counter += 1587# self.curl_stage = None588# self.press_stage = None589 590# # Viz joint angles591# self.viz_joint_angle(image, left_knee_angle, left_knee)592# self.viz_joint_angle(image, left_hip_angle, left_hip)593 594# else:595# pass596# return597 598# @st.cache()599# def prob_viz(self, res, input_frame):600# """601# This function displays the model prediction probability distribution over the set of exercise classes602# as a horizontal bar graph603 604# """605# output_frame = input_frame.copy()606# for num, prob in enumerate(res): 607# cv2.rectangle(output_frame, (0,60+num*40), (int(prob*100), 90+num*40), self.colors[num], -1)608# cv2.putText(output_frame, self.actions[num], (0, 85+num*40), cv2.FONT_HERSHEY_SIMPLEX, 1, (255,255,255), 2, cv2.LINE_AA)609 610# return output_frame611 612 613# # Slider widgets614# threshold1 = st.slider("Minimum Keypoint Detection Confidence", 0.00, 1.00, 0.50)615# threshold2 = st.slider("Minimum Tracking Confidence", 0.00, 1.00, 0.50)616# threshold3 = st.slider("Minimum Activity Classification Confidence", 0.00, 1.00, 0.50)617 618# # Sidebar619# st.sidebar.header("Settings")620# st.sidebar.write("Adjust the confidence thresholds")621 622# # Call process_video_input() method from VideoProcessor623# video_processor.process_video_input(threshold1, threshold2, threshold3)624# # def process_uploaded_file(self, file):625# # """626# # Function to process an uploaded image or video file and run the fitness trainer AI627# # Args:628# # file (BytesIO): uploaded image or video file629# # Returns:630# # numpy array: processed image with keypoint detection and fitness activity classification visualized631# # """632# # # Initialize an empty list to store processed frames633# # processed_frames = []634 635# # # Check if the uploaded file is a video636# # is_video = hasattr(file, 'name') and file.name.endswith(('.mp4', '.avi', '.mov'))637 638# # if is_video:639# # container = av.open(file)640# # for frame in container.decode(video=0):641# # # Convert the frame to OpenCV format642# # image = frame.to_image().convert("RGB")643# # image = np.array(image)644 645# # # Process the frame646# # processed_frame = self.process(image)647 648# # # Append the processed frame to the list649# # processed_frames.append(processed_frame)650 651# # # Close the video file container652# # container.close()653# # else:654# # # If the uploaded file is an image655# # # Load the image from the BytesIO object656# # image = Image.open(file)657# # image = np.array(image)658 659# # # Process the image660# # processed_frame = self.process(image)661 662# # # Append the processed frame to the list663# # processed_frames.append(processed_frame)664 665# # return processed_frames666 667# # def recv_uploaded_file(self, file):668# # """669# # Receive and process an uploaded video file670# # Args:671# # file (BytesIO): uploaded video file672# # Returns:673# # List[av.VideoFrame]: list of processed video frames674# # """675# # # Process the uploaded file676# # processed_frames = self.process_uploaded_file(file)677 678# # # Convert processed frames to av.VideoFrame objects679# # av_frames = []680# # for frame in processed_frames:681# # av_frame = av.VideoFrame.from_ndarray(frame, format="bgr24")682# # av_frames.append(av_frame)683 684# # return av_frames685 686# # # Options687# # RTC_CONFIGURATION = RTCConfiguration(688# # {"iceServers": [{"urls": ["stun:stun.l.google.com:19302"]}]}689# # )690 691# # # Streamer692# # webrtc_ctx = webrtc_streamer(693# # key="AI trainer",694# # mode=WebRtcMode.SENDRECV,695# # rtc_configuration=RTC_CONFIGURATION,696# # media_stream_constraints={"video": True, "audio": False},697# # video_processor_factory=VideoProcessor,698# # async_processing=True,699# # )