肌电信号与姿态检测结合.md

肌电信号与姿态检测结合

• 论文和专利：和老师想的一样，大部分都是两个模块的整合，所以在代码方面我想先把几个识别模型都做一下，肌电信号和姿态，有监督无监督，能够有个比较，整合方面目前没有什么好想法

​ 链接：https://pan.baidu.com/s/1_8a05BFrEKqNQ6YhiPElQg ​ 提取码：xhhb

• 然后我们计划后面每周都开个线上会分享以下进度，9月主要以代码分析和论文为主。因为可能9、10月份有一个中期答辩，前期需要紧凑一点

  每周上传进度：https://trello.com/invite/b/dkYl0tQv/6671da1fc8ccac062040ca43ae807b28/semggesture-measuring

• 按照往年中期答辩之前，需要进行一次费用的报销，我查了一下，咱们在系统里面有3000的额度，如果要做姿态检测，可以买一些摄像头，肌电检测设备等到有实验条件下再考虑

[TOC]

肌电信号

代码

有监督的学习

肌电型号手势识别 https://www.kaggle.com/code/drbilal216/emg-signal-complete-model/notebook

数据集 ：保持同一个手势停留若干秒，记录肌电信号。

导入包、读取数据

import os
import pandas as pd
import tensorflow as tf
import matplotlib.pyplot as plt
import numpy as np

import os
for dirname, _, filenames in os.walk('/kaggle/input'):#filenames 代表遍历的, 内容是该文件夹中所有的文件(不包括子目录)
    for filename in filenames:
         print(os.path.join(dirname, filename))
#读取数据          
import pandas as pd
Input_path = '/kaggle/input/emg-signal-for-gesture-recognition/EMG-data.csv'
df = pd.read_csv(Input_path)#df读取csv文件中的全部数据
print(df.head())#打印每一列标题
print(df.shape)#打印数据的（行，列） 

#数据情况展示
print("class :", df["class"].unique())#unique删除重复项，让类别从小到大排列
print()
print("Value Count :\n",df["class"].value_counts())#对每一个类别出现次数进行计数

#删除label\class\time标签列的情况下输出所有肌电信号数值
features = df.drop(columns=["label","class","time"])
display(features.head())
print(features.shape())
#可以通过.shape()与前面第一次输出进行对比，看到较为明显的差异

#数据类型输出
print(type(Class))
print(type(features))

Class = Class.values
features = features.values

print(type(Class))
print(type(features))

数据预处理：训练集和测试数据划分

此处的划分为训练集0.7，验证集0.1，测试集0.2

from sklearn.model_selection import train_test_split
# 80 and 20
x_train, x_test, y_train, y_test = train_test_split(features, Class, test_size=0.2, random_state=1)

# 归一化数据
mean = x_train.mean(axis=0)#每一列channel的平均值
std = x_train.std(axis=0)#每一列的标准差

x_train -= mean
x_train /= std

x_test -= mean
x_test /= std

# 对标签y编译成一位热码
y_train = tf.keras.utils.to_categorical(y_train)
y_test = tf.keras.utils.to_categorical(y_test)

图像绘制函数

# 训练集、验证集损失函数变化图像绘制函数

def plot(loss,val_loss,acc,val_acc):
    loss = history.history['loss'] #History.history 属性是一个记录了连续迭代的训练/验证（如果存在）损失值和评估值的字典
    val_loss = history.history['val_loss']

    epochs = range(1, len(loss) + 1) #训练的轮数

    plt.plot(epochs, loss, 'bo', label='Training loss') #training的loss值
    plt.plot(epochs, val_loss, 'b', label='Validation loss')#validation的loss值
    plt.title('Training and validation loss')#图像的title
    plt.xlabel('Epochs')#x轴为训练的轮数
    plt.ylabel('Loss')#y轴为训练的loss 下降情况
    plt.legend()#图标
    plt.show()#显示图像

    acc = history.history['accuracy']#准确度
    val_acc = history.history['val_accuracy']#同上一个history

    epochs = range(1, len(acc) + 1)

    plt.plot(epochs, acc, 'bo', label='Training acc')
    plt.plot(epochs, val_acc, 'b', label='Validation acc')
    plt.title('Training and validation acc')
    plt.xlabel('Epochs')
    plt.ylabel('acc')
    plt.legend()
    plt.show()

