4. CAN通信核心库实战:python-can库基础用法

说实话,我刚开始做HIL测试那会儿,最头疼的就是CAN通信这块。那时候还没有python-can这么方便的东西,全是自己手撸底层驱动,一个字节一个字节地拼报文。后来发现python-can这个库,简直像发现了新大陆。今天我就带你把它吃透。

4.1 初识python-can:Bus、Message、Notifier

python-can这个库,说白了就是Python和CAN总线之间的翻译官。它把复杂的CAN通信细节都封装好了,你只需要调用几个简单的接口就行。我个人习惯把它拆成三个核心组件来理解:

组件 作用 我常用的场景
Bus 管理CAN硬件连接 初始化通道、设置波特率
Message 表示一条CAN报文 构造发送数据、解析接收数据
Notifier 异步接收消息 后台监听、不阻塞主流程

先看Bus怎么用。这是最基础的一步,就像打电话要先拨号一样:

import can

# 我习惯用PCAN接口,但vector、kvaser也都支持
bus = can.interface.Bus(
    bustype='pcan',      # 接口类型
    channel='PCAN_USBBUS1',  # 通道号
    bitrate=500000       # 波特率,常用500k
)
print(f"总线已连接: {bus.channel_info}")
小提示: 我在项目中遇到过好几次波特率配错的情况。CAN总线上的所有节点必须用同一个波特率,否则谁也收不到谁的消息。建议先拿CANoe或PCAN-View确认一下DUT的波特率。

接下来是Message。一条CAN报文长什么样?说白了就是ID加数据:

# 构造一条发送报文
msg = can.Message(
    arbitration_id=0x123,  # 报文ID,标准帧用11位
    data=[0x01, 0x02, 0x03, 0x04],  # 数据,最多8字节
    is_extended_id=False,  # 是不是扩展帧?一般用标准帧
    is_remote_frame=False  # 是不是远程帧?一般用数据帧
)

print(f"报文ID: 0x{msg.arbitration_id:X}")
print(f"数据长度: {msg.dlc}")

嗯,这里要注意。DLC(Data Length Code)不一定等于你data数组的长度。如果你data给了4个字节,DLC就是4。但有些ECU会检查DLC,不一致可能会丢帧。我踩过这个坑。

4.2 发送与接收CAN报文

发送报文其实就一行代码。但别小看这一行,里面门道不少:

# 发送一条报文
try:
    bus.send(msg)
    print(f"已发送: ID=0x{msg.arbitration_id:X}, data={msg.data.hex()}")
except can.CanError:
    print("发送失败!检查总线是否正常")
避坑指南: 我曾经在项目里连续发送报文时没加延时,结果总线负载瞬间飙到90%以上,DUT直接罢工了。CAN总线有个仲裁机制,但频繁发送还是会导致丢帧。建议发送间隔至少10ms。

接收报文有两种方式。一种是阻塞式接收,简单粗暴:

# 阻塞等待,直到收到一条报文
received_msg = bus.recv(timeout=1.0)  # 超时1秒
if received_msg:
    print(f"收到: ID=0x{received_msg.arbitration_id:X}")
    print(f"数据: {list(received_msg.data)}")
else:
    print("超时,没收到报文")

另一种是轮询方式,适合需要同时做其他事情的场景:

# 非阻塞轮询
for i in range(100):
    msg = bus.recv(timeout=0.01)  # 10ms超时
    if msg:
        print(f"第{i}次轮询收到: 0x{msg.arbitration_id:X}")
    # 这里可以干别的事

你想想看,如果测试脚本里只有接收逻辑,用阻塞方式没问题。但HIL测试通常要同时发送和接收,这时候轮询或者Notifier就更合适。

4.3 Notifier:后台监听的好帮手

Notifier是我个人最喜欢的组件。它可以在后台一直监听总线,不阻塞主线程。就像请了个小秘书,专门帮你盯着CAN总线:

# 定义一个回调函数
def my_can_listener(msg):
    """收到报文时自动调用"""
    print(f"[监听] ID=0x{msg.arbitration_id:X}, data={msg.data.hex()}")
    # 这里可以加过滤逻辑,比如只处理特定ID

# 创建Notifier
notifier = can.Notifier(bus, [my_can_listener])

# 主线程可以继续做其他事
import time
for i in range(10):
    print(f"主线程工作中... {i}")
    time.sleep(0.5)

# 用完记得关闭
notifier.stop()
实战经验: 我在做ECU唤醒测试时,需要同时监听多个ID的报文。用Notifier配合多个回调函数,一个负责记录日志,一个负责实时判断唤醒状态,互不干扰。比单线程轮询优雅多了。

4.4 过滤与ID配置

CAN总线上报文很多,你不可能全收。过滤就是只收你关心的ID。硬件过滤效率高,软件过滤灵活。我一般这么用:

# 硬件过滤:在Bus初始化时设置
bus_filtered = can.interface.Bus(
    bustype='pcan',
    channel='PCAN_USBBUS1',
    bitrate=500000,
    can_filters=[
        {"can_id": 0x100, "can_mask": 0x7FF},  # 只收ID=0x100
        {"can_id": 0x200, "can_mask": 0x7FF},  # 只收ID=0x200
    ]
)

# 软件过滤:在回调里判断
def smart_listener(msg):
    # 只处理0x100到0x1FF之间的ID
    if 0x100 <= msg.arbitration_id <= 0x1FF:
        print(f"通过软件过滤: 0x{msg.arbitration_id:X}")

为什么会这样?硬件过滤是在CAN控制器层面就筛掉了,CPU负担小。但有些USB-CAN设备不支持硬件过滤,那就只能用软件过滤。我建议能硬件过滤就硬件过滤,省CPU资源。

关于ID配置,还有个常见需求:发送不同ID的报文。比如测试中需要模拟多个节点:

# 模拟多个节点发送
test_ids = [0x100, 0x200, 0x300]
for idx, can_id in enumerate(test_ids):
    msg = can.Message(
        arbitration_id=can_id,
        data=[idx+1, 0x00, 0x00, 0x00],
        is_extended_id=False
    )
    bus.send(msg)
    print(f"模拟节点0x{can_id:X}发送完成")
小技巧: 我在做网关测试时,经常需要同时模拟多个ECU。我会把每个节点的ID和数据做成字典,循环发送。这样代码干净,也方便后期加新节点。

最后提醒一句:用完总线一定要关闭。不然下次运行脚本时会报"设备被占用":

# 善后工作
bus.shutdown()
print("总线已关闭")

好了,python-can的基础用法就这些。说白了就是三板斧:Bus连上去,Message装数据,Notifier收回来。下一章我们讲怎么用这些基础组件搭一个完整的HIL测试脚本。到时候你会发现,今天学的这些就是地基,地基打牢了,上面盖多高的楼都不怕。