brianhansonxyz/clearpilot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

wip

Browse Source
This commit is contained in:
Your Name
2024-04-27 03:19:19 -05:00
parent df418b503f
commit 03af7b6107
306 changed files with 108529 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
panda/python/__init__.py
View File

@@ -6,10 +6,8 @@ import usb1
import struct
import hashlib
import binascii
import datetime
import logging
from functools import wraps, partial
from typing import Optional
from itertools import accumulate
from .base import BaseHandle
@@ -218,6 +216,7 @@ class Panda:
FLAG_TESLA_POWERTRAIN = 1
FLAG_TESLA_LONG_CONTROL = 2
FLAG_TESLA_RAVEN = 4
FLAG_VOLKSWAGEN_LONG_CONTROL = 1
@@ -238,13 +237,13 @@ class Panda:
FLAG_GM_HW_ASCM_LONG = 16
FLAG_GM_NO_CAMERA = 32
FLAG_GM_NO_ACC = 64
FLAG_GM_PEDAL_LONG = 128
FLAG_GM_PEDAL_LONG = 128 # TODO: This can be inferred
FLAG_GM_GAS_INTERCEPTOR = 256
FLAG_FORD_LONG_CONTROL = 1
FLAG_FORD_CANFD = 2
def __init__(self, serial: Optional[str] = None, claim: bool = True, disable_checks: bool = True, can_speed_kbps: int = 500):
def __init__(self, serial: str | None = None, claim: bool = True, disable_checks: bool = True, can_speed_kbps: int = 500):
self._connect_serial = serial
self._disable_checks = disable_checks
@@ -540,7 +539,7 @@ class Panda:
if reconnect:
self.reconnect()
def recover(self, timeout: Optional[int] = 60, reset: bool = True) -> bool:
def recover(self, timeout: int | None = 60, reset: bool = True) -> bool:
dfu_serial = self.get_dfu_serial()
if reset:
@@ -559,7 +558,7 @@ class Panda:
return True
@staticmethod
def wait_for_dfu(dfu_serial: Optional[str], timeout: Optional[int] = None) -> bool:
def wait_for_dfu(dfu_serial: str | None, timeout: int | None = None) -> bool:
t_start = time.monotonic()
dfu_list = PandaDFU.list()
while (dfu_serial is None and len(dfu_list) == 0) or (dfu_serial is not None and dfu_serial not in dfu_list):
@@ -571,7 +570,7 @@ class Panda:
return True
@staticmethod
def wait_for_panda(serial: Optional[str], timeout: int) -> bool:
def wait_for_panda(serial: str | None, timeout: int) -> bool:
t_start = time.monotonic()
serials = Panda.list()
while (serial is None and len(serials) == 0) or (serial is not None and serial not in serials):
@@ -904,21 +903,6 @@ class Panda:
def set_heartbeat_disabled(self):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xf8, 0, 0, b'')
# ******************* RTC *******************
def set_datetime(self, dt):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xa1, int(dt.year), 0, b'')
self._handle.controlWrite(Panda.REQUEST_OUT, 0xa2, int(dt.month), 0, b'')
self._handle.controlWrite(Panda.REQUEST_OUT, 0xa3, int(dt.day), 0, b'')
self._handle.controlWrite(Panda.REQUEST_OUT, 0xa4, int(dt.isoweekday()), 0, b'')
self._handle.controlWrite(Panda.REQUEST_OUT, 0xa5, int(dt.hour), 0, b'')
self._handle.controlWrite(Panda.REQUEST_OUT, 0xa6, int(dt.minute), 0, b'')
self._handle.controlWrite(Panda.REQUEST_OUT, 0xa7, int(dt.second), 0, b'')
def get_datetime(self):
dat = self._handle.controlRead(Panda.REQUEST_IN, 0xa0, 0, 0, 8)
a = struct.unpack("HBBBBBB", dat)
return datetime.datetime(a[0], a[1], a[2], a[4], a[5], a[6])
# ****************** Timer *****************
def get_microsecond_timer(self):
dat = self._handle.controlRead(Panda.REQUEST_IN, 0xa8, 0, 0, 4)

4
panda/python/constants.py
View File