模型建立

from tensorflow.keras import layers, Sequential, optimizers, Input, Model

input_tensor = Input(shape=(8,))#输入形式为8列
x = layers.Dense(1024, activation='relu')(input_tensor)#输出为_*1024,输入为input_tensor,激活函数为relu,以下同理
y = layers.Dense(512, activation='relu')(x)
z = layers.Dense(256, activation='relu')(y)
z = layers.Dense(128, activation='relu')(z)
z = layers.Dense(64, activation='relu')(z)
z = layers.Dense(32, activation='relu')(z)
z = layers.Dense(128, activation='relu')(y) # 层的非循环图？此处可能是处于调参的考虑
z = layers.Dense(64, activation='relu')(z)
z = layers.Dense(32, activation='relu')(z)
output_tensor = layers.Dense(8, activation='softmax')(z)#输出层

model = Model(input_tensor, output_tensor)#模型囊括了从input到output的所有网络

#SGD #RMSprop #Adam #Adadelta #Adagrad ##Adamax ###Nadam #Ftrl
opt = optimizers.Nadam(lr=1e-3)#优化器设置 lr：大或等于0的浮点数，学习率
model.compile(optimizer = opt, 
              loss = "categorical_crossentropy",
              metrics = ["accuracy"])# 编译为字节代码对象 

# 保存模型，为tensorboaed创建日志，并应用少量回调

def callbacks(Log,Dir):
  import tensorflow as tf
  import os

  Filepath = Path
  logdir = os.path.join(Filepath, Dir)#把目录和文件整合成为一条路径，此处应该为 Path/Dir
  
  callbacks_list = [tf.keras.callbacks.TensorBoard( #tensorboard用来提供可视化工具
                    log_dir=logdir,                 # 用来保存被Tensorboaed分析的日志文件的文件名     
                    histogram_freq=1,),             #对于模型中各个层计算激活值和模型权重直方图的频率(训练轮数中)。如果设置为0，直方图不会被计算
                    tf.keras.callbacks.EarlyStopping(   # 如果误差不再随着训练发生明显改变，此时会自动终止训练
                    monitor='val_accuracy',patience=2,),           # 监控验证准确性 monitor: 被监测的数据'val_accuracy'检验集的准确性；patience 在监测2轮不发生变化，没有进度后停止
                    #tf.keras.callbacks.ReduceLROnPlateau(
                    #monitor='val_loss',factor=0.1,         # lr ko .1 se multiply kerdo (kam kerdo)
                    #patience=10,),                # reduce the lrate if val loss stop improving
                    tf.keras.callbacks.ModelCheckpoint(#在每个训练期之后保存模型。
                    filepath= Filepath,             # 保存路径
                    monitor='val_loss',             # 只保留最好的权重参数
                    save_best_only=True,)]
  return callbacks_list

#对“model1/my_log_dir"文件下存储的数据进行callbacks函数访问
Path = "model1"
Dir = "my_log_dir"   
Call_B_Fun = callbacks(Path,Dir)

#模型尺寸，学习轮数
batch_size = 512            
epochs = 200                
#模型拟合
history = model.fit(x_train, y_train,
                    batch_size=batch_size, epochs = epochs,
                    validation_split = 0.2, callbacks=Call_B_Fun)
#误差迭代
loss = history.history['loss']
val_loss = history.history['val_loss']
acc = history.history['accuracy']
val_acc = history.history['val_accuracy']
plot(loss,val_loss,acc,val_acc)

# 保存模型
model.save('model1/emg_1.h5')

