mawei/panda-meb
1
0
Fork 0
mirror of https://github.com/infiniteCable2/panda.git synced 2026-02-18 17:23:52 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'upstream/panda/master' into sync-20251114

Browse Source
This commit is contained in:
Jason Wen
2025-11-14 00:47:48 -05:00
parent 69ab12ee2a 615009cf0f
commit 5dc009be50
34 changed files with 218 additions and 367 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
python/__init__.py
View File

@@ -23,7 +23,7 @@ __version__ = '0.0.10'
CANPACKET_HEAD_SIZE = 0x6
DLC_TO_LEN = [0, 1, 2, 3, 4, 5, 6, 7, 8, 12, 16, 20, 24, 32, 48, 64]
LEN_TO_DLC = {length: dlc for (dlc, length) in enumerate(DLC_TO_LEN)}
PANDA_BUS_CNT = 3
PANDA_CAN_CNT = 3
def calculate_checksum(data):
@@ -32,16 +32,13 @@ def calculate_checksum(data):
res ^= b
return res
def pack_can_buffer(arr, fd=False):
snds = [b'']
def pack_can_buffer(arr, chunk=False, fd=False):
snds = [bytearray(), ]
for address, dat, bus in arr:
assert len(dat) in LEN_TO_DLC
#logger.debug(" W 0x%x: 0x%s", address, dat.hex())
extended = 1 if address >= 0x800 else 0
data_len_code = LEN_TO_DLC[len(dat)]
header = bytearray(CANPACKET_HEAD_SIZE)
word_4b = address << 3 | extended << 2
word_4b = (address << 3) | (extended << 2)
header[0] = (data_len_code << 4) | (bus << 1) | int(fd)
header[1] = word_4b & 0xFF
header[2] = (word_4b >> 8) & 0xFF
@@ -49,9 +46,10 @@ def pack_can_buffer(arr, fd=False):
header[4] = (word_4b >> 24) & 0xFF
header[5] = calculate_checksum(header[:5] + dat)
snds[-1] += header + dat
if len(snds[-1]) > 256: # Limit chunks to 256 bytes
snds.append(b'')
snds[-1].extend(header)
snds[-1].extend(dat)
if chunk and len(snds[-1]) > 256:
snds.append(bytearray())
return snds
@@ -134,7 +132,7 @@ class Panda:
MAX_FAN_RPMs = {
HW_TYPE_TRES: 6600,
HW_TYPE_CUATRO: 12500,
HW_TYPE_CUATRO: 5000,
}
HARNESS_STATUS_NC = 0
@@ -199,9 +197,9 @@ class Panda:
self._handle = None
while self._handle is None:
# try USB first, then SPI
self._context, self._handle, serial, self.bootstub, bcd = self.usb_connect(self._connect_serial, claim=claim, no_error=wait)
self._context, self._handle, serial, self.bootstub = self.usb_connect(self._connect_serial, claim=claim, no_error=wait)
if self._handle is None:
self._context, self._handle, serial, self.bootstub, bcd = self.spi_connect(self._connect_serial)
self._context, self._handle, serial, self.bootstub = self.spi_connect(self._connect_serial)
if not wait:
break
@@ -228,11 +226,11 @@ class Panda:
self.can_reset_communications()
# disable automatic CAN-FD switching
for bus in range(PANDA_BUS_CNT):
for bus in range(PANDA_CAN_CNT):
self.set_canfd_auto(bus, False)
# set CAN speed
for bus in range(PANDA_BUS_CNT):
for bus in range(PANDA_CAN_CNT):
self.set_can_speed_kbps(bus, self._can_speed_kbps)
@property
@@ -241,49 +239,33 @@ class Panda:
@classmethod
def spi_connect(cls, serial, ignore_version=False):
# get UID to confirm slave is present and up
handle = None
spi_serial = None
bootstub = None
spi_version = None
try:
handle = PandaSpiHandle()
# connect by protcol version
try:
dat = handle.get_protocol_version()
spi_serial = binascii.hexlify(dat[:12]).decode()
pid = dat[13]
if pid not in (0xcc, 0xee):
raise PandaSpiException("invalid bootstub status")
bootstub = pid == 0xee
spi_version = dat[14]
except PandaSpiException:
# fallback, we'll raise a protocol mismatch below
dat = handle.controlRead(Panda.REQUEST_IN, 0xc3, 0, 0, 12, timeout=100)
spi_serial = binascii.hexlify(dat).decode()
bootstub = Panda.flasher_present(handle)
spi_version = 0
dat = handle.get_protocol_version()
except PandaSpiException:
pass
return None, None, None, False
# no connection or wrong panda
if None in (spi_serial, bootstub) or (serial is not None and (spi_serial != serial)):
handle = None
spi_serial = None
bootstub = False
spi_serial = binascii.hexlify(dat[:12]).decode()
pid = dat[13]
if pid not in (0xcc, 0xee):
raise PandaProtocolMismatch(f"invalid bootstub status ({pid=}). reflash panda")
bootstub = pid == 0xee
spi_version = dat[14]
# did we get the right panda?
if serial is not None and spi_serial != serial:
return None, None, None, False
# ensure our protocol version matches the panda
if handle is not None and not ignore_version:
if spi_version != handle.PROTOCOL_VERSION:
err = f"panda protocol mismatch: expected {handle.PROTOCOL_VERSION}, got {spi_version}. reflash panda"
raise PandaProtocolMismatch(err)
if (not ignore_version) and spi_version != handle.PROTOCOL_VERSION:
raise PandaProtocolMismatch(f"panda protocol mismatch: expected {handle.PROTOCOL_VERSION}, got {spi_version}. reflash panda")
return None, handle, spi_serial, bootstub, None
# got a device and all good
return None, handle, spi_serial, bootstub
@classmethod
def usb_connect(cls, serial, claim=True, no_error=False):
handle, usb_serial, bootstub, bcd = None, None, None, None
handle, usb_serial, bootstub = None, None, None
context = usb1.USBContext()
context.open()
try:
@@ -309,11 +291,6 @@ class Panda:
handle.claimInterface(0)
# handle.setInterfaceAltSetting(0, 0) # Issue in USB stack
# bcdDevice wasn't always set to the hw type, ignore if it's the old constant
this_bcd = device.getbcdDevice()
if this_bcd is not None and this_bcd != 0x2300:
bcd = bytearray([this_bcd >> 8, ])
break
except Exception:
logger.exception("USB connect error")
@@ -324,7 +301,7 @@ class Panda:
else:
context.close()
return context, usb_handle, usb_serial, bootstub, bcd
return context, usb_handle, usb_serial, bootstub
def is_connected_spi(self):
return isinstance(self._handle, PandaSpiHandle)
@@ -360,7 +337,7 @@ class Panda:
@classmethod
def spi_list(cls):
_, _, serial, _, _ = cls.spi_connect(None, ignore_version=True)
_, _, serial, _ = cls.spi_connect(None, ignore_version=True)
if serial is not None:
return [serial, ]
return []
@@ -732,7 +709,7 @@ class Panda:
@ensure_can_packet_version
def can_send_many(self, arr, *, fd=False, timeout=CAN_SEND_TIMEOUT_MS):
snds = pack_can_buffer(arr, fd=fd)
snds = pack_can_buffer(arr, chunk=(not self.spi), fd=fd)
for tx in snds:
while len(tx) > 0:
bs = self._handle.bulkWrite(3, tx, timeout=timeout)