mawei/StarPilot
1
0
Fork 0
mirror of https://github.com/firestar5683/StarPilot.git synced 2026-07-19 08:12:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

openpilot v0.9.3 release

Browse Source 
date: 2023-06-16T05:25:00
master commit: 8704c1ff952b5c85a44f50143bbd1a4f7b4887e2
This commit is contained in:
Vehicle Researcher
2023-06-16 05:25:01 +00:00
parent cfd8323ccb
commit c39abc00af
188 changed files with 4182 additions and 2011 deletions
Download Patch File Download Diff File Expand all files Collapse all files
panda/python/__init__.py
+26 -15
View File
@@ -14,7 +14,7 @@ from typing import Optional
from itertools import accumulate
from .base import BaseHandle
from .constants import McuType
from .constants import FW_PATH, McuType
from .dfu import PandaDFU
from .isotp import isotp_send, isotp_recv
from .spi import PandaSpiHandle, PandaSpiException
@@ -401,14 +401,16 @@ class Panda:
return []
def reset(self, enter_bootstub=False, enter_bootloader=False, reconnect=True):
# no response is expected since it resets right away
timeout = 5000 if isinstance(self._handle, PandaSpiHandle) else 15000
try:
if enter_bootloader:
self._handle.controlWrite(Panda.REQUEST_IN, 0xd1, 0, 0, b'')
self._handle.controlWrite(Panda.REQUEST_IN, 0xd1, 0, 0, b'', timeout=timeout, expect_disconnect=True)
else:
if enter_bootstub:
self._handle.controlWrite(Panda.REQUEST_IN, 0xd1, 1, 0, b'')
self._handle.controlWrite(Panda.REQUEST_IN, 0xd1, 1, 0, b'', timeout=timeout, expect_disconnect=True)
else:
self._handle.controlWrite(Panda.REQUEST_IN, 0xd8, 0, 0, b'')
self._handle.controlWrite(Panda.REQUEST_IN, 0xd8, 0, 0, b'', timeout=timeout, expect_disconnect=True)
except Exception:
pass
if not enter_bootloader and reconnect:
@@ -477,13 +479,13 @@ class Panda:
# reset
logging.warning("flash: resetting")
try:
handle.controlWrite(Panda.REQUEST_IN, 0xd8, 0, 0, b'')
handle.controlWrite(Panda.REQUEST_IN, 0xd8, 0, 0, b'', expect_disconnect=True)
except Exception:
pass
def flash(self, fn=None, code=None, reconnect=True):
if not fn:
fn = self._mcu_type.config.app_path
fn = os.path.join(FW_PATH, self._mcu_type.config.app_fn)
assert os.path.isfile(fn)
logging.debug("flash: main version is %s", self.get_version())
if not self.bootstub:
@@ -523,18 +525,33 @@ class Panda:
return True
@staticmethod
def wait_for_dfu(dfu_serial: str, timeout: Optional[int] = None) -> bool:
def wait_for_dfu(dfu_serial: Optional[str], timeout: Optional[int] = None) -> bool:
t_start = time.monotonic()
while dfu_serial not in PandaDFU.list():
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):
logging.debug("waiting for DFU...")
time.sleep(0.1)
if timeout is not None and (time.monotonic() - t_start) > timeout:
return False
dfu_list = PandaDFU.list()
return True
@staticmethod
def wait_for_panda(serial: Optional[str], 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):
logging.debug("waiting for panda...")
time.sleep(0.1)
if timeout is not None and (time.monotonic() - t_start) > timeout:
return False
serials = Panda.list()
return True
def up_to_date(self) -> bool:
current = self.get_signature()
expected = Panda.get_signature_from_firmware(self.get_mcu_type().config.app_path)
fn = os.path.join(FW_PATH, self.get_mcu_type().config.app_fn)
expected = Panda.get_signature_from_firmware(fn)
return (current == expected)
def call_control_api(self, msg):
@@ -617,12 +634,6 @@ class Panda:
# ******************* control *******************
def enter_bootloader(self):
try:
self._handle.controlWrite(Panda.REQUEST_OUT, 0xd1, 0, 0, b'')
except Exception:
logging.exception("exception while entering bootloader")
def get_version(self):
return self._handle.controlRead(Panda.REQUEST_IN, 0xd6, 0, 0, 0x40).decode('utf8')
panda/python/constants.py
+7 -7
View File
@@ -3,7 +3,7 @@ import enum
from typing import List, NamedTuple
BASEDIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), "../")
FW_PATH = os.path.join(BASEDIR, "board/obj/")
class McuConfig(NamedTuple):
mcu: str
@@ -13,9 +13,9 @@ class McuConfig(NamedTuple):
sector_sizes: List[int]
serial_number_address: int
app_address: int
app_path: str
app_fn: str
bootstub_address: int
bootstub_path: str
bootstub_fn: str
Fx = (
0x1FFF7A10,
@@ -23,9 +23,9 @@ Fx = (
[0x4000 for _ in range(4)] + [0x10000] + [0x20000 for _ in range(11)],
0x1FFF79C0,
0x8004000,
os.path.join(BASEDIR, "board", "obj", "panda.bin.signed"),
"panda.bin.signed",
0x8000000,
os.path.join(BASEDIR, "board", "obj", "bootstub.panda.bin"),
"bootstub.panda.bin",
)
F2Config = McuConfig("STM32F2", 0x411, *Fx)
F4Config = McuConfig("STM32F4", 0x463, *Fx)
@@ -39,9 +39,9 @@ H7Config = McuConfig(
[0x20000 for _ in range(7)],
0x080FFFC0,
0x8020000,
os.path.join(BASEDIR, "board", "obj", "panda_h7.bin.signed"),
"panda_h7.bin.signed",
0x8000000,
os.path.join(BASEDIR, "board", "obj", "bootstub.panda_h7.bin"),
"bootstub.panda_h7.bin",
)
@enum.unique
panda/python/dfu.py
+5 -4
View File
@@ -1,3 +1,4 @@
import os
import usb1
import struct
import binascii
@@ -6,7 +7,7 @@ from typing import List, Optional
from .base import BaseSTBootloaderHandle
from .spi import STBootloaderSPIHandle, PandaSpiException
from .usb import STBootloaderUSBHandle
from .constants import McuType
from .constants import FW_PATH, McuType
class PandaDFU:
@@ -110,10 +111,10 @@ class PandaDFU:
self._handle.erase_bootstub()
self._handle.erase_app()
self._handle.program(self._mcu_type.config.bootstub_address, code_bootstub)
self.reset()
def recover(self):
with open(self._mcu_type.config.bootstub_path, "rb") as f:
fn = os.path.join(FW_PATH, self._mcu_type.config.bootstub_fn)
with open(fn, "rb") as f:
code = f.read()
self.program_bootstub(code)
self.reset()
panda/python/spi.py
+29 -19
View File
@@ -116,54 +116,64 @@ class PandaSpiHandle(BaseHandle):
raise PandaSpiMissingAck
def _transfer(self, spi, endpoint: int, data, timeout: int, max_rx_len: int = 1000) -> bytes:
def _transfer(self, spi, endpoint: int, data, timeout: int, max_rx_len: int = 1000, expect_disconnect: bool = False) -> bytes:
logging.debug("starting transfer: endpoint=%d, max_rx_len=%d", endpoint, max_rx_len)
logging.debug("==============================================")
n = 0
start_time = time.monotonic()
exc = PandaSpiException()
for n in range(MAX_XFER_RETRY_COUNT):
logging.debug("\ntry #%d", n+1)
while (time.monotonic() - start_time) < timeout*1e-3:
n += 1
logging.debug("\ntry #%d", n)
try:
logging.debug("- send header")
packet = struct.pack("<BBHH", SYNC, endpoint, len(data), max_rx_len)
packet += bytes([reduce(lambda x, y: x^y, packet) ^ CHECKSUM_START])
spi.xfer2(packet)
to = timeout - (time.monotonic() - start_time)*1e3
logging.debug("- waiting for header ACK")
self._wait_for_ack(spi, HACK, timeout, 0x11)
self._wait_for_ack(spi, HACK, int(to), 0x11)
# send data
logging.debug("- sending data")
packet = bytes([*data, self._calc_checksum(data)])
spi.xfer2(packet)
logging.debug("- waiting for data ACK")
self._wait_for_ack(spi, DACK, timeout, 0x13)
if expect_disconnect:
logging.debug("- expecting disconnect, returning")
return b""
else:
to = timeout - (time.monotonic() - start_time)*1e3
logging.debug("- waiting for data ACK")
self._wait_for_ack(spi, DACK, int(to), 0x13)
# get response length, then response
response_len_bytes = bytes(spi.xfer2(b"\x00" * 2))
response_len = struct.unpack("<H", response_len_bytes)[0]
if response_len > max_rx_len:
raise PandaSpiException("response length greater than max")
# get response length, then response
response_len_bytes = bytes(spi.xfer2(b"\x00" * 2))
response_len = struct.unpack("<H", response_len_bytes)[0]
if response_len > max_rx_len:
raise PandaSpiException("response length greater than max")
logging.debug("- receiving response")
dat = bytes(spi.xfer2(b"\x00" * (response_len + 1)))
if self._calc_checksum([DACK, *response_len_bytes, *dat]) != 0:
raise PandaSpiBadChecksum
logging.debug("- receiving response")
dat = bytes(spi.xfer2(b"\x00" * (response_len + 1)))
if self._calc_checksum([DACK, *response_len_bytes, *dat]) != 0:
raise PandaSpiBadChecksum
return dat[:-1]
return dat[:-1]
except PandaSpiException as e:
exc = e
logging.debug("SPI transfer failed, %d retries left", MAX_XFER_RETRY_COUNT - n - 1, exc_info=True)
logging.debug("SPI transfer failed, retrying", exc_info=True)
raise exc
# libusb1 functions
def close(self):
self.dev.close()
def controlWrite(self, request_type: int, request: int, value: int, index: int, data, timeout: int = TIMEOUT):
def controlWrite(self, request_type: int, request: int, value: int, index: int, data, timeout: int = TIMEOUT, expect_disconnect: bool = False):
with self.dev.acquire() as spi:
return self._transfer(spi, 0, struct.pack("<BHHH", request, value, index, 0), timeout)
return self._transfer(spi, 0, struct.pack("<BHHH", request, value, index, 0), timeout, expect_disconnect=expect_disconnect)
def controlRead(self, request_type: int, request: int, value: int, index: int, length: int, timeout: int = TIMEOUT):
with self.dev.acquire() as spi:
panda/python/usb.py
+2 -2
View File
@@ -11,7 +11,7 @@ class PandaUsbHandle(BaseHandle):
def close(self):
self._libusb_handle.close()
def controlWrite(self, request_type: int, request: int, value: int, index: int, data, timeout: int = TIMEOUT):
def controlWrite(self, request_type: int, request: int, value: int, index: int, data, timeout: int = TIMEOUT, expect_disconnect: bool = False):
return self._libusb_handle.controlWrite(request_type, request, value, index, data, timeout)
def controlRead(self, request_type: int, request: int, value: int, index: int, length: int, timeout: int = TIMEOUT):
@@ -77,7 +77,7 @@ class STBootloaderUSBHandle(BaseSTBootloaderHandle):
self._status()
# Program
bs = self._mcu_type.config.block_size
bs = min(len(dat), self._mcu_type.config.block_size)
dat += b"\xFF" * ((bs - len(dat)) % bs)
for i in range(0, len(dat) // bs):
ldat = dat[i * bs:(i + 1) * bs]