CAN通讯进阶-基于Python使用dbc文件解析CAN数据（dbc文件的创建、使用方法）

CAN通讯进阶-基于Python使用DBC文件解析CAN数据配置环境1.环境配置1.1安装CAN通讯需要的包1.2 安装kvaser_drivers以及Kvaser CanKing2.使用DBC文件解析CAN帧2.1DBC文件2.2本博客布局3.DBC文件的创建3.1创建DBC文件代码3.2使用方法4.DBC文件发送指定格式的CAN数据4.1DBC文件发送指定格式的CAN数据代码4.2使用方法5.

Hurri_cane

14602人浏览 · 2021-04-16 20:11:57

Hurri_cane · 2021-04-16 20:11:57 发布

CAN通讯进阶-基于Python使用DBC文件解析CAN数据

配置环境
1.环境配置
- 1.1安装CAN通讯需要的包
- 1.2 安装kvaser_drivers以及Kvaser CanKing
2.使用DBC文件解析CAN帧
- 2.1DBC文件
- 2.2本博客布局
3.DBC文件的创建
- 3.1创建DBC文件代码
- 3.2使用方法
4.DBC文件发送指定格式的CAN数据
- 4.1DBC文件发送指定格式的CAN数据代码
- 4.2使用方法
5.使用DBC文件解析接收到的CAN数据
- 5.1使用DBC文件解析接收到的CAN数据代码
- 5.2使用方法
6.参考文献
7.结束语

配置环境

使用环境：python3.6
平台：Windows10
IDE：PyCharm

1.环境配置

1.1安装CAN通讯需要的包

pip install canlib

1.2 安装kvaser_drivers以及Kvaser CanKing

这些软件可以在官网下载https://www.kvaser.com/download/

官网是英文的，看起来可能有点吃力，这里也给出云盘连接：
链接：https://pan.baidu.com/s/1LDKyIlOV0Ky4d2qxryLZGQ
提取码：vwiv

下载下来之后的文件是这样的，三个文件依次双击安装就行了

在这里插入图片描述
安装完后，在开始栏可以查看到已经安装好了

2.使用DBC文件解析CAN帧

2.1DBC文件

DBC是Database Can的缩写，其代表的是CAN的数据库文件，在这个文件中把CAN通讯的信息定义的非常完整清楚，在Windows下其存在的格式便是dbc，如图所示：
在这里插入图片描述
有了DBC文件便可以对接收到的CAN帧进行解析，分析一系列CAN帧代表的真实物理含义。
正因为如此，每一个DBC文件都对应着一种CAN帧的规则，使用CAN作为通讯手段的设备（如汽车中发动机、车灯、车载空调等等）都有对应的DBC文件。

2.2本博客布局

本博客将从DBC文件的创建、使用DBC文件发送指定格式的CAN数据、使用DBC文件解析接收到的CAN数据来展开。

3.DBC文件的创建

3.1创建DBC文件代码

# author:Hurricane
# date:  2021/4/16
# File : CAN_Create_Database.py 
# E-mail:hurri_cane@qq.com



import argparse
from collections import namedtuple

from canlib import kvadblib

Message = namedtuple('Message', 'name id dlc signals')
Signal = namedtuple('Signal', 'name size scaling limits unit')
EnumSignal = namedtuple('EnumSignal', 'name size scaling limits unit enums')

_messages = [
	Message(
		name='EngineData',
		id=100,
		dlc=8,
		signals=[
			Signal(
				name='PetrolLevel',
				size=(24, 8),
				scaling=(1, 0),
				limits=(0, 255),
				unit="l",
			),
			Signal(
				name='EngPower',
				size=(48, 16),
				scaling=(0.01, 0),
				limits=(0, 150),
				unit="kW",
			),
			Signal(
				name='EngForce',
				size=(32, 16),
				scaling=(1, 0),
				limits=(0, 0),
				unit="N",
			),
			EnumSignal(
				name='IdleRunning',
				size=(23, 1),
				scaling=(1, 0),
				limits=(0, 0),
				unit="",
				enums={'Running': 0, 'Idle': 1},
			),
			Signal(
				name='EngTemp',
				size=(16, 7),
				scaling=(2, -50),
				limits=(-50, 150),
				unit="degC",
			),
			Signal(
				name='EngSpeed',
				size=(0, 16),
				scaling=(1, 0),
				limits=(0, 8000),
				unit="rpm",
			),
		]),
	Message(
		name='GearBoxInfo',
		id=1020,
		dlc=1,
		signals=[
			Signal(
				name='EcoMode',
				size=(6, 2),
				scaling=(1, 0),
				limits=(0, 1),
				unit="",
			),
			EnumSignal(
				name='ShiftRequest',
				size=(3, 1),
				scaling=(1, 0),
				limits=(0, 0),
				unit="",
				enums={'Shift_Request_On': 1, 'Shift_Request_Off': 0},
			),
			EnumSignal(
				name='Gear',
				size=(0, 3),
				scaling=(1, 0),
				limits=(0, 5),
				unit="",
				enums={
					'Idle': 0,
					'Gear_1': 1,
					'Gear_2': 2,
					'Gear_3': 3,
					'Gear_4': 4,
					'Gear_5': 5,
				},
			),
		]),
]


