python姿态识别+Tensflow1.12+pyqt5+UI

import datetime
from PyQt5.QtCore import QCoreApplication
from PyQt5.QtWidgets import QMainWindow, QApplication, QFileDialog
from vedio import vedio
from HumanPoseRec import Ui_MainWindow
from PyQt5 import QtCore, QtGui, QtWidgets
'''
Example of usage:
(1) Test on video file:
python src/s5_test.py \
--model_path model/trained_classifier.pickle \
--data_type video \
--data_path data_test/exercise.avi \
--output_folder output
(2) Test on a folder of images:
python src/s5_test.py \
--model_path model/trained_classifier.pickle \
--data_type folder \
--data_path data_test/apple/ \
--output_folder output
(3) Test on web camera:
python src/s5_test.py \
--model_path model/trained_classifier.pickle \
--data_type webcam \
--data_path 0 \
--output_folder output
'''
SRC_DATA_TYPE = "webcam"
SRC_DATA_PATH = "0"
SRC_MODEL_PATH = r'D:\pysave2023\Action-Recognition\model	rained_classifier.pickle'
output_folder = "output/"
ith_img = -1
predict_label = {}
if True:  # Include project path
import sys
import os
ROOT = os.path.dirname(os.path.abspath(__file__)) + "/../"
CURR_PATH = os.path.dirname(os.path.abspath(__file__)) + "/"
sys.path.append(ROOT)
import utils.lib_images_io as lib_images_io
import utils.lib_plot as lib_plot
import utils.lib_commons as lib_commons
from utils.lib_openpose import SkeletonDetector
from utils.lib_tracker import Tracker
from utils.lib_tracker import Tracker
from utils.lib_classifier import ClassifierOnlineTest
from utils.lib_classifier import *  # Import all sklearn related libraries
# -- Command-line input
def get_dst_folder_name(src_data_type, src_data_path):
global folder_name
try:
if src_data_type == "video":  # /root/data/video.avi --> video
folder_name = os.path.basename(src_data_path).split(".")[-2]
elif src_data_type == "folder":  # /root/data/video/ --> video
folder_name = src_data_path.rstrip("/").split("/")[-1]
elif src_data_type == "webcam":
# month-day-hour-minute-seconds, e.g.: 02-26-15-51-12
folder_name = lib_commons.get_time_string()
except:
pass
return folder_name
class MultiPersonClassifier(object):
''' This is a wrapper around ClassifierOnlineTest
for recognizing actions of multiple people.
'''
def __init__(self, model_path, classes):
self.dict_id2clf = {}  # human id -> classifier of this person
# Define a function for creating classifier for new people.
self._create_classifier = lambda human_id: ClassifierOnlineTest(
model_path, classes, WINDOW_SIZE, human_id)
def classify(self, dict_id2skeleton):
''' Classify the action type of each skeleton in dict_id2skeleton '''
# Clear people not in view
old_ids = set(self.dict_id2clf)
cur_ids = set(dict_id2skeleton)
humans_not_in_view = list(old_ids - cur_ids)
for human in humans_not_in_view:
del self.dict_id2clf[human]
# Predict each person's action
id2label = {}
for id, skeleton in dict_id2skeleton.items():
if id not in self.dict_id2clf:  # add this new person
self.dict_id2clf[id] = self._create_classifier(id)
classifier = self.dict_id2clf[id]
id2label[id] = classifier.predict(skeleton)  # predict label
# print("
Predicting label for human{}".format(id))
# print("  skeleton: {}".format(skeleton))
# print("  label: {}".format(id2label[id]))
return id2label
def get_classifier(self, id):
''' Get the classifier based on the person id.
Arguments:
id {int or "min"}
'''
if len(self.dict_id2clf) == 0:
return None
if id == 'min':
id = min(self.dict_id2clf.keys())
return self.dict_id2clf[id]
def remove_skeletons_with_few_joints(skeletons):
''' Remove bad skeletons before sending to the tracker '''
good_skeletons = []
for skeleton in skeletons:
px = skeleton[2:2 + 13 * 2:2]
py = skeleton[3:2 + 13 * 2:2]
num_valid_joints = len([x for x in px if x != 0])
num_leg_joints = len([x for x in px[-6:] if x != 0])
total_size = max(py) - min(py)
# !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
# IF JOINTS ARE MISSING, TRY CHANGING THESE VALUES:
# !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
if num_valid_joints >= 5 and total_size >= 0.1 and num_leg_joints >= 0:
# add this skeleton only when all requirements are satisfied
good_skeletons.append(skeleton)
return good_skeletons
cfg_all = lib_commons.read_yaml(ROOT + "config/config.yaml")
cfg = cfg_all["s5_test.py"]
CLASSES = np.array(cfg_all["classes"])
SKELETON_FILENAME_FORMAT = cfg_all["skeleton_filename_format"]
# Action recognition: number of frames used to extract features.
WINDOW_SIZE = int(cfg_all["features"]["window_size"])
dict_id2label = {}
scale_h = 0
# Output folder
DST_FOLDER_NAME = get_dst_folder_name(SRC_DATA_TYPE, SRC_DATA_PATH)
DST_FOLDER = output_folder + "/" + DST_FOLDER_NAME + "/"
DST_SKELETON_FOLDER_NAME = cfg["output"]["skeleton_folder_name"]
DST_VIDEO_NAME = cfg["output"]["video_name"]
# framerate of output video.avi
DST_VIDEO_FPS = float(cfg["output"]["video_fps"])
# Video setttings
# If data_type is webcam, set the max frame rate.
SRC_WEBCAM_MAX_FPS = float(cfg["settings"]["source"]
["webcam_max_framerate"])
# If data_type is video, set the sampling interval.
# For example, if it's 3, then the video will be read 3 times faster.
SRC_VIDEO_SAMPLE_INTERVAL = int(cfg["settings"]["source"]
["video_sample_interval"])
# Openpose settings
OPENPOSE_MODEL = cfg["settings"]["openpose"]["model"]
OPENPOSE_IMG_SIZE = cfg["settings"]["openpose"]["img_size"]
# Display settings
img_disp_desired_rows = int(cfg["settings"]["display"]["desired_rows"])
skeleton_detector = SkeletonDetector(OPENPOSE_MODEL, OPENPOSE_IMG_SIZE)
multiperson_tracker = Tracker()
multiperson_classifier = MultiPersonClassifier(SRC_MODEL_PATH, CLASSES)
os.makedirs(DST_FOLDER, exist_ok=True)
os.makedirs(DST_FOLDER + DST_SKELETON_FOLDER_NAME, exist_ok=True)
# video writer
video_writer = lib_images_io.VideoWriter(
DST_FOLDER + DST_VIDEO_NAME, DST_VIDEO_FPS)
video_writer2 = lib_images_io.VideoWriter(
DST_FOLDER + DST_VIDEO_NAME, DST_VIDEO_FPS)
sssr = []
def draw_result_img(img_disp, ith_img, humans, dict_id2skeleton, skeleton_detector, multiperson_classifier):
''' Draw skeletons, labels, and prediction scores onto image for display '''
global sssr
# Resize to a proper size for display
r, c = img_disp.shape[0:2]
desired_cols = int(1.0 * c * (img_disp_desired_rows / r))
img_disp = cv2.resize(img_disp,
dsize=(desired_cols, img_disp_desired_rows))
# Draw all people's skeleton
skeleton_detector.draw(img_disp, humans)
# Draw bounding box and label of each person
if len(dict_id2skeleton):
for id, label in dict_id2label.items():
skeleton = dict_id2skeleton[id]
# scale the y data back to original
skeleton[1::2] = skeleton[1::2] / scale_h
# print("Drawing skeleton: ", dict_id2skeleton[id], "with label:", label, ".")
lib_plot.draw_action_result(img_disp, id, skeleton, label)
# Add blank to the left for displaying prediction scores of each class
# img_disp = lib_plot.add_white_region_to_left_of_image(img_disp)
cv2.putText(img_disp, "Frame:" + str(ith_img),
(20, 20), fontScale=1.5, fontFace=cv2.FONT_HERSHEY_PLAIN,
color=(0, 0, 0), thickness=2)
# Draw predicting score for only 1 person
if len(dict_id2skeleton):
classifier_of_a_person = multiperson_classifier.get_classifier(
id='min')
# 好像是卡在了这里
sssr = classifier_of_a_person.draw_scores_onto_image(img_disp)
print('-------------------------------------------------')
print(sssr)
return img_disp
def get_the_skeleton_data_to_save_to_disk(dict_id2skeleton):
'''
In each image, for each skeleton, save the:
human_id, label, and the skeleton positions of length 18*2.
So the total length per row is 2+36=38
'''
skels_to_save = []
for human_id in dict_id2skeleton.keys():
label = dict_id2label[human_id]
skeleton = dict_id2skeleton[human_id]
skels_to_save.append([[human_id, label] + skeleton.tolist()])
return skels_to_save
class Main(Ui_MainWindow, QMainWindow):
def __init__(self):
super().__init__()
self.setupUi(self)
self.vedio = vedio()
self.timer_camera = QtCore.QTimer()
self.timer_video = QtCore.QTimer()
# 定时器函数
self.timer_camera.timeout.connect(self.show_camera)
self.timer_video.timeout.connect(self.show_video)
self.button()
self.label.setPixmap(QtGui.QPixmap('img.png').scaled(self.label.width(), self.label.height()))
self.pushButton.clicked.connect(self.playVedio)
def playVedio(self):
self.vedio.show()
self.vedio.slotStart()
def setscore(self):
try:
self.actionname1.setText(str(float(sssr[0]) * 100))
self.actionname2.setText(str(float(sssr[1]) * 100))
self.actionname3.setText(str(float(sssr[2]) * 100))
self.actionname4.setText(str(float(sssr[3]) * 100))
self.actionname5.setText(str(float(sssr[4]) * 100))
self.actionname6.setText(str(float(sssr[5]) * 100))
self.actionname7.setText(str(float(sssr[6]) * 100))
self.actionname8.setText(str(float(sssr[7]) * 100))
self.actionname9.setText(str(float(sssr[8]) * 100))
except:
pass
def camera_init(self):
# 打开设置摄像头对象
videoSourceIndex = 0
self.cap1 = cv2.VideoCapture(0, cv2.CAP_DSHOW + videoSourceIndex)
#self.cap = cv2.VideoCapture(0)
self.CAM_NUM = 0
# 显示摄像头界面
def camera(self):
if not self.timer_camera.isActive():
flag = self.cap.open(self.CAM_NUM)
if not flag:
msg = QtWidgets.QMessageBox.warning(self, u"Warning", u"请检测相机与电脑是否连接正确",
buttons=QtWidgets.QMessageBox.Ok,
defaultButton=QtWidgets.QMessageBox.Ok)
else:
self.timer_camera.start(50)
else:
self.timer_camera.stop()
self.cap.release()
def show_camera(self):
global dict_id2label, scale_h, ith_img
try:
# -- Read image
# 这里读视频帧进来
flag, img = self.cap.read()
ith_img += 1
img_disp = img.copy()
print(f"
Processing {ith_img}th image ...")
# -- Detect skeletons
humans = skeleton_detector.detect(img)
skeletons, scale_h = skeleton_detector.humans_to_skels_list(humans)
skeletons = remove_skeletons_with_few_joints(skeletons)
# -- Track people
dict_id2skeleton = multiperson_tracker.track(
skeletons)  # int id -> np.array() skeleton
# -- Recognize action of each person
if len(dict_id2skeleton):
dict_id2label = multiperson_classifier.classify(
dict_id2skeleton)
# -- Draw
# 这里得到处理完后的图像
img_disp = draw_result_img(img_disp, ith_img, humans, dict_id2skeleton,skeleton_detector, multiperson_classifier)
if len(dict_id2skeleton):
print(dict_id2skeleton.keys())
min_id = min(dict_id2skeleton.keys())
if dict_id2label[min_id] != 'LABEL_UNKNOWN':
english_to_chinese = {
'stand': '站姿推举',
'walk': '摆手',
'run': '平板支撑',
'jump': '高抬腿',
'sit': '扎马步',
'squat': '深蹲',
'kick': '俯身飞鸟',
'punch': '招财猫',
'wave': '侧平举'
}
label_index = list(english_to_chinese.keys()).index(dict_id2label[min_id])
print(label_index)
if dict_id2label[min_id] in english_to_chinese:
s = english_to_chinese[dict_id2label[min_id]]
# 指定文本文件的文件名
txt_filename = "data.txt"
# 获取当前时间
current_time = datetime.datetime.now()
# 格式化时间为字符串
formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
# 将每个变量转换为字符串
s_str = str(s)
sssr_str = str(sssr[label_index])
# 创建一个包含当前时间、动作类型和动作分数的字符串
data_point = f"{formatted_time}, {s_str}, {sssr_str}
"
with open(txt_filename, 'a+') as file:
# 检查文件是否为空
if file.tell() == 0:
file.write(data_point)
else:
# 将文件指针移到文件的开头
file.seek(0)
lines = file.readlines()
# 检查最后一行的前19个字符是否与新记录不同
print(lines[-1][:19])
if not lines or lines[-1][:19] != data_point[:19]:
# 将新记录追加到文件中
file.write(data_point)
print("当前动作 :", s)
print('动作分数：', sssr[label_index])
# print("prediced label is :", dict_id2label[min_id])
# -- Display image, and write to video.avi
# 这里把图像img_disp显示到界面，感觉是这的问题，因为img_disp一直都是有数据的
show = cv2.cvtColor(img_disp, cv2.COLOR_BGR2RGB)
showImage = QtGui.QImage(show.data, show.shape[1], show.shape[0],
QtGui.QImage.Format_RGB888)
self.label.setPixmap(QtGui.QPixmap.fromImage(showImage))
self.setscore()
video_writer.write(img_disp)
# -- Get skeleton data and save to file
skels_to_save = get_the_skeleton_data_to_save_to_disk(
dict_id2skeleton)
lib_commons.save_listlist(
DST_FOLDER + DST_SKELETON_FOLDER_NAME +
SKELETON_FILENAME_FORMAT.format(ith_img),
skels_to_save)
finally:
pass
# video_writer.stop()
# print("Program ends")
def show_video(self):
global dict_id2label, scale_h, ith_img
if images_loader.has_image():
ith_img += 1
try:
# -- Read image
img = images_loader.read_image()
img_disp = img.copy()
print(f"
Processing {ith_img}th image ...")
# -- Detect skeletons
humans = skeleton_detector.detect(img)
skeletons, scale_h = skeleton_detector.humans_to_skels_list(humans)
skeletons = remove_skeletons_with_few_joints(skeletons)
# -- Track people
dict_id2skeleton = multiperson_tracker.track(
skeletons)  # int id -> np.array() skeleton
# -- Recognize action of each person
if len(dict_id2skeleton):
dict_id2label = multiperson_classifier.classify(
dict_id2skeleton)
# -- Draw
img_disp = draw_result_img(img_disp, ith_img, humans, dict_id2skeleton,
skeleton_detector, multiperson_classifier)
# Print label of a person
if len(dict_id2skeleton):
print(dict_id2skeleton.keys())
min_id = min(dict_id2skeleton.keys())
if dict_id2label[min_id] != 'LABEL_UNKNOWN':
english_to_chinese = {
'stand': '站姿推举',
'walk': '摆手',
'run': '平板支撑',
'jump': '高抬腿',
'sit': '扎马步',
'squat': '深蹲',
'kick': '俯身飞鸟',
'punch': '招财猫',
'wave': '侧平举'
}
label_index = list(english_to_chinese.keys()).index(dict_id2label[min_id])
print(label_index)
if dict_id2label[min_id] in english_to_chinese:
s = english_to_chinese[dict_id2label[min_id]]
# 指定文本文件的文件名
txt_filename = "data.txt"
# 获取当前时间
current_time = datetime.datetime.now()
# 格式化时间为字符串
formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
# 将每个变量转换为字符串
s_str = str(s)
sssr_str = str(sssr[label_index])
# 创建一个包含当前时间、动作类型和动作分数的字符串
data_point = f"{formatted_time}, {s_str}, {sssr_str}
"
with open(txt_filename, 'a+') as file:
# 检查文件是否为空
if file.tell() == 0:
file.write(data_point)
else:
# 将文件指针移到文件的开头
file.seek(0)
lines = file.readlines()
# 检查最后一行的前19个字符是否与新记录不同
print(lines[-1][:19])
if not lines or lines[-1][:19] != data_point[:19]:
# 将新记录追加到文件中
file.write(data_point)
print("当前动作 :", s)
print('动作分数：', sssr[label_index])
# print("prediced label is :", dict_id2label[min_id])
# -- Display image, and write to video.avi
show = cv2.cvtColor(img_disp, cv2.COLOR_BGR2RGB)
showImage = QtGui.QImage(show.data, show.shape[1], show.shape[0],
QtGui.QImage.Format_RGB888)
self.label.setPixmap(QtGui.QPixmap.fromImage(showImage))
self.setscore()
video_writer2.write(img_disp)
# -- Get skeleton data and save to file
skels_to_save = get_the_skeleton_data_to_save_to_disk(
dict_id2skeleton)
lib_commons.save_listlist(
DST_FOLDER + DST_SKELETON_FOLDER_NAME +
SKELETON_FILENAME_FORMAT.format(ith_img),
skels_to_save)
finally:
pass
def button(self):
self.action_3.triggered.connect(self.videoMode)
self.action_4.triggered.connect(self.cameraMode)
self.action_2.triggered.connect(self.reset)
self.action_5.triggered.connect(self.photoMode)
self.action.triggered.connect(self.save)
self.actionexit.triggered.connect(QCoreApplication.instance().quit)
def save(self):
video_writer.stop()
video_writer2.stop()
print("Program ends")
pass
def videoMode(self):
global ith_img, images_loader, SRC_DATA_type
ith_img = -1
try:
self.timer_camera.stop()
self.cap.release()
except:
pass
# Output folder
SRC_DATA_type = "video"
SRC_DATA_path = QFileDialog.getOpenFileNames(self, '选择动作视频', '', '')[0]
try:
DST_FOLDER_name = get_dst_folder_name(SRC_DATA_type, str(SRC_DATA_path[0]))
DST_folder = output_folder + "/" + DST_FOLDER_name + "/"
DST_SKELETON_FOLDER_name = cfg["output"]["skeleton_folder_name"]
os.makedirs(DST_folder, exist_ok=True)
os.makedirs(DST_folder + DST_SKELETON_FOLDER_name, exist_ok=True)
images_loader = lib_images_io.ReadFromVideo(
SRC_DATA_path[0],
sample_interval=SRC_VIDEO_SAMPLE_INTERVAL)
self.timer_video.start(30)
except:
pass
def cameraMode(self):
global ith_img
ith_img = -1
try:
self.timer_video.stop()
except:
pass
self.camera_init()
self.camera()
def photoMode(self):
global ith_img, dict_id2label, scale_h
ith_img = 0
dict_id2label = {}
scale_h = 0
try:
self.timer_camera.stop()
self.cap.release()
except:
pass
try:
self.timer_video.stop()
except:
pass
try:
self.filename = QFileDialog.getOpenFileNames(self, "打开图片", "./",
"*.jpg;;*.png;;All Files(*)")[0][0]
print(self.filename)
img = cv2.imread(self.filename)
img_disp = img.copy()
# -- Detect skeletons
humans = skeleton_detector.detect(img)
skeletons, scale_h = skeleton_detector.humans_to_skels_list(humans)
skeletons = remove_skeletons_with_few_joints(skeletons)
# -- Track people
dict_id2skeleton = multiperson_tracker.track(
skeletons)  # int id -> np.array() skeleton
# -- Recognize action of each person
if len(dict_id2skeleton):
dict_id2label = multiperson_classifier.classify(
dict_id2skeleton)
# -- Draw
img_disp = draw_result_img(img_disp, ith_img, humans, dict_id2skeleton,
skeleton_detector, multiperson_classifier)
# Print label of a person
if len(dict_id2skeleton):
print(dict_id2skeleton.keys())
min_id = min(dict_id2skeleton.keys())
if dict_id2label[min_id] != 'LABEL_UNKNOWN':
english_to_chinese = {
'stand': '站姿推举',
'walk': '摆手',
'run': '平板支撑',
'jump': '高抬腿',
'sit': '扎马步',
'squat': '深蹲',
'kick': '俯身飞鸟',
'punch': '招财猫',
'wave': '侧平举'
}
label_index = list(english_to_chinese.keys()).index(dict_id2label[min_id])
print(label_index)
if dict_id2label[min_id] in english_to_chinese:
s = english_to_chinese[dict_id2label[min_id]]
# 指定文本文件的文件名
txt_filename = "data.txt"
# 获取当前时间
current_time = datetime.datetime.now()
# 格式化时间为字符串
formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
# 将每个变量转换为字符串
s_str = str(s)
sssr_str = str(sssr[label_index])
# 创建一个包含当前时间、动作类型和动作分数的字符串
data_point = f"{formatted_time}, {s_str}, {sssr_str}
"
with open(txt_filename, 'a+') as file:
# 检查文件是否为空
if file.tell() == 0:
file.write(data_point)
else:
# 将文件指针移到文件的开头
file.seek(0)
lines = file.readlines()
# 检查最后一行的前19个字符是否与新记录不同
print(lines[-1][:19])
if not lines or lines[-1][:19] != data_point[:19]:
# 将新记录追加到文件中
file.write(data_point)
print("当前动作 :", s)
print('动作分数：', sssr[label_index])
# print("prediced label is :", dict_id2label[min_id])
cv2.imwrite('photoMode.png', img_disp)
photo = QtGui.QPixmap('photoMode.png').scaled(self.label.width(), self.label.height())
self.label.setPixmap(photo)
self.setscore()
except:
pass
def reset(self):
global ith_img, dict_id2label, scale_h
ith_img = 0
dict_id2label = {}
scale_h = 0
try:
self.timer_camera.stop()
self.cap.release()
except:
pass
try:
self.timer_video.stop()
except:
pass
self.label.setPixmap(QtGui.QPixmap('img.png'))
if __name__ == "__main__":
app = QApplication(sys.argv)
win = Main()
win.show()
sys.exit(app.exec_())