如何在Python中独立启停multiprocessing进程以分别控制ODrive电机?
问题解决方案
一、独立启停监控单个进程的实现方法
别用multiprocessing.Pool,它是批量进程管理工具,不支持单独控制。直接用multiprocessing.Process创建单个进程,搭配multiprocessing.Event做停止信号,把进程和信号存在字典里,用驱动器唯一标识(比如USB端口号、设备序列号)作为键,就能实现精准控制。
核心实现代码示例
import multiprocessing import odrive from odrive.enums import * def drive_control_loop(device_serial, stop_signal): # 连接指定序列号的驱动器 odrv = odrive.find_any(serial_number=device_serial) # 电机初始化逻辑(按需调整) odrv.axis0.requested_state = AXIS_STATE_CLOSED_LOOP_CONTROL # 运行循环,直到收到停止信号 while not stop_signal.is_set(): # 这里写电机实时控制逻辑,比如读取编码器数据、调整扭矩等 pass # 停止电机并清理连接 odrv.axis0.requested_state = AXIS_STATE_IDLE odrv = None # 存储所有驱动器进程和停止信号的管理器 drive_processes = {} # 启动单台驱动器进程 def start_drive(drive_id, serial_num): stop_event = multiprocessing.Event() proc = multiprocessing.Process(target=drive_control_loop, args=(serial_num, stop_event)) proc.start() drive_processes[drive_id] = {"process": proc, "stop_event": stop_event} # 停止单台驱动器进程 def stop_drive(drive_id): if drive_id not in drive_processes: return info = drive_processes[drive_id] info["stop_event"].set() info["process"].join() info["process"].close() del drive_processes[drive_id] # 监控进程状态 def check_drive_status(): for drive_id, info in drive_processes.items(): status = "运行中" if info["process"].is_alive() else "已停止" print(f"驱动器 {drive_id}: {status}") # 示例:启动三台驱动器 start_drive("drive1", "SERIAL_001") start_drive("drive2", "SERIAL_002") start_drive("drive3", "SERIAL_003") # 示例:单独停止drive2 stop_drive("drive2") # 检查所有驱动器状态 check_drive_status()
二、可选的进程管理工具(含GUI)
- psutil:可以获取进程的CPU、内存占用等系统信息,配合自己的代码能实现更细致的监控,适合轻量自定义管理。
- supervisor:系统级进程管理工具,通过配置文件定义进程,支持自动重启崩溃进程,自带命令行管理界面,还能搭配
supervisor-web实现Web GUI监控,适合生产环境稳定运行。 - PyQt5/PySide2:如果需要可视化GUI,用这两个库快速搭建界面,把启动/停止按钮、状态显示标签和上面的进程控制逻辑绑定,就能实现可视化的单台驱动器管理。
- celery:如果需要任务调度、重试机制等复杂逻辑,可以用它管理异步进程,但对于电机控制场景来说略重,按需选择。
内容的提问来源于stack exchange,提问作者raggot
相关产品推荐
相关产品推荐

