mawei/dragonpilot
1
0
Fork 0
mirror of https://github.com/dragonpilot/dragonpilot.git synced 2026-02-19 21:44:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

openpilot v0.9.5 release

Browse Source 
date: 2023-11-17T23:53:40
master commit: d3aad9ca46
This commit is contained in:
Vehicle Researcher
2023-11-17 23:53:40 +00:00
parent eff388b1b6
commit b2f2dabe71
861 changed files with 33808 additions and 28421 deletions
Download Patch File Download Diff File Expand all files Collapse all files

166
panda/python/__init__.py
View File

@@ -7,9 +7,8 @@ import struct
import hashlib
import binascii
import datetime
import warnings
import logging
from functools import wraps
from functools import wraps, partial
from typing import Optional
from itertools import accumulate
@@ -91,53 +90,21 @@ def unpack_can_buffer(dat):
return (ret, dat)
def ensure_health_packet_version(fn):
def ensure_version(desc, lib_field, panda_field, fn):
@wraps(fn)
def wrapper(self, *args, **kwargs):
if self.health_version < self.HEALTH_PACKET_VERSION:
raise RuntimeError("Panda firmware has outdated health packet definition. Reflash panda firmware.")
elif self.health_version > self.HEALTH_PACKET_VERSION:
raise RuntimeError("Panda python library has outdated health packet definition. Update panda python library.")
lib_version = getattr(self, lib_field)
panda_version = getattr(self, panda_field)
if lib_version != panda_version:
raise RuntimeError(f"{desc} packet version mismatch: panda's firmware v{panda_version}, library v{lib_version}. Reflash panda.")
return fn(self, *args, **kwargs)
return wrapper
ensure_can_packet_version = partial(ensure_version, "CAN", "CAN_PACKET_VERSION", "can_version")
ensure_can_health_packet_version = partial(ensure_version, "CAN health", "CAN_HEALTH_PACKET_VERSION", "can_health_version")
ensure_health_packet_version = partial(ensure_version, "health", "HEALTH_PACKET_VERSION", "health_version")
def ensure_can_packet_version(fn):
@wraps(fn)
def wrapper(self, *args, **kwargs):
if self.can_version < self.CAN_PACKET_VERSION:
raise RuntimeError("Panda firmware has outdated CAN packet definition. Reflash panda firmware.")
elif self.can_version > self.CAN_PACKET_VERSION:
raise RuntimeError("Panda python library has outdated CAN packet definition. Update panda python library.")
return fn(self, *args, **kwargs)
return wrapper
def ensure_can_health_packet_version(fn):
@wraps(fn)
def wrapper(self, *args, **kwargs):
if self.can_health_version < self.CAN_HEALTH_PACKET_VERSION:
raise RuntimeError("Panda firmware has outdated CAN health packet definition. Reflash panda firmware.")
elif self.can_health_version > self.CAN_HEALTH_PACKET_VERSION:
raise RuntimeError("Panda python library has outdated CAN health packet definition. Update panda python library.")
return fn(self, *args, **kwargs)
return wrapper
def parse_timestamp(dat):
a = struct.unpack("HBBBBBB", dat)
if a[0] == 0:
return None
try:
return datetime.datetime(a[0], a[1], a[2], a[4], a[5], a[6])
except ValueError:
return None
def unpack_log(dat):
return {
'id': struct.unpack("H", dat[:2])[0],
'timestamp': parse_timestamp(dat[2:10]),
'uptime': struct.unpack("I", dat[10:14])[0],
'msg': bytes(dat[14:]).decode('utf-8', 'ignore').strip('\x00'),
}
class ALTERNATIVE_EXPERIENCE:
DEFAULT = 0
@@ -184,6 +151,7 @@ class Panda:
GMLAN_CAN2 = 1
GMLAN_CAN3 = 2
USB_PIDS = (0xddee, 0xddcc)
REQUEST_IN = usb1.ENDPOINT_IN | usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE
REQUEST_OUT = usb1.ENDPOINT_OUT | usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE
@@ -201,12 +169,12 @@ class Panda:
CAN_PACKET_VERSION = 4
HEALTH_PACKET_VERSION = 14
CAN_HEALTH_PACKET_VERSION = 5
HEALTH_STRUCT = struct.Struct("<IIIIIIIIIBBBBBBHBBBHfBBHBHH")
HEALTH_STRUCT = struct.Struct("<IIIIIIIIIBBBBBBHBBBHfBBHBHHB")
CAN_HEALTH_STRUCT = struct.Struct("<BIBBBBBBBBIIIIIIIHHBBBIIII")
F2_DEVICES = (HW_TYPE_PEDAL, )
F4_DEVICES = (HW_TYPE_WHITE_PANDA, HW_TYPE_GREY_PANDA, HW_TYPE_BLACK_PANDA, HW_TYPE_UNO, HW_TYPE_DOS)
H7_DEVICES = (HW_TYPE_RED_PANDA, HW_TYPE_RED_PANDA_V2, HW_TYPE_TRES)
F2_DEVICES = [HW_TYPE_PEDAL, ]
F4_DEVICES = [HW_TYPE_WHITE_PANDA, HW_TYPE_GREY_PANDA, HW_TYPE_BLACK_PANDA, HW_TYPE_UNO, HW_TYPE_DOS]
H7_DEVICES = [HW_TYPE_RED_PANDA, HW_TYPE_RED_PANDA_V2, HW_TYPE_TRES]
INTERNAL_DEVICES = (HW_TYPE_UNO, HW_TYPE_DOS, HW_TYPE_TRES)
HAS_OBD = (HW_TYPE_BLACK_PANDA, HW_TYPE_UNO, HW_TYPE_DOS, HW_TYPE_RED_PANDA, HW_TYPE_RED_PANDA_V2, HW_TYPE_TRES)
@@ -238,6 +206,7 @@ class Panda:
FLAG_HYUNDAI_CANFD_HDA2 = 16
FLAG_HYUNDAI_CANFD_ALT_BUTTONS = 32
FLAG_HYUNDAI_ALT_LIMITS = 64
FLAG_HYUNDAI_CANFD_HDA2_ALT_STEERING = 128
FLAG_TESLA_POWERTRAIN = 1
FLAG_TESLA_LONG_CONTROL = 2
@@ -248,6 +217,11 @@ class Panda:
FLAG_CHRYSLER_RAM_HD = 2
FLAG_SUBARU_GEN2 = 1
FLAG_SUBARU_LONG = 2
FLAG_SUBARU_PREGLOBAL_REVERSED_DRIVER_TORQUE = 1
FLAG_NISSAN_ALT_EPS_BUS = 1
FLAG_GM_HW_CAM = 1
FLAG_GM_HW_CAM_LONG = 2
@@ -279,6 +253,8 @@ class Panda:
if self._handle_open:
self._handle.close()
self._handle_open = False
if self._context is not None:
self._context.close()
def connect(self, claim=True, wait=False):
self.close()
@@ -286,9 +262,9 @@ class Panda:
self._handle = None
while self._handle is None:
# try USB first, then SPI
self._handle, serial, self.bootstub, bcd = self.usb_connect(self._connect_serial, claim=claim)
self._context, self._handle, serial, self.bootstub, bcd = self.usb_connect(self._connect_serial, claim=claim)
if self._handle is None:
self._handle, serial, self.bootstub, bcd = self.spi_connect(self._connect_serial)
self._context, self._handle, serial, self.bootstub, bcd = self.spi_connect(self._connect_serial)
if not wait:
break
@@ -324,8 +300,8 @@ class Panda:
self.set_heartbeat_disabled()
self.set_power_save(0)
@staticmethod
def spi_connect(serial, ignore_version=False):
@classmethod
def spi_connect(cls, serial, ignore_version=False):
# get UID to confirm slave is present and up
handle = None
spi_serial = None
@@ -364,16 +340,16 @@ class Panda:
err = f"panda protocol mismatch: expected {handle.PROTOCOL_VERSION}, got {spi_version}. reflash panda"
raise PandaProtocolMismatch(err)
return handle, spi_serial, bootstub, None
return None, handle, spi_serial, bootstub, None
@staticmethod
def usb_connect(serial, claim=True):
@classmethod
def usb_connect(cls, serial, claim=True):
handle, usb_serial, bootstub, bcd = None, None, None, None
context = usb1.USBContext()
context.open()
try:
for device in context.getDeviceList(skip_on_error=True):
if device.getVendorID() == 0xbbaa and device.getProductID() in (0xddcc, 0xddee):
if device.getVendorID() == 0xbbaa and device.getProductID() in cls.USB_PIDS:
try:
this_serial = device.getSerialNumber()
except Exception:
@@ -383,7 +359,7 @@ class Panda:
logging.debug("opening device %s %s", this_serial, hex(device.getProductID()))
usb_serial = this_serial
bootstub = device.getProductID() == 0xddee
bootstub = (device.getProductID() & 0xF0) == 0xe0
handle = device.open()
if sys.platform not in ("win32", "cygwin", "msys", "darwin"):
handle.setAutoDetachKernelDriver(True)
@@ -406,36 +382,36 @@ class Panda:
else:
context.close()
return usb_handle, usb_serial, bootstub, bcd
return context, usb_handle, usb_serial, bootstub, bcd
@staticmethod
def list():
ret = Panda.usb_list()
ret += Panda.spi_list()
@classmethod
def list(cls): # noqa: A003
ret = cls.usb_list()
ret += cls.spi_list()
return list(set(ret))
@staticmethod
def usb_list():
@classmethod
def usb_list(cls):
ret = []
try:
with usb1.USBContext() as context:
for device in context.getDeviceList(skip_on_error=True):
if device.getVendorID() == 0xbbaa and device.getProductID() in (0xddcc, 0xddee):
if device.getVendorID() == 0xbbaa and device.getProductID() in cls.USB_PIDS:
try:
serial = device.getSerialNumber()
if len(serial) == 24:
if len(serial) == 24 or serial == "pedal":
ret.append(serial)
else:
warnings.warn(f"found device with panda descriptors but invalid serial: {serial}", RuntimeWarning)
logging.warning(f"found device with panda descriptors but invalid serial: {serial}", RuntimeWarning)
except Exception:
continue
except Exception:
logging.exception("exception while listing pandas")
return ret
@staticmethod
def spi_list():
_, serial, _, _ = Panda.spi_connect(None, ignore_version=True)
@classmethod
def spi_list(cls):
_, _, serial, _, _ = cls.spi_connect(None, ignore_version=True)
if serial is not None:
return [serial, ]
return []
@@ -463,23 +439,17 @@ class Panda:
def reconnect(self):
if self._handle_open:
self.close()
time.sleep(1.0)
success = False
# wait up to 15 seconds
for i in range(0, 15):
for _ in range(15*10):
try:
self.connect()
success = True
break
except Exception:
logging.debug("reconnecting is taking %d seconds...", i + 1)
try:
dfu = PandaDFU(self.get_dfu_serial())
dfu.recover()
except Exception:
pass
time.sleep(1.0)
pass
time.sleep(0.1)
if not success:
raise Exception("reconnect failed")
@@ -524,6 +494,10 @@ class Panda:
pass
def flash(self, fn=None, code=None, reconnect=True):
if self.up_to_date(fn=fn):
logging.debug("flash: already up to date")
return
if not fn:
fn = os.path.join(FW_PATH, self._mcu_type.config.app_fn)
assert os.path.isfile(fn)
@@ -588,9 +562,10 @@ class Panda:
serials = Panda.list()
return True
def up_to_date(self) -> bool:
def up_to_date(self, fn=None) -> bool:
current = self.get_signature()
fn = os.path.join(FW_PATH, self.get_mcu_type().config.app_fn)
if fn is None:
fn = os.path.join(FW_PATH, self.get_mcu_type().config.app_fn)
expected = Panda.get_signature_from_firmware(fn)
return (current == expected)
@@ -631,6 +606,7 @@ class Panda:
"fan_stall_count": a[24],
"sbu1_voltage_mV": a[25],
"sbu2_voltage_mV": a[26],
"som_reset_triggered": a[27],
}
@ensure_can_health_packet_version
@@ -772,13 +748,6 @@ class Panda:
def enable_deepsleep(self):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xfb, 0, 0, b'')
def set_esp_power(self, on):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xd9, int(on), 0, b'')
def esp_reset(self, bootmode=0):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xda, int(bootmode), 0, b'')
time.sleep(0.2)
def set_safety_mode(self, mode=SAFETY_SILENT, param=0):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xdc, mode, param, b'')
@@ -894,7 +863,7 @@ class Panda:
def serial_write(self, port_number, ln):
ret = 0
if type(ln) == str:
if isinstance(ln, str):
ln = bytes(ln, 'utf-8')
for i in range(0, len(ln), 0x20):
ret += self._handle.bulkWrite(2, struct.pack("B", port_number) + ln[i:i + 0x20])
@@ -987,7 +956,8 @@ class Panda:
def get_datetime(self):
dat = self._handle.controlRead(Panda.REQUEST_IN, 0xa0, 0, 0, 8)
return parse_timestamp(dat)
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):
@@ -1019,14 +989,12 @@ class Panda:
def set_green_led(self, enabled):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xf7, int(enabled), 0, b'')
# ****************** Logging *****************
def get_logs(self, last_id=None, get_all=False):
assert (last_id is None) or (0 <= last_id < 0xFFFF)
def set_clock_source_period(self, period):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xe6, period, 0, b'')
logs = []
dat = self._handle.controlRead(Panda.REQUEST_IN, 0xfd, 1 if get_all else 0, last_id if last_id is not None else 0xFFFF, 0x40)
while len(dat) > 0:
if len(dat) == 0x40:
logs.append(unpack_log(dat))
dat = self._handle.controlRead(Panda.REQUEST_IN, 0xfd, 0, 0xFFFF, 0x40)
return logs
def force_relay_drive(self, intercept_relay_drive, ignition_relay_drive):
self._handle.controlWrite(Panda.REQUEST_OUT, 0xc5, (int(intercept_relay_drive) | int(ignition_relay_drive) << 1), 0, b'')
def read_som_gpio(self) -> bool:
r = self._handle.controlRead(Panda.REQUEST_IN, 0xc6, 0, 0, 1)
return r[0] == 1