@@ -1,6 +1,6 @@
import os
import enum
from typing import List, NamedTuple
from typing import NamedTuple
BASEDIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), "../")
FW_PATH = os.path.join(BASEDIR, "board/obj/")
@@ -10,7 +10,7 @@ USBPACKET_MAX_SIZE = 0x40
class McuConfig(NamedTuple):
mcu: str
mcu_idcode: int
sector_sizes: List[int]
sector_sizes: list[int]
sector_count: int # total sector count, used for MCU identification in DFU mode
uid_address: int
block_size: int

25
panda/python/dfu.py
View File

@@ -2,7 +2,6 @@ import os
import usb1
import struct
import binascii
from typing import List, Optional
from .base import BaseSTBootloaderHandle
from .spi import STBootloaderSPIHandle, PandaSpiException
@@ -11,9 +10,9 @@ from .constants import FW_PATH, McuType
class PandaDFU:
def __init__(self, dfu_serial: Optional[str]):
def __init__(self, dfu_serial: str | None):
# try USB, then SPI
handle: Optional[BaseSTBootloaderHandle]
handle: BaseSTBootloaderHandle | None
self._context, handle = PandaDFU.usb_connect(dfu_serial)
if handle is None:
self._context, handle = PandaDFU.spi_connect(dfu_serial)
@@ -38,7 +37,7 @@ class PandaDFU:
self._context.close()
@staticmethod
def usb_connect(dfu_serial: Optional[str]):
def usb_connect(dfu_serial: str | None):
handle = None
context = usb1.USBContext()
context.open()
@@ -56,7 +55,7 @@ class PandaDFU:
return context, handle
@staticmethod
def spi_connect(dfu_serial: Optional[str]):
def spi_connect(dfu_serial: str | None):
handle = None
this_dfu_serial = None
@@ -72,13 +71,7 @@ class PandaDFU:
return None, handle
@staticmethod
def list() -> List[str]:
ret = PandaDFU.usb_list()
ret += PandaDFU.spi_list()
return list(set(ret))
@staticmethod
def usb_list() -> List[str]:
def usb_list() -> list[str]:
dfu_serials = []
try:
with usb1.USBContext() as context:
@@ -93,7 +86,7 @@ class PandaDFU:
return dfu_serials
@staticmethod
def spi_list() -> List[str]:
def spi_list() -> list[str]:
try:
_, h = PandaDFU.spi_connect(None)
if h is not None:
@@ -134,3 +127,9 @@ class PandaDFU:
code = f.read()
self.program_bootstub(code)
self.reset()
@staticmethod
def list() -> list[str]:
ret = PandaDFU.usb_list()
ret += PandaDFU.spi_list()
return list(set(ret))

4
panda/python/isotp.py
View File

@@ -79,10 +79,10 @@ def isotp_send(panda, x, addr, bus=0, recvaddr=None, subaddr=None, rate=None):
sends = []
while len(x) > 0:
if subaddr:
sends.append(((bytes([subaddr, 0x20 + (idx & 0xF)]) + x[0:6]).ljust(8, b"\x00")))
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")))
sends.append((bytes([0x20 + (idx & 0xF)]) + x[0:7]).ljust(8, b"\x00"))
x = x[7:]
idx += 1

2
panda/python/serial.py
View File

@@ -1,5 +1,5 @@
# mimic a python serial port
class PandaSerial(object):
class PandaSerial:
def __init__(self, panda, port, baud):
self.panda = panda
self.port = port

6
panda/python/spi.py
View File

@@ -9,7 +9,7 @@ import logging
import threading
from contextlib import contextmanager
from functools import reduce
from typing import Callable, List, Optional
from collections.abc import Callable
from .base import BaseHandle, BaseSTBootloaderHandle, TIMEOUT
from .constants import McuType, MCU_TYPE_BY_IDCODE, USBPACKET_MAX_SIZE
@@ -341,7 +341,7 @@ class STBootloaderSPIHandle(BaseSTBootloaderHandle):
elif data != self.ACK:
raise PandaSpiMissingAck
def _cmd_no_retry(self, cmd: int, data: Optional[List[bytes]] = None, read_bytes: int = 0, predata=None) -> bytes:
def _cmd_no_retry(self, cmd: int, data: list[bytes] | None = None, read_bytes: int = 0, predata=None) -> bytes:
ret = b""
with self.dev.acquire() as spi:
# sync + command
@@ -371,7 +371,7 @@ class STBootloaderSPIHandle(BaseSTBootloaderHandle):
return bytes(ret)
def _cmd(self, cmd: int, data: Optional[List[bytes]] = None, read_bytes: int = 0, predata=None) -> bytes:
def _cmd(self, cmd: int, data: list[bytes] | None = None, read_bytes: int = 0, predata=None) -> bytes:
exc = PandaSpiException()
for n in range(MAX_XFER_RETRY_COUNT):
try:

