4. DoIP诊断消息发送与接收

任务目标

在DoIP（Diagnostic over IP）通信中，诊断仪（Tester）通过TCP连接向车辆网关发送UDS诊断请求（如0x10 0x03会话切换），并接收、解析网关返回的诊断响应。本案例实现一个最小可运行的DoIP诊断消息收发闭环：构造DoIP首部（版本、负载类型、长度），封装UDS诊断消息，通过TCP发送，并解析接收到的DoIP响应帧。

环境与依赖

• Python 3.8+，仅使用标准库（socket、struct）
• 无需第三方库

DoIP协议关键字段

完整实现

import socket
import struct

# DoIP常量
DOIP_VERSION = 0x02
DOIP_INV_VERSION = 0xFD
PAYLOAD_TYPE_DIAG = 0x8001
TESTER_ADDR = 0x0E80
GATEWAY_ADDR = 0x0E00

def build_doip_diag_request(uds_data: bytes, source_addr=TESTER_ADDR, target_addr=GATEWAY_ADDR) -> bytes:
    """构造DoIP诊断请求帧"""
    payload = struct.pack('>HH', source_addr, target_addr) + uds_data
    header = struct.pack('>BBHI', DOIP_VERSION, DOIP_INV_VERSION,
                         PAYLOAD_TYPE_DIAG, len(payload))
    return header + payload

def parse_doip_response(frame: bytes) -> dict:
    """解析DoIP帧，返回负载类型和负载内容"""
    if len(frame) < 8:
        raise ValueError(f'帧长度不足: {len(frame)}')
    version, inv_version, payload_type, payload_len = struct.unpack('>BBHI', frame[:8])
    if version != DOIP_VERSION or inv_version != DOIP_INV_VERSION:
        raise ValueError(f'DoIP版本不匹配: ver={version:#x}, inv={inv_version:#x}')
    if len(frame) < 8 + payload_len:
        raise ValueError(f'帧不完整: 声明{payload_len}字节, 实际{len(frame)-8}字节')
    payload = frame[8:8+payload_len]
    return {'payload_type': payload_type, 'payload': payload}

def extract_uds_response(parsed: dict) -> bytes:
    """从解析后的DoIP帧中提取UDS数据（跳过源/目标地址）"""
    if parsed['payload_type'] != PAYLOAD_TYPE_DIAG:
        raise ValueError(f'非诊断消息类型: {parsed["payload_type"]:#x}')
    if len(parsed['payload']) < 4:
        raise ValueError('诊断负载缺少地址字段')
    return parsed['payload'][4:]  # 跳过2字节源地址+2字节目标地址

def run_diag_session(host: str, port: int, uds_request: bytes) -> bytes:
    """建立TCP连接，发送诊断请求，接收并返回UDS响应"""
    frame = build_doip_diag_request(uds_request)
    with socket.create_connection((host, port), timeout=5) as sock:
        sock.sendall(frame)
        # 先接收DoIP首部（8字节）
        header = b''
        while len(header) < 8:
            chunk = sock.recv(8 - len(header))
            if not chunk:
                raise ConnectionError('连接关闭，未收到完整首部')
            header += chunk
        version, inv_version, payload_type, payload_len = struct.unpack('>BBHI', header)
        # 接收剩余负载
        body = b''
        while len(body) < payload_len:
            chunk = sock.recv(payload_len - len(body))
            if not chunk:
                raise ConnectionError('连接关闭，负载不完整')
            body += chunk
        full_frame = header + body
        parsed = parse_doip_response(full_frame)
        return extract_uds_response(parsed)

if __name__ == '__main__':
    # 示例：发送10 03（进入扩展会话），预期响应 50 03
    request = bytes.fromhex('10 03')
    # 实际运行时替换为DoIP网关的IP和端口（默认13400）
    # response = run_diag_session('192.168.1.10', 13400, request)
    # print('UDS响应:', response.hex())
    # 以下为本地自测（不依赖真实网关）
    test_frame = build_doip_diag_request(bytes.fromhex('10 03'))
    parsed = parse_doip_response(test_frame)
    assert parsed['payload_type'] == PAYLOAD_TYPE_DIAG
    assert extract_uds_response(parsed) == bytes.fromhex('10 03')
    print('本地自测通过：封装与解析逻辑正确')

运行与验证

1. 本地自测：直接运行脚本，输出本地自测通过：封装与解析逻辑正确。该自测验证了build_doip_diag_request构造的帧能被parse_doip_response正确解析，且UDS数据往返一致。
2. 真实网关验证：将run_diag_session调用中的IP替换为实际DoIP网关地址（默认端口13400），发送10 03请求。若网关正常响应，控制台打印的UDS响应应为50 03（表示已进入扩展会话）。
3. 判定标准：本地自测无异常即证明封装/解析函数正确；真实网关测试收到50 03即证明端到端收发链路打通。

常见问题

• 连接超时：确认网关IP/端口正确，且车辆网关已激活DoIP（通常需要先通过UDP 13400完成车辆发现和路由激活）。
• 收到负响应：若UDS响应以7F开头，说明请求被拒绝（如7F 10 12表示子功能不支持），需检查UDS请求格式。
• 版本不匹配：部分网关使用DoIP版本0x03，此时需调整DOIP_VERSION和DOIP_INV_VERSION为0x03/0xFC。