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

python lib: allow infinite timeout

Browse Source
This commit is contained in:
Adeeb Shihadeh
2023-07-20 13:07:06 -07:00
parent 1eac0d7634
commit dd78b2bf6c
3 changed files with 8 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
python/spi.py
View File

@@ -32,7 +32,7 @@ CHECKSUM_START = 0xAB
MIN_ACK_TIMEOUT_MS = 100
MAX_XFER_RETRY_COUNT = 5
XFER_SIZE = 1000
XFER_SIZE = 0x40*31
DEV_PATH = "/dev/spidev0.0"
@@ -164,7 +164,6 @@ class PandaSpiHandle(BaseHandle):
logging.debug("- waiting for header ACK")
self._wait_for_ack(spi, HACK, MIN_ACK_TIMEOUT_MS, 0x11)
# send data
logging.debug("- sending data")
packet = bytes([*data, self._calc_checksum(data)])
spi.xfer2(packet)
@@ -187,6 +186,7 @@ class PandaSpiHandle(BaseHandle):
if remaining > 0:
dat += bytes(spi.readbytes(remaining))
dat = dat[:3 + response_len + 1]
if self._calc_checksum(dat) != 0:
raise PandaSpiBadChecksum
@@ -216,7 +216,7 @@ class PandaSpiHandle(BaseHandle):
n = 0
start_time = time.monotonic()
exc = PandaSpiException()
while (time.monotonic() - start_time) < timeout*1e-3:
while (timeout == 0) or (time.monotonic() - start_time) < timeout*1e-3:
n += 1
logging.debug("\ntry #%d", n)
with self.dev.acquire() as spi:
@@ -274,20 +274,19 @@ class PandaSpiHandle(BaseHandle):
def controlRead(self, request_type: int, request: int, value: int, index: int, length: int, timeout: int = TIMEOUT):
return self._transfer(0, struct.pack("<BHHH", request, value, index, length), timeout, max_rx_len=length)
# TODO: implement these properly
def bulkWrite(self, endpoint: int, data: List[int], timeout: int = TIMEOUT) -> int:
def bulkWrite(self, endpoint: int, data: bytes, timeout: int = TIMEOUT) -> int:
for x in range(math.ceil(len(data) / XFER_SIZE)):
self._transfer(endpoint, data[XFER_SIZE*x:XFER_SIZE*(x+1)], timeout)
return len(data)
def bulkRead(self, endpoint: int, length: int, timeout: int = TIMEOUT) -> bytes:
ret: List[int] = []
ret = b""
for _ in range(math.ceil(length / XFER_SIZE)):
d = self._transfer(endpoint, [], timeout, max_rx_len=XFER_SIZE)
ret += d
if len(d) < XFER_SIZE:
break
return bytes(ret)
return ret
class STBootloaderSPIHandle(BaseSTBootloaderHandle):