系列文章目录

【Python搞定车载自动化测试】系列文章目录汇总

---

💯💯💯 前言💯💯💯

在之前的专栏【如何学习CAN总线测试】中介绍了如何通过CAPL语言实现Bootloader刷写，但CAPL语言依赖于Vector CANoe设备，随着智能电动汽车行业的发展，目前也普遍使用Python语言实现Bootloader刷写，本章节主要介绍如何使用Python语言来实现CAN总线UDS Bootloader刷写。
首先回顾一下Bootloader刷写基础知识：

UDS（Unified Diagnostic Services）是一种汽车行业内广泛使用的诊断标准，用于车辆电子控制单元（ECU）的诊断和编程。当提到“UDS Bootloader刷写”，通常是指利用UDS协议对车辆ECU中的Bootloader进行更新或重新编程的过程。这一过程允许通过标准化的服务和子功能访问ECU内部的软件组件，包括其Bootloader，从而实现安全且可靠的远程更新能力。以下是UDS Bootloader刷写的简要流程：


通信建立：
物理连接：首先，需要通过诊断接口（如OBD-II接口）将车辆与编程工具（或编程工作站）相连。
建立诊断会话：使用UDS的会话管理服务（如Diagnostic Session Control service）建立安全或编程会话。


安全访问：
为了保护ECU不被非法访问，刷写前通常需要通过UDS的安全访问服务（Security Access service）进行鉴权，这可能涉及密码或密钥交换。


刷写准备：
读取当前状态：使用Read Data by Identifier (0x22) 或 Request Seed (0x27) 等服务获取ECU当前状态和刷写条件。
编程模式激活：通过Programming Enable (0x2E) 服务使ECU进入可编程模式。


数据传输：
分区擦除：如果需要，先发送请求擦除指定Flash区域的命令（如Request Download (0x34)）。
Bootloader传输：使用Transfer Data (0x36) 服务将新的Bootloader二进制文件分块传输至ECU。


验证与激活：
数据校验：传输完成后，使用Request Transfer Exit (0x37) 结束传输，并通过Control Module Programming (0x31) 服务执行完整性校验。
激活新Bootloader：确认无误后，通过专门的服务命令激活新刷写的Bootloader，可能需要复位ECU。


会话结束与验证：
完成刷写后，退出编程会话，回到正常工作模式，并通过诊断测试验证ECU及新Bootloader的功能完整性。
整个过程中，严格遵循UDS协议的指令和服务，确保操作的可靠性和安全性，防止因错误操作导致ECU功能异常。此外，由于涉及到车辆安全和稳定性，此类操作通常由专业技术人员在符合制造商规范的条件下执行。

---

一、环境搭建

1.软件环境

Python版本：python3.9
第三方库：
pip install allure-pytest2.13.5
pip install can-isotp2.0.4
pip install python-can4.3.1
pip install udsoncan1.23.0
allure：安装allure工具并设置环境变量，https://github.com/allure-framework/allure2/releases

2.硬件环境

支持CAN设备硬件接口：

https://python-can.readthedocs.io/en/stable/configuration.html

二、目录结构

dll目录：存放27服务安全解锁DLL文件（同CANoe中使用的DLL文件）。
public_method目录：存放公共函数方法和27安全解锁工具。
test_case目录：存放python自动化测试用例。
update目录：存放升级包和效验文件。
config.py文件：CAN相关配置参数。
run.py文件：运行入口文件。

三、源码展示

1.诊断基础函数方法

base_can.py文件主要封装了CAN初始化、诊断配置、诊断请求、诊断响应基础方法，源码如下：

