4、Python CAN库入门:python-can库安装、CAN总线初始化、发送与接收CAN报文、CAN报文过滤设置
好,咱们进入第四章。说实话,CAN通信是UDS诊断的底层基石。你想想看,没有CAN报文收发,诊断指令根本发不出去。所以这一章,我带你把python-can这个库彻底搞明白。
我个人习惯,做任何项目前先把工具链摸透。python-can就是咱们和硬件CAN卡打交道的桥梁。它封装了底层驱动,让我们用Python就能轻松操作CAN总线。
4.1 python-can库安装
安装这事儿,其实很简单。但我在项目中遇到过一个问题——版本不兼容。所以这里给你一个稳妥的方案。
推荐安装方式:
# 基础安装
pip install python-can
# 如果你用Vector硬件(比如VN1610、VN1640),需要额外装这个
pip install python-can[vector]
# 如果你用PCAN硬件
pip install python-can[pcan]
# 如果你用Kvaser硬件
pip install python-can[kvaser]
# 如果你用USB转CAN(比如周立功、创芯)
pip install python-can[serial]
# 如果你只是学习,用虚拟CAN总线
pip install python-can[virtual]
嗯,这里要注意。千万别一股脑全装上。我建议你先装基础版,跑通了再根据硬件加装驱动。否则依赖冲突会让你头疼半天。
小技巧: 装完后运行 python -c "import can; print(can.__version__)",如果能正常输出版本号,说明安装成功了。
4.2 CAN总线初始化
初始化CAN总线,说白了就是告诉python-can:你要用哪个硬件、波特率多少、总线模式是什么。我刚开始做UDS诊断时,总在这步卡住——不是找不到硬件,就是波特率配错了。
来看一个典型的初始化代码:
import can
# 方法一:使用配置文件(推荐)
# 创建一个字典,配置总线参数
bus_config = {
'interface': 'vector', # 接口类型:vector, pcan, kvaser, socketcan, virtual等
'channel': 0, # 通道号,通常0或1
'bitrate': 500000, # 波特率,单位bps。CAN通常250k或500k
'receive_own_messages': False, # 是否接收自己发送的报文
}
# 初始化总线
bus = can.interface.Bus(**bus_config)
print(f"CAN总线初始化成功!接口: {bus.config['interface']}, 通道: {bus.channel}")
为什么会这样配置?我解释一下。interface参数决定了你用哪个厂家的硬件。channel是硬件通道号,比如Vector VN1610有两个通道,0和1。bitrate是波特率,UDS诊断通常用500k,但有些老车型用250k。
避坑指南: 我曾经在项目现场折腾了整整一下午,发现CAN报文死活发不出去。最后发现是波特率配错了——客户的车用的是250k,我默认写了500k。所以,拿到车后第一件事,确认波特率!
如果你只是想学习,没有硬件怎么办?用虚拟CAN总线:
# 虚拟CAN总线,适合学习和调试
virtual_bus = can.interface.Bus(bustype='virtual', channel='vcan0', bitrate=500000)
print("虚拟CAN总线已创建,可以开始学习了!")
4.3 发送与接收CAN报文
CAN报文的结构其实很简单。一个标准CAN帧包含:ID(标识符)、DLC(数据长度)、Data(数据)。UDS诊断报文就封装在Data字段里。
先看发送:
import can
# 初始化总线(假设已经初始化好了)
bus = can.interface.Bus(bustype='virtual', channel='vcan0', bitrate=500000)
# 创建一个CAN报文
# 诊断请求通常用扩展帧,ID为0x18DA00F1(物理寻址)
msg = can.Message(
arbitration_id=0x18DA00F1, # CAN ID,UDS诊断常用扩展帧29位ID
data=[0x02, 0x10, 0x01], # 数据:长度2字节,服务0x10(诊断会话控制),子功能0x01(默认会话)
is_extended_id=True, # 是否扩展帧(29位ID)
is_remote_frame=False, # 是否远程帧
dlc=8 # 数据长度,通常固定为8
)
# 发送报文
try:
bus.send(msg)
print(f"报文发送成功!ID: 0x{msg.arbitration_id:X}, 数据: {msg.data.hex()}")
except can.CanError as e:
print(f"发送失败: {e}")
再看接收。接收有两种方式:阻塞式和非阻塞式。我建议初学者先用阻塞式,逻辑清晰。
# 阻塞式接收(等待直到收到报文)
print("等待接收CAN报文...")
received_msg = bus.recv(timeout=1.0) # 超时1秒
if received_msg is not None:
print(f"收到报文!")
print(f" ID: 0x{received_msg.arbitration_id:X}")
print(f" 数据: {received_msg.data.hex()}")
print(f" DLC: {received_msg.dlc}")
print(f" 时间戳: {received_msg.timestamp}")
else:
print("接收超时,没有收到报文")
# 非阻塞式接收(适合循环监听)
def receive_non_blocking(bus):
"""非阻塞接收示例"""
msg = bus.recv(timeout=0) # timeout=0表示立即返回
if msg:
return msg
return None
# 循环监听示例
print("开始循环监听CAN报文(按Ctrl+C停止)...")
try:
while True:
msg = bus.recv(timeout=0.1) # 每100ms检查一次
if msg:
print(f"[{msg.timestamp:.3f}] ID: 0x{msg.arbitration_id:X}, 数据: {msg.data.hex()}")
except KeyboardInterrupt:
print("停止监听")
个人经验: 我在做UDS诊断工具时,接收部分用了多线程。一个线程专门收报文,另一个线程处理业务逻辑。这样不会因为处理数据而漏掉报文。后面我会专门讲多线程设计。
4.4 CAN报文过滤设置
你想想看,CAN总线上报文满天飞。如果不加过滤,你的程序会被无关报文淹没。UDS诊断只关心特定ID的报文,所以过滤是必须的。
python-can支持硬件过滤和软件过滤。硬件过滤效率高,但不同硬件支持程度不同。软件过滤更通用,但CPU开销大。
先看硬件过滤(以Vector为例):
# 硬件过滤设置
# 只接收ID为0x18DA00F1和0x18DAF100的报文
filters = [
{"can_id": 0x18DA00F1, "can_mask": 0x1FFFFFFF, "extended": True},
{"can_id": 0x18DAF100, "can_mask": 0x1FFFFFFF, "extended": True},
]
bus = can.interface.Bus(
bustype='vector',
channel=0,
bitrate=500000,
can_filters=filters # 硬件过滤
)
print("硬件过滤已设置,只接收诊断相关报文")
再看软件过滤(通用方法):
# 软件过滤 - 在接收循环中手动判断
def receive_with_filter(bus, target_ids):
"""
软件过滤接收
:param bus: CAN总线对象
:param target_ids: 感兴趣的ID列表
"""
msg = bus.recv(timeout=0.1)
if msg and msg.arbitration_id in target_ids:
return msg
return None
# 使用示例
diagnostic_ids = [0x18DA00F1, 0x18DAF100, 0x18DBFFF1]
while True:
msg = receive_with_filter(bus, diagnostic_ids)
if msg:
print(f"诊断报文: ID=0x{msg.arbitration_id:X}, 数据={msg.data.hex()}")
避坑指南: 我曾经犯过一个低级错误——硬件过滤设置后,发现收不到任何报文。排查了半天,原来是can_mask写错了。0x1FFFFFFF表示精确匹配,0x00000000表示接收所有。如果你不确定,先用软件过滤调试,稳定了再切硬件过滤。
最后,给你一个完整的收发示例,把今天学的串起来:
import can
import time
def main():
"""CAN通信完整示例"""
# 1. 初始化总线
bus = can.interface.Bus(bustype='virtual', channel='vcan0', bitrate=500000)
# 2. 设置过滤(只收诊断报文)
filters = [{"can_id": 0x18DA00F1, "can_mask": 0x1FFFFFFF, "extended": True}]
bus.set_filters(filters)
# 3. 发送诊断请求(10 01 - 诊断会话控制)
request = can.Message(
arbitration_id=0x18DA00F1,
data=[0x02, 0x10, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00],
is_extended_id=True,
dlc=8
)
bus.send(request)
print(f"发送: {request.data.hex()}")
# 4. 接收响应
time.sleep(0.1) # 给ECU一点响应时间
response = bus.recv(timeout=1.0)
if response:
print(f"接收: {response.data.hex()}")
# 解析响应(50 01 表示成功)
if response.data[0] == 0x50 and response.data[1] == 0x01:
print("诊断会话切换成功!")
else:
print("没有收到响应")
# 5. 关闭总线
bus.shutdown()
if __name__ == "__main__":
main()
嗯,这一章的内容就到这。你把这几个例子跑一遍,CAN通信的基本功就算打牢了。下一章,咱们正式进入UDS诊断协议,看看怎么用这些CAN报文实现真正的诊断功能。