零基础快速实现pyhton 控制CAN通信UDS诊断功能

python-can · PyPI

环境准备

硬件准备

TYPE-C
USBCAN
鲁班猫zero.N开发板

硬件框架

微信截图_20230416193433.png

软件准备

编程语言python
使用python库：
python-can
canalystii
can-isotp
udsoncan
命令行安装库
pip install python-can -i https://pypi.tuna.tsinghua.edu.cn/simple

注意:使用sudo 就都是用要不就都不用，因为使用sudo会按装到root用户对应目录下

功能实现

完成上述软件安装后就可以使用我们的安装的库驱动USBCAN实现我们想要的功能是不是发现很容易。

1.基础CAN通信功能

使用到的库：python-can,canalystii

import can

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

msg = can.Message(arbitration_id=0x181, data=[0xEE, 0xFE, 0xFE, 0xFF, 0xFE, 0xFF, 0xFF, 0xFE],
is_extended_id=False)
bus.send(msg)
bus.shutdown()

通过上述代码既通过USBCAN发送报文到ECU。
接下来我们就可以通过这个库实现复杂的功能。例如周期发送报文只需一个函数即可，多个周期报文发送等等。
CAN的更多参数设置方法可以通过在线文档查找对应方法。
参考库官方在线文档：python-can · PyPI

此库不仅可以用来驱动USBCAN还可以来驱动vector、ETAS、PCAN等设备。

2.多帧通信验证

使用到的库：can-isotp （ISO-15765）

import isotp
import logging
import time

import can

ESC_TX_ID = 0x73E
ESC_RX_ID_PHYS = 0x736
ESC_RX_ID_FUNC = 0x7DF

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

isotpaddr_PHYS = isotp.Address(isotp.AddressingMode.Normal_11bits, txid=ESC_RX_ID_PHYS, rxid = ESC_TX_ID)
isotpaddr_FUNC = isotp.Address(isotp.AddressingMode.Normal_11bits, txid=ESC_RX_ID_FUNC, rxid = ESC_TX_ID)

stack = isotp.CanStack(bus, address=isotpaddr_PHYS, error_handler=my_error_handler)
stack.send(b'Hello, this is a long payload sent in small chunks')

while stack.transmitting():
   stack.process()
   time.sleep(stack.sleep_time())

print("Payload transmission done.")

bus.shutdown()

效果如图：

本验证是通过两个USBCAN实现。如模拟ECU端不发送应答诊，程序会报告详细的警告信息。

通过上述代码既通过USBCAN发送多帧报文到ECU。
参数等设置可以通过在线文档查找到函数
参考库官方在线文档：Python support for IsoTP Transport protocol (ISO-15765) — isotp 1.8 documentation (can-isotp.readthedocs.io)

3.诊断功能验证

# from can.interfaces.vector import VectorBus
import can
import isotp
import udsoncan
from udsoncan.connections import PythonIsoTpConnection
from udsoncan.client import Client
import udsoncan.configs


# Refer to isotp documentation for full details about parameters
isotp_params = {
   'stmin' : 32,                          # Will request the sender to wait 32ms between consecutive frame. 0-127ms or 100-900ns with values from 0xF1-0xF9
   'blocksize' : 8,                       # Request the sender to send 8 consecutives frames before sending a new flow control message
   'wftmax' : 0,                          # Number of wait frame allowed before triggering an error
   'tx_data_length' : 8,                  # Link layer (CAN layer) works with 8 byte payload (CAN 2.0)
   'tx_data_min_length' : None,           # Minimum length of CAN messages. When different from None, messages are padded to meet this length. Works with CAN 2.0 and CAN FD.
   'tx_padding' : 0,                      # Will pad all transmitted CAN messages with byte 0x00.
   'rx_flowcontrol_timeout' : 1000,       # Triggers a timeout if a flow control is awaited for more than 1000 milliseconds
   'rx_consecutive_frame_timeout' : 1000, # Triggers a timeout if a consecutive frame is awaited for more than 1000 milliseconds
   'squash_stmin_requirement' : False,    # When sending, respect the stmin requirement of the receiver. If set to True, go as fast as possible.
   'max_frame_size' : 4095                # Limit the size of receive frame.
}

bus = can.interface.Bus(bustype='canalystii', channel=0, bitrate=500000)            # Link Layer (CAN protocol)
tp_addr = isotp.Address(isotp.AddressingMode.Normal_11bits, txid=0x123, rxid=0x456) # Network layer addressing scheme
stack = isotp.CanStack(bus=bus, address=tp_addr, params=isotp_params)               # Network/Transport layer (IsoTP protocol)
stack.set_sleep_timing(0, 0)                                                        # Speed First (do not sleep)
conn = PythonIsoTpConnection(stack)                                                 # interface between Application and Transport layer

config = dict(udsoncan.configs.default_client_config)
config['data_identifiers'] = {
   0xF190 : udsoncan.AsciiCodec(15)       # Codec that read ASCII string. We must tell the length of the string
   }

with Client(conn, request_timeout=1, config=config) as client:                                     # Application layer (UDS protocol)
    response = client.read_data_by_identifier([0xF190])
    print(response.service_data.values[0xF190]) # This is a dict of DID:Value

bus.shutdown()