# 载入整个模型，并且赋给emg_model
from tensorflow.keras.models import load_model
emg = 'model1/emg_1.h5'
emg_model = load_model(emg)

function_3的情况

#model.layers 是包含模型网络层的展平列表。
#model.inputs 是模型输入张量的列表。
#model.outputs 是模型输出张量的列表。
#model.summary() 打印出模型概述信息。 它是 utils.print_summary 的简捷调用。

new_model = Model(emg_model.inputs, emg_model.layers[-2].output) # removing layers
new_model.summary()
# removed all layers except conv

在fun3的基础上加一层生成fun5

from tensorflow.keras import layers, optimizers, Input, Model

input_tensor = Input(shape=(8,))#输入塑性
x = new_model(input_tensor)       # fun3
#z = layers.Dense(256, activation='relu')(x)
output_tensor = layers.Dense(8, activation='softmax')(x)#新的输出层，softmax为激活函数

model = Model(input_tensor, output_tensor)#根据已知输入输出建立模型

# 传入优化器名称: 默认参数将被采用
opt = optimizers.Nadam(lr=1e-3)
model.compile(optimizer = opt, 
              loss = "categorical_crossentropy",
              metrics = ["accuracy"])

model.summary()

后文重复前述函数，不再赘述

姿态检测：

文献

Realtime Multi-Person 2D Pose Estimation using Part Affinity Fields

doi:https://ieeexplore.ieee.org/document/8099626

Abstract

提出了一种有效检测图像中多人的2D姿势的方法。该方法使用非参数表示，我们将其称为PAFS，以学习将身体部位与图像中的个体相关联。该体系结构是一个贪心的自下而上解析步骤，该步骤在实现实时性能的同时保持较高的精度，而与图像中的人数无关。先锁定肢体，自下而上区分多个人体，与视野范围内人数无关，能够保障实时计算中的效率。

Introduction

自上而下的结构：先确定单个人体再确定躯干

存在问题：

1. 首先，每个图像可能包含可能在任何位置或刻度上发生的人数未知数。
2. 其次，由于接触，遮挡和肢体关节，人们之间的相互作用引起复杂的空间干扰，从而使零件的关联变得困难。
3. 第三，运行时的复杂性往往会随图像中的人数而增长，从而使实时性能成为挑战。
4. 当人像距离摄像头过近时存在识别失效问题，这种问题无法通过算法修复

自下而上结构：先对躯干进行向量化，进而推导整个人体

解决问题：

1. 早期鲁棒性，算法复杂度与视野中人数关系弱

2. 该体系结构通过相同顺序预测过程的两个分支共同学习零件位置及其关联

   但也存在从part推导全局的算法复杂性，NP类问题，通过基于零件的ResNet（残差神经网络）

文章将介绍一种基于贪心自下而上的算法，首先我们通过PAFS得到一组2D矢量字段，该场在图像域上编码了四肢的位置和方向。同时通过自下而上的编码对全局进行预测，以允许贪心的解析以计算成本的一小部分获得高质量的结果。

开源代码：https://github.com/CMU-Perceptual-Computing-Lab/openpose

Method

输入：w*h的图像

输出：2D的关节位置

过程：

代码

姿态识别包应用—视频转化为框架

GitHub包：https://github.com/tryagainconcepts/tf-pose-estimation/blob/master/README.md

源代码链接：https://www.kaggle.com/code/kmader/running-pose-estimate

导入数据、包、数据读取函数

%load_ext autoreload
%autoreload 2
import seaborn as sns# 可视化工具
import matplotlib.pyplot as plt#绘图工具
#关于一些界面的显示风格设置
plt.rcParams["figure.figsize"] = (8, 8)#不改变分辨率的情况下，改变图片尺寸到8*8
plt.rcParams["figure.dpi"] = 125#分辨率
plt.rcParams["font.size"] = 14#文字尺寸
plt.rcParams['font.family'] = ['sans-serif']#字体
plt.rcParams['font.sans-serif'] = ['DejaVu Sans']
plt.style.use('ggplot')
sns.set_style("whitegrid", {'axes.grid': False})

