别再死记硬背了！用Python脚本模拟UDS $34/$36/$37诊断刷写，5分钟搞懂数据流
用Python脚本实战UDS诊断刷写：解密$34/$36/$37服务的数据流奥秘
在汽车电子开发领域，诊断刷写是ECU软件更新的核心技术之一。不同于枯燥的理论学习，本文将带您通过Python脚本构建完整的UDS诊断刷写流程，从数据块分割到序列号循环，从请求构造到响应解析，用可运行的代码还原$34/$36/$37服务的每一个数据交互细节。无论您是刚接触诊断协议的新手，还是希望提升自动化测试能力的工程师，这套脚本工具都能为您打开理解底层通信的新视角。
1. 环境准备与基础概念
1.1 硬件与软件配置
要运行本文的示例代码，您需要准备以下环境：
- CAN硬件接口：PCAN-USB、Kvaser或Vector等支持Python控制的CAN设备
- Python 3.7+环境：推荐使用Anaconda管理Python环境
- 关键Python库：
pip install python-can udsoncan cantools
python-can库提供了与不同CAN硬件的统一接口，udsoncan则是专门处理UDS协议的强大工具库。我们还将使用cantools来解析CAN数据库文件（DBC），但本文示例将聚焦在原始报文交互上。
1.2 UDS诊断刷写核心服务
UDS协议中，$34、$36、$37三个服务构成了诊断刷写的"铁三角"：
| 服务ID | 名称 | 核心功能 |
|---|---|---|
| 0x34 | RequestDownload | 确定单次传输的最大数据块长度 |
| 0x36 | TransferData | 实际数据传输，包含序列号和数据块 |
| 0x37 | RequestTransferExit | 优雅终止传输过程 |
这三个服务必须按严格顺序执行：先通过$34协商参数，然后循环$36传输数据，最后用$37结束会话。任何顺序错误都会导致ECU返回否定响应（Negative Response）。
2. 构建$34服务：协商传输参数
2.1 请求报文构造原理
$34服务的核心目标是确定ECU能接受的最大数据块长度。其请求报文包含三个关键参数：
- dataFormatIdentifier：通常为0x00，表示无压缩无加密
- addressAndLengthFormatIdentifier：高4位指定memorySize长度，低4位指定address长度
- memoryAddress和memorySize：定义刷写区域的起始地址和总大小
以下是构造$34请求的Python函数：
def build_request_download(address, size, addr_len=4, size_len=4): """构造$34服务请求报文 :param address: 刷写起始地址(int) :param size: 待传输数据总大小(int) :param addr_len: 地址字段字节数(默认4) :param size_len: 大小字段字节数(默认4) :return: 完整的请求报文(list) """ service_id = 0x34 data_format = 0x00 addr_format = (size_len << 4) | addr_len # 将地址和大小转换为字节数组 address_bytes = address.to_bytes(addr_len, 'big') size_bytes = size.to_bytes(size_len, 'big') return [service_id, data_format, addr_format] + list(address_bytes) + list(size_bytes)2.2 解析肯定响应
当ECU接受$34请求时，会返回0x74肯定响应，其中包含maxNumberOfBlockLength参数。这个值决定了后续$36服务每次能传输的最大数据量。以下是响应解析示例：
def parse_positive_response(response): """解析$34服务的肯定响应 :param response: ECU返回的报文(list) :return: 最大块长度(int) """ if response[0] != 0x74: raise ValueError("非$34服务肯定响应") length_format = response[1] max_len_bytes = response[2:2+(length_format >> 4)] return int.from_bytes(max_len_bytes, 'big')实际项目中，ECU可能返回否定响应（0x7F）。常见原因包括：
- 0x31：请求超出范围
- 0x33：安全访问未通过
- 0x72：一般编程错误
3. 实现$36服务：数据传输的艺术
3.1 块序列计数器机制
$36服务最精妙的设计在于blockSequenceCounter（块序列计数器），这个1字节的计数器从0x01开始递增，到达0xFF后不是归零而是继续从0x00开始循环。这种设计确保了数据传输的连续性检测。
序列号变化规律：
01 → 02 → ... → FE → FF → 00 → 01 → ... → FF → 00 → ...Python实现序列号生成器：
class BlockCounter: def __init__(self): self._counter = 0 @property def current(self): return self._counter def next(self): self._counter = (self._counter + 1) % 256 return self._counter3.2 数据分块与传输
根据$34服务返回的maxNumberOfBlockLength，我们需要将待刷写文件分割成适当大小的块。以下是分块传输的核心逻辑：
def transfer_data(can_bus, data, max_block_len, ecu_id): """执行$36服务数据传输 :param can_bus: CAN总线对象 :param data: 待传输数据(bytes) :param max_block_len: 每块最大长度 :param ecu_id: 目标ECU地址 """ counter = BlockCounter() total_blocks = (len(data) + max_block_len - 1) // max_block_len for i in range(total_blocks): start = i * max_block_len block = data[start: start+max_block_len] seq_num = counter.next() # 构造$36请求 request = [0x36, seq_num] + list(block) can_bus.send(ecu_id, request) # 等待并验证响应 response = can_bus.recv(timeout=1.0) if not validate_response(response, seq_num): raise RuntimeError(f"块{seq_num:02X}传输失败")响应验证函数需要检查两点：服务ID是否为0x76，以及返回的序列号是否匹配：
def validate_response(response, expected_seq): return (response[0] == 0x76 and response[1] == expected_seq)4. 终止传输：$37服务的实现
4.1 基本请求构造
$37服务通常是最简单的，许多ECU实现只需要发送服务ID即可：
def request_transfer_exit(): """构造$37服务请求""" return [0x37]4.2 响应处理与错误恢复
虽然$37请求简单，但其响应处理不容忽视。肯定响应应为0x77，但实际项目中可能遇到：
- 否定响应0x7F：可能表示之前的数据传输未完成
- 自定义参数：某些厂商要求在请求或响应中包含额外参数
健壮的处理逻辑应包含重试机制：
def exit_transfer(can_bus, ecu_id, max_retries=3): """执行$37服务并处理响应 :param can_bus: CAN总线对象 :param ecu_id: 目标ECU地址 :param max_retries: 最大重试次数 """ for attempt in range(max_retries): can_bus.send(ecu_id, request_transfer_exit()) response = can_bus.recv(timeout=1.0) if response[0] == 0x77: return True # 成功退出 elif response[0] == 0x7F: print(f"否定响应: 0x{response[2]:02X}, 重试 {attempt+1}/{max_retries}") continue return False # 所有重试失败5. 完整流程集成与调试技巧
5.1 组装完整刷写流程
将三个服务按正确顺序组合，形成完整的刷写脚本框架：
def perform_flashing(can_bus, ecu_id, firmware_data): try: # 步骤1: $34服务协商参数 max_block = negotiate_block_size(can_bus, ecu_id, len(firmware_data)) # 步骤2: $36服务传输数据 transfer_data(can_bus, firmware_data, max_block, ecu_id) # 步骤3: $37服务结束传输 if not exit_transfer(can_bus, ecu_id): raise RuntimeError("无法正常结束传输") print("刷写成功完成") except Exception as e: print(f"刷写失败: {str(e)}") # 这里应添加错误恢复或回滚逻辑5.2 常见问题排查指南
当脚本运行不顺利时，可以检查以下方面：
CAN通信基础验证
- 确认总线速率设置正确
- 检查物理连接和终端电阻
- 使用CAN监听工具验证报文是否实际发送
UDS会话层状态
- 确保已进入编程会话（0x10 03）
- 验证安全访问是否已解锁（0x27服务）
数据格式问题
- 检查地址和大小字段的字节序
- 验证数据块是否对齐到maxNumberOfBlockLength
- 确认序列号是否正确循环
ECU特定要求
- 查阅厂商文档确认是否有特殊参数要求
- 检查刷写前置条件（如电压、温度等）
# 调试示例：打印原始CAN报文 def debug_can_message(msg): print(f"ID: 0x{msg.arbitration_id:03X} | Data: {' '.join(f'{b:02X}' for b in msg.data)}")6. 进阶应用：从模拟到实战
6.1 使用CANoe/CANalyzer模拟ECU
在没有真实ECU的情况下，可以使用CANoe等工具模拟诊断响应：
- CAPL脚本示例：
on message 0x700 // 假设ECU响应地址为0x700 { if (this.byte(0) == 0x34) // $34请求 { byte response[8] = {0x74, 0x10, 0x00, 0x40}; // 返回最大块长度0x400 output(response); } else if (this.byte(0) == 0x36) // $36请求 { byte response[2] = {0x76, this.byte(1)}; // 返回相同序列号 output(response); } // 其他服务处理... }
6.2 真实项目中的增强功能
在实际工程应用中，还需要考虑：
- 断点续传：记录已传输块位置，异常后从中断处恢复
- 数据校验：添加CRC校验或使用$31服务验证写入内容
- 并行处理：多线程处理多个ECU的并行刷写
- 进度显示：实时计算和显示传输进度百分比
# 增强版传输进度显示 class ProgressTracker: def __init__(self, total_size): self.total = total_size self.sent = 0 def update(self, block_size): self.sent += block_size percent = (self.sent / self.total) * 100 print(f"\r进度: {percent:.1f}%", end='', flush=True)7. 安全考量与最佳实践
7.1 防止刷写失败的保障措施
预验证机制：
- 检查ECU当前软件版本
- 验证硬件兼容性
- 确认供电稳定
回滚策略：
- 保留原始软件备份
- 实现紧急恢复模式
- 设置看门狗超时
7.2 自动化测试集成
将刷写脚本集成到CI/CD管道中：
# pytest测试用例示例 def test_flashing_sequence(): """验证完整的$34/$36/$37序列""" can_mock = MockCANBus() ecu_sim = ECUSimulator(can_mock) # 执行刷写测试 perform_flashing(can_mock, ecu_sim.address, b'\x01\x02\x03\x04') # 验证ECU模拟器是否正确接收数据 assert ecu_sim.received_data == b'\x01\x02\x03\x04' assert ecu_sim.transfer_complete8. 性能优化技巧
8.1 传输速率提升方案
- 动态块大小调整：根据总线负载实时调整maxNumberOfBlockLength
- 流水线传输：在收到前一块响应前发送下一块请求
- 数据压缩：在$34服务中协商使用压缩算法
8.2 资源占用优化
# 内存高效的数据块生成器 def data_chunks(data, chunk_size): """生成固定大小的数据块，避免内存复制""" for i in range(0, len(data), chunk_size): yield data[i:i + chunk_size]9. 扩展应用：DoIP与CAN FD适配
9.1 适配DoIP协议
对于基于以太网的诊断协议(DoIP)，只需修改传输层：
class DoIPTransporter: def send(self, message): # 将UDS消息封装到DoIP协议中 doip_pkt = build_doip_packet(message) self.socket.send(doip_pkt) def recv(self): doip_pkt = self.socket.recv() return parse_doip_packet(doip_pkt)9.2 CAN FD支持
利用python-can的FD特性：
bus = can.interface.Bus(bitrate=500000, fd=True, data_bitrate=2000000)10. 工具链整合建议
10.1 与现有工具集成
- DBC文件解析：使用cantools加载厂商DBC
- 诊断描述文件：解析CDD或ODX文件自动生成配置
- 测试报告：生成JUnit格式的测试报告
10.2 自定义监控界面
# 使用PyQt创建简单监控界面 class FlashMonitor(QWidget): def __init__(self): super().__init__() self.progress = QProgressBar() self.log = QTextEdit() layout = QVBoxLayout() layout.addWidget(self.progress) layout.addWidget(self.log) self.setLayout(layout) def update_progress(self, percent): self.progress.setValue(percent) def add_log(self, message): self.log.append(message)