PaddlePaddle——多进程实现基于多摄像头的视觉巡线、目标检测、运动控制等任务并且实现数据共享

文章目录[隐藏]

多进程主程序
运行效果

多进程主程序

进程说明

本程序有三个子进程，分别为：

1、巡线进程： 摄像头1的图像处理巡线进程
2、目标检测进程： 摄像头0的目标检测进程
3、决策控制进程： 基于前两个感知进程结果的决策控制进程

核心代码

import multiprocessing as mp
import numpy as np
import cv2
import time 
import math
import argparse

from StereoCamera_Lib import Stereo_Camera
from color_dist import *
from Unitree_Lib import Unitree_Robot

# process 0 摄像头1的图像处理巡线进程
def path_follow(quene):
    # 需要使用进程内部变量的子函数
    def followPath(color_image, color):
        '''
        图像处理巡线程序（省略）
		'''
    cap = Stereo_Camera(camera_id=1)
    cap.camera_init()
    last_target_point = cap.camera_width/2
    while True:
        start = time.time()
        image = cap.rgb_image(1)      
        res, target_point, FollowState, TurnState, SearchState = followPath(image, color="chin_yellow_blind_path")
        if target_point == None:
            target_point = last_target_point
        share_flag_list = [0, [target_point, FollowState, TurnState, SearchState]]
        print("Proess0 FPS:", 1/(time.time()-start), "/s")
        last_target_point = target_point
        quene.put([0, share_flag_list]) # 将消息放入进程队列中

# process 1 摄像头0的目标检测进程
def paddle_predict(quene):
    # paddle inference运行节点
    # 所有和paddlepaddle有关的库必须在进程中import，否则会导致GPU初始化错误，参考：https://blog.csdn.net/qq_45779334/article/details/122024343
    from paddle.inference import PrecisionType
    from PaddleDetection_Inference_Lib import Paddle_inference
    # --------------------------配置区域--------------------------
    infer_img_size = 320        # 自定义模型预测的输入图像尺寸
    use_model_img_size = True   #　是否使用模型默认输入图像尺寸，默认为True
    use_gpu = True              # 是否使用GPU
    gpu_memory = 200            # GPU的显存
    use_tensorrt = False        # 是否使用TensorRT
    precision_mode = PrecisionType.Float32
    # 模型参数
    model_folder_dir = "../model/ppyolo_mbv3_large_coco_270epoch_3class"
    # -----------------------------------------------------------    
    cap = Stereo_Camera(camera_id=0)
    cap.camera_init()
    paddle_infer = Paddle_inference(model_folder_dir, use_model_img_size, 
                                    infer_img_size, use_gpu, gpu_memory, 
                                    use_tensorrt, precision_mode)
    paddle_infer.init(np.zeros(shape=(int(cap.camera_height), int(cap.camera_width),3),dtype="float32"))
    while True:
        start = time.time()          
        image = cap.rgb_image(0)
        result = paddle_infer.infer(image)
        share_flag_list = [1, [result[0][0][0]]]
        image = paddle_infer.draw_bbox_image(image, result[0])
        print("Proess1 FPS:", 1/(time.time()-start), "/s")
        quene.put([1, share_flag_list]) # 将消息放入进程队列中

# process 2 基于前两个感知进程结果的决策控制进程
def robot_control(quene):
    unitree_robot = Unitree_Robot()
    while True:
        start = time.time()
        try:
            process_id, share_flag_list = quene.get() # 获取消息队列
            unitree_robot.robot_control(share_flag_list)
            print(share_flag_list)
        except:
            print("no recive")
        print("Proess2 FPS:", 1/(time.time()-start), "/s")

def main():
    # 初始化进程
    quene = mp.Queue()
    process_list = [path_follow, paddle_predict, robot_control] # 巡线, 预测, 决策
    for process_id, process_function in enumerate(process_list):
        process = mp.Process(target=process_function,args=(quene,)) 
        process_list[process_id] = process
    for process in process_list: # 一次性开始全部进程
        process.start()
    print("process start")

if __name__ == '__main__':    
    main()