Devansh Singhal
Reputation: 11
Kivy video feed freeze
I am building a push up counter, which will count aloud the number of reps after it detects a rep. I am trying to add pyttsx3 with my current code such that it will count out the number of reps. The code without pyttsx3 works fine, however when I add the pyttsx3 module, the video feed tends to freeze when the code detects a rep. Not only that but it would also unexpectedly stop with the long chain of error messages:
[DEBUG ] <comtypes.client._events.CreateEventReceiver.<locals>.Sink object at 0x0000028DFE627860>.Release() -> 1
[DEBUG ] <comtypes.client._events.CreateEventReceiver.<locals>.Sink object at 0x0000028DFE627860>.AddRef() -> 2
[DEBUG ] unimplemented method _ISpeechVoiceEvents_Phoneme called
[DEBUG ] <comtypes.client._events.CreateEventReceiver.<locals>.Sink object at 0x0000028DFE627860>.Release() -> 1
[DEBUG ] <comtypes.client._events.CreateEventReceiver.<locals>.Sink object at 0x0000028DFE627860>.AddRef() -> 2
[DEBUG ] unimplemented method _ISpeechVoiceEvents_Viseme called
[DEBUG ] <comtypes.client._events.CreateEventReceiver.<locals>.Sink object at 0x0000028DFE627860>.Release() -> 1
[DEBUG ] <comtypes.client._events.CreateEventReceiver.<locals>.Sink object at 0x0000028DFE627860>.AddRef() -> 2
[DEBUG ] <comtypes.client._events.CreateEventReceiver.<locals>.Sink object at 0x0000028DFE627860>.Release() -> 1
Process finished with exit code -1
For reference, this is part of my code without pyttsx3:
def load_video(self, dt):
ret, frame = self.capture.read()
if ret:
# Run pose estimation
image_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
results = self.pose.process(image_rgb)
# Extract relevant landmarks for push-up detection
if results.pose_landmarks is not None:
landmarks = results.pose_landmarks.landmark
left_shoulder = landmarks[mp_pose.PoseLandmark.LEFT_SHOULDER.value]
right_shoulder = landmarks[mp_pose.PoseLandmark.RIGHT_SHOULDER.value]
left_elbow = landmarks[mp_pose.PoseLandmark.LEFT_ELBOW.value]
right_elbow = landmarks[mp_pose.PoseLandmark.RIGHT_ELBOW.value]
left_wrist = landmarks[mp_pose.PoseLandmark.LEFT_WRIST.value]
right_wrist = landmarks[mp_pose.PoseLandmark.RIGHT_WRIST.value]
left_hip = landmarks[mp_pose.PoseLandmark.LEFT_HIP.value]
right_hip = landmarks[mp_pose.PoseLandmark.RIGHT_HIP.value]
left_knee = landmarks[mp_pose.PoseLandmark.LEFT_KNEE.value]
right_knee = landmarks[mp_pose.PoseLandmark.RIGHT_KNEE.value]
# Calculate angles for push-up detection
angle_left = self.calculate_angle([left_shoulder.x, left_shoulder.y], [left_elbow.x, left_elbow.y],
[left_wrist.x, left_wrist.y])
angle_right = self.calculate_angle([right_shoulder.x, right_shoulder.y], [right_elbow.x, right_elbow.y],
[right_wrist.x, right_wrist.y])
angle2 = self.calculate_angle([left_shoulder.x, left_shoulder.y], [left_hip.x, left_hip.y],
[left_knee.x, left_knee.y])
angle3 = self.calculate_angle([right_shoulder.x, right_shoulder.y], [right_hip.x, right_hip.y],
[right_knee.x, right_knee.y])
# Update stage based on angle criteria
if angle_left < 90 and angle_right < 90 and angle2 > 165 and angle3 > 165:
self.stage = "down"
else:
self.stage = "up"
# Increment counter if criteria met
if self.stage == "up" and self.prev_stage == "down":
self.counter += 1
self.prev_stage = self.stage
This is the code i tried to come up with:
from kivymd.app import MDApp
from kivymd.uix.boxlayout import MDBoxLayout
from kivymd.uix.button import MDRaisedButton
from kivy.graphics.texture import Texture
from kivy.uix.image import Image
from kivy.clock import Clock
import cv2
import mediapipe as mp
import numpy as np
import pyttsx3
mp_drawing = mp.solutions.drawing_utils
mp_pose = mp.solutions.pose
class App(MDApp):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.engine = pyttsx3.init()
def load_video(self, dt):
ret, frame = self.capture.read()
if ret:
# Run pose estimation
image_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
results = self.pose.process(image_rgb)
# Extract relevant landmarks for push-up detection
if results.pose_landmarks is not None:
landmarks = results.pose_landmarks.landmark
left_shoulder = landmarks[mp_pose.PoseLandmark.LEFT_SHOULDER.value]
right_shoulder = landmarks[mp_pose.PoseLandmark.RIGHT_SHOULDER.value]
left_elbow = landmarks[mp_pose.PoseLandmark.LEFT_ELBOW.value]
right_elbow = landmarks[mp_pose.PoseLandmark.RIGHT_ELBOW.value]
left_wrist = landmarks[mp_pose.PoseLandmark.LEFT_WRIST.value]
right_wrist = landmarks[mp_pose.PoseLandmark.RIGHT_WRIST.value]
left_hip = landmarks[mp_pose.PoseLandmark.LEFT_HIP.value]
right_hip = landmarks[mp_pose.PoseLandmark.RIGHT_HIP.value]
left_knee = landmarks[mp_pose.PoseLandmark.LEFT_KNEE.value]
right_knee = landmarks[mp_pose.PoseLandmark.RIGHT_KNEE.value]
# Calculate angles for push-up detection
angle_left = self.calculate_angle([left_shoulder.x, left_shoulder.y], [left_elbow.x, left_elbow.y], [left_wrist.x, left_wrist.y])
angle_right = self.calculate_angle([right_shoulder.x, right_shoulder.y], [right_elbow.x, right_elbow.y], [right_wrist.x, right_wrist.y])
angle2 = self.calculate_angle([left_shoulder.x, left_shoulder.y], [left_hip.x, left_hip.y], [left_knee.x, left_knee.y])
angle3 = self.calculate_angle([right_shoulder.x, right_shoulder.y], [right_hip.x, right_hip.y], [right_knee.x, right_knee.y])
# Update stage based on angle criteria
if angle_left < 90 and angle_right < 90 and angle2 > 165 and angle3 > 165:
self.stage = "down"
else:
self.stage = "up"
# Increment counter if criteria met
if self.stage == "up" and self.prev_stage == "down":
self.counter += 1
# Speak the current count
self.engine.say(f"Rep number {self.counter}")
self.engine.runAndWait()
self.prev_stage = self.stage
def on_stop(self):
# Release pyttsx3 engine
self.engine.stop()
# Release OpenCV video capture and Mediapipe pose
self.capture.release()
self.pose.close()
super().on_stop()
if __name__ == '__main__':
App().run()
After I faced the errors, I removed the self.engine = pyttsx3.init(), self.engine.stop(), still didnt work.
I also tried creating a whole different function, to no use:
def speak_count_async(self, count):
def speak():
self.engine.say(f"Rep number {count}")
self.engine.runAndWait()
Lastly I also tried logging, also got the same errors:
class App(MDApp):
def __init__(self, **kwargs):
super().__init__(**kwargs)
# Initialize pyttsx3 with 'sapi5' engine and suppress debug logs
self.engine = pyttsx3.init(driverName='sapi5')
logging.getLogger('pyttsx3').setLevel(logging.WARNING)
Im not really sure why these errors are happening anymore, should i use a different tts module? P
lease help me out! Thank you!!
For reference, this is my whole code, without pyttsx3:
from kivymd.app import MDApp
from kivymd.uix.boxlayout import MDBoxLayout
from kivymd.uix.button import MDRaisedButton
from kivy.graphics.texture import Texture
from kivy.uix.image import Image
from kivy.clock import Clock
import cv2
import mediapipe as mp
import numpy as np
mp_drawing = mp.solutions.drawing_utils
mp_pose = mp.solutions.pose
class App(MDApp):
def build(self):
layout = MDBoxLayout(orientation='vertical')
self.image = Image()
layout.add_widget(self.image)
layout.add_widget(MDRaisedButton(
text="Go Back",
pos_hint={'center_x': .5, 'center_y': .5},
size_hint=(None, None))
)
# Initialize OpenCV video capture
self.capture = cv2.VideoCapture(0)
# Initialize Mediapipe pose detection
self.pose = mp_pose.Pose(
min_detection_confidence=0.5,
min_tracking_confidence=0.5
)
# Initialize push-up counter and related variables
self.counter = 0
self.stage = None
self.prev_stage = None
# Schedule the video loading function
Clock.schedule_interval(self.load_video, 1/30) # Adjust frame rate as needed
return layout
def calculate_angle(self, a, b, c):
a = np.array(a)
b = np.array(b)
c = np.array(c)
radians = np.arctan2(c[1] - b[1], c[0] - b[0]) - np.arctan2(a[1] - b[1], a[0] - b[0])
angle = np.abs(radians * 180.0 / np.pi)
if angle > 180.0:
angle = 360.0 - angle
return angle
def load_video(self, dt):
ret, frame = self.capture.read()
if ret:
# Run pose estimation
image_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
results = self.pose.process(image_rgb)
# Extract relevant landmarks for push-up detection
if results.pose_landmarks is not None:
landmarks = results.pose_landmarks.landmark
left_shoulder = landmarks[mp_pose.PoseLandmark.LEFT_SHOULDER.value]
right_shoulder = landmarks[mp_pose.PoseLandmark.RIGHT_SHOULDER.value]
left_elbow = landmarks[mp_pose.PoseLandmark.LEFT_ELBOW.value]
right_elbow = landmarks[mp_pose.PoseLandmark.RIGHT_ELBOW.value]
left_wrist = landmarks[mp_pose.PoseLandmark.LEFT_WRIST.value]
right_wrist = landmarks[mp_pose.PoseLandmark.RIGHT_WRIST.value]
left_hip = landmarks[mp_pose.PoseLandmark.LEFT_HIP.value]
right_hip = landmarks[mp_pose.PoseLandmark.RIGHT_HIP.value]
left_knee = landmarks[mp_pose.PoseLandmark.LEFT_KNEE.value]
right_knee = landmarks[mp_pose.PoseLandmark.RIGHT_KNEE.value]
# Calculate angles for push-up detection
angle_left = self.calculate_angle([left_shoulder.x, left_shoulder.y], [left_elbow.x, left_elbow.y],
[left_wrist.x, left_wrist.y])
angle_right = self.calculate_angle([right_shoulder.x, right_shoulder.y], [right_elbow.x, right_elbow.y],
[right_wrist.x, right_wrist.y])
angle2 = self.calculate_angle([left_shoulder.x, left_shoulder.y], [left_hip.x, left_hip.y],
[left_knee.x, left_knee.y])
angle3 = self.calculate_angle([right_shoulder.x, right_shoulder.y], [right_hip.x, right_hip.y],
[right_knee.x, right_knee.y])
# Update stage based on angle criteria
if angle_left < 90 and angle_right < 90 and angle2 > 165 and angle3 > 165:
self.stage = "down"
else:
self.stage = "up"
# Increment counter if criteria met
if self.stage == "up" and self.prev_stage == "down":
self.counter += 1
self.prev_stage = self.stage
# Display push-up counter on frame
cv2.putText(frame, f'Push-ups: {self.counter}', (10, 50), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2)
# Convert frame to Kivy texture and display in Image widget
buffer = cv2.flip(frame, 0).tobytes()
texture = Texture.create(size=(frame.shape[1], frame.shape[0]), colorfmt='bgr')
texture.blit_buffer(buffer, colorfmt='bgr', bufferfmt='ubyte')
self.image.texture = texture
def on_stop(self):
# Release OpenCV video capture and Mediapipe pose
self.capture.release()
self.pose.close()
super().on_stop()
if __name__ == '__main__':
App().run()
Upvotes: 1
Views: 50
Answers (1)
jbsidis
Reputation: 214
You can move the function that plays the video into a threading function, when there are two threads, kivy one and another one, the app usually freezes, to make sure you can add more threads within the existing thread, add a new thread daemon and run it inside the main Kivy app class:
import threading
class App(MDApp):
def build(self):
return ...
def a_new_thread_for_video(self,the_real_videofunction):
videowilldisplay=threading.Thread(target=the_real_videofunction, args=(dt,)) #dt should be coming from Global or can also be added like a_new_thread_for_video(self,the_real_videofunction,dt)
videowilldisplay.start()
When executing the function to play the video, it can be called like:
app.a_new_thread_for_video(app.load_video) #this moves the cv2 GUI to a new thread in the app lifecylce
Upvotes: 0