How to reshape real-time input data in my prediction code

Viewed 32

This is a real-time sign language detection ,and i reshaped my X_train,y_train,X_test and y_test to add CNN into my architecture .It was LSTM only but am getting errors on the prediction part on how to reshape the real-time input.

import cv2
import numpy as np
import os
from matplotlib import pyplot as plt
import mediapipe as mp
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM,Dense
from tensorflow.keras.callbacks import TensorBoard


mp_holistic = mp.solutions.holistic
mp_drawing = mp.solutions.drawing_utils

def mediapipe_detection(image,model):
  image = cv2.cvtColor(image,cv2.COLOR_BGR2RGB)
  image.flags.writeable = False
  results = model.process(image)
  image.flags.writeable = True
  image = cv2.cvtColor(image,cv2.COLOR_RGB2BGR)
  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)
   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)
  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)
  return np.concatenate([pose,face,lh,rh])


colors = [(245,117,16),(117,245,16),(16,117,245)]

def prob_viz(res,actions,input_frame,colors):
    output_frame = input_frame.copy()
    for num,prob in enumerate(res):
        cv2.rectangle(output_frame, (0,60+num*40), (int(prob*100), 90+num*40),colors[num], -1)
        cv2.putText(output_frame,actions[num],(0,85+num*40), cv2.FONT_HERSHEY_SIMPLEX, 1, 
         (255,255,255),2,cv2.LINE_AA)
    return output_frame


DATA_PATH = os.path.join('MP_Data')
#Actions
actions = np.array(['hello','thanks','iloveyou'])
#30 videos worth of data
no_sequences = 30
#30 frames
sequence_length = 30

for action in actions:
   for sequence in range(no_sequences):
     try:
        os.makedirs(os.path.join(DATA_PATH,action,str(sequence)))
     except:
        pass

