目标行为检测模块

之后是我们的目标检测模块。那么这里的话，我们主要看到这个模块：

这个模块就是一个实现模块。

同样的这里有两个方法。一个是原先的非实时在线检测，还有一个就是我们自己实现的在线检测。

非在线实时检测

这个的话就是一开始给到的方法，这个方法适合对已经录下来了的视频进行操作，对这个视频进行处理，最后得到一个结果。

这里我进行过测试，我的设备是：

戴尔游匣G5
GTX 1650 4GB

对于一个14秒的视频，处理完毕之后大概大概3,40秒左右。
所以这个几乎不能做在线的检测，也就是打开摄像头，你需要实时监控摄像头的画面的那种情况。

    def detect_form_video(self,CameraName,show=False):"""这里的代码执行流程是这样子的:1. 读取到视频当中所有的图像2. 先通过yolo算法,获取到多个目标3. 通过deep_sort对这些目标进行跟踪4. 通过slowfast识别出对应的目标动作5. 调用visualize_yolopreds处理出识别出来的结果,同时绘制图像6. 将处理完毕之后的视频,再次进行合并为一个视频,并输出到指定文件中这里有个毛病就是,需要在处理完一个batch之后,你才能看到这个视频,并且每次都有卡顿这里实现的方式有延迟.:param CameraName::return:"""# self.model = torch.hub.load('path/to/yolov5', 'custom', path='path/to/best.pt', source='local')if DECTION_CONFIG.yolo_classes:self.model.classes = DECTION_CONFIG.yolo_classes# 读取视频video_path = DECTION_CONFIG.input_date# pytorch提供的动作识别器video = pytorchvideo.data.encoded_video.EncodedVideo.from_path(video_path)img_path= DECTION_CONFIG.streamTempBaseChannel+"/"+CameraName+"/01"self.clean_folder(img_path)os.makedirs(img_path,exist_ok=True)print("extracting video...")# 对视频进行切分为图片self.extract_video(video_path,img_path)imgnames=natsort.natsorted(os.listdir(img_path))save_path=DECTION_CONFIG.streamTempBaseChannel+"/"+CameraName+"/02"self.clean_folder(save_path)os.makedirs(save_path,exist_ok=True)process_batch_size = DECTION_CONFIG.solowfast_process_batch_size    # 10 ~ 30 should be fine, the bigger, the fastervideo_clip_length = DECTION_CONFIG.solowfast_video_clip_length   # set 0.8 or 1 or 1.2frames_per_second = DECTION_CONFIG.solowfast_frames_per_second     # usually set 25 or 30print("processing...")a=time.time()for i in range(0,len(imgnames),process_batch_size):imgs=[os.path.join(img_path,name) for name in imgnames[i:i+process_batch_size]]yolo_preds=self.model(imgs, size=self.imsize)mid=(i+process_batch_size/2)/frames_per_secondvideo_clips=video.get_clip(mid - video_clip_length/2, mid + video_clip_length/2 - 0.04)video_clips=video_clips['video']if video_clips is None:continueprint(i/frames_per_second,video_clips.shape,len(imgs))deepsort_outputs=[]for i in range(len(yolo_preds.pred)):temp=self.deepsort_update(self.deepsort_tracker,yolo_preds.pred[i].cpu(),yolo_preds.xywh[i][:,0:4].cpu(),yolo_preds.ims[i])if len(temp)==0:temp=np.ones((0,8))deepsort_outputs.append(temp.astype(np.float32))yolo_preds.pred=deepsort_outputsid_to_ava_labels={}if yolo_preds.pred[len(imgs)//2].shape[0]:inputs,inp_boxes,_= self.ava_inference_transform(video_clips,yolo_preds.pred[len(imgs)//2][:,0:4],crop_size=self.imsize)inp_boxes = torch.cat([torch.zeros(inp_boxes.shape[0],1), inp_boxes], dim=1)if isinstance(inputs, list):inputs = [inp.unsqueeze(0).to(self.device) for inp in inputs]else:inputs = inputs.unsqueeze(0).to(self.device)with torch.no_grad():slowfaster_preds = self.video_model(inputs, inp_boxes.to(self.device))slowfaster_preds = slowfaster_preds.cpu()for tid,avalabel in zip(yolo_preds.pred[len(imgs)//2][:,5].tolist(),np.argmax(slowfaster_preds,axis=1).tolist()):id_to_ava_labels[tid]=self.ava_labelnames[avalabel+1]self.visualize_yolopreds(yolo_preds,id_to_ava_labels,self.coco_color_map,save_path,show)print("total cost: {:.3f}s, video clips length: {}s".format(time.time()-a,len(imgnames)/frames_per_second))vide_save_path = DECTION_CONFIG.output+CameraNameif(not os.path.exists(vide_save_path)):os.makedirs(vide_save_path)vide_save_path = vide_save_path+"/"+CameraName+".mp4"img_list=natsort.natsorted(os.listdir(save_path))im=cv2.imread(os.path.join(save_path,img_list[0]))height, width = im.shape[0], im.shape[1]video = cv2.VideoWriter(vide_save_path,cv2.VideoWriter_fourcc(*'mp4v'), 25, (width,height))for im_name in img_list:img = cv2.imread(os.path.join(save_path,im_name))video.write(img)video.release()self.clean_folder(img_path)self.clean_folder(save_path)print('saved video to:', vide_save_path)

