1、Python-CAN基本

python-can 库为 Python 提供控制器局域网支持，为不同的硬件设备提供通用抽象，以及一套用于在 CAN 总线上发送和接收消息的实用程序。

python-can 可以在任何 Python 运行的地方运行；从 CAN 的高功率计算机到 USB 设备，再到运行 linux 的低功率设备，例如 BeagleBone 或 RaspberryPi。

更具体地说，该库的一些示例用途：

被动记录 CAN 总线上发生的情况。例如，使用 OBD-II 端口监控商用车辆。
测试通过 CAN 交互的硬件。在现代汽车、摩托车、船只甚至轮椅中发现的模块已经使用这个库从 Python 中测试了组件。
在回路中对新的硬件模块或软件算法进行原型设计。轻松与现有总线交互。
创建虚拟模块以原型 CAN 总线通信。

2、python-can 安装

使用pip安装

pip install python-can

同时根据使用的硬件设备需要安装对应的驱动等，参照Installation — python-can 4.0.0 documentation

3、配置

3.1 代码中直接配置

can对象公开了一个rc字典，可用于设置interface和channel

import can
can.rc['interface'] = 'neovi'  # 配置硬件类型
can.rc['channel'] = 2        # 配置通道，根据具体的硬件，int或者str
can.rc['bitrate'] = 500000   # 波特率


from can.interface import Bus

bus = Bus()   # 使用rc字典中的配置实例化can

也可以在代码中直接指定接口和通道实例化can

import can

bus = can.interface.Bus(bustype='socketcan', channel='vcan0', bitrate=500000)

bus2 = bus = can.interface.Bus(bustype='pcan', channel='PCAN_USBBUS1', bitrate=250000)

3.2 使用配置文件配置

在 Linux 系统上，配置文件在以下路径中搜索：

~/can.conf
/etc/can.conf
$HOME/.can
$HOME/.canrc

在 Windows 系统上，配置文件在以下路径中搜索：

%USERPROFILE%/can.conf
can.ini（当前工作目录）
%APPDATA%/can.ini

ps：%USERPROFILE%目录为 C:\Users\用户名

%APPDATA% 目录为 C:\Users\用户名\AppData\Roaming

配置文件可设置默认接口和通道如下：

[default]
interface = <the name of the interface to use>
channel = <the channel to use by default>
bitrate = <the bitrate in bits/s to use by default>

同时也可以添加其他节点如下：

[HS]
# All the values from the 'default' section are inherited
channel = <the channel to use>
bitrate = <the bitrate in bits/s to use. i.e. 500000>

[MS]
# All the values from the 'default' section are inherited
channel = <the channel to use>
bitrate = <the bitrate in bits/s to use. i.e. 125000>

使用配置文件的配置实例化can

from can.interface import Bus

bus = Bus()    # 使用默认（default）配置
hs_bus = Bus(context='HS')        # 使用HS节点配置
ms_bus = Bus(context='MS')        # 使用MS节点配置

3.3 使用环境变量配置

可以设置以下环境变量

CAN_INTERFACE 设置接口类型
CAN_CHANNEL 设置通道
CAN_BITRATE 设置波特率
CAN_CONFIG

CAN_CONFIG允许使用 JSON 设置任何总线配置，例如：

CAN_CONFIG={"receive_own_messages": true, "fd": true}

4、基本使用

4.1发送单帧报文到总线

实例化一条bus总线，使用can.Message()类创建一条message，bus调用send函数将创建的message实例发送到总线。

import can

def send_one():

    # this uses the default configuration (for example from the config file)
    # see https://python-can.readthedocs.io/en/stable/configuration.html
    #bus = can.interface.Bus()

    # Using specific buses works similar:
    bus = can.interface.Bus(bustype='bmcan', channel=0, bitrate=500000, data_bitrate=2000000, tres=True)
    # bus = can.interface.Bus(bustype='socketcan', channel='vcan0', bitrate=250000)
    # bus = can.interface.Bus(bustype='pcan', channel='PCAN_USBBUS1', bitrate=250000)
    # bus = can.interface.Bus(bustype='ixxat', channel=0, bitrate=250000)
    # bus = can.interface.Bus(bustype='vector', app_name='CANalyzer', channel=0, bitrate=250000)
    # ...

    msg = can.Message(arbitration_id=0xc0ffee,
                      data=[0, 25, 0, 1, 3, 1, 4, 1],
                      is_extended_id=True, )

    try:
        bus.send(msg, timeout=None)
        print("Message sent on {}".format(bus.channel_info))
    except can.CanError:
        print("Message NOT sent")

    bus.shutdown()

if __name__ == '__main__':
    send_one()

4.2发送周期报文到总线

以下实例分别演示简单发送周期报文，发送定时的周期报文，和发送周期报文过程中修改发送的内容。

注意：在发送定时的周期报文时，达到超时时间任务结束后，task将仍旧由总线跟踪。除非在创建任务时设置参数store_task=False，或者调用stop()函数结束任务task.stop(）。

import time

import can



def simple_periodic_send(bus):
    """
    发送一条周期为200ms的周期报文到总线
    2s后停止发送报文
    """
    print("Starting to send a message every 200ms for 2s")
    msg = can.Message(arbitration_id=0x123, data=[1, 2, 3, 4, 5, 6], is_extended_id=False)
    task = bus.send_periodic(msg, 0.20)
    assert isinstance(task, can.CyclicSendTaskABC)
    time.sleep(2)
    task.stop()
    print("stopped cyclic send")


def limited_periodic_send(bus):
    """
    发送一条有时间限制的周期报文到总线，达到超时时间后，停止任务
    """
    print("Starting to send a message every 200ms for 1s")
    msg = can.Message(arbitration_id=0x12345678, data=[0, 0, 0, 0, 0, 0], is_extended_id=True)
    task = bus.send_periodic(msg, 0.20, 1, store_task=False)
    if not isinstance(task, can.LimitedDurationCyclicSendTaskABC):
        print("This interface doesn't seem to support a ")
        task.stop()
        return

    time.sleep(2)


def test_periodic_send_with_modifying_data(bus):
    """
    在发送周期报文过程中修改发送的数据
    """
    print("Starting to send a message every 200ms. Initial data is ones")
    msg = can.Message(arbitration_id=0x0cf02200, data=[1, 1, 1, 1])
    task = bus.send_periodic(msg, 0.20)
    if not isinstance(task, can.ModifiableCyclicTaskABC):
        print("This interface doesn't seem to support modification")
        task.stop()
        return
    time.sleep(2)
    print("Changing data of running task to begin with 99")
    msg.data[0] = 0x99
    task.modify_data(msg)
    time.sleep(2)

    task.stop()
    print("stopped cyclic send")
    print("Changing data of stopped task to single ff byte")
    msg.data = bytearray([0xff])
    msg.dlc = 1
    task.modify_data(msg)
    time.sleep(1)
    print("starting again")
    task.start()
    time.sleep(1)
    task.stop()
    print("done")

4.3接收总线的can报文

以下示例演示使用recv()函数接收来自总线的can 报文，然后打印报文。

import can

def receive_all():

    bus = can.interface.Bus(bustype='ixxat', channel=0, bitrate=500000)

    print('Waiting for RX CAN messages ...')
    try:
        while True:
            msg = bus.recv(1)
            if msg is not None:
                print(msg)
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":
    receive_all()

待续。。。

本文含有隐藏内容，请开通VIP 后查看

python-can介绍