8. CAN总线通讯

在我们鲁班猫部分板卡上,带有CAN总线的收发器和接口。通过CAN总线的收发器和接口,我们可以使用鲁班猫板卡 通过CAN总线进行数据的传输。

重要

在做本节实验前,请确保您的鲁班猫板卡上有CAN总线的收发器和接口,才可进行本节实验。 如 鲁班猫I.MX6ULL MINI 板卡上就没有搭载CAN总线的收发器,不能做本节实验。 鲁班猫I.MX6ULL Pro 板则搭载CAN总线的收发器和接口,可做本节实验。

8.1. CAN总线通讯实验

对于我们的鲁班猫板卡而言,它具备了传统工业领域中的一些通讯接口及功能, 比如本节实验中会使用到的CAN总线。 因此在鲁班猫板卡上,我们可以通过其工业通讯接口与许多传统设备耦合, 将鲁班猫板卡丰富的功能,赋予到各种传统的工业设备中。

那么在本讲中,我们可以用Python来编写一些代码,来使用到板卡上的CAN总线资源。

8.2. python-can库

8.2.1. python-can库简介

python-can是Python库下的一个库包,它实现了CAN总线通讯中许多的通讯操作, 它对不同的硬件设备提供了通用的抽象接口,方便Python开发人员使用该库进行CAN通讯中的数据收发。

我们可以在鲁班猫板卡上安装python-can库, 并通过编写一些测试代码来使用该库,进行基于CAN总线的数据收发实验。

8.3. 实验准备

8.3.1. 添加CAN资源

在板卡上的部分资源可能默认未被开启,在使用前请根据需要修改 /boot/uEnv.txt 文件, 可添加对应设备树插件的加载,重启系统,以在系统中添加对应资源。

如本节实验中,通常情况下在鲁班猫系统中默认使能 CAN 的功能, 如大家发现CAN的设备树插件未加载,请做相应修改。

同时如果大家在使用中发现CAN功能引脚被复用或有冲突,请要将对应占用CAN引脚资源的设备树插件取消加载, 否则CAN资源可能无法使用。若已开启对应资源,忽略以下步骤。

本节实验中,笔者使用鲁班猫i.MX6ULL Pro板卡为例演示, 板卡上CAN总线的设备树插件和485总线设备树插件存在引脚冲突, 因此在启用CAN总线设备树插件时,需要关闭485设备树插件。 找到can、485相关设备树插件修改,修改后并重启开发板。

方法参考如下:

broken

添加CAN设备树插件,以使系统支持CAN功能,如下:

broken
# 以鲁班猫i.MX6ULL Pro板卡CAN设备树插件内容为例:
dtoverlay=/usr/lib/linux-image-4.19.35-imx6/overlays/imx-fire-can1.dtbo

如若运行代码时出现“Device or resource busy”或者运行代码卡死等等现象, 请按上述情况检查并按上述步骤操作。

如出现 Permission denied 或类似字样,请注意用户权限,大部分操作硬件外设的功能,几乎都需要root用户权限,简单的解决方案是在执行语句前加入sudo或以root用户运行程序。

在启动文件中添加好设备资源并重启板卡之后,我们可以查看设备是否已经正常添加。

# 在终端中输入如下命令,可以查看到CAN设备资源:
ifconfig -a

检查板卡上的CAN设备资源示例,仅供参考:

broken

在Linux驱动中,CAN设备被当成网络设备来处理。

关于Linux CAN设备相关内容,参考 《Linux CAN》

8.3.2. 硬件连接

硬件连接小节,需要大家根据自身开发环境等情况对本章内容进行参考调整。

本章中笔者使用CAN分析仪来进行CAN数据通讯的展示,各位根据自身的设备情况调整。 如你手中有两块具有CAN功能的板卡,也可进行通讯实验。

注:CAN分析仪是常见的CAN总线调试仪器,和万用表、示波器的属性一致, 非鲁班猫板卡上板载的设备,大家需要的话请自行购买。

笔者连接如下:

broken

8.3.3. python-can库安装

我们使用apt工具安装。

# 在终端中输入如下命令,安装python3-can库:
sudo apt -y install python3-can

等等安装完成后即可。

8.3.4. python-can库使用

安装好对应的库之后,我们就可以利用安装好的python3-can库编写一下测试代码。

8.4. 测试代码

测试代码如下:

配套代码 io/canbus/canbus_test.py文件内容
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
""" python can 测试 """
import sys
import time
import threading
import can


def msg_recv(device_x):
    "接收消息功能"
    print("success: msg_recv Thread is running!")
    # 将can_mask转换为二进制形式,can_mask中为1的位,用于过滤接收到的帧
    # 举例                     id: 0 0
    #                        mask: 1 0 则接收到消息的ID中,mask为1对应id中的位,必须与id一致,为0
    # 如接收到了四个id的消息  id1: 0 0 此条消息被放行
    #                         id2: 0 1 此条消息被放行
    #                         id3: 1 0 此条消息被过滤
    #                         id4: 1 1 此条消息被过滤
    # 过滤器配置示例如下。第一条规则,接收所有标准帧,第二条规则,接收拓展帧中id为0x300的消息。
    can_filters = [
        {"can_id": 1, "can_mask": 0x0, "extended": False},
        {"can_id": 0x300, "can_mask": 0x1FFFFFFF, "extended": True},
    ]
    # 应用过滤器配置
    device_x.set_filters(can_filters)
    # 查询退出线程是否退出,如果为真,则说明用户期望程序退出,退出本线程循环,线程结束
    while tasks_quitThread.is_alive():
        try:
            # 接收can消息
            msg = device_x.recv(1)
            if msg is not None:
                print("success: ", msg)
        except can.CanError:
            print("error: 接收消息时出错,请检查设备是否启用及状态正常")


