Hands-On: DoIP Diagnostic Session

#!/usr/bin/env python3
# Minimal DoIP ECU simulator for lab use
import socket, struct, threading

LOGICAL_ADDR = 0x0001
VIN          = b"WVWZZZ1JZYW000001"  # 17 bytes
SW_VERSION   = b"HW12_SW03.04.01"

def handle_client(conn, addr):
    print(f"Tester connected from {addr}")
    activated = False
    try:
        while True:
            data = conn.recv(65536)
            if not data: break
            ver, inv, ptype, plen = struct.unpack(">BBHI", data[:8])
            payload = data[8:8+plen]

            if ptype == 0x0005:   # RoutingActivationRequest
                sa = struct.unpack(">H", payload[:2])[0]
                response = struct.pack(">BBHIHHBBL",
                    0x02, 0xFD, 0x0006, 13,
                    sa, LOGICAL_ADDR, 0x10, 0x00, 0)
                conn.sendall(response)
                activated = True

            elif ptype == 0x8001 and activated:  # DiagnosticMessage
                sa_t, ta_t = struct.unpack(">HH", payload[:4])
                uds = payload[4:]

                # ACK first
                ack = struct.pack(">BBHIHH", 0x02, 0xFD, 0x8002, 5, sa_t, ta_t) + b''
                conn.sendall(ack)

                # UDS response
                if uds[0] == 0x22 and uds[1:3] == b'ñ':  # ReadDID VIN
                    uds_resp = bytes([0x62, 0xF1, 0x90]) + VIN
                elif uds[0] == 0x22 and uds[1:3] == b'ñ‰':  # SW version
                    uds_resp = bytes([0x62, 0xF1, 0x89]) + SW_VERSION
                elif uds[0] == 0x3E:   # TesterPresent
                    uds_resp = bytes([0x7E, uds[1]]) if uds[1] == 0x00 else b''
                else:
                    uds_resp = bytes([0x7F, uds[0], 0x11])  # NRC serviceNotSupported

                if uds_resp:
                    resp_payload = struct.pack(">HH", LOGICAL_ADDR, sa_t) + uds_resp
                    hdr = struct.pack(">BBHI", 0x02, 0xFD, 0x8001, len(resp_payload))
                    conn.sendall(hdr + resp_payload)
    finally:
        conn.close()

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(("0.0.0.0", 13400))
server.listen(5)
print("DoIP ECU simulator listening on port 13400")
while True:
    conn, addr = server.accept()
    threading.Thread(target=handle_client, args=(conn, addr), daemon=True).start()

#!/usr/bin/env python3
import time, socket, struct

# Benchmark: read VIN 100 times via DoIP; compare to simulated CAN timing

def benchmark_doip(ecu_ip: str, iterations: int = 100) -> float:
    client = DoipUdsClient(ecu_ip)
    client.connect()

    times = []
    for _ in range(iterations):
        t0 = time.perf_counter()
        _ = client.send_uds(bytes([0x22, 0xF1, 0x90]))  # read VIN
        t1 = time.perf_counter()
        times.append((t1 - t0) * 1000)  # ms

    client.sock.close()
    return sum(times) / len(times)  # avg ms

# Simulated CAN timing: VIN = 20 bytes UDS response
# CAN 500 kbit/s, ISO 15765-2: ~5 CAN frames × 8 bytes = 40 bytes
# 40 bytes × 8 bits / 500000 bps = 0.64 ms + overhead ≈ 3–8 ms typical
SIMULATED_CAN_MS = 5.0  # ms typical

doip_avg_ms = benchmark_doip("127.0.0.1")
speedup = SIMULATED_CAN_MS / doip_avg_ms

print(f"DoIP avg latency:    {doip_avg_ms:.2f} ms")
print(f"CAN typical latency: {SIMULATED_CAN_MS:.2f} ms")
print(f"DoIP speedup:        {speedup:.1f}×")
# Expected: DoIP ~0.3–1 ms → ~5–15× faster for small reads
# For large flash: DoIP 100 Mbit/s vs CAN 0.5 Mbit/s = 200× faster