系列文章目录
【Python搞定车载自动化测试】系列文章目录汇总
文章目录
- 系列文章目录
- 💯💯💯 前言💯💯💯
- 一、环境搭建
- 1.软件环境
- 2.硬件环境
- 二、目录结构
- 三、源码展示
- 1.诊断基础函数方法
- 2.诊断业务函数方法
- 3.27服务安全解锁
- 4.Bootloader刷写
- 5.配置参数
- 四、测试报告
- 五、完整源码链接
💯💯💯 前言💯💯💯
在之前的专栏【如何学习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 秘钥长度
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