在线实时检测

之后就是这个在线的检测。这个在线的意思不是说，处理速度非常快，这个毕竟硬件就摆在那里不可能有加速的。所以的话，我们唯一可以做的就是不遗漏，那么要想实现这个就只能做异步处理了。那么我在这里做了一个简单的消费者生产者模型，来实现这个功能。

这里也就是两个队列。
这样的话就可以实现这个在线实时的功能。

 def readTimeCamera(self,video_path,CameraName):"""读取摄像头,并且按照帧数读取,并且把这读取的图像先合成视频,因为这边会用到这个pytorchvideo优化:param CameraName::return:"""def readCamera(video_path,CameraName):camera = cv2.VideoCapture(video_path)wait_time = int((1 / DECTION_CONFIG.realTimeFps) * 1000)count = 0imgs = []split_count = 0try:while camera.isOpened() and self.going:success,img = camera.read()imgs.append(img)count+=1if(count==DECTION_CONFIG.realTimeFps*DECTION_CONFIG.tempStream):# 此时读取了长度为tempStream秒的视频vide_save_path = DECTION_CONFIG.output + CameraNameif (not os.path.exists(vide_save_path)):os.makedirs(vide_save_path)split_count+=1# 这里达到队列长度之后进行重复覆盖if(split_count==DECTION_CONFIG.max_queue_size):split_count = 1vide_save_path = vide_save_path + "/" + CameraName +"-"+str(split_count)+ ".mp4"height = len(img)width = len(img[0])video = cv2.VideoWriter(vide_save_path, cv2.VideoWriter_fourcc(*'mp4v'),DECTION_CONFIG.realTimeFps, (width, height))for img in imgs:video.write(img)video.release()imgs = []count=0self.exitTempStreamVideo = True# 如果没有消耗掉，这里会进行一个等待，阻塞self.read_camera_videos.put(vide_save_path)key = cv2.waitKey(wait_time)if key == ord('q'):self.going = Falsereturnexcept Exception as e:self.going = Falseprint(e)returnfinally:camera.release()print("摄像头终止并释放")return"""开启多线程进行调用"""t = threading.Thread(target=readCamera,args=(video_path,CameraName,))t.start()# readCamera(video_path,CameraName)def readTimeProcessing(self,CameraName):"""在这里完成算法处理:return:"""def processing(CameraName):while(1):if(not self.going and self.read_camera_videos.empty()):print("---计算完毕---")self.finish_process = Truereturn# 加载识别类型if DECTION_CONFIG.yolo_classes:self.model.classes = DECTION_CONFIG.yolo_classes# 加载路径视频if(self.read_camera_videos.empty() and not self.exitTempStreamVideo):continuevide_path = self.read_camera_videos.get()# pytorch提供的动作识别器,就是因为这个所以我们还需要存储一下视频文件# 虽然会消耗IO的时间,但是这玩意有对视频的优化video = pytorchvideo.data.encoded_video.EncodedVideo.from_path(vide_path)# 这里还是和先前一样img_path = DECTION_CONFIG.streamTempBaseChannel + "/" + CameraName + "/01"self.clean_folder(img_path)os.makedirs(img_path, exist_ok=True)# 对视频进行切分为图片self.extract_video(vide_path, img_path)imgnames = natsort.natsorted(os.listdir(img_path))process_batch_size = DECTION_CONFIG.realTimeFps  # 10 ~ 30 should be fine, the bigger, the fastervideo_clip_length = DECTION_CONFIG.solowfast_video_clip_length  # set 0.8 or 1 or 1.2frames_per_second = DECTION_CONFIG.solowfast_frames_per_second  # usually set 25 or 30for i in range(0, len(imgnames), process_batch_size):imgs = [os.path.join(img_path, name) for name in imgnames[i:i + process_batch_size]]yolo_preds = self.model(imgs, size=self.imsize)mid = (i + process_batch_size / 2) / frames_per_secondvideo_clips = video.get_clip(mid - video_clip_length / 2, mid + video_clip_length / 2 - 0.04)video_clips = video_clips['video']if video_clips is None:continue# print("*"*100)# print(i / frames_per_second, video_clips.shape, len(imgs))deepsort_outputs = []for i in range(len(yolo_preds.pred)):temp = self.deepsort_update(self.deepsort_tracker, yolo_preds.pred[i].cpu(),yolo_preds.xywh[i][:, 0:4].cpu(), yolo_preds.ims[i])if len(temp) == 0:temp = np.ones((0, 8))deepsort_outputs.append(temp.astype(np.float32))yolo_preds.pred = deepsort_outputsid_to_ava_labels = {}if yolo_preds.pred[len(imgs) // 2].shape[0]:inputs, inp_boxes, _ = self.ava_inference_transform(video_clips,yolo_preds.pred[len(imgs) // 2][:, 0:4],crop_size=self.imsize)inp_boxes = torch.cat([torch.zeros(inp_boxes.shape[0], 1), inp_boxes], dim=1)if isinstance(inputs, list):inputs = [inp.unsqueeze(0).to(self.device) for inp in inputs]else:inputs = inputs.unsqueeze(0).to(self.device)with torch.no_grad():slowfaster_preds = self.video_model(inputs, inp_boxes.to(self.device))slowfaster_preds = slowfaster_preds.cpu()for tid, avalabel in zip(yolo_preds.pred[len(imgs) // 2][:, 5].tolist(),np.argmax(slowfaster_preds, axis=1).tolist()):id_to_ava_labels[tid] = self.ava_labelnames[avalabel + 1]"""在这里,我们把这个结果放在队列当中"""print("-" * 100)print("已完成当前视频计算，剩余：",self.read_camera_videos.qsize())print("-" * 100)self.read_process_data.put((yolo_preds, id_to_ava_labels))self.exitTempStreamVideo = False"""开启多线程进行调用"""t = threading.Thread(target=processing, args=(CameraName,))t.start()def readTime_visualize_yolopreds(self,CameraName,show=False,process=None):"""显示yolo预测的结果，绘制图像,咱们在这里进行逻辑处理在这里如果需要进行逻辑处理的话,在这里传入一个process方法这个方法需要接收三个参数,第一个参数是,当前的动作,另一个是当前对象的bounding box 已经image对象:param yolo_preds::param id_to_ava_labels::param color_map::param save_folder::return:"""def visulize_readTime(CameraName,show=False,process=None):while(1):if(self.finish_process and self.read_process_data.empty()):print("---计算处理完毕---")breakdata = self.read_process_data.get()print("*"*100)print("正在处理实时结果，剩余：",self.read_process_data.qsize())print("*" * 100)yolo_preds = data[0]id_to_ava_labels = data[1]for i, (im, pred) in enumerate(zip(yolo_preds.ims, yolo_preds.pred)):im=cv2.cvtColor(im,cv2.COLOR_BGR2RGB)if pred.shape[0]:for j, (*box, cls, trackid, vx, vy) in enumerate(pred):if int(cls) != 0:ava_label = ''elif trackid in id_to_ava_labels.keys():ava_label = id_to_ava_labels[trackid].split(' ')[0]"""在这里得到动作信息和对应的bounding box并且处理一些逻辑"""if(process!=None):process(im,ava_label,box)else:ava_label = 'Unknow'text = '{} {} {}'.format(int(trackid),yolo_preds.names[int(cls)],ava_label)color = self.coco_color_map[int(cls)]im = self.plot_one_box(box,im,color,text)if(show):cv2.namedWindow('camera', 0)cv2.imshow('camera',im)cv2.waitKey(1)folder_name_temp = DECTION_CONFIG.output + CameraNameif (os.path.exists(folder_name_temp)):shutil.rmtree(folder_name_temp)print("!"*100)print("视频处理完毕！")print("!" * 100)t = threading.Thread(target=visulize_readTime, args=(CameraName,show,process,))t.start()def detect_from_video_realTime(self,camera,CameraName,show=False,process=None):"""这里需要传入一个camera对象,从而实现实时读取:param camera::param CameraName::param show::param save_path::return:"""# 实时读取摄像头self.readTimeCamera(camera,CameraName)# 进行处理运算self.readTimeProcessing(CameraName)# 对预测结果进行处理self.readTime_visualize_yolopreds(CameraName,show,process)