%matplotlib inline
#一些关于图像监测的库
import tf_pose
import cv2
from glob import glob
from tqdm import tqdm_notebook
from PIL import Image
import numpy as np
import os
#读取视频的函数
def video_gen(in_path):
    c_cap = cv2.VideoCapture(in_path)#打开摄像头获取视频数据
    while c_cap.isOpened():#监测以属于打开状态，读取视频信息
        ret, frame = c_cap.read()
        #读取两个返回值
        #ret是布尔值，读取帧数为正返回True，读取到视频末尾返回值为False
        #frame就是每一帧的画面
        if not ret:
            break
        yield c_cap.get(cv2.CAP_PROP_POS_MSEC), frame[:, :, ::-1]
        #cv2.CAP_PROP_POS_MSEC视频文件的当前位置（播放）以毫秒为单位
    c_cap.release()#及时释放资源
    
video_paths = glob('../input/*.mp4')#视频路径
c_video = video_gen(video_paths[0])#依次读取视频文件
for _ in range(300):
    c_ts, c_frame = next(c_video)
plt.imshow(c_frame)

姿态估计包的效果

#导入姿态监测包，可以直接对图像和视频进行处理
from tf_pose.estimator import TfPoseEstimator
from tf_pose.networks import get_graph_path, model_wh
tfpe = tf_pose.get_estimator()

humans = tfpe.inference(npimg=c_frame, upsample_size=4.0)#对hunman变量赋予读取到的c_frame值
print(humans)

#对图像上的人姿态进行估计
new_image = TfPoseEstimator.draw_humans(c_frame[:, :, ::-1], humans, imgcopy=False)
fig, ax1 = plt.subplots(1, 1, figsize=(10, 10))
ax1.imshow(new_image[:, :, ::-1])#显示视频的最后一帧

#lambda 参数列表:返回值 此处代表c_fig对应的函数的返回值
#format()功能很强大，它把字符串当成一个模板，通过传入的参数进行格式化，并且使用大括号‘{}’作为特殊字符代替‘%’。
#此处可以理解为bp_{k}_{vec_name}
#zip() 函数用于将可迭代的对象作为参数，将对象中对应的元素打包成一个个元组，然后返回由这些元组组成的列表。
#此处输出每一个部位的坐标与得分
body_to_dict = lambda c_fig: {'bp_{}_{}'.format(k, vec_name): vec_val 
                              for k, part_vec in c_fig.body_parts.items() 
                              for vec_name, vec_val in zip(['x', 'y', 'score'],
                                                           (part_vec.x, 1-part_vec.y, part_vec.score))}
c_fig = humans[0]
body_to_dict(c_fig)

MAX_FRAMES = 200
#创建一个身体姿态字典
body_pose_list = []
for vid_path in tqdm_notebook(video_paths, desc='Files'):
    c_video = video_gen(vid_path)#读取视频文件，使用之前定义的函数video_gen
    c_ts, c_frame = next(c_video)#信息指针转向下一个视频文件
    out_path = '{}_out.avi'.format(os.path.split(vid_path)[1])#标记视频文件的存储路径,"os.path.split(vid_path)[1]_out.avi"
    
    out = cv2.VideoWriter(out_path,
                          cv2.VideoWriter_fourcc('M','J','P','G'),#文件存储格式MJPG,指定视频大小为10
                          10, 
                          (c_frame.shape[1], c_frame.shape[0]))
    for (c_ts, c_frame), _ in zip(c_video, 
                                  tqdm_notebook(range(MAX_FRAMES), desc='Frames')):
        bgr_frame = c_frame[:,:,::-1]
        humans = tfpe.inference(npimg=bgr_frame, upsample_size=4.0)#压缩图片
        for c_body in humans:
            body_pose_list += [dict(video=out_path, time=c_ts, **body_to_dict(c_body))]#存储信息
        new_image = TfPoseEstimator.draw_humans(bgr_frame, humans, imgcopy=False)
        out.write(new_image)
    out.release()
