I am trying to extract keypoints using mediapipe to train an LSTM model to detect sign language actions(only for a set of adjectives for now). The training dataset is being parsed and passed to VideoCapture, which is then used to extract keypoints through mediapipe's holistic model, which are then saved as numpy arrays for each frame in .npy files in respective directories. However, due to varying video lengths, the total number of frames captured and thus the number of numpy arrays generated are varying for each video, which is not suitable for the training architecture. It would be straightforward to simply break out of the loop when 30 frames have been captured, but that would exract data only for first 30 frames, which is not desirable. Is there a way to not skip over data whilst getting only a fixed number of frames per video.
import cv2
import numpy as np
import os
from matplotlib import pyplot as plt
import time
import mediapipe as mp
DATASET_PATH = "/home/kuna71/Dev/HearMySign/Datasets/Adjectives_1of8/Adjectives"
KEYPOINT_PATH = "/home/kuna71/Dev/HearMySign/Keypoints"
sequence_len = 30
mp_holistic = mp.solutions.holistic
mp_drawing = mp.solutions.drawing_utils
def mediapipe_detection(image, model):
print("In mediapipe_detection function")
image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) # COLOR CONVERSION BGR 2 RGB
image.flags.writeable = False # Image is no longer writeable (just to save on some memory)
results = model.process(image) # Make prediction
image.flags.writeable = True # Image is now writeable
image = cv2.cvtColor(image, cv2.COLOR_RGB2BGR) # COLOR COVERSION RGB 2 BGR
return image, results
def extract_keypoints(results):
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)
face = np.array([[res.x, res.y, res.z] for res in results.face_landmarks.landmark]).flatten() if results.face_landmarks else np.zeros(468*3)
lh = np.array([[res.x, res.y, res.z] for res in results.left_hand_landmarks.landmark]).flatten() if results.left_hand_landmarks else np.zeros(21*3)
rh = np.array([[res.x, res.y, res.z] for res in results.right_hand_landmarks.landmark]).flatten() if results.right_hand_landmarks else np.zeros(21*3)
return np.concatenate([pose, face, lh, rh])
#opencv code to capture frames
directories = os.listdir(DATASET_PATH)
for d in directories:
vids = os.listdir(os.path.join(DATASET_PATH, d))
for v in vids:
videopath = os.path.join(DATASET_PATH, d, v)
print("\n\n\n\n\n__________________CHANGING VIDEO__________________\n\n\n\n\n")
print("\n\n________VIDEOPATH:"+videopath)
cap = cv2.VideoCapture(videopath)
length = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
print("LENGTH:" + str(length))
# Set mediapipe model
with mp_holistic.Holistic(min_detection_confidence=0.5, min_tracking_confidence=0.5) as holistic:
count =0
while cap.isOpened():
count=count+1
print("Reading feed")
ret, frame = cap.read()
if(frame is None):
break
image, result = mediapipe_detection(frame, holistic)
result_test = extract_keypoints(result)
print(result_test)
if not os.path.exists(os.path.join(KEYPOINT_PATH, d, v)):
os.makedirs(os.path.join(KEYPOINT_PATH, d, v))
NPY_PATH = os.path.join(KEYPOINT_PATH, d, v, str(count))
np.save(NPY_PATH, result_test)
# Show to screen
cv2.imshow('OpenCV Feed', image)
# Break gracefully
if cv2.waitKey(10) & 0xFF == ord('q'):
break
cap.release()
cv2.destroyAllWindows()