#-*-coding:utf-8-*- ''' DpCas-Light |||| ||||| |||| || ||||||| || || || || || || |||| || || || || || || || || || || || || || || || || ||====|| |||||| || || ||||| || || ||======|| || || || || || || || || || || |||| || |||| || || ||||||| /--------------------- HandPose_X ---------------------/ ''' # date:2021-03-12 # Author: Eric.Lee # function: handpose demo import os import cv2 import time from multiprocessing import Process from multiprocessing import Manager import cv2 import numpy as np import random import time # 加载模型组件库 from hand_detect.yolo_v3_hand import yolo_v3_hand_model from hand_keypoints.handpose_x import handpose_x_model from classify_imagenet.imagenet_c import classify_imagenet_model # 加载工具库 import sys sys.path.append("./lib/hand_lib/") from cores.handpose_fuction import handpose_track_keypoints21_pipeline from cores.handpose_fuction import hand_tracking,audio_recognize,judge_click_stabel,draw_click_lines from utils.utils import parse_data_cfg from playsound import playsound def audio_process_dw_edge_cnt(info_dict): while (info_dict["handpose_procss_ready"] == False): # 等待 模型加载 time.sleep(2) gesture_names = ["click"] gesture_dict = {} for k_ in gesture_names: gesture_dict[k_] = None # time.sleep(1) # playsound("./materials/audio/sentences/WelcomeAR.mp3") # time.sleep(0.01) # playsound("./materials/audio/sentences/MorningEric.mp3") # time.sleep(1) reg_cnt = 0 while True: time.sleep(0.01) try: reg_cnt = info_dict["click_dw_cnt"] for i in range(reg_cnt): # playsound("./materials/audio/cue/winwin-1.mp3") playsound("./materials/audio/sentences/welldone.mp3") info_dict["click_dw_cnt"] = info_dict["click_dw_cnt"] - reg_cnt except Exception as inst: print(type(inst),inst) # exception instance if info_dict["break"] == True: break def audio_process_up_edge_cnt(info_dict): while (info_dict["handpose_procss_ready"] == False): # 等待 模型加载 time.sleep(2) gesture_names = ["click"] gesture_dict = {} for k_ in gesture_names: gesture_dict[k_] = None reg_cnt = 0 while True: time.sleep(0.01) # print(" --->>> audio_process") try: reg_cnt = info_dict["click_up_cnt"] for i in range(reg_cnt): # playsound("./materials/audio/cue/m2-0.mp3") playsound("./materials/audio/sentences/Click.mp3") info_dict["click_up_cnt"] = info_dict["click_up_cnt"] - reg_cnt except Exception as inst: print(type(inst),inst) # the exception instance if info_dict["break"] == True: break def audio_process_dw_edge(info_dict): while (info_dict["handpose_procss_ready"] == False): # 等待 模型加载 time.sleep(2) gesture_names = ["click"] gesture_dict = {} for k_ in gesture_names: gesture_dict[k_] = None while True: time.sleep(0.01) # print(" --->>> audio_process") try: for g_ in gesture_names: if gesture_dict[g_] is None: gesture_dict[g_] = info_dict[g_] else: if ("click"==g_): if (info_dict[g_]^gesture_dict[g_]) and info_dict[g_]==False:# 判断Click手势信号为下降沿，Click动作结束 playsound("./materials/audio/cue/winwin.mp3") # playsound("./materials/audio/sentences/welldone.mp3") gesture_dict[g_] = info_dict[g_] except Exception as inst: print(type(inst),inst) # the exception instance if info_dict["break"] == True: break def audio_process_up_edge(info_dict): while (info_dict["handpose_procss_ready"] == False): # 等待 模型加载 time.sleep(2) gesture_names = ["click"] gesture_dict = {} for k_ in gesture_names: gesture_dict[k_] = None while True: time.sleep(0.01) # print(" --->>> audio_process") try: for g_ in gesture_names: if gesture_dict[g_] is None: gesture_dict[g_] = info_dict[g_] else: if ("click"==g_): if (info_dict[g_]^gesture_dict[g_]) and info_dict[g_]==True:# 判断Click手势信号为上升沿，Click动作开始 playsound("./materials/audio/cue/m2.mp3") # playsound("./materials/audio/sentences/clik_quick.mp3") gesture_dict[g_] = info_dict[g_] except Exception as inst: print(type(inst),inst) # the exception instance if info_dict["break"] == True: break ''' 启动识别语音进程 ''' def audio_process_recognize_up_edge(info_dict): while (info_dict["handpose_procss_ready"] == False): # 等待 模型加载 time.sleep(2) gesture_names = ["double_en_pts"] gesture_dict = {} for k_ in gesture_names: gesture_dict[k_] = None while True: time.sleep(0.01) # print(" --->>> audio_process") try: for g_ in gesture_names: if gesture_dict[g_] is None: gesture_dict[g_] = info_dict[g_] else: if ("double_en_pts"==g_): if (info_dict[g_]^gesture_dict[g_]) and info_dict[g_]==True:# 判断Click手势信号为上升沿，Click动作开始 playsound("./materials/audio/sentences/IdentifyingObjectsWait.mp3") playsound("./materials/audio/sentences/ObjectMayBeIdentified.mp3") if info_dict["reco_msg"] is not None: print("process - (audio_process_recognize_up_edge) reco_msg : {} ".format(info_dict["reco_msg"])) doc_name = info_dict["reco_msg"]["label_msg"]["doc_name"] reco_audio_file = "./materials/audio/imagenet_2012/{}.mp3".format(doc_name) if os.access(reco_audio_file,os.F_OK):# 判断语音文件是否存在 playsound(reco_audio_file) info_dict["reco_msg"] = None gesture_dict[g_] = info_dict[g_] except Exception as inst: print(type(inst),inst) # exception instance if info_dict["break"] == True: break ''' /*****************************************/ 算法 pipeline /*****************************************/ ''' def handpose_x_process(info_dict,config): # 模型初始化 print("load model component ...") # yolo v3 手部检测模型初始化 hand_detect_model = yolo_v3_hand_model(conf_thres=float(config["detect_conf_thres"]),nms_thres=float(config["detect_nms_thres"]), model_arch = config["detect_model_arch"],model_path = config["detect_model_path"],yolo_anchor_scale = float(config["yolo_anchor_scale"]), img_size = float(config["detect_input_size"]), ) # handpose_x 21 关键点回归模型初始化 handpose_model = handpose_x_model(model_arch = config["handpose_x_model_arch"],model_path = config["handpose_x_model_path"]) # gesture_model = None # 目前缺省 # object_recognize_model = classify_imagenet_model(model_arch = config["classify_model_arch"],model_path = config["classify_model_path"]) # 识别分类模型 # img_reco_crop = None cap = cv2.VideoCapture(int(config["camera_id"])) # 开启摄像机 cap.set(cv2.CAP_PROP_EXPOSURE, -8) # 设置相机曝光，（注意：不是所有相机有效） # url="http://admin:admin@192.168.43.1:8081" # cap=cv2.VideoCapture(url) print("start handpose process ~") info_dict["handpose_procss_ready"] = True #多进程间的开始同步信号 gesture_lines_dict = {} # 点击使能时的轨迹点 hands_dict = {} # 手的信息 hands_click_dict = {} #手的按键信息计数 track_index = 0 # 跟踪的全局索引 while True: ret, img = cap.read()# 读取相机图像 if ret:# 读取相机图像成功 # img = cv2.flip(img,-1) algo_img = img.copy() st_ = time.time() #------ hand_bbox =hand_detect_model.predict(img,vis = True) # 检测手，获取手的边界框 hands_dict,track_index = hand_tracking(data = hand_bbox,hands_dict = hands_dict,track_index = track_index) # 手跟踪，目前通过IOU方式进行目标跟踪 # 检测每个手的关键点及相关信息 handpose_list = handpose_track_keypoints21_pipeline(img,hands_dict = hands_dict,hands_click_dict = hands_click_dict,track_index = track_index,algo_img = algo_img, handpose_model = handpose_model,gesture_model = gesture_model, icon = None,vis = True) et_ = time.time() fps_ = 1./(et_-st_+1e-8) #------------------------------------------ 跟踪手的 信息维护 #------------------ 获取跟踪到的手ID id_list = [] for i in range(len(handpose_list)): _,_,_,dict_ = handpose_list[i] id_list.append(dict_["id"]) # print(id_list) #----------------- 获取需要删除的手ID id_del_list = [] for k_ in gesture_lines_dict.keys(): if k_ not in id_list:#去除过往已经跟踪失败的目标手的相关轨迹 id_del_list.append(k_) #----------------- 删除无法跟踪到的手的相关信息 for k_ in id_del_list: del gesture_lines_dict[k_] del hands_click_dict[k_] #----------------- 更新检测到手的轨迹信息,及手点击使能时的上升沿和下降沿信号 double_en_pts = [] for i in range(len(handpose_list)): _,_,_,dict_ = handpose_list[i] id_ = dict_["id"] if dict_["click"]: if id_ not in gesture_lines_dict.keys(): gesture_lines_dict[id_] = {} gesture_lines_dict[id_]["pts"]=[] gesture_lines_dict[id_]["line_color"] = (random.randint(100,255),random.randint(100,255),random.randint(100,255)) gesture_lines_dict[id_]["click"] = None #判断是否上升沿 if gesture_lines_dict[id_]["click"] is not None: if gesture_lines_dict[id_]["click"] == False:#上升沿计数器 info_dict["click_up_cnt"] += 1 #获得点击状态 gesture_lines_dict[id_]["click"] = True #---获得坐标 gesture_lines_dict[id_]["pts"].append(dict_["choose_pt"]) double_en_pts.append(dict_["choose_pt"]) else: if id_ not in gesture_lines_dict.keys(): gesture_lines_dict[id_] = {} gesture_lines_dict[id_]["pts"]=[] gesture_lines_dict[id_]["line_color"] = (random.randint(100,255),random.randint(100,255),random.randint(100,255)) gesture_lines_dict[id_]["click"] = None elif id_ in gesture_lines_dict.keys(): gesture_lines_dict[id_]["pts"]=[]# 清除轨迹 #判断是否上升沿 if gesture_lines_dict[id_]["click"] == True:#下降沿计数器 info_dict["click_dw_cnt"] += 1 # 更新点击状态 gesture_lines_dict[id_]["click"] = False #绘制手click 状态时的大拇指和食指中心坐标点轨迹 draw_click_lines(img,gesture_lines_dict,vis = bool(config["vis_gesture_lines"])) # 判断各手的click状态是否稳定，且满足设定阈值 flag_click_stable = judge_click_stabel(img,handpose_list,int(config["charge_cycle_step"])) # 判断是否启动识别语音,且进行选中目标识别 img_reco_crop,reco_msg = audio_recognize(img,algo_img,img_reco_crop,object_recognize_model,info_dict,double_en_pts,flag_click_stable) # print(reco_msg) cv2.putText(img, 'HandNum:[{}]'.format(len(hand_bbox)), (5,25),cv2.FONT_HERSHEY_COMPLEX, 0.7, (255, 0, 0),5) cv2.putText(img, 'HandNum:[{}]'.format(len(hand_bbox)), (5,25),cv2.FONT_HERSHEY_COMPLEX, 0.7, (0, 0, 255)) cv2.namedWindow("image",0) cv2.imshow("image",img) if cv2.waitKey(1) == 27: info_dict["break"] = True break else: break cap.release() cv2.destroyAllWindows() def main_handpose_x(cfg_file): config = parse_data_cfg(cfg_file) print("\n/---------------------- main_handpose_x config ------------------------/\n") for k_ in config.keys(): print("{} : {}".format(k_,config[k_])) print("\n/------------------------------------------------------------------------/\n") print(" loading handpose_x local demo ...") g_info_dict = Manager().dict()# 多进程共享字典初始化：用于多进程间的 key：value 操作 g_info_dict["handpose_procss_ready"] = False # 进程间的开启同步信号 g_info_dict["break"] = False # 进程间的退出同步信号 g_info_dict["double_en_pts"] = False # 双手选中动作使能信号 g_info_dict["click_up_cnt"] = 0 g_info_dict["click_dw_cnt"] = 0 g_info_dict["reco_msg"] = None print(" multiprocessing dict key:\n") for key_ in g_info_dict.keys(): print( " -> ",key_) print() #-------------------------------------------------- 初始化各进程 process_list = [] t = Process(target=handpose_x_process,args=(g_info_dict,config,)) process_list.append(t) t = Process(target=audio_process_recognize_up_edge,args=(g_info_dict,)) # 上升沿播放 process_list.append(t) # t = Process(target=audio_process_dw_edge_cnt,args=(g_info_dict,)) # 下降沿播放 # process_list.append(t) # t = Process(target=audio_process_up_edge_cnt,args=(g_info_dict,)) # 上升沿播放 # process_list.append(t) for i in range(len(process_list)): process_list[i].start() for i in range(len(process_list)): process_list[i].join()# 设置主线程等待子线程结束 del process_list