def create_database(name, filename):
	db = kvadblib.Dbc(name=name)

	for _msg in _messages:
		message = db.new_message(
			name=_msg.name,
			id=_msg.id,
			dlc=_msg.dlc,
		)

		for _sig in _msg.signals:
			if isinstance(_sig, EnumSignal):
				_type = kvadblib.SignalType.ENUM_UNSIGNED
				_enums = _sig.enums
			else:
				_type = kvadblib.SignalType.UNSIGNED
				_enums = {}
			message.new_signal(
				name=_sig.name,
				type=_type,
				byte_order=kvadblib.SignalByteOrder.INTEL,
				mode=kvadblib.SignalMultiplexMode.MUX_INDEPENDENT,
				size=kvadblib.ValueSize(*_sig.size),
				scaling=kvadblib.ValueScaling(*_sig.scaling),
				limits=kvadblib.ValueLimits(*_sig.limits),
				unit=_sig.unit,
				enums=_enums,
			)

	db.write_file(filename)
	db.close()


if __name__ == '__main__':
	parser = argparse.ArgumentParser(
		description="Create a database from scratch.")
	parser.add_argument('--filename', default=r'..\engine_example.dbc', help=(
		"The filename to save the database to."))
	parser.add_argument('-n', '--name', default='Engine example', help=(
		"The name of the database (not the filename, the internal name."))
	args = parser.parse_args()

	create_database(args.name, args.filename)

3.2使用方法

代码中if __name__ == '__main__':下：

parser.add_argument('--filename', default=r'..\engine_example.dbc', help=(
		"The filename to save the database to."))

指定的filename为创建的DBC文件的路径以及文件名，本代码中便是将DBC文件存在代码所在目录的上层目录中，以engine_example.dbc文件名保存

代码中_messages列表
在这里插入图片描述
便是定义CAN帧解析的规则，可以定义：

CAN帧的id
CAN帧的数据长度
CAN帧的解析single规则
等等…

创建方式如下：

在这里插入图片描述

4.DBC文件发送指定格式的CAN数据

4.1DBC文件发送指定格式的CAN数据代码

# author:Hurricane
# date:  2021/4/12
# File : CAN_Random_Send.py 
# E-mail:hurri_cane@qq.com

import argparse
import time
import random

from canlib import canlib, kvadblib


bitrates = {
    '1M': canlib.canBITRATE_1M,
    '500K': canlib.canBITRATE_500K,
    '250K': canlib.canBITRATE_250K,
    '125K': canlib.canBITRATE_125K,
    '100K': canlib.canBITRATE_100K,
    '62K': canlib.canBITRATE_62K,
    '50K': canlib.canBITRATE_50K,
    '83K': canlib.canBITRATE_83K,
    '10K': canlib.canBITRATE_10K,
}


# 随机在dbc结构中抽取一个结构
def set_random_framebox_signal(db, framebox, signals):
    sig = random.choice(signals)
    value = get_random_value(db, sig)
    framebox.signal(sig.name).phys = value

# 随机在抽取到的结构帧的数值范围中产生一个值
def get_random_value(db, sig):
    limits = sig.limits
    value = random.uniform(limits.min, limits.max)

    # round value depending on type...
    if (
            sig.type is kvadblib.SignalType.UNSIGNED or
            sig.type is kvadblib.SignalType.SIGNED
    ):
        # ...remove decimals if the signal was of type unsigned
        value = int(round(value))
    else:
        # ...otherwise, round to get only one decimal
        value = round(value, 1)

    return value