label_map = {label:num for num, label in enumerate(actions)}
sequences, labels = [], []
for action in actions:
   for sequence in range(no_sequences):
      window = []
      for frame_num in range(sequence_length):
         res = np.load(os.path.join(DATA_PATH,action, str(sequence)," 
                {}.npy".format(frame_num)))
         window.append(res)
         sequences.append(window)
         labels.append(label_map[action])


 #------------------------------------------------------------------------------------
  #this above codes are to show what my code looks like.But my question starts from here below
 #------------------------------------------------------------------------------------

x_train, x_test, y_train, y_test=train_test_split(x, y, test_size=0.05)
x_train.shape ---->(85, 30, 1662)


# reshaping the input
x_train = x_train.reshape(-1, 300, 1662,1) ; x_test = x_test.reshape(-1, 30, 1662,1)






log_dir = os.path.join('Logs')
tb_callback = TensorBoard(log_dir = log_dir)

# define the model
model = Sequential()

model.add(TimeDistributed(Conv1D(3, 3, 1,activation='relu', input_shape=[30,1662,1])) ) # 
 (3, 128, 216, 1)
# model.add(TimeDistributed(Conv1D(3,3,1,activation='relu')))
model.add(TimeDistributed(MaxPooling1D(pool_size=(3,))))
model.add(TimeDistributed(Flatten()))

model.add(LSTM(320, return_sequences=True, activation='relu'))
model.add(LSTM(640, return_sequences=True, activation='relu'))
model.add(LSTM(320, return_sequences=False, activation='relu'))
model.add(Dense(320, activation='relu'))
model.add(Dense(180, activation='relu'))
model.add(Dense(np.array(actions).shape[0], activation='softmax'))

res = [.2,0.7,.01]

actions[np.argmax(res)]
model.compile(optimizer = 'Adam',loss='categorical_crossentropy',metrics= 
            ['categorical_accuracy'])
actions[np.argmax(res[1])]

model.load_weights('action.h5')

############################################################################################# #Prediction ########################################################################################

#New Detection Variables
sequence = []
sentence = []
threshold = .4

cap = cv2.VideoCapture(0)
#Mediapipe Model
with mp_holistic.Holistic(min_detection_confidence=0.5, min_tracking_confidence=0.5) as holistic:
while cap.isOpened():
    
    #Read Feed
    ret, frame = cap.read()
    
    #Make detections
    image,results = mediapipe_detection(frame,holistic)
    
    #Prediciton Logic
    keypoints = extract_keypoints(results)
    sequence.insert(0,keypoints)
    sequence = sequence[:30]

    if len(sequence) == 30:
        res = model.predict(np.expand_dims(sequence,axis=0))[0]
    
    #Visualization
    if res[np.argmax(res)] > threshold:
        if len(sentence) > 0:
            if actions[np.argmax(res)] != sentence[-1]:
                sentence.append(actions[np.argmax(res)])
        else:
            sentence.append(actions[np.argmax(res)])
    
    if len(sentence)>5:
        sentence = sentence[-5:]
    
    
    #Viz probability
    image = prob_viz(res,actions,image,colors)
    
        
    cv2.rectangle(image,(0,0),(640,40),(245,117,16),-1)
    cv2.putText(image, ' '.join(sentence),(3,30),
                   cv2.FONT_HERSHEY_SIMPLEX, 1,(255,255,255),2,cv2.LINE_AA)
    
    #Show to Screen
    cv2.imshow('OpenCV feed', image)
    
    #Breaking the Feed
    if cv2.waitKey(10) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows() 

But am getting an error on the prediction part

1 Answers

I adjust a bit that is only matching the dataset and expecting values you can modify the target value from my sample. It is possible working with the streams when input is concatenated you can do it with transform functions or queue then transform functions.

[ Sample ]:

import tensorflow as tf
import cv2
import numpy as np
import os
from matplotlib import pyplot as plt
import mediapipe as mp
from tensorflow.keras.callbacks import TensorBoard

"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
: Variables
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
DATA_PATH = os.path.join('MP_Data')
#Actions
actions = np.array(['hello','thanks','iloveyou'])
#30 videos worth of data
no_sequences = 2
# no_sequences = 30
#30 frames
# sequence_length = 30
sequence_length = 2

mp_holistic = mp.solutions.holistic
mp_drawing = mp.solutions.drawing_utils

"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
: Definication / Class
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
def load_target_wave( wave_file='F:\\temp\\Python\\Speech\\Piano\\Berklee44v4\\piano_G2.wav' ) :

    test_file = tf.io.read_file( wave_file )    
    test_audio, sample_rates = tf.audio.decode_wav(contents=test_file)

    return test_audio

def mediapipe_detection(image,model):
    image = cv2.cvtColor(image,cv2.COLOR_BGR2RGB)
    image.flags.writeable = False
    results = model.process(image)
    image.flags.writeable = True
    image = cv2.cvtColor(image,cv2.COLOR_RGB2BGR)
    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)
    lh = np.array([[res.x, res.y, res.z] for res in 
    results.left_hand_landmarks.landmark]).flatten() 
    if results.left_hand_landmarks :
        pass
    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 :
        pass
    else :          
        np.zeros(21*3)

    face = np.array([[res.x, res.y, res.z] for res in 
    results.face_landmarks.landmark]).flatten()     
    if results.face_landmarks :
        pass
    else :
        np.zeros(468*3)
    return np.concatenate([pose,face,lh,rh])


colors = [(245,117,16),(117,245,16),(16,117,245)]

def prob_viz(res,actions,input_frame,colors):
    output_frame = input_frame.copy()
    for num,prob in enumerate(res):
        cv2.rectangle(output_frame, (0,60+num*40), (int(prob*100), 90+num*40),colors[num], -1)
        cv2.putText(output_frame,actions[num],(0,85+num*40), cv2.FONT_HERSHEY_SIMPLEX, 1, 
         (255,255,255),2,cv2.LINE_AA)
    return output_frame


"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
: Generate DATA
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
wav_G2 = load_target_wave( )

for action in actions:
   for sequence in range(no_sequences):
     try:
        os.makedirs(os.path.join(DATA_PATH,action,str(sequence)))
     except:
        pass

label_map = {label:num for num, label in enumerate(actions)}
sequences, labels = [], []
for action in actions:
    for sequence in range(no_sequences):
        window = []
        for frame_num in range(sequence_length):
            # res = np.load(os.path.join(DATA_PATH,action, str(sequence), "{}.npy".format(frame_num)))
            res = wav_G2
            window.append(res)
            sequences.append(window)
            labels.append(label_map[action])


"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
: DataSet
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
sequences = tf.cast( sequences, dtype=tf.int32 )
sequences = tf.constant( sequences, shape=( 12, 1, 2, 287232, 1 ) )
labels = tf.cast( labels, dtype=tf.int32 )
labels = tf.constant( labels, shape=( 12, 1, 1, ) )

dataset = tf.data.Dataset.from_tensor_slices(( sequences, labels ))

log_dir = os.path.join('Logs')
tb_callback = TensorBoard(log_dir = log_dir)

"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
: Model Initialize
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
model = tf.keras.models.Sequential([
    tf.keras.layers.InputLayer(input_shape=( 2, 287232, 1 )),
    tf.keras.layers.Reshape(( 256, 2244, 1 )),
    tf.keras.layers.Normalization(mean=3., variance=2.),
    tf.keras.layers.Normalization(mean=4., variance=6.),
    tf.keras.layers.Conv2D(32, (4, 4), activation='relu'),
    tf.keras.layers.MaxPooling2D((2, 2)),
    tf.keras.layers.Dense(128, activation='relu'),
    tf.keras.layers.Reshape((16128, 1120)),
    tf.keras.layers.Bidirectional(tf.keras.layers.LSTM(96, return_sequences=True, return_state=False)),
    tf.keras.layers.Bidirectional(tf.keras.layers.LSTM(96)),
    tf.keras.layers.Flatten(),
    tf.keras.layers.Dense(192, activation='relu'),
    tf.keras.layers.Dense(1),
])

model.summary()

res = [.2,0.7,.01]

actions[np.argmax(res)]
model.compile(optimizer = 'Adam',loss='categorical_crossentropy',metrics= 
            ['categorical_accuracy'])
actions[np.argmax(res[1])]

"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
: Training
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
history = model.fit(dataset, epochs=1 ,validation_data=(dataset))

# model.load_weights('action.h5')

[ My other sample ]: For idea Sample

Related