结合人脸识别

之后的话，我们就可以结合人脸识别了。
其实仔细看一下这个上面的这个代码：

  def readTime_visualize_yolopreds(self,CameraName,show=False,process=None):"""显示yolo预测的结果，绘制图像,咱们在这里进行逻辑处理在这里如果需要进行逻辑处理的话,在这里传入一个process方法这个方法需要接收三个参数,第一个参数是,当前的动作,另一个是当前对象的bounding box 已经image对象:param yolo_preds::param id_to_ava_labels::param color_map::param save_folder::return:"""def visulize_readTime(CameraName,show=False,process=None):while(1):if(self.finish_process and self.read_process_data.empty()):print("---计算处理完毕---")breakdata = self.read_process_data.get()print("*"*100)print("正在处理实时结果，剩余：",self.read_process_data.qsize())print("*" * 100)yolo_preds = data[0]id_to_ava_labels = data[1]for i, (im, pred) in enumerate(zip(yolo_preds.ims, yolo_preds.pred)):im=cv2.cvtColor(im,cv2.COLOR_BGR2RGB)if pred.shape[0]:for j, (*box, cls, trackid, vx, vy) in enumerate(pred):if int(cls) != 0:ava_label = ''elif trackid in id_to_ava_labels.keys():ava_label = id_to_ava_labels[trackid].split(' ')[0]"""在这里得到动作信息和对应的bounding box并且处理一些逻辑"""if(process!=None):process(im,ava_label,box)else:ava_label = 'Unknow'text = '{} {} {}'.format(int(trackid),yolo_preds.names[int(cls)],ava_label)color = self.coco_color_map[int(cls)]im = self.plot_one_box(box,im,color,text)if(show):cv2.namedWindow('camera', 0)cv2.imshow('camera',im)cv2.waitKey(1)folder_name_temp = DECTION_CONFIG.output + CameraNameif (os.path.exists(folder_name_temp)):shutil.rmtree(folder_name_temp)print("!"*100)print("视频处理完毕！")print("!" * 100)

这里有一个process，其实就是你写你的处理函数的地方。

def detect_pose():pose = PoseRec()source_path = 0# pose.detect_form_video("Test01",True)pose.detect_from_video_realTime(source_path,"Test02",True,process=None)

你直接调用我们人脸识别模块的识别代码，然后放在process里面就好了。

总结

以上就是咱们全部的内容，其实给到这里就可以做很多事情了。