def ping_loop(channel_number, db_name, num_messages, quantity, interval, bitrate, seed=0):
    db = kvadblib.Dbc(filename=db_name)

    ch = canlib.openChannel(channel_number, canlib.canOPEN_ACCEPT_VIRTUAL)
    ch.setBusOutputControl(canlib.canDRIVER_NORMAL)
    ch.setBusParams(bitrate)
    ch.busOn()

    random.seed(seed)

    if num_messages == -1:
        # used_messages为dbc文件定义的所有帧结构
        used_messages = list(db)
    else:
        used_messages = random.sample(list(db), num_messages)

    print()
    print("Randomly selecting signals from the following messages:")
    print(used_messages)
    print("Seed used was " + repr(seed))
    print()

    while True:
        # Create an empty framebox each time, ignoring previously set signal
        # values.
        framebox = kvadblib.FrameBox(db)

        # Add all messages to the framebox, as we may use send any signal from
        # any of them.
        for msg in db:
            framebox.add_message(msg.name)

        # Make a list of all signals (which framebox has found in all messages
        # we gave it), so that set_random_framebox_signal() can pick a random
        # one.
        signals = [bsig.signal for bsig in framebox.signals()]

        # Set some random signals to random values
        for i in range(quantity):
            set_random_framebox_signal(db, framebox, signals)

        # Send all messages/frames
        for frame in framebox.frames():
            print('Sending frame', frame)
            ch.writeWait(frame, timeout=5000)

        time.sleep(interval)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        description="Send random CAN message based on a database.")
    parser.add_argument('channel', type=int, default=0, nargs='?', help=(
        "The channel to send messages on."))
    parser.add_argument('--bitrate', '-b', default='500k', help=(
        "Bitrate, one of " + ', '.join(bitrates.keys())))
    parser.add_argument('--db', default="../Mobileye.dbc", help=(
        "The database file to base messages on."))
    parser.add_argument('-Q', '--quantity', type=int, default=5, help=(
        "The number of signals to send each tick."))
    parser.add_argument('-I', '--interval', type=float, default=0.2, help=(
        "The time, in seconds, between ticks."))
    parser.add_argument('-n', '--num-messages', type=int, default=-1, help=(
        "The number of message from the database to use, or -1 to use all."))
    parser.add_argument('-s', '--seed', nargs='?', default='0', help=(
        "The seed used for choosing messages. If possible, will be converted to an int. If no argument is given, a random seed will be used."))
    args = parser.parse_args()

    if args.seed is None:
        seed = None
    else:
        try:
            seed = int(args.seed)
        except ValueError:
            seed = args.seed

    ping_loop(
        channel_number=args.channel,
        db_name=args.db,
        num_messages=args.num_messages,
        quantity=args.quantity,
        interval=args.interval,
        bitrate=bitrates[args.bitrate.upper()],
        seed=args.seed,
    )

4.2使用方法

这份代码的功能为：
以DBC文件规定的数据格式，随机发送一些数据帧出去

代码中if __name__ == '__main__':下：

    parser.add_argument('--db', default="../Mobileye.dbc", help=(
        "The database file to base messages on."))

指定的filename为读取的DBC文件的路径，本代码中读取的DBC文件存在代码所在目录的上层目录中，以Mobileye.dbc文件名存在

PS：这里面的DBC文件也可以改为3.1创建DBC文件代码中生成的engine_example.dbc文件

此处附上两个DBC文件的云盘链接

链接：https://pan.baidu.com/s/1JAT2o2fPzto8555qU-lVKg
提取码：49bv

为了测试发送数据是否成功，采用上篇博客2.2.1使用Kvaser Can King接收数据中的方法。链接如下：
https://blog.csdn.net/ShakalakaPHD/article/details/115767739

运行4.1代码：
在这里插入图片描述

可以看到Kvaser Can King接收到各式各样的数据，但是其中的具含义我们并不清楚，这边涉及到第5节，使用DBC文件解析接收到的CAN帧。

5.使用DBC文件解析接收到的CAN数据

5.1使用DBC文件解析接收到的CAN数据代码

# author:Hurricane
# date:  2021/4/16
# File : CAN_Using_Database.py
# E-mail:hurri_cane@qq.com

import argparse

from canlib import canlib, kvadblib


