如何在ROS2 Python启动文件中根据CAN节点输出启动对应车辆节点

先调整CAN节点的输出逻辑 #

让CAN节点获取到车辆类型后，要么直接打印到标准输出，要么写入临时文件/本地套接字。比如在CAN节点代码里，拿到类型后直接输出 print(f"vehicle_type: {vehicle_type}")，或者把内容写入/tmp/vehicle_type.tmp文件。

用subprocess同步捕获CAN节点输出 #

Python脚本里借助subprocess模块启动CAN节点，实时抓取它的输出，拿到车辆类型后再启动对应业务节点：

import subprocess

# 启动CAN节点，捕获标准输出和错误输出
can_proc = subprocess.Popen(
    ["python3", "can_reader_node.py"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# 循环读取输出，直到拿到车辆类型标识
vehicle_type = None
while True:
    line = can_proc.stdout.readline()
    if not line:
        break
    # 假设CAN节点输出格式为"vehicle_type: SUV"
    if line.startswith("vehicle_type:"):
        vehicle_type = line.strip().split(": ")[1]
        break

# 关闭CAN节点（如果不需要它持续运行）
can_proc.terminate()
can_proc.wait()

# 根据车辆类型启动对应节点
if vehicle_type:
    match vehicle_type:
        case "SUV":
            subprocess.run(["python3", "suv_control_node.py", "--max_speed", "180"])
        case "SEDAN":
            subprocess.run(["python3", "sedan_control_node.py", "--max_speed", "200"])
        case _:
            print(f"未知车辆类型: {vehicle_type}")

用IPC实现长期通信（若CAN节点需持续运行） #

如果CAN节点要长期监听CAN总线，用本地UDP套接字做进程间通信：

CAN节点代码片段 #

import socket

# 绑定本地UDP端口，准备发送数据
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server_addr = ("127.0.0.1", 9000)

# 实际逻辑：从CAN接口读取车辆类型
vehicle_type = "SUV"  # 替换为真实CAN读取逻辑
# 发送给启动脚本
sock.sendto(vehicle_type.encode(), server_addr)

# 继续监听CAN总线的其他数据...

启动脚本代码片段 #

import socket
import subprocess

# 绑定本地UDP端口接收数据
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(("127.0.0.1", 9000))

# 启动CAN节点
subprocess.Popen(["python3", "can_reader_node.py"])

# 等待接收车辆类型
data, _ = sock.recvfrom(1024)
vehicle_type = data.decode().strip()

# 启动对应节点
# ... 同之前的分支逻辑

简单场景用临时文件兜底 #

如果逻辑不复杂，CAN节点拿到类型后写入临时文件，启动脚本轮询读取：

import time
import os

temp_file = "/tmp/can_vehicle_type.tmp"
# 启动CAN节点
subprocess.Popen(["python3", "can_reader_node.py"])

# 轮询直到文件有内容
vehicle_type = None
while not vehicle_type:
    if os.path.exists(temp_file):
        with open(temp_file, "r") as f:
            vehicle_type = f.read().strip()
        # 读完后删除临时文件（可选）
        os.remove(temp_file)
    time.sleep(0.2)

# 启动对应节点
# ...

内容的提问来源于stack exchange，提问作者A. R.