[FL-2269] Core2 OTA (#1144)
* C2OTA: wip * Update Cube to 1.13.3 * Fixed prio * Functional Core2 updater * Removed hardware CRC usage; code cleanup & linter fixes * Moved hardcoded stack params to copro.mk * Fixing CI bundling of core2 fw * Removed last traces of hardcoded radio stack * OB processing draft * Python scripts cleanup * Support for comments in ob data * Sacrificed SD card icon in favor of faster update. Waiting for Storage fix * Additional handling for OB mismatched values * Description for new furi_hal apis; spelling fixes * Rework of OB write, WIP * Properly restarting OB verification loop * Split update_task_workers.c * Checking OBs after enabling post-update mode * Moved OB verification before flashing * Removed ob.data for custom stacks * Fixed progress calculation for OB * Removed unnecessary OB mask cast Co-authored-by: Aleksandr Kutuzov <alleteam@gmail.com>
This commit is contained in:
hedger
GitHub
@@ -2,9 +2,12 @@ import logging
|
||||
import datetime
|
||||
import shutil
|
||||
import json
|
||||
from os.path import basename
|
||||
|
||||
import xml.etree.ElementTree as ET
|
||||
from flipper.utils import *
|
||||
from flipper.assets.coprobin import CoproBinary, get_stack_type
|
||||
|
||||
|
||||
CUBE_COPRO_PATH = "Projects/STM32WB_Copro_Wireless_Binaries"
|
||||
|
||||
@@ -13,14 +16,7 @@ MANIFEST_TEMPLATE = {
|
||||
"copro": {
|
||||
"fus": {"version": {"major": 1, "minor": 2, "sub": 0}, "files": []},
|
||||
"radio": {
|
||||
"version": {
|
||||
"type": 3,
|
||||
"major": 1,
|
||||
"minor": 13,
|
||||
"sub": 0,
|
||||
"branch": 0,
|
||||
"release": 5,
|
||||
},
|
||||
"version": {},
|
||||
"files": [],
|
||||
},
|
||||
},
|
||||
@@ -35,7 +31,7 @@ class Copro:
|
||||
self.mcu_copro = None
|
||||
self.logger = logging.getLogger(self.__class__.__name__)
|
||||
|
||||
def loadCubeInfo(self, cube_dir):
|
||||
def loadCubeInfo(self, cube_dir, cube_version):
|
||||
if not os.path.isdir(cube_dir):
|
||||
raise Exception(f'"{cube_dir}" doesn\'t exists')
|
||||
self.cube_dir = cube_dir
|
||||
@@ -51,8 +47,8 @@ class Copro:
|
||||
if not cube_version or not cube_version.startswith("FW.WB"):
|
||||
raise Exception(f"Incorrect Cube package or version info")
|
||||
cube_version = cube_version.replace("FW.WB.", "", 1)
|
||||
if cube_version != "1.13.1":
|
||||
raise Exception(f"Unknonwn cube version")
|
||||
if cube_version != cube_version:
|
||||
raise Exception(f"Unsupported cube version")
|
||||
self.version = cube_version
|
||||
|
||||
def addFile(self, array, filename, **kwargs):
|
||||
@@ -63,14 +59,32 @@ class Copro:
|
||||
{"name": filename, "sha256": file_sha256(destination_file), **kwargs}
|
||||
)
|
||||
|
||||
def bundle(self, output_dir):
|
||||
def bundle(self, output_dir, stack_file_name, stack_type, stack_addr=None):
|
||||
if not os.path.isdir(output_dir):
|
||||
raise Exception(f'"{output_dir}" doesn\'t exists')
|
||||
self.output_dir = output_dir
|
||||
stack_file = os.path.join(self.mcu_copro, stack_file_name)
|
||||
manifest_file = os.path.join(self.output_dir, "Manifest.json")
|
||||
# Form Manifest
|
||||
manifest = dict(MANIFEST_TEMPLATE)
|
||||
manifest["manifest"]["timestamp"] = timestamp()
|
||||
copro_bin = CoproBinary(stack_file)
|
||||
self.logger.info(f"Bundling {copro_bin.img_sig.get_version()}")
|
||||
stack_type_code = get_stack_type(stack_type)
|
||||
manifest["copro"]["radio"]["version"].update(
|
||||
{
|
||||
"type": stack_type_code,
|
||||
"major": copro_bin.img_sig.version_major,
|
||||
"minor": copro_bin.img_sig.version_minor,
|
||||
"sub": copro_bin.img_sig.version_sub,
|
||||
"branch": copro_bin.img_sig.version_branch,
|
||||
"release": copro_bin.img_sig.version_build,
|
||||
}
|
||||
)
|
||||
if not stack_addr:
|
||||
stack_addr = copro_bin.get_flash_load_addr()
|
||||
self.logger.info(f"Using guessed flash address 0x{stack_addr:x}")
|
||||
|
||||
# Old FUS Update
|
||||
self.addFile(
|
||||
manifest["copro"]["fus"]["files"],
|
||||
@@ -88,8 +102,8 @@ class Copro:
|
||||
# BLE Full Stack
|
||||
self.addFile(
|
||||
manifest["copro"]["radio"]["files"],
|
||||
"stm32wb5x_BLE_Stack_light_fw.bin",
|
||||
address="0x080D7000",
|
||||
stack_file_name,
|
||||
address=f"0x{stack_addr:X}",
|
||||
)
|
||||
# Save manifest to
|
||||
json.dump(manifest, open(manifest_file, "w"))
|
||||
|
||||
187
scripts/flipper/assets/coprobin.py
Normal file
187
scripts/flipper/assets/coprobin.py
Normal file
@@ -0,0 +1,187 @@
|
||||
import struct
|
||||
import math
|
||||
import os, os.path
|
||||
import sys
|
||||
|
||||
|
||||
# From STM32CubeWB\Middlewares\ST\STM32_WPAN\interface\patterns\ble_thread\shci\shci.h
|
||||
__STACK_TYPE_CODES = {
|
||||
"BLE_FULL": 0x01,
|
||||
"BLE_HCI": 0x02,
|
||||
"BLE_LIGHT": 0x03,
|
||||
"BLE_BEACON": 0x04,
|
||||
"BLE_BASIC": 0x05,
|
||||
"BLE_FULL_EXT_ADV": 0x06,
|
||||
"BLE_HCI_EXT_ADV": 0x07,
|
||||
"THREAD_FTD": 0x10,
|
||||
"THREAD_MTD": 0x11,
|
||||
"ZIGBEE_FFD": 0x30,
|
||||
"ZIGBEE_RFD": 0x31,
|
||||
"MAC": 0x40,
|
||||
"BLE_THREAD_FTD_STATIC": 0x50,
|
||||
"BLE_THREAD_FTD_DYAMIC": 0x51,
|
||||
"802154_LLD_TESTS": 0x60,
|
||||
"802154_PHY_VALID": 0x61,
|
||||
"BLE_PHY_VALID": 0x62,
|
||||
"BLE_LLD_TESTS": 0x63,
|
||||
"BLE_RLV": 0x64,
|
||||
"802154_RLV": 0x65,
|
||||
"BLE_ZIGBEE_FFD_STATIC": 0x70,
|
||||
"BLE_ZIGBEE_RFD_STATIC": 0x71,
|
||||
"BLE_ZIGBEE_FFD_DYNAMIC": 0x78,
|
||||
"BLE_ZIGBEE_RFD_DYNAMIC": 0x79,
|
||||
"RLV": 0x80,
|
||||
"BLE_MAC_STATIC": 0x90,
|
||||
}
|
||||
|
||||
|
||||
class CoproException(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
# Formats based on AN5185
|
||||
class CoproFooterBase:
|
||||
SIG_BIN_SIZE = 5 * 4
|
||||
_SIG_BIN_COMMON_SIZE = 2 * 4
|
||||
|
||||
def get_version(self):
|
||||
return f"Version {self.version_major}.{self.version_minor}.{self.version_sub}, branch {self.version_branch}, build {self.version_build} (magic {self.magic:X})"
|
||||
|
||||
def get_details(self):
|
||||
raise CoproException("Not implemented")
|
||||
|
||||
def __init__(self, raw: bytes):
|
||||
if len(raw) != self.SIG_BIN_SIZE:
|
||||
raise CoproException("Invalid footer size")
|
||||
sig_common_part = raw[-self._SIG_BIN_COMMON_SIZE :]
|
||||
parts = struct.unpack("BBBBI", sig_common_part)
|
||||
self.version_major = parts[3]
|
||||
self.version_minor = parts[2]
|
||||
self.version_sub = parts[1]
|
||||
# AN5185 mismatch: swapping byte halves
|
||||
self.version_build = parts[0] & 0x0F
|
||||
self.version_branch = (parts[0] & 0xF0) >> 4
|
||||
self.magic = parts[4]
|
||||
|
||||
|
||||
class CoproFusFooter(CoproFooterBase):
|
||||
FUS_MAGIC_IMG_STACK = 0x23372991
|
||||
FUS_MAGIC_IMG_FUS = 0x32279221
|
||||
FUS_MAGIC_IMG_OTHER = 0x42769811
|
||||
|
||||
FUS_BASE = 0x80F4000
|
||||
FLASH_PAGE_SIZE = 4 * 1024
|
||||
|
||||
def __init__(self, raw: bytes):
|
||||
super().__init__(raw)
|
||||
if self.magic not in (
|
||||
self.FUS_MAGIC_IMG_OTHER,
|
||||
self.FUS_MAGIC_IMG_FUS,
|
||||
self.FUS_MAGIC_IMG_STACK,
|
||||
):
|
||||
raise CoproException(f"Invalid FUS img magic {self.magic:x}")
|
||||
own_data = raw[: -self._SIG_BIN_COMMON_SIZE]
|
||||
parts = struct.unpack("IIBBBB", own_data)
|
||||
self.info1 = parts[0]
|
||||
self.info2 = parts[1]
|
||||
self.sram2b_1ks = parts[5]
|
||||
self.sram2a_1ks = parts[4]
|
||||
self.flash_4ks = parts[2]
|
||||
|
||||
def get_details(self):
|
||||
return f"SRAM2b={self.sram2b_1ks}k SRAM2a={self.sram2a_1ks}k flash={self.flash_4ks}p"
|
||||
|
||||
def is_stack(self):
|
||||
return self.magic == self.FUS_MAGIC_IMG_STACK
|
||||
|
||||
def get_flash_pages(self, fullsize):
|
||||
return math.ceil(fullsize / self.FLASH_PAGE_SIZE)
|
||||
|
||||
def get_flash_base(self, fullsize):
|
||||
if not self.is_stack():
|
||||
raise CoproException("Not a stack image")
|
||||
return self.FUS_BASE - self.get_flash_pages(fullsize) * self.FLASH_PAGE_SIZE
|
||||
|
||||
|
||||
class CoproSigFooter(CoproFooterBase):
|
||||
SIG_MAGIC_ST = 0xD3A12C5E
|
||||
SIG_MAGIC_CUSTOMER = 0xE2B51D4A
|
||||
|
||||
def __init__(self, raw: bytes):
|
||||
super().__init__(raw)
|
||||
if self.magic not in (self.SIG_MAGIC_ST, self.SIG_MAGIC_CUSTOMER):
|
||||
raise CoproException(f"Invalid FUS img magic {self.magic:x}")
|
||||
own_data = raw[: -self._SIG_BIN_COMMON_SIZE]
|
||||
parts = struct.unpack("IIBBH", own_data)
|
||||
self.reserved_1 = parts[0]
|
||||
self.reserved_2 = parts[1]
|
||||
self.size = parts[2]
|
||||
self.source = parts[3]
|
||||
self.reserved_34 = parts[4]
|
||||
|
||||
def get_details(self):
|
||||
return f"Signature Src {self.source:x} size {self.size:x}"
|
||||
|
||||
|
||||
class CoproBinary:
|
||||
def __init__(self, binary_path):
|
||||
self.binary_path = binary_path
|
||||
self.img_sig_footer = None
|
||||
self.img_sig = None
|
||||
self.binary_size = -1
|
||||
self._load()
|
||||
|
||||
def _load(self):
|
||||
with open(self.binary_path, "rb") as fin:
|
||||
whole_file = fin.read()
|
||||
self.binary_size = len(whole_file)
|
||||
|
||||
img_sig_footer_bin = whole_file[-CoproFooterBase.SIG_BIN_SIZE :]
|
||||
self.img_sig_footer = CoproSigFooter(img_sig_footer_bin)
|
||||
img_sig_size = self.img_sig_footer.size + CoproSigFooter.SIG_BIN_SIZE
|
||||
img_sig_bin = whole_file[
|
||||
-(img_sig_size + CoproFusFooter.SIG_BIN_SIZE) : -img_sig_size
|
||||
]
|
||||
self.img_sig = CoproFusFooter(img_sig_bin)
|
||||
|
||||
def is_valid(self):
|
||||
return self.img_sig_footer is not None and self.img_sig is not None
|
||||
|
||||
def is_stack(self):
|
||||
return self.img_sig and self.img_sig.is_stack()
|
||||
|
||||
def get_flash_load_addr(self):
|
||||
if not self.is_stack():
|
||||
raise CoproException("Not a stack image")
|
||||
return self.img_sig.get_flash_base(self.binary_size)
|
||||
|
||||
|
||||
def get_stack_type(typestr: str):
|
||||
stack_code = __STACK_TYPE_CODES.get(typestr.upper(), None)
|
||||
if stack_code is None:
|
||||
raise CoproException(f"Unknown stack type {typestr}. See shci.h")
|
||||
return stack_code
|
||||
|
||||
|
||||
def _load_bin(binary_path: str):
|
||||
print(binary_path)
|
||||
copro_bin = CoproBinary(binary_path)
|
||||
print(copro_bin.img_sig.get_version())
|
||||
if copro_bin.img_sig.is_stack():
|
||||
print(f"\t>> FLASH AT {copro_bin.get_flash_load_addr():X}\n")
|
||||
|
||||
|
||||
def main():
|
||||
coprodir = (
|
||||
sys.argv[1]
|
||||
if len(sys.argv) > 1
|
||||
else "../../../lib/STM32CubeWB/Projects/STM32WB_Copro_Wireless_Binaries/STM32WB5x"
|
||||
)
|
||||
for fn in os.listdir(coprodir):
|
||||
if not fn.endswith(".bin"):
|
||||
continue
|
||||
_load_bin(os.path.join(coprodir, fn))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
208
scripts/flipper/assets/obdata.py
Normal file
208
scripts/flipper/assets/obdata.py
Normal file
@@ -0,0 +1,208 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import logging
|
||||
import struct
|
||||
|
||||
from enum import Enum
|
||||
from dataclasses import dataclass
|
||||
from typing import Tuple
|
||||
from array import array
|
||||
|
||||
|
||||
class OBException(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass
|
||||
class OBParams:
|
||||
word_idx: int
|
||||
bits: Tuple[int, int]
|
||||
name: str
|
||||
|
||||
|
||||
_OBS_descr = (
|
||||
OBParams(0, (0, 8), "RDP"),
|
||||
OBParams(0, (8, 9), "ESE"),
|
||||
OBParams(0, (9, 12), "BOR_LEV"),
|
||||
OBParams(0, (12, 13), "nRST_STOP"),
|
||||
OBParams(0, (13, 14), "nRST_STDBY"),
|
||||
OBParams(0, (14, 15), "nRSTSHDW"),
|
||||
OBParams(0, (15, 16), "UNUSED1"),
|
||||
OBParams(0, (16, 17), "IWDGSW"),
|
||||
OBParams(0, (17, 18), "IWDGSTOP"),
|
||||
OBParams(0, (18, 19), "IWGDSTDBY"), # ST's typo: IWDGSTDBY
|
||||
OBParams(0, (18, 19), "IWDGSTDBY"), # ST's typo: IWDGSTDBY
|
||||
OBParams(0, (19, 20), "WWDGSW"),
|
||||
OBParams(0, (20, 23), "UNUSED2"),
|
||||
OBParams(0, (23, 24), "nBOOT1"),
|
||||
OBParams(0, (24, 25), "SRAM2PE"),
|
||||
OBParams(0, (25, 26), "SRAM2RST"),
|
||||
OBParams(0, (26, 27), "nSWBOOT0"),
|
||||
OBParams(0, (27, 28), "nBOOT0"),
|
||||
OBParams(0, (28, 29), "UNUSED3"),
|
||||
OBParams(0, (29, 32), "AGC_TRIM"),
|
||||
OBParams(1, (0, 9), "PCROP1A_STRT"),
|
||||
OBParams(1, (9, 32), "UNUSED"),
|
||||
OBParams(2, (0, 9), "PCROP1A_END"),
|
||||
OBParams(2, (9, 31), "UNUSED"),
|
||||
OBParams(2, (31, 32), "PCROP_RDP"),
|
||||
OBParams(3, (0, 8), "WRP1A_STRT"),
|
||||
OBParams(3, (8, 16), "UNUSED1"),
|
||||
OBParams(3, (16, 24), "WRP1A_END"),
|
||||
OBParams(3, (24, 32), "UNUSED2"),
|
||||
OBParams(4, (0, 8), "WRP1B_STRT"),
|
||||
OBParams(4, (8, 16), "UNUSED1"),
|
||||
OBParams(4, (16, 24), "WRP1B_END"),
|
||||
OBParams(4, (24, 32), "UNUSED2"),
|
||||
OBParams(5, (0, 9), "PCROP1B_STRT"),
|
||||
OBParams(5, (9, 32), "UNUSED"),
|
||||
OBParams(6, (0, 9), "PCROP1B_END"),
|
||||
OBParams(6, (9, 32), "UNUSED"),
|
||||
OBParams(13, (0, 14), "IPCCDBA"),
|
||||
OBParams(13, (14, 32), "UNUSED"),
|
||||
OBParams(14, (0, 8), "SFSA"),
|
||||
OBParams(14, (8, 9), "FSD"),
|
||||
OBParams(14, (9, 12), "UNUSED1"),
|
||||
OBParams(14, (12, 13), "DDS"),
|
||||
OBParams(14, (13, 32), "UNUSED2"),
|
||||
OBParams(15, (0, 18), "SBRV"),
|
||||
OBParams(15, (18, 23), "SBRSA"),
|
||||
OBParams(15, (23, 24), "BRSD"),
|
||||
OBParams(15, (24, 25), "UNUSED1"),
|
||||
OBParams(15, (25, 30), "SNBRSA"),
|
||||
OBParams(15, (30, 31), "NBRSD"),
|
||||
OBParams(15, (31, 32), "C2OPT"),
|
||||
)
|
||||
|
||||
|
||||
_OBS = dict((param.name, param) for param in _OBS_descr)
|
||||
|
||||
|
||||
@dataclass
|
||||
class EncodedOBValue:
|
||||
value: int
|
||||
mask: int
|
||||
params: OBParams
|
||||
|
||||
|
||||
class OptionByte:
|
||||
class OBMode(Enum):
|
||||
IGNORE = 0
|
||||
READ = 1
|
||||
READ_WRITE = 2
|
||||
|
||||
@classmethod
|
||||
def from_str(cls, value):
|
||||
if value == "r":
|
||||
return cls.READ
|
||||
elif value == "rw":
|
||||
return cls.READ_WRITE
|
||||
else:
|
||||
raise OBException(f"Unknown OB check mode '{value}'")
|
||||
|
||||
def __init__(self, obstr):
|
||||
parts = obstr.split(":")
|
||||
if len(parts) != 3:
|
||||
raise OBException(f"Invalid OB value definition {obstr}")
|
||||
self.name = parts[0]
|
||||
self.value = int(parts[1], 16)
|
||||
self.mode = OptionByte.OBMode.from_str(parts[2].strip())
|
||||
self.descr = _OBS.get(self.name, None)
|
||||
if self.descr is None:
|
||||
raise OBException(f"Missing OB descriptor for {self.name}")
|
||||
|
||||
def encode(self):
|
||||
startbit, endbit = self.descr.bits
|
||||
value_mask = 2 ** (endbit - startbit) - 1
|
||||
value_corrected = self.value & value_mask
|
||||
|
||||
value_shifted = value_corrected << startbit
|
||||
value_mask_shifted = value_mask << startbit
|
||||
return EncodedOBValue(value_shifted, value_mask_shifted, self)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<OB {self.name}, 0x{self.value:x}, {self.mode} at 0x{id(self):X}>"
|
||||
|
||||
|
||||
@dataclass
|
||||
class ObReferenceValues:
|
||||
reference: bytes
|
||||
compare_mask: bytes
|
||||
write_mask: bytes
|
||||
|
||||
|
||||
class ObReferenceValuesGenerator:
|
||||
def __init__(self):
|
||||
self.compare_mask = array("I", [0] * 16)
|
||||
self.write_mask = array("I", [0] * 16)
|
||||
self.ref_values = array("I", [0] * 16)
|
||||
|
||||
def __repr__(self):
|
||||
return (
|
||||
f"<OBRefs REFS=[{' '.join(hex(v) for v in self.ref_values)}] "
|
||||
f"CMPMASK=[{' '.join(hex(v) for v in self.compare_mask)}] "
|
||||
f"WRMASK=[{' '.join(hex(v) for v in self.write_mask)}] "
|
||||
)
|
||||
|
||||
def export_values(self):
|
||||
export_cmpmask = array("I")
|
||||
for value in self.compare_mask:
|
||||
export_cmpmask.append(value)
|
||||
export_cmpmask.append(value)
|
||||
export_wrmask = array("I")
|
||||
for value in self.write_mask:
|
||||
export_wrmask.append(value)
|
||||
export_wrmask.append(value)
|
||||
export_refvals = array("I")
|
||||
for cmpmask, refval in zip(self.compare_mask, self.ref_values):
|
||||
export_refvals.append(refval)
|
||||
export_refvals.append((refval ^ 0xFFFFFFFF) & cmpmask)
|
||||
return export_refvals, export_cmpmask, export_wrmask
|
||||
|
||||
def export(self):
|
||||
return ObReferenceValues(*map(lambda a: a.tobytes(), self.export_values()))
|
||||
|
||||
def apply(self, ob):
|
||||
ob_params = ob.descr
|
||||
encoded_ob = ob.encode()
|
||||
self.compare_mask[ob_params.word_idx] |= encoded_ob.mask
|
||||
self.ref_values[ob_params.word_idx] |= encoded_ob.value
|
||||
if ob.mode == OptionByte.OBMode.READ_WRITE:
|
||||
self.write_mask[ob_params.word_idx] |= encoded_ob.mask
|
||||
|
||||
|
||||
class OptionBytesData:
|
||||
def __init__(self, obfname):
|
||||
self.obs = list()
|
||||
with open(obfname, "rt") as obfin:
|
||||
self.obs = list(
|
||||
OptionByte(line) for line in obfin if not line.startswith("#")
|
||||
)
|
||||
|
||||
def gen_values(self):
|
||||
obref = ObReferenceValuesGenerator()
|
||||
converted_refs = list(obref.apply(ob) for ob in self.obs)
|
||||
return obref
|
||||
|
||||
|
||||
def main():
|
||||
with open("../../../../logs/obs.bin", "rb") as obsbin:
|
||||
ob_sample = obsbin.read(128)
|
||||
ob_sample_arr = array("I", ob_sample)
|
||||
print(ob_sample_arr)
|
||||
|
||||
obd = OptionBytesData("../../ob.data")
|
||||
print(obd.obs)
|
||||
# print(obd.gen_values().export())
|
||||
ref, mask, wrmask = obd.gen_values().export_values()
|
||||
for idx in range(len(ob_sample_arr)):
|
||||
real_masked = ob_sample_arr[idx] & mask[idx]
|
||||
print(
|
||||
f"#{idx}: ref {ref[idx]:08x} real {real_masked:08x} ({ob_sample_arr[idx]:08x} & {mask[idx]:08x}) match {ref[idx]==real_masked}"
|
||||
)
|
||||
|
||||
# print(ob_sample)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user