Python解析CAN报文

Python的CAN总线库

Python处理CAN总线的库主要有python-can和cantools。这里我的CAN总线数据保存为asc格式，database保存为dbc格式。

from can import ASCReader
from cantools.database import load_file

# 我的数据
asc_path = 'xx.asc'
dbc_path = 'xx.dbc'

# 读取dbc文件
database = load_file(dbc_path) # 变量database的数据类型cantools.db.can.database.Database

效果相同的读取方法

第一种：cantools.db.can.database.add_dbc_file(filename: StringPathLike,
                                             encoding: str = 'cp1252') -> None:
                     
    >>> db = cantools.database.Database()
    >>> db.add_dbc_file('foo.dbc')
    
第二种：cantools.db.can.database.add_dbc_string(string: str) -> None:

    >>> db = cantools.database.Database()
    >>> with open ('foo.dbc', 'r') as fin:
    ...     db.add_dbc_string(fin.read())
    
值得注意的是，以上两种方法可以在原有database上增加，如：
    >>> db = cantools.database.Database()
    >>> db.add_dbc_file('foo1.dbc')
    >>> db.add_dbc_file('foo2.dbc')
这里的db就包含了'foo1.dbc'和'foo2.dbc'两个dbc文件的信息。
如果两个dbc有ID相同的报文，则后面添加的dbc会覆盖前面id相同的报文。

cantools.db.can.database.Database包含CAN网络的所有报文、信号和定义，是我们经常会用到的一个类。

cantools.db.can.database.Database常用的类属性有messages，nodes和buses，分别返回所有报文的list，所有节点的list和所有总线的list。如果想查看某个报文，则可以使用.get_message_by_frame_id()或.get_message_by_name()

>>> dbc_path = 'xx.dbc'
>>> db = cantools.database.load_file(dbc_path)
>>> db.messages
[message('msg1', 0x18ffffff, True, 8, None), message('smg2', 0x18000000, True, 8, None)]

# 通过id检索
>>> db.get_message_by_frame_id(0x18ffffff)
message('msg1', 0x18ffffff, True, 8, None)

# 通过名称检索
>>> db.get_message_by_name('msg1')
message('msg1', 0x18ffffff, True, 8, None)

解析CAN报文的字节数据

def hex_bin(hex_data):
 
    byte_li = []
    for hex_str in range(0, len(hex_data), 2):
        data = hex_data[hex_str:hex_str+2]
        binary_str = bin(int(data, 16))[2:]
        if len(binary_str) < 8:
            binary_str = '0' * (8 - len(binary_str)) + binary_str
        byte_li.append(binary_str[::-1])
    return byte_li
 
 
def start_byte(bin_data, start_bit, length):
 
    data = "".join(bin_data)
    print(int(data[start_bit:start_bit+length][::-1], 2))
 
 
if __name__ == '__main__':
 
    while True:
        hex_data = input("请输入源字节数据：")
        Start_Bit_Position = int(input("请输入起始位："))
        Signal_Length = int(input("请输入信号长度："))
        start_byte(hex_bin(hex_data), Start_Bit_Position, Signal_Length)

读取两个报文时间戳超过1s的数据

import can
import cantools
import os
import sys

# 加载 BLF 文件
dbc_data = cantools.db.load_file(dbc文件)

target_id = 0x572

frame_572_count = 0
interval_morethan1_1_count = 0
max_interval = 0
# print(sys.argv[0],sys.argv[1])
# 检查是否输入了所要查询的文件夹路径
# if len(sys.argv) < 2:
#     print("请提供输入文件路径作为第一个参数。")
#     sys.exit(1)
# 手动输入所要识别的文件夹路径
# dst_path = sys.argv[1]

# 获取执行文件的路径
executable_path = sys.argv[0]
dst_path = os.path.dirname(os.path.abspath(executable_path))
# os.listdir() 方法用于返回指定的文件夹包含的文件或文件夹的名字的列表
for file in os.listdir( dst_path ):
    if file.endswith( 'blf' ):
        print( '正在处理' + file )
        blf_data = can.BLFReader(dst_path + '\\' + file)
        previous_timestamp = 0
        for message in blf_data:
            if message.arbitration_id == target_id:
                frame_572_count = frame_572_count + 1
                #print('time:{}, channel:{}, data:{}\n'.format(message.timestamp, message.channel, message.data))
                data_572 = dbc_data.decode_message(target_id, message.data, decode_choices=False)
                #print(previous_timestamp)
                if previous_timestamp == 0:
                    previous_timestamp = message.timestamp
                    continue
                #print(message.timestamp - previous_timestamp)
                interval = message.timestamp - previous_timestamp
                if interval > max_interval:
                    max_interval = interval
                if interval > 1:
                    interval_morethan1_1_count += 1
                    print(interval)
                    print('time:{}, channel:{}, data:{}\n'.format(message.timestamp, message.channel, message.data))
                    #读取字节数组的具体数值，秒在第28个字节，纳秒在32个字节
                    bytearray = message.data
                    print('数组长度:{},秒:{},纳秒:{}\n'.format(len(message.data), bytearray[28], bytearray[32]))
                    #遍历数组数据
                    print('读取的数据:\n')
                    for byte in bytearray:  
                    # 打印每个字节的十六进制表示（可选添加前缀'0x'），输出：Byte: 12, Byte: 34, Byte: AB, Byte: CD   
                    #  print(f"Byte: {byte:02X}")    
                    # 如果你想要输出前缀'0x'，可以使用format方法或字符串格式化，输出：Byte: 0x12, Byte: 0x34, Byte: 0xab  
                    #  print("Byte:", hex(byte)) 
                    # 输出：12 34 AB CD（每个字节后有一个空格）                      
                      print(f"{byte:02X}", end=' ')  

                previous_timestamp = message.timestamp

print('\n\n0x572总帧数:{}, 间隔超过1s数量:{}, 最大间隔:{}\n'.format(frame_572_count, interval_morethan1_1_count, max_interval))
#print('0x572总帧数:{}, 间隔超过1s数量:{}, 最大间隔:{}\n'.format(frame_572_count, interval_morethan1_1_count, max_interval))