2. CAN通信基础脚本

任务目标

编写一个Python脚本，通过CAN接口周期性发送指定ID的报文，同时接收总线上的CAN帧并解析关键字段，用于验证ECU之间的基本通信链路是否正常。

环境依赖

本案例基于python-can库实现，需安装：

pip install python-can

硬件层使用PCAN或Kvaser等CAN接口卡，驱动安装完成后，通过can.interface.Bus()指定通道与接口类型。

关键完整代码

import can
import time
import threading

# ========== 配置参数 ==========
CHANNEL = 'PCAN_USBBUS1'      # 根据实际硬件修改
INTERFACE = 'pcan'            # 可选: 'pcan', 'kvaser', 'socketcan'
BITRATE = 500000              # 500 kbps

SEND_ID = 0x123               # 发送报文ID
SEND_DATA = [0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08]
SEND_PERIOD = 0.1             # 发送周期(秒)

RECV_TIMEOUT = 5.0            # 接收超时(秒)
EXPECTED_IDS = [0x123, 0x456] # 期望接收的报文ID列表

# ========== 发送线程 ==========
def send_loop(bus):
    msg = can.Message(arbitration_id=SEND_ID, data=SEND_DATA, is_extended_id=False)
    while not stop_event.is_set():
        try:
            bus.send(msg)
            print(f"[TX] ID=0x{SEND_ID:03X} Data={SEND_DATA.hex()}")
        except can.CanError as e:
            print(f"[TX ERROR] {e}")
        time.sleep(SEND_PERIOD)

# ========== 接收验证 ==========
def receive_and_verify(bus):
    received_ids = set()
    end_time = time.time() + RECV_TIMEOUT
    while time.time() < end_time:
        msg = bus.recv(timeout=0.5)
        if msg is None:
            continue
        print(f"[RX] ID=0x{msg.arbitration_id:03X} "
              f"Data={bytes(msg.data).hex()} "
              f"DLC={msg.dlc}")
        received_ids.add(msg.arbitration_id)
        # 检查是否收到所有期望ID
        if EXPECTED_IDS.issubset(received_ids):
            print("PASS: 所有期望ID均已收到")
            return True
    missing = set(EXPECTED_IDS) - received_ids
    print(f"FAIL: 未收到ID: {[hex(x) for x in missing]}")
    return False

# ========== 主程序 ==========
stop_event = threading.Event()

# 初始化CAN总线
bus = can.interface.Bus(channel=CHANNEL, interface=INTERFACE, bitrate=BITRATE)
print(f"CAN总线已初始化: {bus}")

# 启动发送线程
tx_thread = threading.Thread(target=send_loop, args=(bus,), daemon=True)
tx_thread.start()

# 执行接收验证
try:
    result = receive_and_verify(bus)
    print(f"\n验证结果: {'通过' if result else '失败'}")
finally:
    stop_event.set()
    tx_thread.join(timeout=1.0)
    bus.shutdown()

运行方式

1. 将CAN接口卡连接至PC，并确认驱动已正确安装；
2. 根据实际硬件修改CHANNEL和INTERFACE参数；
3. 将总线上的对端节点配置为发送0x456报文（用于验证接收），或使用另一个CAN设备回环发送；
4. 执行脚本：python can_basic_test.py。

验证方法与判定标准

• 发送验证：脚本周期性打印[TX]日志，表明报文已成功写入总线驱动；
• 接收验证：若在RECV_TIMEOUT时间内收到EXPECTED_IDS中所有ID，脚本输出PASS；否则输出FAIL并列出缺失ID；
• 通过条件：控制台出现验证结果: 通过，且无CanError异常。

常见问题

• 总线初始化失败：检查接口卡连接、通道号是否正确，以及驱动是否安装；
• 收不到对端报文：确认对端节点波特率与本脚本一致（500kbps），且报文ID在EXPECTED_IDS中；
• 发送报错：确认总线无短路或终端电阻配置正确（120Ω）。