使用python-can和cantools实现arxml报文解析、发送和接收的完整指南

文章目录

背景
一、硬件支持
二、环境准备
- 1、python解释器安装
- 2、python库安装
三、收发案例
四、方法拓展
- 1、canoe硬件调用
- 2、回调函数介绍
结论

背景

在汽车行业中，CAN (Controller Area Network) 总线是用于车辆内部通信的关键技术。arxml文件是一种用于描述CAN消息和信号的标准格式。在本指南中，我们将介绍如何使用python-can和cantools库来解析arxml文件，并在Python中发送和接收CAN消息，我大概以下面的逻辑去实现整个发送和接收的框架，CANMsgSender和CANMsgReceiver建议通过线程去实现。

在这里插入图片描述

一、硬件支持

1、python-can的库支持多种硬件，另外还支持虚拟接口，方便调试的时候使用。

点击查看python-can硬件介绍

在这里插入图片描述

二、环境准备

1、python解释器安装

1）请确保安装的python解释器大于3.8，因为我们在发送报文时会用到send periodic()函数，这个函数在python-can4.3.0版本之后才开始增加回调函数的方法（如下图示），当设定send_periodic(register_message,period = cycle_time,modifier_callback = update_msg_data)，发送报文时会自动回调update_msg_data函数，我们可以在update_msg_data函数中更新报文数据。

点击查看python-can release信息

在这里插入图片描述

2、python库安装

1）首先，确保你的Python环境中已经安装了python-can和cantools库，这里的python-can的版本一定要大于等于4.3.0。如果没有安装，可以使用pip进行安装：

pip install python-can
pip install cantools

三、收发案例

这个例子可以简单地实现报文的发送与接收，这里我就不展示完整的代码了，因为每个人的需求都不一样，可以按上面提供的框架进行开发。

import can
import time
import cantools


#创建虚拟总线对象
send_bus = can.interface.Bus('test', interface='virtual', preserve_timestamps=True)
recv_bus = can.interface.Bus('test', interface='virtual', preserve_timestamps=True)

#cantools解析arxml
database = cantools.database.load_file('demo.arxml' , strict=False)

#这个是send_periodic的回调函数，python-can会周期回调这个函数
#另外这个message还有其他属性
def update_msg_data(message):
    for msg in database.messages:
        if message.arbitration_id == msg.frame_id:
            #这里可以处理message的data
            message.data = message.data
            print(f"正在回调{message.arbitration_id}报文")                        


if __name__ == "__main__":

    #注册周期报文
    for msg in database.messages:
        encode_data = []
        msg_data = None
        #判断是否含有pdu
        if msg.is_container:
            for contain_msg in msg.contained_messages:
                #获取signal的字典，值就默认填0
                signals_dict = {signal.name : 0 for signal in contain_msg.signals}
                contain_msg_data = contain_msg.encode(signals_dict)
                encode_data.append((contain_msg,contain_msg_data))
            #获取报文的发送周期
            cycle_time = contain_msg.cycle_time if contain_msg.cycle_time is not None else 20
            #encode每个pdu的data
            msg_data = msg.encode(encode_data)
        #如果是报文类型
        else:
            signals_dict = {signal.name : 0 for signal in msg.signals}
            cycle_time = msg.cycle_time if msg.cycle_time is not None else 20
            msg_data = msg.encode(signals_dict)
        #注册周期发送报文
        register_message = can.Message(timestamp=time.time(), arbitration_id=msg.frame_id, is_extended_id=msg.is_extended_frame, is_fd=True, data=msg_data)
        send_bus.send_periodic(register_message,period = cycle_time /1000,modifier_callback = update_msg_data)


    #循环接收报文
    database_msg_dict = {msg.frame_id : msg for msg in database.messages}
    while True:
        decode_msg_dict = {}
        recv_msg = recv_bus.recv()
        frame_id = recv_msg.arbitration_id
        data = recv_msg.data
        try:
            if database_msg_dict[frame_id].is_container:
                contain_msgs = database.decode_message(frame_id,data,decode_choices=False,decode_containers=True,allow_truncated=True)
                for contain_msg in contain_msgs:
                    contain_msg_signals = contain_msg[1]
                    if isinstance(contain_msg_signals,dict):
                        decode_msg_dict.update(contain_msg[1])
            else:
                decode_msg_dict =  database.decode_message(frame_id,data,decode_choices=False,decode_containers=False)
            print(decode_msg_dict)
        except Exception as e:
            print(e)
            continue

循环打印结果：
在这里插入图片描述

四、方法拓展

1、canoe硬件调用

1）这里以canoe为例，首先在canoe中定义Application pycan(这个名字随便取)，然后再分配到真实的canoe通道中去。
在这里插入图片描述

在这里插入图片描述

2）修改通道定义

#send_bus = can.interface.Bus('test', interface='virtual', preserve_timestamps=True)
send_bus= can.interfaces.vector.canlib.VectorBus(app_name='pycan', channel=0, bitrate=500000, data_bitrate=2000000, fd=True)

3）其他硬件调用
点击查看python-can硬件介绍

2、回调函数介绍

1）回调函数中的message有其他属性，其他属性可以搭配cantools一起使用。

def update_msg_data(message):
    for msg in database.messages:
        if message.arbitration_id == msg.frame_id:
            #这里可以处理message的data
            message.data = message.data
            print(f"正在回调{message.arbitration_id}报文")