9
panda/python/base.py
View File

@@ -15,7 +15,7 @@ class BaseHandle(ABC):
...
@abstractmethod
def controlWrite(self, request_type: int, request: int, value: int, index: int, data, timeout: int = TIMEOUT) -> int:
def controlWrite(self, request_type: int, request: int, value: int, index: int, data, timeout: int = TIMEOUT, expect_disconnect: bool = False):
...
@abstractmethod
@@ -53,14 +53,9 @@ class BaseSTBootloaderHandle(ABC):
...
@abstractmethod
def erase_app(self) -> None:
...
@abstractmethod
def erase_bootstub(self) -> None:
def erase_sector(self, sector: int) -> None:
...
@abstractmethod
def jump(self, address: int) -> None:
...

53
panda/python/canhandle.py Normal file
View File

@@ -0,0 +1,53 @@
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)

1
panda/python/ccp.py
View File

@@ -1,4 +1,3 @@
#!/usr/bin/env python3
import sys
import time
import struct

15
panda/python/constants.py
View File

@@ -10,35 +10,40 @@ USBPACKET_MAX_SIZE = 0x40
class McuConfig(NamedTuple):
mcu: str
mcu_idcode: int
sector_sizes: List[int]
sector_count: int # total sector count, used for MCU identification in DFU mode
uid_address: int
block_size: int
sector_sizes: List[int]
serial_number_address: int
app_address: int
app_fn: str
bootstub_address: int
bootstub_fn: str
def sector_address(self, i):
# assume bootstub is in sector 0
return self.bootstub_address + sum(self.sector_sizes[:i])
Fx = (
0x1FFF7A10,
0x800,
[0x4000 for _ in range(4)] + [0x10000] + [0x20000 for _ in range(11)],
0x1FFF79C0,
0x8004000,
"panda.bin.signed",
0x8000000,
"bootstub.panda.bin",
)
F2Config = McuConfig("STM32F2", 0x411, *Fx)
F4Config = McuConfig("STM32F4", 0x463, *Fx)
F2Config = McuConfig("STM32F2", 0x411, [0x4000 for _ in range(4)] + [0x10000] + [0x20000 for _ in range(7)], 12, *Fx)
F4Config = McuConfig("STM32F4", 0x463, [0x4000 for _ in range(4)] + [0x10000] + [0x20000 for _ in range(11)], 16, *Fx)
H7Config = McuConfig(
"STM32H7",
0x483,
[0x20000 for _ in range(7)],
8,
0x1FF1E800,
0x400,
# there is an 8th sector, but we use that for the provisioning chunk, so don't program over that!
[0x20000 for _ in range(7)],
0x080FFFC0,
0x8020000,
"panda_h7.bin.signed",

36
panda/python/dfu.py
View File

@@ -14,9 +14,9 @@ class PandaDFU:
def __init__(self, dfu_serial: Optional[str]):
# try USB, then SPI
handle: Optional[BaseSTBootloaderHandle]
handle = PandaDFU.usb_connect(dfu_serial)
self._context, handle = PandaDFU.usb_connect(dfu_serial)
if handle is None:
handle = PandaDFU.spi_connect(dfu_serial)
self._context, handle = PandaDFU.spi_connect(dfu_serial)
if handle is None:
raise Exception(f"failed to open DFU device {dfu_serial}")
@@ -24,8 +24,21 @@ class PandaDFU:
self._handle: BaseSTBootloaderHandle = handle
self._mcu_type: McuType = self._handle.get_mcu_type()
def __enter__(self):
return self
def __exit__(self, *args):
self.close()
def close(self):
if self._handle is not None:
self._handle.close()
self._handle = None
if self._context is not None:
self._context.close()
@staticmethod
def usb_connect(dfu_serial: Optional[str]) -> Optional[STBootloaderUSBHandle]:
def usb_connect(dfu_serial: Optional[str]):
handle = None
context = usb1.USBContext()
context.open()
@@ -40,10 +53,10 @@ class PandaDFU:
handle = STBootloaderUSBHandle(device, device.open())
break
return handle
return context, handle
@staticmethod
def spi_connect(dfu_serial: Optional[str]) -> Optional[STBootloaderSPIHandle]:
def spi_connect(dfu_serial: Optional[str]):
handle = None
this_dfu_serial = None
@@ -56,10 +69,10 @@ class PandaDFU:
if dfu_serial is not None and dfu_serial != this_dfu_serial:
handle = None
return handle
return None, handle
@staticmethod
def list() -> List[str]:
def list() -> List[str]: # noqa: A003
ret = PandaDFU.usb_list()
ret += PandaDFU.spi_list()
return list(set(ret))
@@ -82,7 +95,7 @@ class PandaDFU:
@staticmethod
def spi_list() -> List[str]:
try:
h = PandaDFU.spi_connect(None)
_, h = PandaDFU.spi_connect(None)
if h is not None:
dfu_serial = PandaDFU.st_serial_to_dfu_serial(h.get_uid(), h.get_mcu_type())
return [dfu_serial, ]
@@ -108,8 +121,11 @@ class PandaDFU:
def program_bootstub(self, code_bootstub):
self._handle.clear_status()
self._handle.erase_bootstub()
self._handle.erase_app()
# erase all sectors
for i in range(len(self._mcu_type.config.sector_sizes)):
self._handle.erase_sector(i)
self._handle.program(self._mcu_type.config.bootstub_address, code_bootstub)
def recover(self):

10
panda/python/isotp.py
View File

@@ -6,10 +6,8 @@ DEBUG = False
def msg(x):
if DEBUG:
print("S:", binascii.hexlify(x))
if len(x) <= 7:
ret = bytes([len(x)]) + x
else:
assert False
assert len(x) <= 7
ret = bytes([len(x)]) + x
return ret.ljust(8, b"\x00")
kmsgs = []
@@ -56,7 +54,7 @@ def isotp_recv_subaddr(panda, addr, bus, sendaddr, subaddr):
dat = msg[2:]
else:
print(binascii.hexlify(msg))
assert False
raise AssertionError
return dat[0:tlen]
@@ -133,7 +131,7 @@ def isotp_recv(panda, addr, bus=0, sendaddr=None, subaddr=None):
tlen = msg[0] & 0xf
dat = msg[1:]
else:
assert False
raise AssertionError
dat = dat[0:tlen]
if DEBUG:

2
panda/python/serial.py
View File

@@ -8,7 +8,7 @@ class PandaSerial(object):
self.panda.set_uart_baud(self.port, baud)
self.buf = b""
def read(self, l=1): # noqa: E741
def read(self, l=1):
tt = self.panda.serial_read(self.port)
if len(tt) > 0:
self.buf += tt

13
panda/python/spi.py
View File

@@ -17,10 +17,8 @@ from .utils import crc8_pedal
try:
import spidev
import spidev2
except ImportError:
spidev = None
spidev2 = None
# Constants
SYNC = 0x5A
@@ -194,6 +192,7 @@ class PandaSpiHandle(BaseHandle):
return dat[3:-1]
def _transfer_kernel_driver(self, spi, endpoint: int, data, timeout: int, max_rx_len: int = 1000, expect_disconnect: bool = False) -> bytes:
import spidev2
self.tx_buf[:len(data)] = data
self.ioctl_data.endpoint = endpoint
self.ioctl_data.tx_length = len(data)
@@ -314,7 +313,7 @@ class STBootloaderSPIHandle(BaseSTBootloaderHandle):
self._mcu_type = MCU_TYPE_BY_IDCODE[self.get_chip_id()]
except PandaSpiException:
raise PandaSpiException("failed to connect to panda") # pylint: disable=W0707
raise PandaSpiException("failed to connect to panda") from None
def _get_ack(self, spi, timeout=1.0):
data = 0x00
@@ -403,12 +402,6 @@ class STBootloaderSPIHandle(BaseSTBootloaderHandle):
# *** PandaDFU API ***
def erase_app(self):
self.erase_sector(1)
def erase_bootstub(self):
self.erase_sector(0)
def get_mcu_type(self):
return self._mcu_type
@@ -421,7 +414,7 @@ class STBootloaderSPIHandle(BaseSTBootloaderHandle):
def program(self, address, dat):
bs = 256 # max block size for writing to flash over SPI
dat += b"\xFF" * ((bs - len(dat)) % bs)
for i in range(0, len(dat) // bs):
for i in range(len(dat) // bs):
block = dat[i * bs:(i + 1) * bs]
self._cmd(0x31, data=[
struct.pack('>I', address + i*bs),

1
panda/python/uds.py
View File

@@ -1,4 +1,3 @@
#!/usr/bin/env python3
import time
import struct
from collections import deque

22
panda/python/usb.py
View File

@@ -34,9 +34,16 @@ class STBootloaderUSBHandle(BaseSTBootloaderHandle):
def __init__(self, libusb_device, libusb_handle):
self._libusb_handle = libusb_handle
# TODO: Find a way to detect F4 vs F2
# TODO: also check F4 BCD, don't assume in else
self._mcu_type = McuType.H7 if libusb_device.getbcdDevice() == 512 else McuType.F4
# example from F4: lsusb -v | grep Flash
# iInterface 4 @Internal Flash /0x08000000/04*016Kg,01*064Kg,011*128Kg
for i in range(20):
desc = libusb_handle.getStringDescriptor(i, 0)
if desc is not None and desc.startswith("@Internal Flash"):
sector_count = sum([int(s.split('*')[0]) for s in desc.split('/')[-1].split(',')])
break
mcu_by_sector_count = {m.config.sector_count: m for m in McuType}
assert sector_count in mcu_by_sector_count, f"Unkown MCU: {sector_count=}"
self._mcu_type = mcu_by_sector_count[sector_count]
def _status(self) -> None:
while 1:
@@ -51,11 +58,8 @@ class STBootloaderUSBHandle(BaseSTBootloaderHandle):
def get_mcu_type(self):
return self._mcu_type
def erase_app(self):
self._erase_page_address(self._mcu_type.config.app_address)
def erase_bootstub(self):
self._erase_page_address(self._mcu_type.config.bootstub_address)
def erase_sector(self, sector: int):
self._erase_page_address(self._mcu_type.config.sector_address(sector))
def clear_status(self):
# Clear status
@@ -78,7 +82,7 @@ class STBootloaderUSBHandle(BaseSTBootloaderHandle):
# Program
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):
for i in range(len(dat) // bs):
ldat = dat[i * bs:(i + 1) * bs]
print("programming %d with length %d" % (i, len(ldat)))
self._libusb_handle.controlWrite(0x21, self.DFU_DNLOAD, 2 + i, 0, ldat)