def msg_send(device_x):
    "发送消息功能"
    print("success: msg_send Thread is running!")
    # 构造发送的CAN消息结构,ID为0xC0FFEE,数据内容包含在data中,is_extended_id为拓展ID标识
    msg = can.Message(
        arbitration_id=0xC0FFEE, data=[0, 25, 0, 1, 3, 1, 4, 1], is_extended_id=True
    )
    # 查询退出线程是否退出,如果为真,则说明用户期望程序退出,退出本线程循环,线程结束
    while tasks_quitThread.is_alive():
        try:
            # 发送构造的CAN消息
            device_x.send(msg)
            # 打印发送提示
            print(f"success: 消息已发送至 {device_x.channel_info}")
        except can.CanError:
            print("error: 消息发送出错,请检查设备是否启用及状态正常!")
        # 两秒后再次发送
        time.sleep(2)


def tasks_quit():
    "程序退出功能"
    print("success: tasks_quit Thread is running!")
    exitright = "e"
    while exitright not in ["q", "Q"]:
        # 获取用户输入,如果为q则退出程序
        exitright = input(
            """
***********************************
**输入字母q后,按下回车以退出程序**
***********************************
"""
        )
        # 线程退出


# 打印运行程序前提示信息
print(
    "information: 执行本程序前,请先启用can设备。命令如下:\
    \nsudo ip link set can0 type can bitrate 1000000\nsudo ip link set can0 up"
)
# 打开CAN设备,CAN设备类型为socketcan,channel为can0,可使用ifconfig -a命令查看。
with can.interface.Bus(
    bustype="socketcan", channel="can0", bitrate=1000000
) as device_can0:
    # 创建线程:监听程序退出线程、发送can消息线程、接收can消息线程
    try:
        print("information: 开始创建 tasks_quitThread 线程!")
        tasks_quitThread = threading.Thread(target=tasks_quit, daemon=True)
        print("information: 开始创建 msg_sendThread 线程!")
        msg_sendThread = threading.Thread(
            target=msg_send, daemon=True, args=(device_can0,)
        )
        print("information: 开始创建 msg_recvThread 线程!")
        msg_recvThread = threading.Thread(
            target=msg_recv, daemon=True, args=(device_can0,)
        )
        # 开启线程
        print("information: 开始启动 tasks_quitThread 线程!")
        tasks_quitThread.start()
        print("information: 开始启动 msg_sendThread 线程!")
        msg_sendThread.start()
        print("information: 开始启动 msg_recvThread 线程!")
        msg_recvThread.start()
    except:
        print("error: 创建或启动线程中出错!")
        sys.exit()

    # 等待线程结束
    tasks_quitThread.join()
    print("information: tasks_quitThread结束")
    msg_sendThread.join()
    print("information: msg_sendThread结束")
    msg_recvThread.join()
    print("information: msg_recvThread结束")
    # 所有正常线程结束,退出程序
    sys.exit()

代码功能已经在注释中详细给出。

8.4.1. 实验步骤

在测试代码前,我们需要将CAN总线设备启动,在前面我们说过Linux下CAN设备被当成网络设备处理。

CAN总线设备未启动时,状态为 NOARP

broken

所以我们可以输入如下命令来启用CAN总线设备:

# 在终端中输入如下命令,可以启用CAN总线设备:
sudo ip link set can0 type can bitrate 1000000;sudo ip link set can0 up

该命令的作用是设置CAN设备的通讯速率为1Mbps,并启动CAN设备。

重要

CAN设备的波特率以终端输入命令配置的为准!Python代码中配置的波特率需要与终端输入保持一致! 否则可能会出现意料之外的效果。

启动后,状态为 UP,RUNNING,NOARP ,图如下:

broken

下面我们开始运行测试代码。请先将配套代码文件夹io\canbus_test\下,再运行程序。

编写的测试代码示例中,我们的CAN设备会监听CAN总线上所有的CAN标准帧,监听帧头为0x300的拓展帧。

所以我们可以运行代码来查看一下现象。

# 在终端中输入如下命令:
python3 canbus_test.py

运行程序后,终端打印出程序运行信息:

broken

程序里创建了三个线程,分别用于监控程序退出、间隔2秒发送一次CAN消息到总线上、 监听总线上对应的数据。

程序运行后,就可以在CAN分析仪中查看到对应的数据包了:

broken

下面我们尝试使用CAN分析仪发送CAN数据包到总线上,看我们鲁班猫板卡上的CAN设备是否会监听到消息。

实验现象如下(点击图片可放大):

broken

根据图片中的文字说明,大家就可以知道,实现现象与我们的程序设计是一致的了!

更多关于python-can库的使用,可参考 python-can库说明文档 。 还可以上作者的Github仓库学习更多内容, python-can Github