#zip() 函数用于将可迭代的对象作为参数，将对象中对应的元素打包成一个个元组，然后返回由这些元组组成的列表。

可视化对比两段视频的差异性

import pandas as pd
body_pose_df = pd.DataFrame(body_pose_list)
body_pose_df.describe()

fig, m_axs = plt.subplots(1, 2, figsize=(15, 5))
for c_ax, (c_name, c_rows) in zip(m_axs, body_pose_df.groupby('video')):
    #随着时间变化不同躯干的Y变化
    for i in range(17):
        c_ax.plot(c_rows['time'], c_rows['bp_{}_y'.format(i)], label='x {}'.format(i))
    c_ax.legend()
    c_ax.set_title(c_name)
    
fig, m_axs = plt.subplots(1, 2, figsize=(15, 5))
for c_ax, (c_name, n_rows) in zip(m_axs, body_pose_df.groupby('video')):
    for i in range(17):
        #随着时间变化不同躯干的bp_得分变化
        c_rows = n_rows.query('bp_{}_score>0.6'.format(i)) # only keep confident results
        c_ax.plot(c_rows['bp_{}_x'.format(i)], c_rows['bp_{}_y'.format(i)], label='BP {}'.format(i))
    c_ax.legend()
    c_ax.set_title(c_name)
    
body_pose_df.to_csv('body_pose.csv', index=False)

有监督的学习-CNN

以手势红外图片为例https://www.kaggle.com/code/benenharrington/hand-gesture-recognition-database-with-cnn/notebook?q=gesture+recognition

数据集：手势识别数据库是十种不同手势的近红外图像的集合。在这个笔记本中，我们使用端到端深度学习来为这些图像构建分类器。 我们将首先加载一些读取和绘制图像所需的包。

链接🔗https://www.kaggle.com/datasets/gti-upm/leapgestrecog

导入包

import numpy as np
import os #访问文件
from PIL import Image # 处理图片
import matplotlib.pyplot as plt
import matplotlib.image as mpimg # 显示图片

建立字典

如数据概述中所述，有 10 个文件夹标记为 00 到 09，每个文件夹包含来自给定主题的图像。在每个文件夹中，每个手势都有子文件夹。我们将构建一个字典查找来存储我们需要识别的手势名称，并为每个手势提供一个数字标识符。我们还将构建一个字典反向查找，它告诉我们什么手势与给定的标识符相关联。

lookup = dict()#建立字典
reverselookup = dict()#建立逆字典
count = 0
for j in os.listdir('../input/leapgestrecog/leapGestRecog/00/'):
    if not j.startswith('.'): #*防止被隐藏的文件被访问
        lookup[j] = count #字典形式 文件名：序号
        reverselookup[count] = j#逆字典形式 序号：文件
        count = count + 1

图像预处理

红外传感器采集图像过大，转换为灰度并调整大小，其中包含一个遍历文件的过程

x_data = [] #图像集合
y_data = [] #分类集合
datacount = 0 # 图片计数器
for i in range(0, 10): # 访问十个文件
    for j in os.listdir('../input/leapgestrecog/leapGestRecog/0' + str(i) + '/'):#将i字符串，这样能够循环表达文件名称，循环访问
        if not j.startswith('.'): # 同样为了避免访问到隐藏文件
            count = 0  
            for k in os.listdir('../input/leapgestrecog/leapGestRecog/0' + 
                                str(i) + '/' + j + '/'):
                                
                img = Image.open('../input/leapgestrecog/leapGestRecog/0' + 
                                 str(i) + '/' + j + '/' + k).convert('L')
                                # 访问'0i'文件中的每一项，并将其灰度图像赋予img
                img = img.resize((320, 120))#调整图像尺寸
                arr = np.array(img)#矩阵赋值
                x_data.append(arr) #在x中存储灰度值
                count = count + 1
            y_values = np.full((count, 1), lookup[j]) #对于count即oi文件中所有的的图像的类别标记为字典lookup中的类别
            y_data.append(y_values)
            datacount = datacount + count#总数=总数+当前文件图片数