bitrates = {
    '1M': canlib.canBITRATE_1M,
    '500K': canlib.canBITRATE_500K,
    '250K': canlib.canBITRATE_250K,
    '125K': canlib.canBITRATE_125K,
    '100K': canlib.canBITRATE_100K,
    '62K': canlib.canBITRATE_62K,
    '50K': canlib.canBITRATE_50K,
    '83K': canlib.canBITRATE_83K,
    '10K': canlib.canBITRATE_10K,
}


def printframe(db, frame):
    try:
        bmsg = db.interpret(frame)
    except kvadblib.KvdNoMessage:
        print("<<< No message found for frame with id %s >>>" % frame.id)
        return

    msg = bmsg._message

    # form = '═^' + str(width)
    # print(format(" %s " % msg.name, form))

    print('┏', msg.name)

    if msg.comment:
        print('┃', '"%s"' % msg.comment)

    for bsig in bmsg:
        print('┃', bsig.name + ':', bsig.value, bsig.unit)

    print('┗')


def monitor_channel(channel_number, db_name, bitrate, ticktime):
    db = kvadblib.Dbc(filename=db_name)

    ch = canlib.openChannel(channel_number, canlib.canOPEN_ACCEPT_VIRTUAL)
    ch.setBusOutputControl(canlib.canDRIVER_NORMAL)
    ch.setBusParams(bitrate)
    ch.busOn()

    timeout = 0.5
    tick_countup = 0
    if ticktime <= 0:
        ticktime = None
    elif ticktime < timeout:
        timeout = ticktime

    print("Listening...")
    while True:
        try:
            frame = ch.read(timeout=int(timeout * 1000))
            printframe(db, frame)
        except canlib.CanNoMsg:
            if ticktime is not None:
                tick_countup += timeout
                while tick_countup > ticktime:
                    print("tick")
                    tick_countup -= ticktime
        except KeyboardInterrupt:
            print("Stop.")
            break

if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        description="Listen on a CAN channel and print all signals received, as specified by a database.")
    parser.add_argument('channel', type=int, default=1, nargs='?', help=(
        "The channel to listen on."))
    parser.add_argument('--db', default="../Mobileye.dbc", help=(
        "The database file to look up messages and signals in."))
    parser.add_argument('--bitrate', '-b', default='500k', help=(
        "Bitrate, one of " + ', '.join(bitrates.keys())))
    parser.add_argument('--ticktime', '-t', type=float, default=0, help=(
        "If greater than zero, display 'tick' every this many seconds"))
    args = parser.parse_args()

    monitor_channel(args.channel, args.db, bitrates[args.bitrate.upper()], args.ticktime)

5.2使用方法

代码中if __name__ == '__main__':下：

    parser.add_argument('--db', default="../Mobileye.dbc", help=(
        "The database file to look up messages and signals in."))

指定的filename为读取的DBC文件的路径，本代码中读取的DBC文件存在代码所在目录的上层目录中，以Mobileye.dbc文件名存在

PS：这里面的DBC文件也可以改为3.1创建DBC文件代码中生成的engine_example.dbc文件

为了测试解析数据是否有效，采用此博客中4.DBC文件发送指定格式的CAN数据节下的随机发送数据的代码发送数据。

运行4.1节代码后，再运行5.1节代码来解析数据：

在这里插入图片描述
可以看到相较于Kvaser Can King接收到各式各样的数据，使用DBC文件解析后的数据都已准换为了真实的物理含义：

6.参考文献

Python Canlib Documentation
file:///D:/Program%20Files%20(x86)/kvaserCAN/canlib/python/pycanlib/docs/index.html

7.结束语

如果本文对你有帮助的话还请点赞、收藏一键带走哦，你的支持是我最大的动力！(づ｡◕ᴗᴗ◕｡)づ
在这里插入图片描述

华为云开发者联盟

为开发者提供学习成长、分享交流、生态实践、资源工具等服务，帮助开发者快速成长。

更多推荐

解锁HDC 2024之旅：从购票到报名，全程攻略

华为云开发者联盟

从原始边列表到邻接矩阵Python实现图数据处理的完整指南

华为云开发者联盟

GaussDB数据库SQL系列-复合查询

华为云开发者联盟

所有评论(0)

查看更多评论

Hurri_cane

@ShakalakaPHD

已为社区贡献6条内容