【Python搞定车载自动化测试】——Python实现CAN总线Bootloader刷写（含Python源码）

系列文章目录

【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 = interface
        self.channel = channel
        self.bitrate = bitrate
        self.fd = fd
        self.data_bitrate = data_bitrate
        self.can_filters = can_filters
        try:
            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.conn

    def 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:
                    break
                req_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 None
            resp1 = 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 time

import pytest

from config import Parameter
from public_method.base_can import CanBus


class 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 = 0

    def __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 seed

    def 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 key

    def 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" % key
                self.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 response

    def 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 result

    def 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 = 1
        transmitted_size = 0
        try:
            status = True
            while 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 + 1
                    else:
                        status = False
                        break
                print("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 += 1
                if sequence == 256:
                    sequence = 0
        finally:
            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 = 0
                while response2[:2] == '7F' and response2[4:6] == '78':
                    self.sign_nrc78 = 1
                    response = self.diag_respond(timeout1=timeout1)
                    if response is not None:
                        response2 = response.replace(' ', '')
                    cyc += 1
                    if cyc > 20:
                        break
            except Exception as e:
                print("异常:%s" % e)
        return response

    def __diag_test_response_judge(self, exp=None):
        """
        断言响应结果与预期结果是否一致
        """
        response = self.uds_respond_0x78()
        response_return = response
        if (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 = exp2
            if len(exp) < len(response):
                response = response[0: len(exp)]
        if response == exp:
            return response_return
        else:
            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 time

import allure

from public_method.fun_can import uds


class 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,
        }
    }