mawei/panda-meb
1
0
Fork 0
mirror of https://github.com/infiniteCable2/panda.git synced 2026-02-19 09:43:51 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master-new' of https://github.com/sunnypilot/panda into master-new

Browse Source
This commit is contained in:
infiniteCable2
2025-04-06 15:07:38 +02:00
parent 26d8562257 503f5dd6c3
commit 1d1daaf7b6
136 changed files with 499 additions and 14416 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
python/__init__.py
View File

@@ -9,10 +9,11 @@ import binascii
from functools import wraps, partial
from itertools import accumulate
from opendbc.car.structs import CarParams
from .base import BaseHandle
from .constants import FW_PATH, McuType
from .dfu import PandaDFU
from .isotp import isotp_send, isotp_recv
from .spi import PandaSpiHandle, PandaSpiException, PandaProtocolMismatch
from .usb import PandaUsbHandle
from .utils import logger
@@ -101,48 +102,7 @@ ensure_health_packet_version = partial(ensure_version, "health", "HEALTH_PACKET_
class ALTERNATIVE_EXPERIENCE:
DEFAULT = 0
DISABLE_DISENGAGE_ON_GAS = 1
DISABLE_STOCK_AEB = 2
RAISE_LONGITUDINAL_LIMITS_TO_ISO_MAX = 8
ALLOW_AEB = 16
# sunnypilot
ENABLE_MADS = 2 ** 10
DISENGAGE_LATERAL_ON_BRAKE = 2 ** 11
class Panda:
# matches cereal.car.CarParams.SafetyModel
SAFETY_SILENT = 0
SAFETY_HONDA_NIDEC = 1
SAFETY_TOYOTA = 2
SAFETY_ELM327 = 3
SAFETY_GM = 4
SAFETY_HONDA_BOSCH_GIRAFFE = 5
SAFETY_FORD = 6
SAFETY_HYUNDAI = 8
SAFETY_CHRYSLER = 9
SAFETY_TESLA = 10
SAFETY_SUBARU = 11
SAFETY_MAZDA = 13
SAFETY_NISSAN = 14
SAFETY_VOLKSWAGEN_MQB = 15
SAFETY_ALLOUTPUT = 17
SAFETY_GM_ASCM = 18
SAFETY_NOOUTPUT = 19
SAFETY_HONDA_BOSCH = 20
SAFETY_VOLKSWAGEN_PQ = 21
SAFETY_SUBARU_PREGLOBAL = 22
SAFETY_HYUNDAI_LEGACY = 23
SAFETY_HYUNDAI_COMMUNITY = 24
SAFETY_STELLANTIS = 25
SAFETY_FAW = 26
SAFETY_BODY = 27
SAFETY_HYUNDAI_CANFD = 28
SAFETY_VOLKSWAGEN_MEB = 33
SERIAL_DEBUG = 0
SERIAL_ESP = 1
SERIAL_LIN1 = 2
@@ -169,7 +129,7 @@ class Panda:
CAN_PACKET_VERSION = 4
HEALTH_PACKET_VERSION = 16
CAN_HEALTH_PACKET_VERSION = 5
HEALTH_STRUCT = struct.Struct("<IIIIIIIIBBBBBHBBBHfBBHBHHBB")
HEALTH_STRUCT = struct.Struct("<IIIIIIIIBBBBBHBBBHfBBHBHHB")
CAN_HEALTH_STRUCT = struct.Struct("<BIBBBBBBBBIIIIIIIHHBBBIIII")
F4_DEVICES = [HW_TYPE_WHITE_PANDA, HW_TYPE_GREY_PANDA, HW_TYPE_BLACK_PANDA, HW_TYPE_UNO, HW_TYPE_DOS]
@@ -189,50 +149,6 @@ class Panda:
HARNESS_STATUS_NORMAL = 1
HARNESS_STATUS_FLIPPED = 2
# first byte is for EPS scaling factor
FLAG_TOYOTA_ALT_BRAKE = (1 << 8)
FLAG_TOYOTA_STOCK_LONGITUDINAL = (2 << 8)
FLAG_TOYOTA_LTA = (4 << 8)
FLAG_TOYOTA_SECOC = (8 << 8)
FLAG_HONDA_ALT_BRAKE = 1
FLAG_HONDA_BOSCH_LONG = 2
FLAG_HONDA_NIDEC_ALT = 4
FLAG_HONDA_RADARLESS = 8
FLAG_HYUNDAI_EV_GAS = 1
FLAG_HYUNDAI_HYBRID_GAS = 2
FLAG_HYUNDAI_LONG = 4
FLAG_HYUNDAI_CAMERA_SCC = 8
FLAG_HYUNDAI_CANFD_HDA2 = 16
FLAG_HYUNDAI_CANFD_ALT_BUTTONS = 32
FLAG_HYUNDAI_ALT_LIMITS = 64
FLAG_HYUNDAI_CANFD_HDA2_ALT_STEERING = 128
FLAG_HYUNDAI_ESCC = 512
FLAG_HYUNDAI_LONG_MAIN_CRUISE_TOGGLEABLE = 1024
FLAG_TESLA_POWERTRAIN = 1
FLAG_TESLA_LONG_CONTROL = 2
FLAG_TESLA_RAVEN = 4
FLAG_VOLKSWAGEN_LONG_CONTROL = 1
FLAG_CHRYSLER_RAM_DT = 1
FLAG_CHRYSLER_RAM_HD = 2
FLAG_SUBARU_GEN2 = 1
FLAG_SUBARU_LONG = 2
FLAG_SUBARU_PREGLOBAL_REVERSED_DRIVER_TORQUE = 4
FLAG_NISSAN_ALT_EPS_BUS = 1
FLAG_NISSAN_LEAF = 512
FLAG_GM_HW_CAM = 1
FLAG_GM_HW_CAM_LONG = 2
FLAG_FORD_LONG_CONTROL = 1
FLAG_FORD_CANFD = 2
def __init__(self, serial: str | None = None, claim: bool = True, disable_checks: bool = True, can_speed_kbps: int = 500, cli: bool = True):
self._disable_checks = disable_checks
@@ -660,7 +576,6 @@ class Panda:
"sbu1_voltage_mV": a[23],
"sbu2_voltage_mV": a[24],
"som_reset_triggered": a[25],
"controls_allowed_lat": a[26],
}
@ensure_can_health_packet_version
@@ -794,10 +709,13 @@ class Panda:
# ******************* configuration *******************
def set_alternative_experience(self, alternative_experience, safety_param_sp=0):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xdf, int(alternative_experience), int(safety_param_sp), b'')
def set_power_save(self, power_save_enabled=0):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xe7, int(power_save_enabled), 0, b'')
def set_safety_mode(self, mode=SAFETY_SILENT, param=0):
def set_safety_mode(self, mode=CarParams.SafetyModel.silent, param=0):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xdc, mode, param, b'')
def set_obd(self, obd):
@@ -878,14 +796,6 @@ class Panda:
"""
self._handle.controlWrite(Panda.REQUEST_OUT, 0xf1, bus, 0, b'')
# ******************* isotp *******************
def isotp_send(self, addr, dat, bus, recvaddr=None, subaddr=None):
return isotp_send(self, dat, addr, bus, recvaddr, subaddr)
def isotp_recv(self, addr, bus=0, sendaddr=None, subaddr=None):
return isotp_recv(self, addr, bus, sendaddr, subaddr)
# ******************* serial *******************
def serial_read(self, port_number):

53
python/canhandle.py
View File

@@ -1,53 +0,0 @@
import struct
import signal
from .base import BaseHandle
class CanHandle(BaseHandle):
def __init__(self, p, bus):
self.p = p
self.bus = bus
def transact(self, dat):
def _handle_timeout(signum, frame):
# will happen on reset or can error
raise TimeoutError
signal.signal(signal.SIGALRM, _handle_timeout)
signal.alarm(1)
try:
self.p.isotp_send(1, dat, self.bus, recvaddr=2)
finally:
signal.alarm(0)
signal.signal(signal.SIGALRM, _handle_timeout)
signal.alarm(1)
try:
ret = self.p.isotp_recv(2, self.bus, sendaddr=1)
finally:
signal.alarm(0)
return ret
def close(self):
pass
def controlWrite(self, request_type, request, value, index, data, timeout=0, expect_disconnect=False):
# ignore data in reply, panda doesn't use it
return self.controlRead(request_type, request, value, index, 0, timeout)
def controlRead(self, request_type, request, value, index, length, timeout=0):
dat = struct.pack("HHBBHHH", 0, 0, request_type, request, value, index, length)
return self.transact(dat)
def bulkWrite(self, endpoint, data, timeout=0):
if len(data) > 0x10:
raise ValueError("Data must not be longer than 0x10")
dat = struct.pack("HH", endpoint, len(data)) + data
return self.transact(dat)
def bulkRead(self, endpoint, length, timeout=0):
dat = struct.pack("HH", endpoint, 0)
return self.transact(dat)

374
python/ccp.py
View File

@@ -1,374 +0,0 @@
import sys
import time
import struct
from enum import IntEnum, Enum
from dataclasses import dataclass
from typing import Optional
@dataclass
class ExchangeStationIdsReturn:
id_length: int
data_type: int
available: int
protected: int
@dataclass
class GetDaqListSizeReturn:
list_size: int
first_pid: int
@dataclass
class GetSessionStatusReturn:
status: int
info: Optional[int]
@dataclass
class DiagnosticServiceReturn:
length: int
type: int
@dataclass
class ActionServiceReturn:
length: int
type: int
class COMMAND_CODE(IntEnum):
CONNECT = 0x01
SET_MTA = 0x02
DNLOAD = 0x03
UPLOAD = 0x04
TEST = 0x05
START_STOP = 0x06
DISCONNECT = 0x07
START_STOP_ALL = 0x08
GET_ACTIVE_CAL_PAGE = 0x09
SET_S_STATUS = 0x0C
GET_S_STATUS = 0x0D
BUILD_CHKSUM = 0x0E
SHORT_UP = 0x0F
CLEAR_MEMORY = 0x10
SELECT_CAL_PAGE = 0x11
GET_SEED = 0x12
UNLOCK = 0x13
GET_DAQ_SIZE = 0x14
SET_DAQ_PTR = 0x15
WRITE_DAQ = 0x16
EXCHANGE_ID = 0x17
PROGRAM = 0x18
MOVE = 0x19
GET_CCP_VERSION = 0x1B
DIAG_SERVICE = 0x20
ACTION_SERVICE = 0x21
PROGRAM_6 = 0x22
DNLOAD_6 = 0x23
COMMAND_RETURN_CODES = {
0x00: "acknowledge / no error",
0x01: "DAQ processor overload",
0x10: "command processor busy",
0x11: "DAQ processor busy",
0x12: "internal timeout",
0x18: "key request",
0x19: "session status request",
0x20: "cold start request",
0x21: "cal. data init. request",
0x22: "DAQ list init. request",
0x23: "code update request",
0x30: "unknown command",
0x31: "command syntax",
0x32: "parameter(s) out of range",
0x33: "access denied",
0x34: "overload",
0x35: "access locked",
0x36: "resource/function not available",
}
class BYTE_ORDER(Enum):
LITTLE_ENDIAN = '<'
BIG_ENDIAN = '>'
class CommandTimeoutError(Exception):
pass
class CommandCounterError(Exception):
pass
class CommandResponseError(Exception):
def __init__(self, message, return_code):
super().__init__()
self.message = message
self.return_code = return_code
def __str__(self):
return self.message
class CcpClient():
def __init__(self, panda, tx_addr: int, rx_addr: int, bus: int=0, byte_order: BYTE_ORDER=BYTE_ORDER.BIG_ENDIAN, debug=False):
self.tx_addr = tx_addr
self.rx_addr = rx_addr
self.can_bus = bus
self.byte_order = byte_order
self.debug = debug
self._panda = panda
self._command_counter = -1
def _send_cro(self, cmd: int, dat: bytes = b"") -> None:
self._command_counter = (self._command_counter + 1) & 0xFF
tx_data = (bytes([cmd, self._command_counter]) + dat).ljust(8, b"\x00")
if self.debug:
print(f"CAN-TX: {hex(self.tx_addr)} - 0x{bytes.hex(tx_data)}")
assert len(tx_data) == 8, "data is not 8 bytes"
self._panda.can_clear(self.can_bus)
self._panda.can_clear(0xFFFF)
self._panda.can_send(self.tx_addr, tx_data, self.can_bus)
def _recv_dto(self, timeout: float) -> bytes:
start_time = time.time()
while time.time() - start_time < timeout:
msgs = self._panda.can_recv() or []
if len(msgs) >= 256:
print("CAN RX buffer overflow!!!", file=sys.stderr)
for rx_addr, rx_data_bytearray, rx_bus in msgs:
if rx_bus == self.can_bus and rx_addr == self.rx_addr:
rx_data = bytes(rx_data_bytearray)
if self.debug:
print(f"CAN-RX: {hex(rx_addr)} - 0x{bytes.hex(rx_data)}")
assert len(rx_data) == 8, f"message length not 8: {len(rx_data)}"
pid = rx_data[0]
if pid == 0xFF or pid == 0xFE:
err = rx_data[1]
err_desc = COMMAND_RETURN_CODES.get(err, "unknown error")
ctr = rx_data[2]
dat = rx_data[3:]
if pid == 0xFF and self._command_counter != ctr:
raise CommandCounterError(f"counter invalid: {ctr} != {self._command_counter}")
if err >= 0x10 and err <= 0x12:
if self.debug:
print(f"CCP-WAIT: {hex(err)} - {err_desc}")
start_time = time.time()
continue
if err >= 0x30:
raise CommandResponseError(f"{hex(err)} - {err_desc}", err)
else:
dat = rx_data[1:]
return dat
time.sleep(0.001)
raise CommandTimeoutError("timeout waiting for response")
# commands
def connect(self, station_addr: int) -> None:
if station_addr > 65535:
raise ValueError("station address must be less than 65536")
# NOTE: station address is always little endian
self._send_cro(COMMAND_CODE.CONNECT, struct.pack("<H", station_addr))
self._recv_dto(0.025)
def exchange_station_ids(self, device_id_info: bytes = b"") -> ExchangeStationIdsReturn:
self._send_cro(COMMAND_CODE.EXCHANGE_ID, device_id_info)
resp = self._recv_dto(0.025)
return ExchangeStationIdsReturn(id_length=resp[0], data_type=resp[1], available=resp[2], protected=resp[3])
def get_seed(self, resource_mask: int) -> bytes:
if resource_mask > 255:
raise ValueError("resource mask must be less than 256")
self._send_cro(COMMAND_CODE.GET_SEED, bytes([resource_mask]))
resp = self._recv_dto(0.025)
# protected = resp[0] == 0
seed = resp[1:]
return seed
def unlock(self, key: bytes) -> int:
if len(key) > 6:
raise ValueError("max key size is 6 bytes")
self._send_cro(COMMAND_CODE.UNLOCK, key)
resp = self._recv_dto(0.025)
status = resp[0]
return status
def set_memory_transfer_address(self, mta_num: int, addr_ext: int, addr: int) -> None:
if mta_num > 255:
raise ValueError("MTA number must be less than 256")
if addr_ext > 255:
raise ValueError("address extension must be less than 256")
self._send_cro(COMMAND_CODE.SET_MTA, bytes([mta_num, addr_ext]) + struct.pack(f"{self.byte_order.value}I", addr))
self._recv_dto(0.025)
def download(self, data: bytes) -> int:
if len(data) > 5:
raise ValueError("max data size is 5 bytes")
self._send_cro(COMMAND_CODE.DNLOAD, bytes([len(data)]) + data)
resp = self._recv_dto(0.025)
# mta_addr_ext = resp[0]
mta_addr = struct.unpack(f"{self.byte_order.value}I", resp[1:5])[0]
return mta_addr # type: ignore
def download_6_bytes(self, data: bytes) -> int:
if len(data) != 6:
raise ValueError("data size must be 6 bytes")
self._send_cro(COMMAND_CODE.DNLOAD_6, data)
resp = self._recv_dto(0.025)
# mta_addr_ext = resp[0]
mta_addr = struct.unpack(f"{self.byte_order.value}I", resp[1:5])[0]
return mta_addr # type: ignore
def upload(self, size: int) -> bytes:
if size > 5:
raise ValueError("size must be less than 6")
self._send_cro(COMMAND_CODE.UPLOAD, bytes([size]))
return self._recv_dto(0.025)[:size]
def short_upload(self, size: int, addr_ext: int, addr: int) -> bytes:
if size > 5:
raise ValueError("size must be less than 6")
if addr_ext > 255:
raise ValueError("address extension must be less than 256")
self._send_cro(COMMAND_CODE.SHORT_UP, bytes([size, addr_ext]) + struct.pack(f"{self.byte_order.value}I", addr))
return self._recv_dto(0.025)[:size]
def select_calibration_page(self) -> None:
self._send_cro(COMMAND_CODE.SELECT_CAL_PAGE)
self._recv_dto(0.025)
def get_daq_list_size(self, list_num: int, can_id: int = 0) -> GetDaqListSizeReturn:
if list_num > 255:
raise ValueError("list number must be less than 256")
self._send_cro(COMMAND_CODE.GET_DAQ_SIZE, bytes([list_num, 0]) + struct.pack(f"{self.byte_order.value}I", can_id))
resp = self._recv_dto(0.025)
return GetDaqListSizeReturn(list_size=resp[0], first_pid=resp[1])
def set_daq_list_pointer(self, list_num: int, odt_num: int, element_num: int) -> None:
if list_num > 255:
raise ValueError("list number must be less than 256")
if odt_num > 255:
raise ValueError("ODT number must be less than 256")
if element_num > 255:
raise ValueError("element number must be less than 256")
self._send_cro(COMMAND_CODE.SET_DAQ_PTR, bytes([list_num, odt_num, element_num]))
self._recv_dto(0.025)
def write_daq_list_entry(self, size: int, addr_ext: int, addr: int) -> None:
if size > 255:
raise ValueError("size must be less than 256")
if addr_ext > 255:
raise ValueError("address extension must be less than 256")
self._send_cro(COMMAND_CODE.WRITE_DAQ, bytes([size, addr_ext]) + struct.pack(f"{self.byte_order.value}I", addr))
self._recv_dto(0.025)
def start_stop_transmission(self, mode: int, list_num: int, odt_num: int, channel_num: int, rate_prescaler: int = 0) -> None:
if mode > 255:
raise ValueError("mode must be less than 256")
if list_num > 255:
raise ValueError("list number must be less than 256")
if odt_num > 255:
raise ValueError("ODT number must be less than 256")
if channel_num > 255:
raise ValueError("channel number must be less than 256")
if rate_prescaler > 65535:
raise ValueError("rate prescaler must be less than 65536")
self._send_cro(COMMAND_CODE.START_STOP, bytes([mode, list_num, odt_num, channel_num]) + struct.pack(f"{self.byte_order.value}H", rate_prescaler))
self._recv_dto(0.025)
def disconnect(self, station_addr: int, temporary: bool = False) -> None:
if station_addr > 65535:
raise ValueError("station address must be less than 65536")
# NOTE: station address is always little endian
self._send_cro(COMMAND_CODE.DISCONNECT, bytes([int(not temporary), 0x00]) + struct.pack("<H", station_addr))
self._recv_dto(0.025)
def set_session_status(self, status: int) -> None:
if status > 255:
raise ValueError("status must be less than 256")
self._send_cro(COMMAND_CODE.SET_S_STATUS, bytes([status]))
self._recv_dto(0.025)
def get_session_status(self) -> GetSessionStatusReturn:
self._send_cro(COMMAND_CODE.GET_S_STATUS)
resp = self._recv_dto(0.025)
info = resp[2] if resp[1] else None
return GetSessionStatusReturn(status=resp[0], info=info)
def build_checksum(self, size: int) -> bytes:
self._send_cro(COMMAND_CODE.BUILD_CHKSUM, struct.pack(f"{self.byte_order.value}I", size))
resp = self._recv_dto(30.0)
chksum_size = resp[0]
assert chksum_size <= 4, "checksum more than 4 bytes"
chksum = resp[1:1+chksum_size]
return chksum
def clear_memory(self, size: int) -> None:
self._send_cro(COMMAND_CODE.CLEAR_MEMORY, struct.pack(f"{self.byte_order.value}I", size))
self._recv_dto(30.0)
def program(self, size: int, data: bytes) -> int:
if size > 5:
raise ValueError("size must be less than 6")
if len(data) > 5:
raise ValueError("max data size is 5 bytes")
self._send_cro(COMMAND_CODE.PROGRAM, bytes([size]) + data)
resp = self._recv_dto(0.1)
# mta_addr_ext = resp[0]
mta_addr = struct.unpack(f"{self.byte_order.value}I", resp[1:5])[0]
return mta_addr # type: ignore
def program_6_bytes(self, data: bytes) -> int:
if len(data) != 6:
raise ValueError("data size must be 6 bytes")
self._send_cro(COMMAND_CODE.PROGRAM_6, data)
resp = self._recv_dto(0.1)
# mta_addr_ext = resp[0]
mta_addr = struct.unpack(f"{self.byte_order.value}I", resp[1:5])[0]
return mta_addr # type: ignore
def move_memory_block(self, size: int) -> None:
self._send_cro(COMMAND_CODE.MOVE, struct.pack(f"{self.byte_order.value}I", size))
self._recv_dto(0.025)
def diagnostic_service(self, service_num: int, data: bytes = b"") -> DiagnosticServiceReturn:
if service_num > 65535:
raise ValueError("service number must be less than 65536")
if len(data) > 4:
raise ValueError("max data size is 4 bytes")
self._send_cro(COMMAND_CODE.DIAG_SERVICE, struct.pack(f"{self.byte_order.value}H", service_num) + data)
resp = self._recv_dto(0.025)
return DiagnosticServiceReturn(length=resp[0], type=resp[1])
def action_service(self, service_num: int, data: bytes = b"") -> ActionServiceReturn:
if service_num > 65535:
raise ValueError("service number must be less than 65536")
if len(data) > 4:
raise ValueError("max data size is 4 bytes")
self._send_cro(COMMAND_CODE.ACTION_SERVICE, struct.pack(f"{self.byte_order.value}H", service_num) + data)
resp = self._recv_dto(0.025)
return ActionServiceReturn(length=resp[0], type=resp[1])
def test_availability(self, station_addr: int) -> None:
if station_addr > 65535:
raise ValueError("station address must be less than 65536")
# NOTE: station address is always little endian
self._send_cro(COMMAND_CODE.TEST, struct.pack("<H", station_addr))
self._recv_dto(0.025)
def start_stop_synchronised_transmission(self, mode: int) -> None:
if mode > 255:
raise ValueError("mode must be less than 256")
self._send_cro(COMMAND_CODE.START_STOP_ALL, bytes([mode]))
self._recv_dto(0.025)
def get_active_calibration_page(self):
self._send_cro(COMMAND_CODE.GET_ACTIVE_CAL_PAGE)
resp = self._recv_dto(0.025)
# cal_addr_ext = resp[0]
cal_addr = struct.unpack(f"{self.byte_order.value}I", resp[1:5])[0]
return cal_addr
def get_version(self, desired_version: float = 2.1) -> float:
major, minor = map(int, str(desired_version).split("."))
self._send_cro(COMMAND_CODE.GET_CCP_VERSION, bytes([major, minor]))
resp = self._recv_dto(0.025)
return float(f"{resp[0]}.{resp[1]}")

140
python/isotp.py
View File

@@ -1,140 +0,0 @@
import binascii
import time
DEBUG = False
def msg(x):
if DEBUG:
print("S:", binascii.hexlify(x))
assert len(x) <= 7
ret = bytes([len(x)]) + x
return ret.ljust(8, b"\x00")
kmsgs = []
def recv(panda, cnt, addr, nbus):
global kmsgs
ret = []
while len(ret) < cnt:
kmsgs += panda.can_recv()
nmsgs = []
for ids, dat, bus in kmsgs:
if ids == addr and bus == nbus and len(ret) < cnt:
ret.append(dat)
else:
# leave around
nmsgs.append((ids, dat, bus))
kmsgs = nmsgs[-256:]
return ret
def isotp_recv_subaddr(panda, addr, bus, sendaddr, subaddr):
msg = recv(panda, 1, addr, bus)[0]
# TODO: handle other subaddr also communicating
assert msg[0] == subaddr
if msg[1] & 0xf0 == 0x10:
# first
tlen = ((msg[1] & 0xf) << 8) | msg[2]
dat = msg[3:]
# 0 block size?
CONTINUE = bytes([subaddr]) + b"\x30" + b"\x00" * 6
panda.can_send(sendaddr, CONTINUE, bus)
idx = 1
for mm in recv(panda, (tlen - len(dat) + 5) // 6, addr, bus):
assert mm[0] == subaddr
assert mm[1] == (0x20 | (idx & 0xF))
dat += mm[2:]
idx += 1
elif msg[1] & 0xf0 == 0x00:
# single
tlen = msg[1] & 0xf
dat = msg[2:]
else:
print(binascii.hexlify(msg))
raise AssertionError
return dat[0:tlen]
# **** import below this line ****
def isotp_send(panda, x, addr, bus=0, recvaddr=None, subaddr=None, rate=None):
if recvaddr is None:
recvaddr = addr + 8
if len(x) <= 7 and subaddr is None:
panda.can_send(addr, msg(x), bus)
elif len(x) <= 6 and subaddr is not None:
panda.can_send(addr, bytes([subaddr]) + msg(x)[0:7], bus)
else:
if subaddr:
ss = bytes([subaddr, 0x10 + (len(x) >> 8), len(x) & 0xFF]) + x[0:5]
x = x[5:]
else:
ss = bytes([0x10 + (len(x) >> 8), len(x) & 0xFF]) + x[0:6]
x = x[6:]
idx = 1
sends = []
while len(x) > 0:
if subaddr:
sends.append((bytes([subaddr, 0x20 + (idx & 0xF)]) + x[0:6]).ljust(8, b"\x00"))
x = x[6:]
else:
sends.append((bytes([0x20 + (idx & 0xF)]) + x[0:7]).ljust(8, b"\x00"))
x = x[7:]
idx += 1
# actually send
panda.can_send(addr, ss, bus)
rr = recv(panda, 1, recvaddr, bus)[0]
if rr.find(b"\x30\x01") != -1:
for s in sends[:-1]:
panda.can_send(addr, s, 0)
rr = recv(panda, 1, recvaddr, bus)[0]
panda.can_send(addr, sends[-1], 0)
else:
if rate is None:
panda.can_send_many([(addr, s, bus) for s in sends])
else:
for dat in sends:
panda.can_send(addr, dat, bus)
time.sleep(rate)
def isotp_recv(panda, addr, bus=0, sendaddr=None, subaddr=None):
if sendaddr is None:
sendaddr = addr - 8
if subaddr is not None:
dat = isotp_recv_subaddr(panda, addr, bus, sendaddr, subaddr)
else:
msg = recv(panda, 1, addr, bus)[0]
if msg[0] & 0xf0 == 0x10:
# first
tlen = ((msg[0] & 0xf) << 8) | msg[1]
dat = msg[2:]
# 0 block size?
CONTINUE = b"\x30" + b"\x00" * 7
panda.can_send(sendaddr, CONTINUE, bus)
idx = 1
for mm in recv(panda, (tlen - len(dat) + 6) // 7, addr, bus):
assert mm[0] == (0x20 | (idx & 0xF))
dat += mm[1:]
idx += 1
elif msg[0] & 0xf0 == 0x00:
# single
tlen = msg[0] & 0xf
dat = msg[1:]
else:
raise AssertionError
dat = dat[0:tlen]
if DEBUG:
print("R:", binascii.hexlify(dat))
return dat

943
python/uds.py
View File

@@ -1,943 +0,0 @@
import time
import struct
from collections import deque
from typing import NamedTuple, cast
from collections.abc import Callable, Generator
from enum import IntEnum
from functools import partial
class SERVICE_TYPE(IntEnum):
DIAGNOSTIC_SESSION_CONTROL = 0x10
ECU_RESET = 0x11
SECURITY_ACCESS = 0x27
COMMUNICATION_CONTROL = 0x28
TESTER_PRESENT = 0x3E
ACCESS_TIMING_PARAMETER = 0x83
SECURED_DATA_TRANSMISSION = 0x84
CONTROL_DTC_SETTING = 0x85
RESPONSE_ON_EVENT = 0x86
LINK_CONTROL = 0x87
READ_DATA_BY_IDENTIFIER = 0x22
READ_MEMORY_BY_ADDRESS = 0x23
READ_SCALING_DATA_BY_IDENTIFIER = 0x24
READ_DATA_BY_PERIODIC_IDENTIFIER = 0x2A
DYNAMICALLY_DEFINE_DATA_IDENTIFIER = 0x2C
WRITE_DATA_BY_IDENTIFIER = 0x2E
WRITE_MEMORY_BY_ADDRESS = 0x3D
CLEAR_DIAGNOSTIC_INFORMATION = 0x14
READ_DTC_INFORMATION = 0x19
INPUT_OUTPUT_CONTROL_BY_IDENTIFIER = 0x2F
ROUTINE_CONTROL = 0x31
REQUEST_DOWNLOAD = 0x34
REQUEST_UPLOAD = 0x35
TRANSFER_DATA = 0x36
REQUEST_TRANSFER_EXIT = 0x37
class SESSION_TYPE(IntEnum):
DEFAULT = 1
PROGRAMMING = 2
EXTENDED_DIAGNOSTIC = 3
SAFETY_SYSTEM_DIAGNOSTIC = 4
class RESET_TYPE(IntEnum):
HARD = 1
KEY_OFF_ON = 2
SOFT = 3
ENABLE_RAPID_POWER_SHUTDOWN = 4
DISABLE_RAPID_POWER_SHUTDOWN = 5
class ACCESS_TYPE(IntEnum):
REQUEST_SEED = 1
SEND_KEY = 2
class CONTROL_TYPE(IntEnum):
ENABLE_RX_ENABLE_TX = 0
ENABLE_RX_DISABLE_TX = 1
DISABLE_RX_ENABLE_TX = 2
DISABLE_RX_DISABLE_TX = 3
class MESSAGE_TYPE(IntEnum):
NORMAL = 1
NETWORK_MANAGEMENT = 2
NORMAL_AND_NETWORK_MANAGEMENT = 3
class TIMING_PARAMETER_TYPE(IntEnum):
READ_EXTENDED_SET = 1
SET_TO_DEFAULT_VALUES = 2
READ_CURRENTLY_ACTIVE = 3
SET_TO_GIVEN_VALUES = 4
class DTC_SETTING_TYPE(IntEnum):
ON = 1
OFF = 2
class RESPONSE_EVENT_TYPE(IntEnum):
STOP_RESPONSE_ON_EVENT = 0
ON_DTC_STATUS_CHANGE = 1
ON_TIMER_INTERRUPT = 2
ON_CHANGE_OF_DATA_IDENTIFIER = 3
REPORT_ACTIVATED_EVENTS = 4
START_RESPONSE_ON_EVENT = 5
CLEAR_RESPONSE_ON_EVENT = 6
ON_COMPARISON_OF_VALUES = 7
class LINK_CONTROL_TYPE(IntEnum):
VERIFY_BAUDRATE_TRANSITION_WITH_FIXED_BAUDRATE = 1
VERIFY_BAUDRATE_TRANSITION_WITH_SPECIFIC_BAUDRATE = 2
TRANSITION_BAUDRATE = 3
class BAUD_RATE_TYPE(IntEnum):
PC9600 = 1
PC19200 = 2
PC38400 = 3
PC57600 = 4
PC115200 = 5
CAN125000 = 16
CAN250000 = 17
CAN500000 = 18
CAN1000000 = 19
class DATA_IDENTIFIER_TYPE(IntEnum):
BOOT_SOFTWARE_IDENTIFICATION = 0xF180
APPLICATION_SOFTWARE_IDENTIFICATION = 0xF181
APPLICATION_DATA_IDENTIFICATION = 0xF182
BOOT_SOFTWARE_FINGERPRINT = 0xF183
APPLICATION_SOFTWARE_FINGERPRINT = 0xF184
APPLICATION_DATA_FINGERPRINT = 0xF185
ACTIVE_DIAGNOSTIC_SESSION = 0xF186
VEHICLE_MANUFACTURER_SPARE_PART_NUMBER = 0xF187
VEHICLE_MANUFACTURER_ECU_SOFTWARE_NUMBER = 0xF188
VEHICLE_MANUFACTURER_ECU_SOFTWARE_VERSION_NUMBER = 0xF189
SYSTEM_SUPPLIER_IDENTIFIER = 0xF18A
ECU_MANUFACTURING_DATE = 0xF18B
ECU_SERIAL_NUMBER = 0xF18C
SUPPORTED_FUNCTIONAL_UNITS = 0xF18D
VEHICLE_MANUFACTURER_KIT_ASSEMBLY_PART_NUMBER = 0xF18E
VIN = 0xF190
VEHICLE_MANUFACTURER_ECU_HARDWARE_NUMBER = 0xF191
SYSTEM_SUPPLIER_ECU_HARDWARE_NUMBER = 0xF192
SYSTEM_SUPPLIER_ECU_HARDWARE_VERSION_NUMBER = 0xF193
SYSTEM_SUPPLIER_ECU_SOFTWARE_NUMBER = 0xF194
SYSTEM_SUPPLIER_ECU_SOFTWARE_VERSION_NUMBER = 0xF195
EXHAUST_REGULATION_OR_TYPE_APPROVAL_NUMBER = 0xF196
SYSTEM_NAME_OR_ENGINE_TYPE = 0xF197
REPAIR_SHOP_CODE_OR_TESTER_SERIAL_NUMBER = 0xF198
PROGRAMMING_DATE = 0xF199
CALIBRATION_REPAIR_SHOP_CODE_OR_CALIBRATION_EQUIPMENT_SERIAL_NUMBER = 0xF19A
CALIBRATION_DATE = 0xF19B
CALIBRATION_EQUIPMENT_SOFTWARE_NUMBER = 0xF19C
ECU_INSTALLATION_DATE = 0xF19D
ODX_FILE = 0xF19E
ENTITY = 0xF19F
class TRANSMISSION_MODE_TYPE(IntEnum):
SEND_AT_SLOW_RATE = 1
SEND_AT_MEDIUM_RATE = 2
SEND_AT_FAST_RATE = 3
STOP_SENDING = 4
class DYNAMIC_DEFINITION_TYPE(IntEnum):
DEFINE_BY_IDENTIFIER = 1
DEFINE_BY_MEMORY_ADDRESS = 2
CLEAR_DYNAMICALLY_DEFINED_DATA_IDENTIFIER = 3
class ISOTP_FRAME_TYPE(IntEnum):
SINGLE = 0
FIRST = 1
CONSECUTIVE = 2
FLOW = 3
class DynamicSourceDefinition(NamedTuple):
data_identifier: int
position: int
memory_size: int
memory_address: int
class DTC_GROUP_TYPE(IntEnum):
EMISSIONS = 0x000000
ALL = 0xFFFFFF
class DTC_REPORT_TYPE(IntEnum):
NUMBER_OF_DTC_BY_STATUS_MASK = 0x01
DTC_BY_STATUS_MASK = 0x02
DTC_SNAPSHOT_IDENTIFICATION = 0x03
DTC_SNAPSHOT_RECORD_BY_DTC_NUMBER = 0x04
DTC_SNAPSHOT_RECORD_BY_RECORD_NUMBER = 0x05
DTC_EXTENDED_DATA_RECORD_BY_DTC_NUMBER = 0x06
NUMBER_OF_DTC_BY_SEVERITY_MASK_RECORD = 0x07
DTC_BY_SEVERITY_MASK_RECORD = 0x08
SEVERITY_INFORMATION_OF_DTC = 0x09
SUPPORTED_DTC = 0x0A
FIRST_TEST_FAILED_DTC = 0x0B
FIRST_CONFIRMED_DTC = 0x0C
MOST_RECENT_TEST_FAILED_DTC = 0x0D
MOST_RECENT_CONFIRMED_DTC = 0x0E
MIRROR_MEMORY_DTC_BY_STATUS_MASK = 0x0F
MIRROR_MEMORY_DTC_EXTENDED_DATA_RECORD_BY_DTC_NUMBER = 0x10
NUMBER_OF_MIRROR_MEMORY_DTC_BY_STATUS_MASK = 0x11
NUMBER_OF_EMISSIONS_RELATED_OBD_DTC_BY_STATUS_MASK = 0x12
EMISSIONS_RELATED_OBD_DTC_BY_STATUS_MASK = 0x13
DTC_FAULT_DETECTION_COUNTER = 0x14
DTC_WITH_PERMANENT_STATUS = 0x15
class DTC_STATUS_MASK_TYPE(IntEnum):
TEST_FAILED = 0x01
TEST_FAILED_THIS_OPERATION_CYCLE = 0x02
PENDING_DTC = 0x04
CONFIRMED_DTC = 0x08
TEST_NOT_COMPLETED_SINCE_LAST_CLEAR = 0x10
TEST_FAILED_SINCE_LAST_CLEAR = 0x20
TEST_NOT_COMPLETED_THIS_OPERATION_CYCLE = 0x40
WARNING_INDICATOR_REQUESTED = 0x80
ALL = 0xFF
class DTC_SEVERITY_MASK_TYPE(IntEnum):
MAINTENANCE_ONLY = 0x20
CHECK_AT_NEXT_HALT = 0x40
CHECK_IMMEDIATELY = 0x80
ALL = 0xE0
class CONTROL_PARAMETER_TYPE(IntEnum):
RETURN_CONTROL_TO_ECU = 0
RESET_TO_DEFAULT = 1
FREEZE_CURRENT_STATE = 2
SHORT_TERM_ADJUSTMENT = 3
class ROUTINE_CONTROL_TYPE(IntEnum):
START = 1
STOP = 2
REQUEST_RESULTS = 3
class ROUTINE_IDENTIFIER_TYPE(IntEnum):
ERASE_MEMORY = 0xFF00
CHECK_PROGRAMMING_DEPENDENCIES = 0xFF01
ERASE_MIRROR_MEMORY_DTCS = 0xFF02
class MessageTimeoutError(Exception):
pass
class NegativeResponseError(Exception):
def __init__(self, message, service_id, error_code):
super().__init__()
self.message = message
self.service_id = service_id
self.error_code = error_code
def __str__(self):
return self.message
class InvalidServiceIdError(Exception):
pass
class InvalidSubFunctionError(Exception):
pass
class InvalidSubAddressError(Exception):
pass
_negative_response_codes = {
0x00: 'positive response',
0x10: 'general reject',
0x11: 'service not supported',
0x12: 'sub-function not supported',
0x13: 'incorrect message length or invalid format',
0x14: 'response too long',
0x21: 'busy repeat request',
0x22: 'conditions not correct',
0x24: 'request sequence error',
0x25: 'no response from subnet component',
0x26: 'failure prevents execution of requested action',
0x31: 'request out of range',
0x33: 'security access denied',
0x35: 'invalid key',
0x36: 'exceed number of attempts',
0x37: 'required time delay not expired',
0x70: 'upload download not accepted',
0x71: 'transfer data suspended',
0x72: 'general programming failure',
0x73: 'wrong block sequence counter',
0x78: 'request correctly received - response pending',
0x7e: 'sub-function not supported in active session',
0x7f: 'service not supported in active session',
0x81: 'rpm too high',
0x82: 'rpm too low',
0x83: 'engine is running',
0x84: 'engine is not running',
0x85: 'engine run time too low',
0x86: 'temperature too high',
0x87: 'temperature too low',
0x88: 'vehicle speed too high',
0x89: 'vehicle speed too low',
0x8a: 'throttle/pedal too high',
0x8b: 'throttle/pedal too low',
0x8c: 'transmission not in neutral',
0x8d: 'transmission not in gear',
0x8f: 'brake switch(es) not closed',
0x90: 'shifter lever not in park',
0x91: 'torque converter clutch locked',
0x92: 'voltage too high',
0x93: 'voltage too low',
}
def get_dtc_num_as_str(dtc_num_bytes):
# ISO 15031-6
designator = {
0b00: "P",
0b01: "C",
0b10: "B",
0b11: "U",
}
d = designator[dtc_num_bytes[0] >> 6]
n = bytes([dtc_num_bytes[0] & 0x3F]) + dtc_num_bytes[1:]
return d + n.hex()
def get_dtc_status_names(status):
result = list()
for m in DTC_STATUS_MASK_TYPE:
if m == DTC_STATUS_MASK_TYPE.ALL:
continue
if status & m.value:
result.append(m.name)
return result
class CanClient:
def __init__(self, can_send: Callable[[int, bytes, int], None], can_recv: Callable[[], list[tuple[int, bytes, int]]],
tx_addr: int, rx_addr: int, bus: int, sub_addr: int | None = None, debug: bool = False):
self.tx = can_send
self.rx = can_recv
self.tx_addr = tx_addr
self.rx_addr = rx_addr
self.rx_buff: deque[bytes] = deque()
self.sub_addr = sub_addr
self.bus = bus
self.debug = debug
def _recv_filter(self, bus: int, addr: int) -> bool:
# handle functional addresses (switch to first addr to respond)
if self.tx_addr == 0x7DF:
is_response = addr >= 0x7E8 and addr <= 0x7EF
if is_response:
if self.debug:
print(f"switch to physical addr {hex(addr)}")
self.tx_addr = addr - 8
self.rx_addr = addr
return is_response
if self.tx_addr == 0x18DB33F1:
is_response = addr >= 0x18DAF100 and addr <= 0x18DAF1FF
if is_response:
if self.debug:
print(f"switch to physical addr {hex(addr)}")
self.tx_addr = 0x18DA00F1 + (addr << 8 & 0xFF00)
self.rx_addr = addr
return bus == self.bus and addr == self.rx_addr
def _recv_buffer(self, drain: bool = False) -> None:
while True:
msgs = self.rx()
if drain:
if self.debug:
print(f"CAN-RX: drain - {len(msgs)}")
self.rx_buff.clear()
else:
for rx_addr, rx_data, rx_bus in msgs or []:
if self._recv_filter(rx_bus, rx_addr) and len(rx_data) > 0:
rx_data = bytes(rx_data) # convert bytearray to bytes
if self.debug:
print(f"CAN-RX: {hex(rx_addr)} - 0x{bytes.hex(rx_data)}")
# Cut off sub addr in first byte
if self.sub_addr is not None:
if rx_data[0] != self.sub_addr:
raise InvalidSubAddressError(f"isotp - rx: invalid sub-address: {rx_data[0]}, expected: {self.sub_addr}")
rx_data = rx_data[1:]
self.rx_buff.append(rx_data)
# break when non-full buffer is processed
if len(msgs) < 254:
return
def recv(self, drain: bool = False) -> Generator[bytes, None, None]:
# buffer rx messages in case two response messages are received at once
# (e.g. response pending and success/failure response)
self._recv_buffer(drain)
try:
while True:
yield self.rx_buff.popleft()
except IndexError:
pass # empty
def send(self, msgs: list[bytes], delay: float = 0) -> None:
for i, msg in enumerate(msgs):
if delay and i != 0:
if self.debug:
print(f"CAN-TX: delay - {delay}")
time.sleep(delay)
if self.sub_addr is not None:
msg = bytes([self.sub_addr]) + msg
if self.debug:
print(f"CAN-TX: {hex(self.tx_addr)} - 0x{bytes.hex(msg)}")
assert len(msg) <= 8
self.tx(self.tx_addr, msg, self.bus)
# prevent rx buffer from overflowing on large tx
if i % 10 == 9:
self._recv_buffer()
class IsoTpMessage:
def __init__(self, can_client: CanClient, timeout: float = 1, single_frame_mode: bool = False, separation_time: float = 0,
debug: bool = False, max_len: int = 8):
self._can_client = can_client
self.timeout = timeout
self.single_frame_mode = single_frame_mode
self.debug = debug
self.max_len = max_len
# <= 127, separation time in milliseconds
# 0xF1 to 0xF9 UF, 100 to 900 microseconds
if 1e-4 <= separation_time <= 9e-4:
offset = int(round(separation_time, 4) * 1e4) - 1
separation_time = 0xF1 + offset
elif 0 <= separation_time <= 0.127:
separation_time = round(separation_time * 1000)
else:
raise Exception("Separation time not in range")
self.flow_control_msg = bytes([
0x30, # flow control
0x01 if self.single_frame_mode else 0x00, # block size
separation_time,
]).ljust(self.max_len, b"\x00")
def send(self, dat: bytes, setup_only: bool = False) -> None:
# throw away any stale data
self._can_client.recv(drain=True)
self.tx_dat = dat
self.tx_len = len(dat)
self.tx_idx = 0
self.tx_done = False
self.rx_dat = b""
self.rx_len = 0
self.rx_idx = 0
self.rx_done = False
if self.debug and not setup_only:
print(f"ISO-TP: REQUEST - {hex(self._can_client.tx_addr)} 0x{bytes.hex(self.tx_dat)}")
self._tx_first_frame(setup_only=setup_only)
def _tx_first_frame(self, setup_only: bool = False) -> None:
if self.tx_len < self.max_len:
# single frame (send all bytes)
if self.debug and not setup_only:
print(f"ISO-TP: TX - single frame - {hex(self._can_client.tx_addr)}")
msg = (bytes([self.tx_len]) + self.tx_dat).ljust(self.max_len, b"\x00")
self.tx_done = True
else:
# first frame (send first 6 bytes)
if self.debug and not setup_only:
print(f"ISO-TP: TX - first frame - {hex(self._can_client.tx_addr)}")
msg = (struct.pack("!H", 0x1000 | self.tx_len) + self.tx_dat[:self.max_len - 2]).ljust(self.max_len - 2, b"\x00")
if not setup_only:
self._can_client.send([msg])
def recv(self, timeout=None) -> tuple[bytes | None, bool]:
if timeout is None:
timeout = self.timeout
start_time = time.monotonic()
rx_in_progress = False
try:
while True:
for msg in self._can_client.recv():
frame_type = self._isotp_rx_next(msg)
start_time = time.monotonic()
# Anything that signifies we're building a response
rx_in_progress = frame_type in (ISOTP_FRAME_TYPE.FIRST, ISOTP_FRAME_TYPE.CONSECUTIVE)
if self.tx_done and self.rx_done:
return self.rx_dat, False
# no timeout indicates non-blocking
if timeout == 0:
return None, rx_in_progress
if time.monotonic() - start_time > timeout:
raise MessageTimeoutError("timeout waiting for response")
finally:
if self.debug and self.rx_dat:
print(f"ISO-TP: RESPONSE - {hex(self._can_client.rx_addr)} 0x{bytes.hex(self.rx_dat)}")
def _isotp_rx_next(self, rx_data: bytes) -> ISOTP_FRAME_TYPE:
# TODO: Handle CAN frame data optimization, which is allowed with some frame types
# # ISO 15765-2 specifies an eight byte CAN frame for ISO-TP communication
# assert len(rx_data) == self.max_len, f"isotp - rx: invalid CAN frame length: {len(rx_data)}"
if rx_data[0] >> 4 == ISOTP_FRAME_TYPE.SINGLE:
assert self.rx_dat == b"" or self.rx_done, "isotp - rx: single frame with active frame"
self.rx_len = rx_data[0] & 0x0F
assert self.rx_len < self.max_len, f"isotp - rx: invalid single frame length: {self.rx_len}"
self.rx_dat = rx_data[1:1 + self.rx_len]
self.rx_idx = 0
self.rx_done = True
if self.debug:
print(f"ISO-TP: RX - single frame - {hex(self._can_client.rx_addr)} idx={self.rx_idx} done={self.rx_done}")
return ISOTP_FRAME_TYPE.SINGLE
elif rx_data[0] >> 4 == ISOTP_FRAME_TYPE.FIRST:
# Once a first frame is received, further frames must be consecutive
assert self.rx_dat == b"" or self.rx_done, "isotp - rx: first frame with active frame"
self.rx_len = ((rx_data[0] & 0x0F) << 8) + rx_data[1]
assert self.rx_len >= self.max_len, f"isotp - rx: invalid first frame length: {self.rx_len}"
assert len(rx_data) == self.max_len, f"isotp - rx: invalid CAN frame length: {len(rx_data)}"
self.rx_dat = rx_data[2:]
self.rx_idx = 0
self.rx_done = False
if self.debug:
print(f"ISO-TP: RX - first frame - {hex(self._can_client.rx_addr)} idx={self.rx_idx} done={self.rx_done}")
if self.debug:
print(f"ISO-TP: TX - flow control continue - {hex(self._can_client.tx_addr)}")
# send flow control message
self._can_client.send([self.flow_control_msg])
return ISOTP_FRAME_TYPE.FIRST
elif rx_data[0] >> 4 == ISOTP_FRAME_TYPE.CONSECUTIVE:
assert not self.rx_done, "isotp - rx: consecutive frame with no active frame"
self.rx_idx += 1
assert self.rx_idx & 0xF == rx_data[0] & 0xF, "isotp - rx: invalid consecutive frame index"
rx_size = self.rx_len - len(self.rx_dat)
self.rx_dat += rx_data[1:1 + rx_size]
if self.rx_len == len(self.rx_dat):
self.rx_done = True
elif self.single_frame_mode:
# notify ECU to send next frame
self._can_client.send([self.flow_control_msg])
if self.debug:
print(f"ISO-TP: RX - consecutive frame - {hex(self._can_client.rx_addr)} idx={self.rx_idx} done={self.rx_done}")
return ISOTP_FRAME_TYPE.CONSECUTIVE
elif rx_data[0] >> 4 == ISOTP_FRAME_TYPE.FLOW:
assert not self.tx_done, "isotp - rx: flow control with no active frame"
assert rx_data[0] != 0x32, "isotp - rx: flow-control overflow/abort"
assert rx_data[0] == 0x30 or rx_data[0] == 0x31, "isotp - rx: flow-control transfer state indicator invalid"
if rx_data[0] == 0x30:
if self.debug:
print(f"ISO-TP: RX - flow control continue - {hex(self._can_client.tx_addr)}")
delay_ts = rx_data[2] & 0x7F
# scale is 1 milliseconds if first bit == 0, 100 micro seconds if first bit == 1
delay_div = 1000. if rx_data[2] & 0x80 == 0 else 10000.
delay_sec = delay_ts / delay_div
# first frame = 6 bytes, each consecutive frame = 7 bytes
num_bytes = self.max_len - 1
start = self.max_len - 2 + self.tx_idx * num_bytes
count = rx_data[1]
end = start + count * num_bytes if count > 0 else self.tx_len
tx_msgs = []
for i in range(start, end, num_bytes):
self.tx_idx += 1
# consecutive tx messages
msg = (bytes([0x20 | (self.tx_idx & 0xF)]) + self.tx_dat[i:i + num_bytes]).ljust(self.max_len, b"\x00")
tx_msgs.append(msg)
# send consecutive tx messages
self._can_client.send(tx_msgs, delay=delay_sec)
if end >= self.tx_len:
self.tx_done = True
if self.debug:
print(f"ISO-TP: TX - consecutive frame - {hex(self._can_client.tx_addr)} idx={self.tx_idx} done={self.tx_done}")
elif rx_data[0] == 0x31:
# wait (do nothing until next flow control message)
if self.debug:
print(f"ISO-TP: TX - flow control wait - {hex(self._can_client.tx_addr)}")
return ISOTP_FRAME_TYPE.FLOW
# 4-15 - reserved
else:
raise Exception(f"isotp - rx: invalid frame type: {rx_data[0] >> 4}")
FUNCTIONAL_ADDRS = [0x7DF, 0x18DB33F1]
def get_rx_addr_for_tx_addr(tx_addr, rx_offset=0x8):
if tx_addr in FUNCTIONAL_ADDRS:
return None
if tx_addr < 0xFFF8:
# pseudo-standard 11 bit response addr (add 8) works for most manufacturers
# allow override; some manufacturers use other offsets for non-OBD2 access
return tx_addr + rx_offset
if tx_addr > 0x10000000 and tx_addr < 0xFFFFFFFF:
# standard 29 bit response addr (flip last two bytes)
return (tx_addr & 0xFFFF0000) + (tx_addr << 8 & 0xFF00) + (tx_addr >> 8 & 0xFF)
raise ValueError(f"invalid tx_addr: {tx_addr}")
class UdsClient:
def __init__(self, panda, tx_addr: int, rx_addr: int | None = None, bus: int = 0, sub_addr: int | None = None, timeout: float = 1,
debug: bool = False, tx_timeout: float = 1, response_pending_timeout: float = 10):
self.bus = bus
self.tx_addr = tx_addr
self.rx_addr = rx_addr if rx_addr is not None else get_rx_addr_for_tx_addr(tx_addr)
self.sub_addr = sub_addr
self.timeout = timeout
self.debug = debug
can_send_with_timeout = partial(panda.can_send, timeout=int(tx_timeout*1000))
self._can_client = CanClient(can_send_with_timeout, panda.can_recv, self.tx_addr, self.rx_addr, self.bus, self.sub_addr, debug=self.debug)
self.response_pending_timeout = response_pending_timeout
# generic uds request
def _uds_request(self, service_type: SERVICE_TYPE, subfunction: int | None = None, data: bytes | None = None) -> bytes:
req = bytes([service_type])
if subfunction is not None:
req += bytes([subfunction])
if data is not None:
req += data
# send request, wait for response
max_len = 8 if self.sub_addr is None else 7
isotp_msg = IsoTpMessage(self._can_client, timeout=self.timeout, debug=self.debug, max_len=max_len)
isotp_msg.send(req)
response_pending = False
while True:
timeout = self.response_pending_timeout if response_pending else self.timeout
resp, _ = isotp_msg.recv(timeout)
if resp is None:
continue
response_pending = False
resp_sid = resp[0] if len(resp) > 0 else None
# negative response
if resp_sid == 0x7F:
service_id = resp[1] if len(resp) > 1 else -1
try:
service_desc = SERVICE_TYPE(service_id).name
except BaseException:
service_desc = 'NON_STANDARD_SERVICE'
error_code = resp[2] if len(resp) > 2 else -1
try:
error_desc = _negative_response_codes[error_code]
except BaseException:
error_desc = resp[3:].hex()
# wait for another message if response pending
if error_code == 0x78:
response_pending = True
if self.debug:
print("UDS-RX: response pending")
continue
raise NegativeResponseError(f'{service_desc} - {error_desc}', service_id, error_code)
# positive response
if service_type + 0x40 != resp_sid:
resp_sid_hex = hex(resp_sid) if resp_sid is not None else None
raise InvalidServiceIdError(f'invalid response service id: {resp_sid_hex}')
if subfunction is not None:
resp_sfn = resp[1] if len(resp) > 1 else None
if subfunction != resp_sfn:
resp_sfn_hex = hex(resp_sfn) if resp_sfn is not None else None
raise InvalidSubFunctionError(f'invalid response subfunction: {resp_sfn_hex}')
# return data (exclude service id and sub-function id)
return resp[(1 if subfunction is None else 2):]
# services
def diagnostic_session_control(self, session_type: SESSION_TYPE):
self._uds_request(SERVICE_TYPE.DIAGNOSTIC_SESSION_CONTROL, subfunction=session_type)
def ecu_reset(self, reset_type: RESET_TYPE):
resp = self._uds_request(SERVICE_TYPE.ECU_RESET, subfunction=reset_type)
power_down_time = None
if reset_type == RESET_TYPE.ENABLE_RAPID_POWER_SHUTDOWN:
power_down_time = resp[0]
return power_down_time
def security_access(self, access_type: ACCESS_TYPE, security_key: bytes = b'', data_record: bytes = b''):
request_seed = access_type % 2 != 0
if request_seed and len(security_key) != 0:
raise ValueError('security_key not allowed')
if not request_seed and len(security_key) == 0:
raise ValueError('security_key is missing')
if not request_seed and len(data_record) != 0:
raise ValueError('data_record not allowed')
data = security_key + data_record
resp = self._uds_request(SERVICE_TYPE.SECURITY_ACCESS, subfunction=access_type, data=data)
if request_seed:
security_seed = resp
return security_seed
def communication_control(self, control_type: CONTROL_TYPE, message_type: MESSAGE_TYPE):
data = bytes([message_type])
self._uds_request(SERVICE_TYPE.COMMUNICATION_CONTROL, subfunction=control_type, data=data)
def tester_present(self, ):
self._uds_request(SERVICE_TYPE.TESTER_PRESENT, subfunction=0x00)
def access_timing_parameter(self, timing_parameter_type: TIMING_PARAMETER_TYPE, parameter_values: bytes | None = None):
write_custom_values = timing_parameter_type == TIMING_PARAMETER_TYPE.SET_TO_GIVEN_VALUES
read_values = (timing_parameter_type == TIMING_PARAMETER_TYPE.READ_CURRENTLY_ACTIVE or
timing_parameter_type == TIMING_PARAMETER_TYPE.READ_EXTENDED_SET)
if not write_custom_values and parameter_values is not None:
raise ValueError('parameter_values not allowed')
if write_custom_values and parameter_values is None:
raise ValueError('parameter_values is missing')
resp = self._uds_request(SERVICE_TYPE.ACCESS_TIMING_PARAMETER, subfunction=timing_parameter_type, data=parameter_values)
if read_values:
# TODO: parse response into values?
parameter_values = resp
return parameter_values
def secured_data_transmission(self, data: bytes):
# TODO: split data into multiple input parameters?
resp = self._uds_request(SERVICE_TYPE.SECURED_DATA_TRANSMISSION, subfunction=None, data=data)
# TODO: parse response into multiple output values?
return resp
def control_dtc_setting(self, dtc_setting_type: DTC_SETTING_TYPE):
self._uds_request(SERVICE_TYPE.CONTROL_DTC_SETTING, subfunction=dtc_setting_type)
def response_on_event(self, response_event_type: RESPONSE_EVENT_TYPE, store_event: bool, window_time: int,
event_type_record: int, service_response_record: int):
if store_event:
response_event_type |= 0x20 # type: ignore
# TODO: split record parameters into arrays
data = bytes([window_time, event_type_record, service_response_record])
resp = self._uds_request(SERVICE_TYPE.RESPONSE_ON_EVENT, subfunction=response_event_type, data=data)
if response_event_type == RESPONSE_EVENT_TYPE.REPORT_ACTIVATED_EVENTS:
return {
"num_of_activated_events": resp[0],
"data": resp[1:], # TODO: parse the reset of response
}
return {
"num_of_identified_events": resp[0],
"event_window_time": resp[1],
"data": resp[2:], # TODO: parse the reset of response
}
def link_control(self, link_control_type: LINK_CONTROL_TYPE, baud_rate_type: BAUD_RATE_TYPE | None = None):
data: bytes | None
if link_control_type == LINK_CONTROL_TYPE.VERIFY_BAUDRATE_TRANSITION_WITH_FIXED_BAUDRATE:
# baud_rate_type = BAUD_RATE_TYPE
data = bytes([cast(int, baud_rate_type)])
elif link_control_type == LINK_CONTROL_TYPE.VERIFY_BAUDRATE_TRANSITION_WITH_SPECIFIC_BAUDRATE:
# baud_rate_type = custom value (3 bytes big-endian)
data = struct.pack('!I', baud_rate_type)[1:]
else:
data = None
self._uds_request(SERVICE_TYPE.LINK_CONTROL, subfunction=link_control_type, data=data)
def read_data_by_identifier(self, data_identifier_type: DATA_IDENTIFIER_TYPE):
# TODO: support list of identifiers
data = struct.pack('!H', data_identifier_type)
resp = self._uds_request(SERVICE_TYPE.READ_DATA_BY_IDENTIFIER, subfunction=None, data=data)
resp_id = struct.unpack('!H', resp[0:2])[0] if len(resp) >= 2 else None
if resp_id != data_identifier_type:
raise ValueError(f'invalid response data identifier: {hex(resp_id)} expected: {hex(data_identifier_type)}')
return resp[2:]
def read_memory_by_address(self, memory_address: int, memory_size: int, memory_address_bytes: int = 4, memory_size_bytes: int = 1):
if memory_address_bytes < 1 or memory_address_bytes > 4:
raise ValueError(f'invalid memory_address_bytes: {memory_address_bytes}')
if memory_size_bytes < 1 or memory_size_bytes > 4:
raise ValueError(f'invalid memory_size_bytes: {memory_size_bytes}')
data = bytes([memory_size_bytes << 4 | memory_address_bytes])
if memory_address >= 1 << (memory_address_bytes * 8):
raise ValueError(f'invalid memory_address: {memory_address}')
data += struct.pack('!I', memory_address)[4 - memory_address_bytes:]
if memory_size >= 1 << (memory_size_bytes * 8):
raise ValueError(f'invalid memory_size: {memory_size}')
data += struct.pack('!I', memory_size)[4 - memory_size_bytes:]
resp = self._uds_request(SERVICE_TYPE.READ_MEMORY_BY_ADDRESS, subfunction=None, data=data)
return resp
def read_scaling_data_by_identifier(self, data_identifier_type: DATA_IDENTIFIER_TYPE):
data = struct.pack('!H', data_identifier_type)
resp = self._uds_request(SERVICE_TYPE.READ_SCALING_DATA_BY_IDENTIFIER, subfunction=None, data=data)
resp_id = struct.unpack('!H', resp[0:2])[0] if len(resp) >= 2 else None
if resp_id != data_identifier_type:
raise ValueError(f'invalid response data identifier: {hex(resp_id)}')
return resp[2:] # TODO: parse the response
def read_data_by_periodic_identifier(self, transmission_mode_type: TRANSMISSION_MODE_TYPE, periodic_data_identifier: int):
# TODO: support list of identifiers
data = bytes([transmission_mode_type, periodic_data_identifier])
self._uds_request(SERVICE_TYPE.READ_DATA_BY_PERIODIC_IDENTIFIER, subfunction=None, data=data)
def dynamically_define_data_identifier(self, dynamic_definition_type: DYNAMIC_DEFINITION_TYPE, dynamic_data_identifier: int,
source_definitions: list[DynamicSourceDefinition], memory_address_bytes: int = 4, memory_size_bytes: int = 1):
if memory_address_bytes < 1 or memory_address_bytes > 4:
raise ValueError(f'invalid memory_address_bytes: {memory_address_bytes}')
if memory_size_bytes < 1 or memory_size_bytes > 4:
raise ValueError(f'invalid memory_size_bytes: {memory_size_bytes}')
data = struct.pack('!H', dynamic_data_identifier)
if dynamic_definition_type == DYNAMIC_DEFINITION_TYPE.DEFINE_BY_IDENTIFIER:
for s in source_definitions:
data += struct.pack('!H', s.data_identifier) + bytes([s.position, s.memory_size])
elif dynamic_definition_type == DYNAMIC_DEFINITION_TYPE.DEFINE_BY_MEMORY_ADDRESS:
data += bytes([memory_size_bytes << 4 | memory_address_bytes])
for s in source_definitions:
if s.memory_address >= 1 << (memory_address_bytes * 8):
raise ValueError(f'invalid memory_address: {s.memory_address}')
data += struct.pack('!I', s.memory_address)[4 - memory_address_bytes:]
if s.memory_size >= 1 << (memory_size_bytes * 8):
raise ValueError(f'invalid memory_size: {s.memory_size}')
data += struct.pack('!I', s.memory_size)[4 - memory_size_bytes:]
elif dynamic_definition_type == DYNAMIC_DEFINITION_TYPE.CLEAR_DYNAMICALLY_DEFINED_DATA_IDENTIFIER:
pass
else:
raise ValueError(f'invalid dynamic identifier type: {hex(dynamic_definition_type)}')
self._uds_request(SERVICE_TYPE.DYNAMICALLY_DEFINE_DATA_IDENTIFIER, subfunction=dynamic_definition_type, data=data)
def write_data_by_identifier(self, data_identifier_type: DATA_IDENTIFIER_TYPE, data_record: bytes):
data = struct.pack('!H', data_identifier_type) + data_record
resp = self._uds_request(SERVICE_TYPE.WRITE_DATA_BY_IDENTIFIER, subfunction=None, data=data)
resp_id = struct.unpack('!H', resp[0:2])[0] if len(resp) >= 2 else None
if resp_id != data_identifier_type:
raise ValueError(f'invalid response data identifier: {hex(resp_id)}')
def write_memory_by_address(self, memory_address: int, memory_size: int, data_record: bytes, memory_address_bytes: int = 4, memory_size_bytes: int = 1):
if memory_address_bytes < 1 or memory_address_bytes > 4:
raise ValueError(f'invalid memory_address_bytes: {memory_address_bytes}')
if memory_size_bytes < 1 or memory_size_bytes > 4:
raise ValueError(f'invalid memory_size_bytes: {memory_size_bytes}')
data = bytes([memory_size_bytes << 4 | memory_address_bytes])
if memory_address >= 1 << (memory_address_bytes * 8):
raise ValueError(f'invalid memory_address: {memory_address}')
data += struct.pack('!I', memory_address)[4 - memory_address_bytes:]
if memory_size >= 1 << (memory_size_bytes * 8):
raise ValueError(f'invalid memory_size: {memory_size}')
data += struct.pack('!I', memory_size)[4 - memory_size_bytes:]
data += data_record
self._uds_request(SERVICE_TYPE.WRITE_MEMORY_BY_ADDRESS, subfunction=None, data=data)
def clear_diagnostic_information(self, dtc_group_type: DTC_GROUP_TYPE):
data = struct.pack('!I', dtc_group_type)[1:] # 3 bytes
self._uds_request(SERVICE_TYPE.CLEAR_DIAGNOSTIC_INFORMATION, subfunction=None, data=data)
def read_dtc_information(self, dtc_report_type: DTC_REPORT_TYPE, dtc_status_mask_type: DTC_STATUS_MASK_TYPE = DTC_STATUS_MASK_TYPE.ALL,
dtc_severity_mask_type: DTC_SEVERITY_MASK_TYPE = DTC_SEVERITY_MASK_TYPE.ALL, dtc_mask_record: int = 0xFFFFFF,
dtc_snapshot_record_num: int = 0xFF, dtc_extended_record_num: int = 0xFF):
data = b''
# dtc_status_mask_type
if dtc_report_type == DTC_REPORT_TYPE.NUMBER_OF_DTC_BY_STATUS_MASK or \
dtc_report_type == DTC_REPORT_TYPE.DTC_BY_STATUS_MASK or \
dtc_report_type == DTC_REPORT_TYPE.MIRROR_MEMORY_DTC_BY_STATUS_MASK or \
dtc_report_type == DTC_REPORT_TYPE.NUMBER_OF_MIRROR_MEMORY_DTC_BY_STATUS_MASK or \
dtc_report_type == DTC_REPORT_TYPE.NUMBER_OF_EMISSIONS_RELATED_OBD_DTC_BY_STATUS_MASK or \
dtc_report_type == DTC_REPORT_TYPE.EMISSIONS_RELATED_OBD_DTC_BY_STATUS_MASK:
data += bytes([dtc_status_mask_type])
# dtc_mask_record
if dtc_report_type == DTC_REPORT_TYPE.DTC_SNAPSHOT_IDENTIFICATION or \
dtc_report_type == DTC_REPORT_TYPE.DTC_SNAPSHOT_RECORD_BY_DTC_NUMBER or \
dtc_report_type == DTC_REPORT_TYPE.DTC_EXTENDED_DATA_RECORD_BY_DTC_NUMBER or \
dtc_report_type == DTC_REPORT_TYPE.MIRROR_MEMORY_DTC_EXTENDED_DATA_RECORD_BY_DTC_NUMBER or \
dtc_report_type == DTC_REPORT_TYPE.SEVERITY_INFORMATION_OF_DTC:
data += struct.pack('!I', dtc_mask_record)[1:] # 3 bytes
# dtc_snapshot_record_num
if dtc_report_type == DTC_REPORT_TYPE.DTC_SNAPSHOT_IDENTIFICATION or \
dtc_report_type == DTC_REPORT_TYPE.DTC_SNAPSHOT_RECORD_BY_DTC_NUMBER or \
dtc_report_type == DTC_REPORT_TYPE.DTC_SNAPSHOT_RECORD_BY_RECORD_NUMBER:
data += bytes([dtc_snapshot_record_num])
# dtc_extended_record_num
if dtc_report_type == DTC_REPORT_TYPE.DTC_EXTENDED_DATA_RECORD_BY_DTC_NUMBER or \
dtc_report_type == DTC_REPORT_TYPE.MIRROR_MEMORY_DTC_EXTENDED_DATA_RECORD_BY_DTC_NUMBER:
data += bytes([dtc_extended_record_num])
# dtc_severity_mask_type
if dtc_report_type == DTC_REPORT_TYPE.NUMBER_OF_DTC_BY_SEVERITY_MASK_RECORD or \
dtc_report_type == DTC_REPORT_TYPE.DTC_BY_SEVERITY_MASK_RECORD:
data += bytes([dtc_severity_mask_type, dtc_status_mask_type])
resp = self._uds_request(SERVICE_TYPE.READ_DTC_INFORMATION, subfunction=dtc_report_type, data=data)
# TODO: parse response
return resp
def input_output_control_by_identifier(self, data_identifier_type: DATA_IDENTIFIER_TYPE, control_parameter_type: CONTROL_PARAMETER_TYPE,
control_option_record: bytes = b'', control_enable_mask_record: bytes = b''):
data = struct.pack('!H', data_identifier_type) + bytes([control_parameter_type]) + control_option_record + control_enable_mask_record
resp = self._uds_request(SERVICE_TYPE.INPUT_OUTPUT_CONTROL_BY_IDENTIFIER, subfunction=None, data=data)
resp_id = struct.unpack('!H', resp[0:2])[0] if len(resp) >= 2 else None
if resp_id != data_identifier_type:
raise ValueError(f'invalid response data identifier: {hex(resp_id)}')
return resp[2:]
def routine_control(self, routine_control_type: ROUTINE_CONTROL_TYPE, routine_identifier_type: ROUTINE_IDENTIFIER_TYPE, routine_option_record: bytes = b''):
data = struct.pack('!H', routine_identifier_type) + routine_option_record
resp = self._uds_request(SERVICE_TYPE.ROUTINE_CONTROL, subfunction=routine_control_type, data=data)
resp_id = struct.unpack('!H', resp[0:2])[0] if len(resp) >= 2 else None
if resp_id != routine_identifier_type:
raise ValueError(f'invalid response routine identifier: {hex(resp_id)}')
return resp[2:]
def request_download(self, memory_address: int, memory_size: int, memory_address_bytes: int = 4, memory_size_bytes: int = 4, data_format: int = 0x00):
data = bytes([data_format])
if memory_address_bytes < 1 or memory_address_bytes > 4:
raise ValueError(f'invalid memory_address_bytes: {memory_address_bytes}')
if memory_size_bytes < 1 or memory_size_bytes > 4:
raise ValueError(f'invalid memory_size_bytes: {memory_size_bytes}')
data += bytes([memory_size_bytes << 4 | memory_address_bytes])
if memory_address >= 1 << (memory_address_bytes * 8):
raise ValueError(f'invalid memory_address: {memory_address}')
data += struct.pack('!I', memory_address)[4 - memory_address_bytes:]
if memory_size >= 1 << (memory_size_bytes * 8):
raise ValueError(f'invalid memory_size: {memory_size}')
data += struct.pack('!I', memory_size)[4 - memory_size_bytes:]
resp = self._uds_request(SERVICE_TYPE.REQUEST_DOWNLOAD, subfunction=None, data=data)
max_num_bytes_len = resp[0] >> 4 if len(resp) > 0 else 0
if max_num_bytes_len >= 1 and max_num_bytes_len <= 4:
max_num_bytes = struct.unpack('!I', (b"\x00" * (4 - max_num_bytes_len)) + resp[1:max_num_bytes_len + 1])[0]
else:
raise ValueError(f'invalid max_num_bytes_len: {max_num_bytes_len}')
return max_num_bytes # max number of bytes per transfer data request
def request_upload(self, memory_address: int, memory_size: int, memory_address_bytes: int = 4, memory_size_bytes: int = 4, data_format: int = 0x00):
data = bytes([data_format])
if memory_address_bytes < 1 or memory_address_bytes > 4:
raise ValueError(f'invalid memory_address_bytes: {memory_address_bytes}')
if memory_size_bytes < 1 or memory_size_bytes > 4:
raise ValueError(f'invalid memory_size_bytes: {memory_size_bytes}')
data += bytes([memory_size_bytes << 4 | memory_address_bytes])
if memory_address >= 1 << (memory_address_bytes * 8):
raise ValueError(f'invalid memory_address: {memory_address}')
data += struct.pack('!I', memory_address)[4 - memory_address_bytes:]
if memory_size >= 1 << (memory_size_bytes * 8):
raise ValueError(f'invalid memory_size: {memory_size}')
data += struct.pack('!I', memory_size)[4 - memory_size_bytes:]
resp = self._uds_request(SERVICE_TYPE.REQUEST_UPLOAD, subfunction=None, data=data)
max_num_bytes_len = resp[0] >> 4 if len(resp) > 0 else 0
if max_num_bytes_len >= 1 and max_num_bytes_len <= 4:
max_num_bytes = struct.unpack('!I', (b"\x00" * (4 - max_num_bytes_len)) + resp[1:max_num_bytes_len + 1])[0]
else:
raise ValueError(f'invalid max_num_bytes_len: {max_num_bytes_len}')
return max_num_bytes # max number of bytes per transfer data request
def transfer_data(self, block_sequence_count: int, data: bytes = b''):
data = bytes([block_sequence_count]) + data
resp = self._uds_request(SERVICE_TYPE.TRANSFER_DATA, subfunction=None, data=data)
resp_id = resp[0] if len(resp) > 0 else None
if resp_id != block_sequence_count:
raise ValueError(f'invalid block_sequence_count: {resp_id}')
return resp[1:]
def request_transfer_exit(self):
self._uds_request(SERVICE_TYPE.REQUEST_TRANSFER_EXIT, subfunction=None)

258
python/xcp.py
View File

@@ -1,258 +0,0 @@
import sys
import time
import struct
from enum import IntEnum
class COMMAND_CODE(IntEnum):
CONNECT = 0xFF
DISCONNECT = 0xFE
GET_STATUS = 0xFD
SYNCH = 0xFC
GET_COMM_MODE_INFO = 0xFB
GET_ID = 0xFA
SET_REQUEST = 0xF9
GET_SEED = 0xF8
UNLOCK = 0xF7
SET_MTA = 0xF6
UPLOAD = 0xF5
SHORT_UPLOAD = 0xF4
BUILD_CHECKSUM = 0xF3
TRANSPORT_LAYER_CMD = 0xF2
USER_CMD = 0xF1
DOWNLOAD = 0xF0
DOWNLOAD_NEXT = 0xEF
DOWNLOAD_MAX = 0xEE
SHORT_DOWNLOAD = 0xED
MODIFY_BITS = 0xEC
SET_CAL_PAGE = 0xEB
GET_CAL_PAGE = 0xEA
GET_PAG_PROCESSOR_INFO = 0xE9
GET_SEGMENT_INFO = 0xE8
GET_PAGE_INFO = 0xE7
SET_SEGMENT_MODE = 0xE6
GET_SEGMENT_MODE = 0xE5
COPY_CAL_PAGE = 0xE4
CLEAR_DAQ_LIST = 0xE3
SET_DAQ_PTR = 0xE2
WRITE_DAQ = 0xE1
SET_DAQ_LIST_MODE = 0xE0
GET_DAQ_LIST_MODE = 0xDF
START_STOP_DAQ_LIST = 0xDE
START_STOP_SYNCH = 0xDD
GET_DAQ_CLOCK = 0xDC
READ_DAQ = 0xDB
GET_DAQ_PROCESSOR_INFO = 0xDA
GET_DAQ_RESOLUTION_INFO = 0xD9
GET_DAQ_LIST_INFO = 0xD8
GET_DAQ_EVENT_INFO = 0xD7
FREE_DAQ = 0xD6
ALLOC_DAQ = 0xD5
ALLOC_ODT = 0xD4
ALLOC_ODT_ENTRY = 0xD3
PROGRAM_START = 0xD2
PROGRAM_CLEAR = 0xD1
PROGRAM = 0xD0
PROGRAM_RESET = 0xCF
GET_PGM_PROCESSOR_INFO = 0xCE
GET_SECTOR_INFO = 0xCD
PROGRAM_PREPARE = 0xCC
PROGRAM_FORMAT = 0xCB
PROGRAM_NEXT = 0xCA
PROGRAM_MAX = 0xC9
PROGRAM_VERIFY = 0xC8
ERROR_CODES = {
0x00: "Command processor synchronization",
0x10: "Command was not executed",
0x11: "Command rejected because DAQ is running",
0x12: "Command rejected because PGM is running",
0x20: "Unknown command or not implemented optional command",
0x21: "Command syntax invalid",
0x22: "Command syntax valid but command parameter(s) out of range",
0x23: "The memory location is write protected",
0x24: "The memory location is not accessible",
0x25: "Access denied, Seed & Key is required",
0x26: "Selected page not available",
0x27: "Selected page mode not available",
0x28: "Selected segment not valid",
0x29: "Sequence error",
0x2A: "DAQ configuration not valid",
0x30: "Memory overflow error",
0x31: "Generic error",
0x32: "The slave internal program verify routine detects an error",
}
class CONNECT_MODE(IntEnum):
NORMAL = 0x00,
USER_DEFINED = 0x01,
class GET_ID_REQUEST_TYPE(IntEnum):
ASCII = 0x00,
ASAM_MC2_FILE = 0x01,
ASAM_MC2_PATH = 0x02,
ASAM_MC2_URL = 0x03,
ASAM_MC2_UPLOAD = 0x04,
# 128-255 user defined
class CommandTimeoutError(Exception):
pass
class CommandCounterError(Exception):
pass
class CommandResponseError(Exception):
def __init__(self, message, return_code):
super().__init__()
self.message = message
self.return_code = return_code
def __str__(self):
return self.message
class XcpClient():
def __init__(self, panda, tx_addr: int, rx_addr: int, bus: int=0, timeout: float=0.1, debug=False, pad=True):
self.tx_addr = tx_addr
self.rx_addr = rx_addr
self.can_bus = bus
self.timeout = timeout
self.debug = debug
self._panda = panda
self._byte_order = ">"
self._max_cto = 8
self._max_dto = 8
self.pad = pad
def _send_cto(self, cmd: int, dat: bytes = b"") -> None:
tx_data = (bytes([cmd]) + dat)
# Some ECUs don't respond if the packets are not padded to 8 bytes
if self.pad:
tx_data = tx_data.ljust(8, b"\x00")
if self.debug:
print("CAN-CLEAR: TX")
self._panda.can_clear(self.can_bus)
if self.debug:
print("CAN-CLEAR: RX")
self._panda.can_clear(0xFFFF)
if self.debug:
print(f"CAN-TX: {hex(self.tx_addr)} - 0x{bytes.hex(tx_data)}")
self._panda.can_send(self.tx_addr, tx_data, self.can_bus)
def _recv_dto(self, timeout: float) -> bytes:
start_time = time.time()
while time.time() - start_time < timeout:
msgs = self._panda.can_recv() or []
if len(msgs) >= 256:
print("CAN RX buffer overflow!!!", file=sys.stderr)
for rx_addr, rx_data, rx_bus in msgs:
if rx_bus == self.can_bus and rx_addr == self.rx_addr:
rx_data = bytes(rx_data) # convert bytearray to bytes
if self.debug:
print(f"CAN-RX: {hex(rx_addr)} - 0x{bytes.hex(rx_data)}")
pid = rx_data[0]
if pid == 0xFE:
err = rx_data[1]
err_desc = ERROR_CODES.get(err, "unknown error")
dat = rx_data[2:]
raise CommandResponseError(f"{hex(err)} - {err_desc} {dat}", err)
return bytes(rx_data[1:])
time.sleep(0.001)
raise CommandTimeoutError("timeout waiting for response")
# commands
def connect(self, connect_mode: CONNECT_MODE=CONNECT_MODE.NORMAL) -> dict:
self._send_cto(COMMAND_CODE.CONNECT, bytes([connect_mode]))
resp = self._recv_dto(self.timeout)
assert len(resp) == 7, f"incorrect data length: {len(resp)}"
self._byte_order = ">" if resp[1] & 0x01 else "<"
self._slave_block_mode = resp[1] & 0x40 != 0
self._max_cto = resp[2]
self._max_dto = struct.unpack(f"{self._byte_order}H", resp[3:5])[0]
return {
"cal_support": resp[0] & 0x01 != 0,
"daq_support": resp[0] & 0x04 != 0,
"stim_support": resp[0] & 0x08 != 0,
"pgm_support": resp[0] & 0x10 != 0,
"byte_order": self._byte_order,
"address_granularity": 2**((resp[1] & 0x06) >> 1),
"slave_block_mode": self._slave_block_mode,
"optional": resp[1] & 0x80 != 0,
"max_cto": self._max_cto,
"max_dto": self._max_dto,
"protocol_version": resp[5],
"transport_version": resp[6],
}
def disconnect(self) -> None:
self._send_cto(COMMAND_CODE.DISCONNECT)
resp = self._recv_dto(self.timeout)
assert len(resp) == 0, f"incorrect data length: {len(resp)}"
def get_id(self, req_id_type: GET_ID_REQUEST_TYPE = GET_ID_REQUEST_TYPE.ASCII) -> dict:
if req_id_type > 255:
raise ValueError("request id type must be less than 255")
self._send_cto(COMMAND_CODE.GET_ID, bytes([req_id_type]))
resp = self._recv_dto(self.timeout)
return {
# mode = 0 means MTA was set
# mode = 1 means data is at end (only CAN-FD has space for this)
"mode": resp[0],
"length": struct.unpack(f"{self._byte_order}I", resp[3:7])[0],
"identifier": resp[7:] if self._max_cto > 8 else None
}
def get_seed(self, mode: int = 0) -> bytes:
if mode > 255:
raise ValueError("mode must be less than 255")
self._send_cto(COMMAND_CODE.GET_SEED, bytes([0, mode]))
# TODO: add support for longer seeds spread over multiple blocks
ret = self._recv_dto(self.timeout)
length = ret[0]
return ret[1:length+1]
def unlock(self, key: bytes) -> bytes:
# TODO: add support for longer keys spread over multiple blocks
self._send_cto(COMMAND_CODE.UNLOCK, bytes([len(key)]) + key)
return self._recv_dto(self.timeout)
def set_mta(self, addr: int, addr_ext: int = 0) -> bytes:
if addr_ext > 255:
raise ValueError("address extension must be less than 256")
# TODO: this looks broken (missing addr extension)
self._send_cto(COMMAND_CODE.SET_MTA, bytes([0x00, 0x00, addr_ext]) + struct.pack(f"{self._byte_order}I", addr))
return self._recv_dto(self.timeout)
def upload(self, size: int) -> bytes:
if size > 255:
raise ValueError("size must be less than 256")
if not self._slave_block_mode and size > self._max_dto - 1:
raise ValueError("block mode not supported")
self._send_cto(COMMAND_CODE.UPLOAD, bytes([size]))
resp = b""
while len(resp) < size:
resp += self._recv_dto(self.timeout)[:size - len(resp) + 1]
return resp[:size] # trim off bytes with undefined values
def short_upload(self, size: int, addr_ext: int, addr: int) -> bytes:
if size > 6:
raise ValueError("size must be less than 7")
if addr_ext > 255:
raise ValueError("address extension must be less than 256")
self._send_cto(COMMAND_CODE.SHORT_UPLOAD, bytes([size, 0x00, addr_ext]) + struct.pack(f"{self._byte_order}I", addr))
return self._recv_dto(self.timeout)[:size] # trim off bytes with undefined values
def download(self, data: bytes) -> bytes:
size = len(data)
if size > 255:
raise ValueError("size must be less than 256")
if not self._slave_block_mode and size > self._max_dto - 2:
raise ValueError("block mode not supported")
self._send_cto(COMMAND_CODE.DOWNLOAD, bytes([size]) + data)
return self._recv_dto(self.timeout)[:size]