OBD-II & Emission Diagnostics

#!/usr/bin/env python3
# OBD-II Mode 0x01: read standard PIDs (no security required)
import can, struct, time

def obd_request(bus, pid: int) -> bytes:
    frame = can.Message(arbitration_id=0x7DF,
                        data=[0x02, 0x01, pid, 0,0,0,0,0],
                        is_extended_id=False)
    bus.send(frame)
    deadline = time.time() + 0.05  # 50ms P2
    while time.time() < deadline:
        msg = bus.recv(timeout=0.01)
        if msg and 0x7E8 <= msg.arbitration_id <= 0x7EF:
            if msg.data[1] == 0x41 and msg.data[2] == pid:
                return bytes(msg.data[3:])
    return b''

def decode_pid(pid: int, data: bytes) -> str:
    if pid == 0x0C:  # Engine RPM
        rpm = ((data[0] << 8) | data[1]) / 4.0
        return f"Engine RPM: {rpm:.0f} RPM"
    elif pid == 0x0D:  # Vehicle Speed
        return f"Vehicle Speed: {data[0]} km/h"
    elif pid == 0x05:  # Coolant Temperature
        return f"Coolant Temp: {data[0] - 40} °C"
    elif pid == 0x2F:  # Fuel Tank Level
        return f"Fuel Level: {data[0] * 100 / 255:.0f}%"
    elif pid == 0x42:  # Control module voltage
        voltage = ((data[0] << 8) | data[1]) / 1000.0
        return f"Battery: {voltage:.2f} V"
    return f"PID 0x{pid:02X}: {data.hex()}"

# bus = can.Bus('vcan0', bustype='socketcan')
# for pid in [0x0C, 0x0D, 0x05, 0x2F, 0x42]:
#     raw = obd_request(bus, pid)
#     print(decode_pid(pid, raw))