89
panda/python/uds.py
View File

@@ -1,7 +1,8 @@
import time
import struct
from collections import deque
from typing import Callable, NamedTuple, Tuple, List, Deque, Generator, Optional, cast
from typing import NamedTuple, Deque, cast
from collections.abc import Callable, Generator
from enum import IntEnum
from functools import partial
@@ -300,8 +301,8 @@ def get_dtc_status_names(status):
return result
class CanClient():
def __init__(self, can_send: Callable[[int, bytes, int], None], can_recv: Callable[[], List[Tuple[int, int, bytes, int]]],
tx_addr: int, rx_addr: int, bus: int, sub_addr: Optional[int] = None, debug: bool = False):
def __init__(self, can_send: Callable[[int, bytes, int], None], can_recv: Callable[[], list[tuple[int, 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
@@ -335,7 +336,7 @@ class CanClient():
msgs = self.rx()
if drain:
if self.debug:
print("CAN-RX: drain - {}".format(len(msgs)))
print(f"CAN-RX: drain - {len(msgs)}")
self.rx_buff.clear()
else:
for rx_addr, _, rx_data, rx_bus in msgs or []:
@@ -366,7 +367,7 @@ class CanClient():
except IndexError:
pass # empty
def send(self, msgs: List[bytes], delay: float = 0) -> None:
def send(self, msgs: list[bytes], delay: float = 0) -> None:
for i, msg in enumerate(msgs):
if delay and i != 0:
if self.debug:
@@ -443,7 +444,7 @@ class IsoTpMessage():
if not setup_only:
self._can_client.send([msg])
def recv(self, timeout=None) -> Tuple[Optional[bytes], bool]:
def recv(self, timeout=None) -> tuple[bytes | None, bool]:
if timeout is None:
timeout = self.timeout
@@ -566,11 +567,11 @@ def get_rx_addr_for_tx_addr(tx_addr, rx_offset=0x8):
# standard 29 bit response addr (flip last two bytes)
return (tx_addr & 0xFFFF0000) + (tx_addr << 8 & 0xFF00) + (tx_addr >> 8 & 0xFF)
raise ValueError("invalid tx_addr: {}".format(tx_addr))
raise ValueError(f"invalid tx_addr: {tx_addr}")
class UdsClient():
def __init__(self, panda, tx_addr: int, rx_addr: Optional[int] = None, bus: int = 0, sub_addr: Optional[int] = None, timeout: float = 1,
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
@@ -583,7 +584,7 @@ class UdsClient():
self.response_pending_timeout = response_pending_timeout
# generic uds request
def _uds_request(self, service_type: SERVICE_TYPE, subfunction: Optional[int] = None, data: Optional[bytes] = None) -> bytes:
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])
@@ -623,12 +624,12 @@ class UdsClient():
if self.debug:
print("UDS-RX: response pending")
continue
raise NegativeResponseError('{} - {}'.format(service_desc, error_desc), service_id, error_code)
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('invalid response service id: {}'.format(resp_sid_hex))
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
@@ -671,7 +672,7 @@ class UdsClient():
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: Optional[bytes] = None):
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)
@@ -714,8 +715,8 @@ class UdsClient():
"data": resp[2:], # TODO: parse the reset of response
}
def link_control(self, link_control_type: LINK_CONTROL_TYPE, baud_rate_type: Optional[BAUD_RATE_TYPE] = None):
data: Optional[bytes]
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
@@ -733,21 +734,21 @@ class UdsClient():
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('invalid response data identifier: {} expected: {}'.format(hex(resp_id), hex(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('invalid memory_address_bytes: {}'.format(memory_address_bytes))
raise ValueError(f'invalid memory_address_bytes: {memory_address_bytes}')
if memory_size_bytes < 1 or memory_size_bytes > 4:
raise ValueError('invalid memory_size_bytes: {}'.format(memory_size_bytes))
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('invalid memory_address: {}'.format(memory_address))
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('invalid memory_size: {}'.format(memory_size))
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)
@@ -758,7 +759,7 @@ class UdsClient():
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('invalid response data identifier: {}'.format(hex(resp_id)))
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):
@@ -767,11 +768,11 @@ class UdsClient():
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):
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('invalid memory_address_bytes: {}'.format(memory_address_bytes))
raise ValueError(f'invalid memory_address_bytes: {memory_address_bytes}')
if memory_size_bytes < 1 or memory_size_bytes > 4:
raise ValueError('invalid memory_size_bytes: {}'.format(memory_size_bytes))
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:
@@ -781,15 +782,15 @@ class UdsClient():
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('invalid memory_address: {}'.format(s.memory_address))
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('invalid memory_size: {}'.format(s.memory_size))
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('invalid dynamic identifier type: {}'.format(hex(dynamic_definition_type)))
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):
@@ -797,20 +798,20 @@ class UdsClient():
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('invalid response data identifier: {}'.format(hex(resp_id)))
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('invalid memory_address_bytes: {}'.format(memory_address_bytes))
raise ValueError(f'invalid memory_address_bytes: {memory_address_bytes}')
if memory_size_bytes < 1 or memory_size_bytes > 4:
raise ValueError('invalid memory_size_bytes: {}'.format(memory_size_bytes))
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('invalid memory_address: {}'.format(memory_address))
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('invalid memory_size: {}'.format(memory_size))
raise ValueError(f'invalid memory_size: {memory_size}')
data += struct.pack('!I', memory_size)[4 - memory_size_bytes:]
data += data_record
@@ -864,7 +865,7 @@ class UdsClient():
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('invalid response data identifier: {}'.format(hex(resp_id)))
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''):
@@ -872,23 +873,23 @@ class UdsClient():
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('invalid response routine identifier: {}'.format(hex(resp_id)))
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('invalid memory_address_bytes: {}'.format(memory_address_bytes))
raise ValueError(f'invalid memory_address_bytes: {memory_address_bytes}')
if memory_size_bytes < 1 or memory_size_bytes > 4:
raise ValueError('invalid memory_size_bytes: {}'.format(memory_size_bytes))
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('invalid memory_address: {}'.format(memory_address))
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('invalid memory_size: {}'.format(memory_size))
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)
@@ -896,7 +897,7 @@ class UdsClient():
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('invalid max_num_bytes_len: {}'.format(max_num_bytes_len))
raise ValueError(f'invalid max_num_bytes_len: {max_num_bytes_len}')
return max_num_bytes # max number of bytes per transfer data request
@@ -904,16 +905,16 @@ class UdsClient():
data = bytes([data_format])
if memory_address_bytes < 1 or memory_address_bytes > 4:
raise ValueError('invalid memory_address_bytes: {}'.format(memory_address_bytes))
raise ValueError(f'invalid memory_address_bytes: {memory_address_bytes}')
if memory_size_bytes < 1 or memory_size_bytes > 4:
raise ValueError('invalid memory_size_bytes: {}'.format(memory_size_bytes))
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('invalid memory_address: {}'.format(memory_address))
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('invalid memory_size: {}'.format(memory_size))
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)
@@ -921,7 +922,7 @@ class UdsClient():
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('invalid max_num_bytes_len: {}'.format(max_num_bytes_len))
raise ValueError(f'invalid max_num_bytes_len: {max_num_bytes_len}')
return max_num_bytes # max number of bytes per transfer data request
@@ -930,7 +931,7 @@ class UdsClient():
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('invalid block_sequence_count: {}'.format(resp_id))
raise ValueError(f'invalid block_sequence_count: {resp_id}')
return resp[1:]
def request_transfer_exit(self):

258
panda/python/xcp.py Normal file
View File

@@ -0,0 +1,258 @@
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]