class CanBus:def __init__(self, interface: str = None, channel: int = None, bitrate: int = None, fd: bool = None,data_bitrate: int = None, can_filters: CanFilters = None, *args, **kwargs):self.interface = interfaceself.channel = channelself.bitrate = bitrateself.fd = fdself.data_bitrate = data_bitrateself.can_filters = can_filterstry:self.bus = can.interface.Bus(channel=self.channel, interface=self.interface, app_name="CANoe",bitrate=self.bitrate, fd=self.fd, data_bitrate=self.data_bitrate,can_filters=self.can_filters, *args, **kwargs)except Exception as e:raise Exception("初始化失败：%s" % e)else:print("初始化成功")def diag_congfig(self, tx: int, rx: int, addressingmode=isotp.AddressingMode.Normal_11bits):"""诊断配置函数:param tx: 诊断请求ID，功能寻址、物理寻址:param rx: 诊断响应ID:return:"""self.isotp_params = {'stmin': 20,  # 流控帧间隔时间'blocksize': 8,  # 流控帧单包大小，0表示不限制'tx_padding': 0,  # 当 notNone表示用于填充发送的消息的字节。'rx_flowcontrol_timeout': 1000,  # 在停止接收和触发之前等待流控制帧的毫秒数'rx_consecutive_frame_timeout': 1000,  # 在停止接收和触发 a 之前等待连续帧的毫秒数}try:self.tp_addr = isotp.Address(addressing_mode=addressingmode, txid=tx, rxid=rx)  # 网络层寻址方案tp_stack = isotp.CanStack(bus=self.bus, address=self.tp_addr, params=self.isotp_params)  # 网络/传输层（IsoTP 协议）self.conn = PythonIsoTpConnection(tp_stack)  # 应用层和传输层之间的接口except Exception as e:print("UDS配置失败:%s" % e)return self.conndef diag_request(self, request_command: str, request_data_log_flag=True):"""诊断请求"""requestPdu = binascii.a2b_hex(request_command.replace(' ', ''))if not self.conn.is_open():self.conn.open()try:self.conn.send(requestPdu)except Exception as e:print("诊断请求失败:%s" % e)else:req_info = ''request_command = request_command.replace(' ', '')for i in range(len(request_command)):if i >= len(request_command) / 2:breakreq_info += request_command[2 * i:2 * i + 2] + ' 'if request_data_log_flag:print("诊断请求：%s" % req_info)def diag_respond(self, timeout1=1):"""诊断响应"""try:respPdu = self.conn.wait_frame(timeout=timeout1)except Exception as e:print(e)else:if respPdu is None:return Noneresp1 = respPdu.hex().upper()resp2 = ''for i in range(len(resp1)):if i != 0 and i % 2 == 0:resp2 += ' 'resp2 += resp1[i]print("诊断响应：%s" % resp2)return resp2

2.诊断业务函数方法

fun_can.py主要二次封装UDS诊断函数，包括：27安全解锁，34服务、36服务、诊断78响应处理、UDS诊断测试、CRC效验等函数，源码如下：

import binascii
import os
import subprocess
import timeimport pytestfrom config import Parameter
from public_method.base_can import CanBusclass CanMethod(CanBus):def __init__(self, config):self.interface = config['can']['interface']self.channel = config['can']['channel']self.bitrate = config['can']['bitrate']self.fd = config['can']['canfd']self.data_bitrate = config['can']['data_bitrate']self.addressingmode = config['can']['addressing_mode']self.tx = config['can']['physics_id_default']self.rx = config['can']['response_id_default']CanBus.__init__(self, interface=self.interface, channel=self.channel, bitrate=self.bitrate, fd=self.fd,data_bitrate=self.data_bitrate, )self.diag_congfig(addressingmode=self.addressingmode, tx=self.tx, rx=self.rx)self.sign_nrc78 = 0def __diag_get_seed(self, req_data="27 01"):"""27服务获取种子并解析"""self.diag_request(req_data)try:uds_res_data = self.conn.specific_wait_frame(timeout=2)while uds_res_data[0] == 0x7F and uds_res_data[2] == 0x78:print("已收到 %d bytes : [%s]" % (len(uds_res_data), binascii.hexlify(uds_res_data)))uds_res_data = self.conn.specific_wait_frame(timeout=3)resp1 = uds_res_data.hex().upper()resp2 = ''for i in range(len(resp1)):if i != 0 and i % 2 == 0:resp2 += ' 'resp2 += resp1[i]print("诊断响应：%s" % resp2)except:print("响应数据失败")else:seed = []res_seed = resp2.split(' ')[2:]for i in range(len(res_seed)):seed.append(eval('0x' + res_seed[i]))print("seed：%s" % seed)return seeddef get_key_level(self, seed):"""dll_security_unlock.exe工具解锁语法：dll_security_unlock.exe --dll_path dome.dll --seed [11,22,33,44] --seedsize 4 --level 1 --keysize 4--dll_path DLL路径--seed 请求种子--seedsize 种子长度--level 安全级别--keysize 秘钥长度"""seed = str(seed).replace(' ', '')tool = os.path.join(os.path.dirname(__file__), 'dll_security_unlock.exe')cmd = '{} --dll_path {} --seed {}'.format(tool, os.path.join(os.path.dirname(os.path.dirname(__file__)), r'dll\dome.dll'), seed)key = subprocess.getoutput(cmd)return keydef unlock_level(self):"""27安全解锁"""seed = self.__diag_get_seed(req_data="27 01")if seed is not None:if seed != 0 and len(seed) > 1:key = self.get_key_level(seed)print("key= %s" % key)req_data = "27 02 %s" % keyself.diag_request(req_data)self.uds_respond_0x78()time.sleep(0.1)else:print("seed响应不正确")def diag_send(self, req_data="3E 80"):"""发送诊断请求，不断言诊断响应"""self.diag_request(req_data)response = self.uds_respond_0x78()time.sleep(0.1)return responsedef diag_send_exp(self, req_data="3E 80", exp=None):"""发送诊断请求，并断言诊断响应"""self.diag_request(req_data)result = self.__diag_test_response_judge(exp=exp)time.sleep(0.1)return resultdef diag_send_0x34(self, req_data="34 00 44", address="00 00 00 00", size=0, exp=None):"""刷写时使用，请求传输升级包"""print("传输包大小= %s" % size)self.diag_request(req_data + address + "{:08X}".format(size))self.__diag_test_response_judge(exp=exp)time.sleep(0.1)def diag_send_0x36(self, req_data="36", trans_size=255, path="", exp=None):"""36服务传包"""total_size = os.path.getsize(path)print("size = %s" % total_size)with open(path, "rb") as f:file_read = f.read()print("CRC= %s" % "{:02X}".format(binascii.crc32(file_read)))file_byte = []for i in range(len(file_read)):file_byte.append("{:02X}".format(file_read[i]))sequence = 1transmitted_size = 0try:status = Truewhile status:trans_data = ""for i in range(trans_size):if transmitted_size < total_size:trans_data = trans_data + file_byte[transmitted_size]transmitted_size = transmitted_size + 1else:status = Falsebreakprint("data_num=%s" % transmitted_size)self.diag_request(request_command=req_data + "{:02X}".format(sequence) + trans_data,request_data_log_flag=False,)print(req_data + "{:02X}".format(sequence) + "...")self.__diag_test_response_judge(exp=exp)sequence += 1if sequence == 256:sequence = 0finally:print("36传输结束")def diag_crc32(self, req_data="31 01 02 02", path="", exp=None):"""刷写时使用，CRC32校验"""size = os.path.getsize(path)print("size = %s" % size)with open(path, "rb") as f:file_read = f.read()crc32 = "{:08X}".format(binascii.crc32(file_read))print("crc 32 = %s " % crc32)self.diag_send_exp(req_data=req_data + crc32, exp=exp)def __diag_session_mode(self, session):"""诊断会话模式"""if session == "01":self.diag_send(req_data="10 01")elif session == "03":self.diag_send(req_data="10 03")elif session == "02":self.diag_send(req_data="10 03")self.unlock_level()self.diag_send(req_data="10 02")def uds_respond_0x78(self, timeout1=2):"""78响应处理"""response = self.diag_respond(timeout1=timeout1)if response is not None:try:response2 = response.replace(' ', '')cyc = 0while response2[:2] == '7F' and response2[4:6] == '78':self.sign_nrc78 = 1response = self.diag_respond(timeout1=timeout1)if response is not None:response2 = response.replace(' ', '')cyc += 1if cyc > 20:breakexcept Exception as e:print("异常:%s" % e)return responsedef __diag_test_response_judge(self, exp=None):"""断言响应结果与预期结果是否一致"""response = self.uds_respond_0x78()response_return = responseif (exp is not None) & (response is not None):exp = exp.replace(" ", "").upper()exp2 = ""for i in range(len(exp)):if i != 0 and i % 2 == 0:exp2 += " "exp2 += exp[i]exp = exp2if len(exp) < len(response):response = response[0: len(exp)]if response == exp:return response_returnelse:print("诊断结果与预期结果不匹配")pytest.fail("诊断结果与预期结果不匹配")def diag_test(self, session="01", req_data=None, exp=None):"""诊断测试:param session: 执行前会话模式，01默认会话，02编程会话，03扩展会话:param req_data:请求数据:param exp:预期结果:return:"""self.__diag_session_mode(session=session)if req_data is not None:self.diag_request(req_data)self.__diag_test_response_judge(exp=exp)uds = CanMethod(Parameter.config)

3.27服务安全解锁

dll_security_unlock.exe文件可实现DLL安全解锁，使用方法如下：
语法：
举例：dll_security_unlock.exe --dll_path dome.dll --seed [11,22,33,44] --seedsize 4 --level 1 --keysize 4
–dll_path DLL路径
–seed 请求种子
–seedsize 种子长度
–level 安全级别
–keysize 秘钥长度

Bootloader_399">4.Bootloader刷写

test_uds.py主要是自动化测试用例举例，包括10服务测试、11服务测试、14服务测试、19服务测试、22服务测试、28服务测试、31服务测试、85服务测试等，源码如下：

import os
import timeimport allurefrom public_method.fun_can import udsclass TestBootloader:@allure.title("Bootloader刷写")def test_diag_Bootloader(self):print("#####Bootloader刷写#####")path_driver=r'update\driver.bin'path_app=r"update\001_app.bin"uds.diag_send_exp(req_data='10 03', exp='50 03')time.sleep(0.2)uds.diag_send_exp(req_data='31 01 02 03', exp='71 01 02 03')time.sleep(0.2)uds.diag_send(req_data='85 82')time.sleep(1)uds.diag_send(req_data='28 81 01')time.sleep(1)uds.unlock_level()time.sleep(0.2)uds.diag_send_exp(req_data='10 02', exp='50 02')time.sleep(0.2)uds.diag_send_0x34(req_data='34 00 44 ', address='00 00 00 01 ', size=os.path.getsize(path_driver),exp='74')time.sleep(0.2)uds.diag_send_0x36(req_data='36', path=path_driver, trans_size=1024, exp='76')time.sleep(0.2)uds.diag_send_exp(req_data='37', exp='77')time.sleep(0.2)uds.diag_crc32(req_data='31 01 02 02 ', path=path_driver, exp='71 01 02 02')time.sleep(0.2)uds.diag_send_exp(req_data='31 01 ff 00 44 00 00 00 01 00 00 00 02', exp='71 01 ff 00')time.sleep(0.2)uds.diag_send_0x34(req_data='34 00 44 ', address='00 00 00 01 ', size=os.path.getsize(path_app), exp='74')time.sleep(0.2)uds.diag_send_0x36(req_data='36',path=path_app, trans_size=102400, exp='76')time.sleep(0.2)uds.diag_send_exp(req_data='37', exp='77')time.sleep(0.2)uds.diag_crc32(req_data='31 01 02 02 ', path=path_app, exp='71 01 02 02')time.sleep(1)uds.diag_send_exp(req_data='31 01 ff 01', exp='71 01 ff 01')time.sleep(0.2)uds.diag_send_exp(req_data='11 01', exp='51 01')print('重启中,等待5分钟..')time.sleep(300)uds.diag_send(req_data='10 83')time.sleep(1)uds.diag_send(req_data='28 80 01')time.sleep(1)uds.diag_send(req_data='85 81')

5.配置参数

config主要配置CAN和诊断相关的参数：
interface：配置can设备类型（支持python-can三方库的设备）
channel：通道
bitrate：波特率
addressing_mode：数据比特率
physics_id_default：物理寻址
response_id_default：响应寻址

class Parameter():"""CAN参数配置"""config = {"can": {"interface": "vector","channel": 0,"bitrate": 500000,"data_bitrate": 2000000,"canfd": False,  # 是否canfd"addressing_mode": 0,"physics_id_default": 0x56A,"response_id_default": 0x56B,"function_id_default": 0x56C,}}

四、测试报告

五、完整源码链接

如下载源码链接失效，请将购买专栏截图和用户名截图通过CSDN私信发送给博主，博主更新源码链接：
链接：https://pan.baidu.com/s/1EIx0upnVz-ZiudXE9Ki8Bg
提取码：4kdj