代码收藏家技术教程 2023-07-05

使用Python进行UDS诊断操作

主要是通过python-can模块与pcan等支持的硬件通讯，uds协议层使用udsoncan模块和can-isotp模块实现uds诊断。

1、模块安装及相关文档

python-can模块

pip install python-can

相关文档链接：Installation – python-can 4.1.0 documentation

udsoncan模块

pip install udsoncan

can-isotp模块

pip install can-isotp

2、相关示例

下面示例展示了如何将PythonIsoTpConnection与Vector接口一起使用。

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

# 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 = VectorBus(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
with Client(conn, request_timeout=1) as client:                                     # Application layer (UDS protocol)
   client.change_session(1)
   # ...

其他uds服务相关使用示例：

import udsoncan
from udsoncan.connections import IsoTPSocketConnection
from udsoncan.client import Client
from udsoncan.exceptions import *
from udsoncan.services import *

udsoncan.setup_logging()

conn = IsoTPSocketConnection('can0', rxid=0x123, txid=0x456)
with Client(conn,  request_timeout=2, config=MyCar.config) as client:
   try:
      client.change_session(DiagnosticSessionControl.Session.extendedDiagnosticSession)  # integer with value of 3
      client.unlock_security_access(MyCar.debug_level)   # Fictive security level. Integer coming from fictive lib, let's say its value is 5
      client.write_data_by_identifier(udsoncan.DataIdentifier.VIN, 'ABC123456789')       # Standard ID for VIN is 0xF190. Codec is set in the client configuration
      print('Vehicle Identification Number successfully changed.')
      client.ecu_reset(ECUReset.ResetType.hardReset)  # HardReset = 0x01
   except NegativeResponseException as e:
      print('Server refused our request for service %s with code "%s" (0x%02x)' % (e.response.service.get_name(), e.response.code_name, e.response.code))
   except InvalidResponseException, UnexpectedResponseException as e:
      print('Server sent an invalid payload : %s' % e.response.original_payload)

安全算法示例，需要实现myalgo函数，并且uds的configs中的security_algo配置为myalgo，同时配置security_algo_params。

def myalgo(level, seed, params):
"""
Builds the security key to unlock a security level. Returns the seed xor'ed with pre-shared key.
"""
   output_key = bytearray(seed)
   xorkey = bytearray(params['xorkey'])

   for i in range(len(seed)):
      output_key[i] = seed[i] ^ xorkey[i%len(xorkey)]
   return bytes(output_key)

client.config['security_algo'] = myalgo
client.config['security_algo_params'] = dict(xorkey=b'\x12\x34\x56\x78')

使用DID配置配置客户端，并使用ReadDataByIdentifier请求服务器

import udsoncan
from udsoncan.connections import IsoTPSocketConnection
from udsoncan.client import Client
import udsoncan.configs
import struct

class MyCustomCodecThatShiftBy4(udsoncan.DidCodec):
   def encode(self, val):
      val = (val << 4) & 0xFFFFFFFF # Do some stuff
      return struct.pack('<L', val) # Little endian, 32 bit value

   def decode(self, payload):
      val = struct.unpack('<L', payload)[0]  # decode the 32 bits value
      return val >> 4                        # Do some stuff (reversed)

   def __len__(self):
      return 4    # encoded payload is 4 byte long.


config = dict(udsoncan.configs.default_client_config)
config['data_identifiers'] = {
   0x1234 : MyCustomCodecThatShiftBy4,    # Uses own custom defined codec. Giving the class is ok
   0x1235 : MyCustomCodecThatShiftBy4(),  # Same as 0x1234, giving an instance is good also
   0xF190 : udsoncan.AsciiCodec(15)       # Codec that read ASCII string. We must tell the length of the string
   }

# IsoTPSocketconnection only works with SocketCAN under Linux. Use another connection if needed.
conn = IsoTPSocketConnection('vcan0', rxid=0x123, txid=0x456)
with Client(conn,  request_timeout=2, config=config) as client:
   response = client.read_data_by_identifier([0xF190])
   print(response.service_data.values[0xF190]) # This is a dict of DID:Value

   # Or, if a single DID is expected, a shortcut to read the value of the first DID
   vin = client.read_data_by_identifier_first(0xF190)
   print(vin)  # 'ABCDE0123456789' (15 chars)