x_data = np.array(x_data, dtype = 'float32')
y_data = np.array(y_data)
y_data = y_data.reshape(datacount, 1) # 改变一下y的形状

数据预处理

import keras
from keras.utils import to_categorical
y_data = to_categorical(y_data)#把数据变成二进制

x_data = x_data.reshape((datacount, 120, 320, 1))#重塑x_data的形状，使它的值位于0和1中间
x_data /= 255

#训练集与测试集划分
from sklearn.model_selection import train_test_split
x_train,x_further,y_train,y_further = train_test_split(x_data,y_data,test_size = 0.2)
x_validate,x_test,y_validate,y_test = train_test_split(x_further,y_further,test_size = 0.5)

模型训练

kears有两种不同的建模方法

• Sequential models

  Sequential模型字面上的翻译是顺序模型，给人的第一感觉是那种简单的线性模型，但实际上Sequential模型可以构建非常复杂的神经网络，包括全连接神经网络、卷积神经网络(CNN)、循环神经网络(RNN)、等等。这里的Sequential更准确的应该理解为堆叠，通过堆叠许多层，构建出深度神经网络。

  Sequential模型的核心操作是添加layers（图层）

• Functional API

from keras import layers
from keras import models

model=models.Sequential()
#输入层
model.add(layers.Conv2D(32, (5, 5), strides=(2, 2),activation='relu', input_shape=(120, 320,1))) #添加一个带有32个5*5的过滤器的卷积层，水平滑动步长和垂直滑动步长都为2，其获取120* 320 * 1的输入图像

#最大池化层 把图像压缩从5*5到2*2 选择最大值
model.add(layers.MaxPooling2D((2, 2)))
#重复压缩操作
model.add(layers.Conv2D(64, (3, 3), activation='relu')) 
model.add(layers.MaxPooling2D((2, 2)))
model.add(layers.Conv2D(64, (3, 3), activation='relu'))
model.add(layers.MaxPooling2D((2, 2)))
#展平层，求和平均
model.add(layers.Flatten())
#全连接层：和上一层数据量一样
model.add(layers.Dense(128, activation='relu'))#(神经元节点数，激活函数)
model.add(layers.Dense(10, activation='softmax'))

建模成功

optimizer 优化器 基于训练数据和损失函数来更新网络的机制，常用的有Adam， RMSprop、SGD等

loss 损失函数 网络衡量在训练数据上的性能，即网络如何朝着正确的方向前进。 BinaryCrossentropy, CategoricalCrossentropy,KLDivergence等

metrics 监控指标 训练和测试过程中需要监控的指标。常用的有AUC、Accuracy、BinaryAccuracy、BinaryCrossentropy, CategoricalCrossentropy, KLDivergence、Precision等等

loss和metrics关系 loss和metrics都是用来评价训练过程中模型的预测性能； optimizer是根据loss值进行反向误差方向传播，计算更新网络权值； metrics不参与网络的训练过程，只作为一个监控指标，方便直观显示模型的预测，选择范围相比loss更多； 比如在分类问题中，交叉熵是模型训练的loss，但是我们难以直观通过交叉熵的数值进行判断，需要一个更为直观的质保，因此选择精度作为metrics

model.compile(optimizer='rmsprop',
              loss='categorical_crossentropy',
              metrics=['accuracy'])
model.fit(x_train, y_train, epochs=10, batch_size=64, verbose=1, validation_data=(x_validate, y_validate))
#(输入数据，标签，训练总轮数=10，梯度下降每个batch样本数=64，输出进度条记录，验证集)

#误差计算
[loss, acc] = model.evaluate(x_test,y_test,verbose=1)
print("Accuracy:" + str(acc))