# %% [markdown] # # CNN-VIT 视频动态手势识别 # # ### 隔空手势 # %% [markdown] # ### 下载数据 # %% # import os # import moxing as mox # if not os.path.exists('hand_gesture'): # mox.file.copy_parallel('obs://modelbox-course/hand_gesture', 'hand_gesture') # %% [markdown] # ### 准备环境 # %% # conda clean -i -y # %% # conda install cudatoolkit=11.3.1 cudnn=8.2.1 -y --override-channels --channel https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main # %% # !pip install --upgrade pip # %% # !pip install tensorflow-gpu==2.5.0 imageio # %% # !pip install opencv-python -i https://pypi.tuna.tsinghua.edu.cn/simple # %% # conda install tqdm matplotlib==3.5.3 # %% [markdown] # 运行完成请务必点击左上角Kernel->Restart Kernel重启kernel # %% [markdown] # ### 模型训练 # %% import cv2 import glob import numpy as np from tqdm import tqdm import tensorflow as tf from tensorflow import keras from tensorflow.keras import layers import matplotlib.pyplot as plt %matplotlib inline # %% [markdown] # 打印TensorFlow版本并显示可用GPU # %% print('Tensorflow version: {}'.format(tf.__version__)) print('GPU available: {}'.format(tf.config.list_physical_devices('GPU'))) # %% [markdown] # 创建视频输入管道获取视频类别标签 # %% videos = glob.glob('hand_gesture/*.mp4') np.random.shuffle(videos) labels = [int(video.split('_')[-2]) for video in videos] videos[:5], len(videos), labels[:5], len(videos) # %% [markdown] # 显示数据分布情况 # %% from collections import Counter counts = Counter(labels) print(counts) plt.figure(figsize=(8, 4)) plt.bar(counts.keys(), counts.values()) plt.xlabel('Class label') plt.ylabel('Number of samples') plt.title('Class distribution in videos') plt.show() # %% [markdown] # 图像中心裁剪 # %% def crop_center_square(img): h, w = img.shape[:2] square_w = min(h, w) start_x = w // 2 - square_w // 2 end_x = start_x + square_w start_y = h // 2 - square_w // 2 end_y = start_y + square_w result = img[start_y:end_y, start_x:end_x] return result # %% MAX_SEQUENCE_LENGTH = 40 IMG_SIZE = 299 NUM_FEATURES = 1536 # %% [markdown] # 视频抽帧预处理 # %% def load_video(file_name): cap = cv2.VideoCapture(file_name) # 每隔多少帧抽取一次 frame_interval = 4 frames = [] count = 0 while True: ret, frame = cap.read() if not ret: break # 每隔frame_interval帧保存一次 if count % frame_interval == 0: # 中心裁剪 frame = crop_center_square(frame) # 缩放 frame = cv2.resize(frame, (IMG_SIZE, IMG_SIZE)) # BGR -> RGB [0,1,2] -> [2,1,0] frame = frame[:, :, [2, 1, 0]] frames.append(frame) count += 1 return np.array(frames) # %% [markdown] # 显示视频 # %% import random import imageio from IPython.display import Image label_to_name = {0:'无效手势', 1:'上滑', 2:'下滑', 3:'左滑', 4:'右滑', 5:'打开', 6:'关闭', 7:'放大', 8:'缩小'} print(label_to_name.get(labels[0])) frames = load_video(videos[0]) frames = frames[:MAX_SEQUENCE_LENGTH].astype(np.uint8) imageio.mimsave('test.gif', frames, durations=10) display(Image(open('test.gif', 'rb').read())) frames.shape # %% [markdown] # #### InceptionResNetV2 # %% [markdown] # 创建图像特征提取器 # %% def get_feature_extractor(): feature_extractor = keras.applications.inception_resnet_v2.InceptionResNetV2( weights = 'imagenet', include_top = False, pooling = 'avg', input_shape = (IMG_SIZE, IMG_SIZE, 3) ) preprocess_input = keras.applications.inception_resnet_v2.preprocess_input inputs = keras.Input((IMG_SIZE, IMG_SIZE, 3)) preprocessed = preprocess_input(inputs) outputs = feature_extractor(preprocessed) model = keras.Model(inputs, outputs, name = 'feature_extractor') return model # %% feature_extractor = get_feature_extractor() feature_extractor.summary() # %% # from tensorflow.keras.applications import InceptionResNetV2 # weights_path = 'inception_resnet_v2_weights_tf_dim_ordering_tf_kernels_notop.h5' # model = InceptionResNetV2(weights=None, include_top=False, input_shape=(IMG_SIZE, IMG_SIZE, 3)) # model.load_weights(weights_path) # feature_extractor = model # feature_extractor.summary() # %% [markdown] # 提取视频图像特征 # %% def load_data(videos, labels): video_features = [] for video in tqdm(videos): frames = load_video(video) counts = len(frames) # 如果帧数小于MAX_SEQUENCE_LENGTH if counts < MAX_SEQUENCE_LENGTH: # 补白 diff = MAX_SEQUENCE_LENGTH - counts # 创建全0的numpy数组 padding = np.zeros((diff, IMG_SIZE, IMG_SIZE, 3)) # 数组拼接 frames = np.concatenate((frames, padding)) # 获取前MAX_SEQUENCE_LENGTH帧画面 frames = frames[:MAX_SEQUENCE_LENGTH, :] # 批量提取特征 video_feature = feature_extractor.predict(frames) video_features.append(video_feature) return np.array(video_features), np.array(labels) # %% video_features, classes = load_data(videos, labels) video_features.shape, classes.shape # %% [markdown] # #### Dataset # %% batch_size = 16 dataset = tf.data.Dataset.from_tensor_slices((video_features, classes)) dataset = dataset.shuffle(len(videos)) test_count = int(len(videos) * 0.2) train_count = len(videos) - test_count dataset_train = dataset.skip(test_count).cache().repeat() dataset_test = dataset.take(test_count).cache().repeat() train_dataset = dataset_train.shuffle(train_count).batch(batch_size) test_dataset = dataset_test.shuffle(test_count).batch(batch_size) train_dataset, train_count, test_dataset, test_count # %% [markdown] # #### VIT Model # %% # 位置编码 class PositionalEmbedding(layers.Layer): def __init__(self, seq_length, output_dim): super().__init__() # 构造从0~MAX_SEQUENCE_LENGTH的列表 self.positions = tf.range(0, limit=MAX_SEQUENCE_LENGTH) self.positional_embedding = layers.Embedding(input_dim=seq_length, output_dim=output_dim) def call(self,x): # 位置编码 positions_embedding = self.positional_embedding(self.positions) # 输入相加 return x + positions_embedding # %% # 编码器 class TransformerEncoder(layers.Layer): def __init__(self, num_heads, embed_dim): super().__init__() self.p_embedding = PositionalEmbedding(MAX_SEQUENCE_LENGTH, NUM_FEATURES) self.attention = layers.MultiHeadAttention(num_heads=num_heads, key_dim=embed_dim, dropout=0.1) self.layernorm = layers.LayerNormalization() def call(self,x): # positional embedding positional_embedding = self.p_embedding(x) # self attention attention_out = self.attention( query = positional_embedding, value = positional_embedding, key = positional_embedding, attention_mask = None ) # layer norm with residual connection output = self.layernorm(positional_embedding + attention_out) return output # %% def video_cls_model(class_vocab): # 类别数量 classes_num = len(class_vocab) # 定义模型 model = keras.Sequential([ layers.InputLayer(input_shape=(MAX_SEQUENCE_LENGTH, NUM_FEATURES)), TransformerEncoder(2, NUM_FEATURES), layers.GlobalMaxPooling1D(), layers.Dropout(0.1), layers.Dense(classes_num, activation="softmax") ]) # 编译模型 model.compile(optimizer = keras.optimizers.Adam(1e-5), loss = keras.losses.SparseCategoricalCrossentropy(from_logits=False), metrics = ['accuracy'] ) return model # %% # 模型实例化 model = video_cls_model(np.unique(labels)) # 打印模型结构 model.summary() # %% from tensorflow.keras.callbacks import ModelCheckpoint, EarlyStopping, ReduceLROnPlateau # 保存检查点 checkpoint = ModelCheckpoint(filepath='best.h5', monitor='val_loss', save_weights_only=True, save_best_only=True, verbose=1, mode='min') # 提前终止 earlyStopping = EarlyStopping(monitor='loss', patience=50, mode='min', baseline=None) # 减少learning rate rlp = ReduceLROnPlateau(monitor='loss', factor=0.7, patience=30, min_lr=1e-15, mode='min', verbose=1) # %% [markdown] # #### 开始训练 # %% history = model.fit(train_dataset, epochs = 1000, steps_per_epoch = train_count // batch_size, validation_steps = test_count // batch_size, validation_data = test_dataset, callbacks = [checkpoint, earlyStopping, rlp]) # %% [markdown] # #### 绘制结果 # %% plt.plot(history.epoch, history.history['loss'], 'r', label='loss') plt.plot(history.epoch, history.history['val_loss'], 'g--', label='val_loss') plt.title('VIT Model') plt.xlabel('Epoch') plt.ylabel('Loss') plt.legend() # %% plt.plot(history.epoch, history.history['accuracy'], 'r', label='acc') plt.plot(history.epoch, history.history['val_accuracy'], 'g--', label='val_acc') plt.title('VIT Model') plt.xlabel('Epoch') plt.ylabel('Accuracy') plt.legend() # %% [markdown] # 加载训练最优权重 # %% model.load_weights('best.h5') # %% [markdown] # 模型评估 # %% model.evaluate(dataset.batch(batch_size)) # %% [markdown] # 保存模型 # %% model.save('saved_model') # %% [markdown] # ### 手势识别 # %% import random # 加载模型 model = tf.keras.models.load_model('saved_model') # 类别标签 label_to_name = {0:'无效手势', 1:'上滑', 2:'下滑', 3:'左滑', 4:'右滑', 5:'打开', 6:'关闭', 7:'放大', 8:'缩小'} # %% # 获取视频特征 def getVideoFeat(frames): frames_count = len(frames) # 如果帧数小于MAX_SEQUENCE_LENGTH if frames_count < MAX_SEQUENCE_LENGTH: # 补白 diff = MAX_SEQUENCE_LENGTH - frames_count # 创建全0的numpy数组 padding = np.zeros((diff, IMG_SIZE, IMG_SIZE, 3)) # 数组拼接 frames = np.concatenate((frames, padding)) # 取前MAX_SEQ_LENGTH帧 frames = frames[:MAX_SEQUENCE_LENGTH,:] # 计算视频特征 N, 1536 video_feat = feature_extractor.predict(frames) return video_feat # %% # 视频预测 def testVideo(): test_file = random.sample(videos, 1)[0] label = test_file.split('_')[-2] print('文件名:{}'.format(test_file) ) print('真实类别:{}'.format(label_to_name.get(int(label))) ) # 读取视频每一帧 frames = load_video(test_file) # 挑选前帧MAX_SEQUENCE_LENGTH显示 frames = frames[:MAX_SEQUENCE_LENGTH].astype(np.uint8) # 保存为GIF imageio.mimsave('animation.gif', frames, duration=10) # 获取特征 feat = getVideoFeat(frames) # 模型推理 prob = model.predict(tf.expand_dims(feat, axis=0))[0] print('预测类别：') for i in np.argsort(prob)[::-1][:5]: print('{}: {}%'.format(label_to_name[i], round(prob[i]*100, 2))) return display(Image(open('animation.gif', 'rb').read())) # %% [markdown] # 视频推理 # %% for i in range(20): testVideo() 添加usb视频推理