From 67a9718ed6c07f45386f1af55aafd51698cd489c Mon Sep 17 00:00:00 2001
From: Benjamin Koch <bbbsnowball@gmail.com>
Date: Tue, 30 May 2023 04:01:37 +0200
Subject: [PATCH] test+fix firmware download
---
.../rust1/download_firmware_via_modbus.py | 98 ++++++++++++
firmware/rust1/memory.x | 2 +
firmware/rust1/rtumaster_pymodbus.py | 1 -
firmware/rust1/src/bin/heizung.rs | 10 +-
firmware/rust1/src/modbus_server.rs | 9 +-
firmware/rust1/src/uf2updater.rs | 79 +++++++---
firmware/rust1/uf2.py | 145 +++++++++++++-----
7 files changed, 276 insertions(+), 68 deletions(-)
create mode 100644 firmware/rust1/download_firmware_via_modbus.py
diff --git a/firmware/rust1/download_firmware_via_modbus.py b/firmware/rust1/download_firmware_via_modbus.py
new file mode 100644
index 0000000..a1f287a
--- /dev/null
+++ b/firmware/rust1/download_firmware_via_modbus.py
@@ -0,0 +1,98 @@
+#!/usr/bin/env python
+# -*- coding: utf_8 -*-
+
+# based on https://github.com/pymodbus-dev/pymodbus/blob/dev/examples/client_sync.py
+
+import logging
+import time
+import sys
+import struct
+import pymodbus
+from pymodbus.exceptions import ModbusException
+from pymodbus.client import ModbusSerialClient
+from pymodbus.file_message import FileRecord
+from uf2 import *
+
+PORT = '/dev/ttyUSB0'
+DEVICE_ADDR = 1
+FIRMWARE = "./heizung-release.uf2"
+ACTIVE_PARTITION_ADDRESS = 0x10007000
+
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+log_handler = logging.StreamHandler(sys.stdout)
+logger.addHandler(log_handler)
+
+def main():
+ firmware_blocks = read_blocks(FIRMWARE)
+ firmware_streams = [[firmware_blocks[0]]]
+ prev = firmware_blocks[0]
+ for block in firmware_blocks[1:]:
+ if block.block_no == 0 or block.num_blocks != prev.num_blocks:
+ firmware_streams.append([])
+ firmware_streams[-1].append(block)
+ prev = block
+ logger.info("found %s streams in firmware file %s" % (len(firmware_streams), FIRMWARE))
+
+ firmware_streams2 = []
+ for i, stream in enumerate(firmware_streams):
+ valid = all(block.valid for block in stream)
+ relevant = any(block.target_addr >= ACTIVE_PARTITION_ADDRESS for block in stream)
+ if not valid:
+ raise Exception("stream %s contains invalid blocks" % i)
+ if relevant:
+ firmware_streams2.append(stream)
+ else:
+ logger.info("skipping stream %s because it contains only irrelevant blocks (e.g. bootloader)" % i)
+ if len(firmware_streams2) == 0:
+ raise Exception("no relevant streams")
+
+ try:
+ client = ModbusSerialClient(
+ port=PORT,
+ timeout=0.2,
+ # Common optional paramers:
+ # framer=ModbusRtuFramer,
+ # timeout=10,
+ # retries=3,
+ # retry_on_empty=False,
+ # close_comm_on_error=False,
+ # strict=True,
+ baudrate=19200,
+ parity="E",
+ )
+ client.connect()
+ logger.info("connected")
+
+ blocks_progress = 0
+ blocks_total = sum(len(stream) for stream in firmware_streams2) * 3
+ for stream in firmware_streams2:
+ #bytes = b"".join(block.bytes for block in stream)
+ for block in stream:
+ maxlen = (0xf5-7) // 2 * 2
+ offset = 0
+ for j in range(3):
+ x = client.write_file_record([
+ FileRecord(file_number=1, record_number=offset//2, record_data=block.bytes[offset:offset+maxlen]),
+ ], slave=DEVICE_ADDR)
+ blocks_progress += 1
+ logger.info("%3s / %3s, %08x, %s", blocks_progress, blocks_total, block.target_addr, x)
+ offset += maxlen
+
+ x = client.read_input_registers(25, 4, slave=DEVICE_ADDR)
+ logger.info(x)
+ logger.info(x.registers)
+ if x.registers != [0, 1, 0, 1]:
+ raise Exception("We haven't implemented re-sending blocks, yet.")
+
+ x = client.write_registers(1, struct.unpack(">H", b"UP")[0], slave=DEVICE_ADDR)
+ logger.info("write to mark for update: %s", x)
+
+ x = client.write_registers(1, struct.unpack(">H", b"RE")[0], slave=DEVICE_ADDR)
+ logger.info("write to reset: %s", x)
+
+ except ModbusException as exc:
+ logger.error("%s", exc)
+
+if __name__ == "__main__":
+ main()
diff --git a/firmware/rust1/memory.x b/firmware/rust1/memory.x
index 7578153..5e049f0 100644
--- a/firmware/rust1/memory.x
+++ b/firmware/rust1/memory.x
@@ -8,6 +8,8 @@ MEMORY
RAM : ORIGIN = 0x20000000, LENGTH = 256K
}
+__bootloader_flash_start = ORIGIN(BOOT2);
+
__bootloader_state_start = ORIGIN(BOOTLOADER_STATE) - ORIGIN(BOOT2);
__bootloader_state_end = ORIGIN(BOOTLOADER_STATE) + LENGTH(BOOTLOADER_STATE) - ORIGIN(BOOT2);
diff --git a/firmware/rust1/rtumaster_pymodbus.py b/firmware/rust1/rtumaster_pymodbus.py
index 08b36a4..bd890c2 100644
--- a/firmware/rust1/rtumaster_pymodbus.py
+++ b/firmware/rust1/rtumaster_pymodbus.py
@@ -18,7 +18,6 @@ DEVICE_ADDR = 1
logger = logging.getLogger()
logger.setLevel(logging.INFO)
log_handler = logging.StreamHandler(sys.stdout)
-#log_handler.setFormatter(logging.Formatter("%(asctime)s\t%(levelname)s\t%(module)s.%(funcName)s\t%(threadName)s\t%(message)s"))
logger.addHandler(log_handler)
def main():
diff --git a/firmware/rust1/src/bin/heizung.rs b/firmware/rust1/src/bin/heizung.rs
index d9bd06f..e14ef8b 100644
--- a/firmware/rust1/src/bin/heizung.rs
+++ b/firmware/rust1/src/bin/heizung.rs
@@ -361,19 +361,19 @@ impl<'a> ModBusRegs<'a> {
fn read_reg_2x_u32(self: &mut Self, read_type: ReadType, base_addr: u16, addr: u16, value: [u32; 2]) -> Result<u16, ModbusErrorCode> {
self.read_reg(read_type, base_addr, addr, &[
- ((value[0] >> 0) & 0xffff) as u16,
((value[0] >> 16) & 0xffff) as u16,
- ((value[1] >> 0) & 0xffff) as u16,
+ ((value[0] >> 0) & 0xffff) as u16,
((value[1] >> 16) & 0xffff) as u16,
+ ((value[1] >> 0) & 0xffff) as u16,
])
}
fn read_reg_u64(self: &mut Self, read_type: ReadType, base_addr: u16, addr: u16, value: u64) -> Result<u16, ModbusErrorCode> {
self.read_reg(read_type, base_addr, addr, &[
- ((value >> 0) & 0xffff) as u16,
- ((value >> 16) & 0xffff) as u16,
- ((value >> 32) & 0xffff) as u16,
((value >> 48) & 0xffff) as u16,
+ ((value >> 32) & 0xffff) as u16,
+ ((value >> 16) & 0xffff) as u16,
+ ((value >> 0) & 0xffff) as u16,
])
}
}
diff --git a/firmware/rust1/src/modbus_server.rs b/firmware/rust1/src/modbus_server.rs
index fa1c6d2..bd13b5b 100644
--- a/firmware/rust1/src/modbus_server.rs
+++ b/firmware/rust1/src/modbus_server.rs
@@ -47,7 +47,7 @@ impl<'a, E: Clone> Cursor<'a, E> {
}
fn read_bytes(self: &mut Self, len: usize) -> Result<&'a [u8], E> {
- if self.1 + len >= self.0.len() {
+ if self.1 + len > self.0.len() {
return Err(self.2.clone())
}
let data = &self.0[self.1 .. self.1 + len];
@@ -181,7 +181,7 @@ impl<REGS: ModbusRegisters> ModbusServer<REGS> {
let rxbuf = &self.rxbuf;
let mut rx = Cursor::new(&rxbuf[0..rxbuf.len()-2], ModbusErrorCode::IllegalDataValue);
let txbuf = &mut self.txbuf;
- info!("Modbus frame: {:?}", rxbuf.as_slice());
+ //info!("Modbus frame: {:?}", rxbuf.as_slice());
txbuf.clear();
let capacity = txbuf.capacity();
@@ -408,7 +408,8 @@ impl<REGS: ModbusRegisters> ModbusServer<REGS> {
let file_number = rx.read_u16be()?;
let record_number = rx.read_u16be()?;
let record_length = rx.read_u16be()?;
- if record_length > 127 || record_length as usize * 2 > capacity - txbuf.len() {
+ if record_length > 127 {
+ info!("Modbus write file record: too long, record_length={}, capacity={}", record_length, capacity - txbuf.len());
return Err(IllegalDataValue)
}
let data = rx.read_bytes(record_length as usize * 2)?;
@@ -534,7 +535,7 @@ impl<REGS: ModbusRegisters> RS485Handler for ModbusServer<REGS> {
self.handle_modbus_frame();
if !self.txbuf.is_empty() {
- info!("Modbus reply: {:?}", self.txbuf);
+ //info!("Modbus reply: {:?}", self.txbuf);
match reply {
Option::Some(reply) => reply(self.txbuf.as_slice()),
Option::None => warn!("Cannot send reply because a reply is already in progress!"),
diff --git a/firmware/rust1/src/uf2updater.rs b/firmware/rust1/src/uf2updater.rs
index 17db334..af26301 100644
--- a/firmware/rust1/src/uf2updater.rs
+++ b/firmware/rust1/src/uf2updater.rs
@@ -9,15 +9,16 @@ use zerocopy::FromBytes;
use crate::{modbus_server::ModbusErrorCode, uf2::*};
-#[derive(PartialEq, Eq)]
+#[derive(PartialEq, Eq, Debug, Format)]
enum PositionInSector {
Start, StartPartial, Middle, End
}
struct BootLoaderPartitions {
- _state: Partition,
+ state: Partition,
active: Partition,
dfu: Partition,
+ flash_start_addr: u32,
}
// copied from embassy/embassy-boot/rp/src/lib.rs because fields are private for BootLoader
impl Default for BootLoaderPartitions {
@@ -30,6 +31,7 @@ impl Default for BootLoaderPartitions {
static __bootloader_active_end: u32;
static __bootloader_dfu_start: u32;
static __bootloader_dfu_end: u32;
+ static __bootloader_flash_start: u32;
}
let active = unsafe {
@@ -44,14 +46,17 @@ impl Default for BootLoaderPartitions {
&__bootloader_dfu_end as *const u32 as u32,
)
};
- let _state = unsafe {
+ let state = unsafe {
Partition::new(
&__bootloader_state_start as *const u32 as u32,
&__bootloader_state_end as *const u32 as u32,
)
};
+ let flash_start_addr = unsafe {
+ &__bootloader_flash_start as *const u32 as u32
+ };
- BootLoaderPartitions{ active, dfu, _state }
+ BootLoaderPartitions{ active, dfu, state, flash_start_addr }
}
}
@@ -67,6 +72,7 @@ pub struct UF2UpdateHandler<const FLASH_SIZE: usize> {
uf2_seen_bitmask: BitArr!(for MAX_UF2_SECTORS, in u32),
uf2_num_blocks: u32,
flash_erased_address_and_first_block: Option<(u32, usize)>,
+ bootloader_state_erased: bool,
}
impl<const FLASH_SIZE: usize> UF2UpdateHandler<FLASH_SIZE> {
@@ -79,6 +85,7 @@ impl<const FLASH_SIZE: usize> UF2UpdateHandler<FLASH_SIZE> {
uf2_seen_bitmask: bitarr!(u32, Lsb0; 0; MAX_UF2_SECTORS),
uf2_num_blocks: 0,
flash_erased_address_and_first_block: None,
+ bootloader_state_erased: false,
}
}
@@ -135,8 +142,10 @@ impl<const FLASH_SIZE: usize> UF2UpdateHandler<FLASH_SIZE> {
fn process_sector(self: &mut Self) -> Result<(), ModbusErrorCode> {
defmt::assert!(self.write_pos == 512);
- let uf2_header = Uf2BlockHeader::read_from_prefix(self.buf.0.as_slice()).unwrap();
- let uf2_footer = Uf2BlockFooter::read_from_suffix(self.buf.0.as_slice()).unwrap();
+ let uf2_block = &self.buf.0[0..512];
+
+ let uf2_header = Uf2BlockHeader::read_from_prefix(uf2_block).unwrap();
+ let uf2_footer = Uf2BlockFooter::read_from_suffix(uf2_block).unwrap();
if uf2_header.magic_start0 != UF2_MAGIC_START0 || uf2_header.magic_start1 != UF2_MAGIC_START1 || uf2_footer.magic_end != UF2_MAGIC_END {
warn!("Invalid magic in UF2 block");
return Err(ModbusErrorCode::IllegalDataValue)
@@ -155,26 +164,28 @@ impl<const FLASH_SIZE: usize> UF2UpdateHandler<FLASH_SIZE> {
}
if (uf2_header.flags & UF2_FLAG_FAMILY_ID_PRESENT) == 0 || uf2_header.file_size != RP2040_FAMILY_ID {
// not for us but that shouldn't be treated as an error
+ warn!("Ignoring UF2 block for different family");
self.uf2_seen_bitmask.set(uf2_header.block_no as usize, true);
return Ok(())
}
if (uf2_header.flags & (UF2_FLAG_NOT_MAIN_FLASH | UF2_FLAG_FILE_CONTAINER)) != 0 {
// not for DFU partition
+ warn!("Ignoring UF2 block that is not for main flash");
self.uf2_seen_bitmask.set(uf2_header.block_no as usize, true);
return Ok(())
}
- if uf2_header.block_no != self.uf2_num_blocks {
+ if uf2_header.block_no == 0 || uf2_header.num_blocks != self.uf2_num_blocks {
self.uf2_seen_bitmask.fill_with(|_| false);
- self.uf2_num_blocks = uf2_header.block_no;
+ self.uf2_num_blocks = uf2_header.num_blocks;
self.flash_erased_address_and_first_block = None;
}
let data_start = size_of::<Uf2BlockHeader>();
let data_end = data_start + uf2_header.payload_size as usize;
let addr = uf2_header.target_addr as usize;
- let til_end_of_page = 4096 - (addr & 4096);
- if (addr & 4096) == 0 {
+ let til_end_of_page = 4096 - (addr % 4096);
+ if (addr % 4096) == 0 {
self.process_sector_part(PositionInSector::Start, uf2_header.block_no as usize, addr as u32, data_start, data_end);
} else if (uf2_header.payload_size as usize) < til_end_of_page {
self.process_sector_part(PositionInSector::Middle, uf2_header.block_no as usize, addr as u32, data_start, data_end);
@@ -183,7 +194,7 @@ impl<const FLASH_SIZE: usize> UF2UpdateHandler<FLASH_SIZE> {
let remaining = uf2_header.payload_size as usize - til_end_of_page;
if remaining > 0 {
self.process_sector_part(PositionInSector::StartPartial, uf2_header.block_no as usize,
- (addr + til_end_of_page) as u32, til_end_of_page, data_end);
+ (addr + til_end_of_page) as u32, data_start + til_end_of_page, data_end);
}
}
@@ -195,16 +206,20 @@ impl<const FLASH_SIZE: usize> UF2UpdateHandler<FLASH_SIZE> {
let data = &self.buf.0[data_start .. data_end];
+ info!("process_sector_part: block {}, {:?}, addr {:08x}, len {}", block_no, pos, addr, data.len());
+
let partitions = BootLoaderPartitions::default();
- if addr < partitions.active.from || addr >= partitions.active.to {
+ if addr < partitions.flash_start_addr + partitions.active.from || addr >= partitions.flash_start_addr + partitions.active.to {
// We don't want to write this.
if pos != StartPartial {
self.uf2_seen_bitmask.set(block_no, true);
}
+ //info!("not in active partition: not {:08x} <= {:08x} < {:08x}",
+ // partitions.flash_start_addr + partitions.active.from, addr, partitions.flash_start_addr + partitions.active.to);
return;
}
- //FIXME We have to use FirmwareUpdater, I think.
- let addr = addr - partitions.active.from + partitions.dfu.to;
+ let addr_orig = addr;
+ let addr = addr - partitions.flash_start_addr - partitions.active.from + partitions.dfu.to;
let abort_previous: bool;
let process_current: bool;
@@ -230,6 +245,8 @@ impl<const FLASH_SIZE: usize> UF2UpdateHandler<FLASH_SIZE> {
// If there is a partially written page, mark the blocks as not seen so we will later try again.
if abort_previous {
+ info!("aborting from block {} to {} because flash_erased_address_and_first_block={:?} and current is {:?} {}",
+ first_block, block_no, self.flash_erased_address_and_first_block, pos, block_no);
for i in first_block .. block_no {
self.uf2_seen_bitmask.set(i, false);
}
@@ -258,15 +275,33 @@ impl<const FLASH_SIZE: usize> UF2UpdateHandler<FLASH_SIZE> {
}
if is_start {
- // We have to erase the block.
+ // We have to erase the block. However, let's erase the bootloader state first
+ // because we don't want to swap into a partially cleared DFU partition.
+ if !self.bootloader_state_erased {
+ info!("erasing state partition at {:08x}", partitions.state.from);
+ match self.flash.erase(partitions.state.from, partitions.state.to) {
+ Ok(()) => (),
+ Err(e) => {
+ error!("erase: {:?} at {:08x}", e, partitions.state.from);
+ return;
+ }
+ }
+
+ self.bootloader_state_erased = true;
+ }
+
+ info!("erasing at {:08x} -> {:08x}", addr_orig, addr);
match self.flash.erase(addr, addr+4096) {
Ok(()) => (),
Err(e) => {
- error!("erase: {:?}", e);
+ error!("erase: {:?} at {:08x} -> {:08x}", e, addr_orig, addr);
return;
}
}
+ // From which block should we start marking as unseen if we later encounter an error
+ // for this block? This is usually the current block but if this block also contains
+ // the end of another sector, we use the next one.
let first_block = match pos {
Start => block_no,
StartPartial => block_no+1,
@@ -279,9 +314,10 @@ impl<const FLASH_SIZE: usize> UF2UpdateHandler<FLASH_SIZE> {
let (mut prev_addr, first_block) = self.flash_erased_address_and_first_block.unwrap();
defmt::assert!(prev_addr == addr);
//FIXME We might have to handle alignment concerns for address and data.
+ info!("writing to {:08x} -> {:08x}, len={}", addr_orig, addr, data.len());
let r = self.flash.write(addr, data);
if let Err(e) = r {
- error!("write: {:?}", e);
+ error!("write: {:?} at {:08x} -> {:08x}", e, addr_orig, addr);
// abort current block
for i in first_block .. block_no {
@@ -292,7 +328,7 @@ impl<const FLASH_SIZE: usize> UF2UpdateHandler<FLASH_SIZE> {
}
prev_addr += data.len() as u32;
- if (prev_addr & 4096) == 0 {
+ if (prev_addr % 4096) == 0 {
self.flash_erased_address_and_first_block = None;
} else {
self.flash_erased_address_and_first_block = Some((prev_addr, first_block));
@@ -317,6 +353,9 @@ impl<const FLASH_SIZE: usize> UF2UpdateHandler<FLASH_SIZE> {
let first_present = first_present.first_one()
.map(|x| (x + first_missing) as u32)
.unwrap_or(self.uf2_num_blocks);
+ let first_missing = first_missing as u32;
+ // subtract one because a partial sector might be in the previous block
+ let first_missing = if first_missing > 0 { first_missing - 1 } else { first_missing };
(first_missing as u32, first_present)
}
}
@@ -328,6 +367,8 @@ impl<const FLASH_SIZE: usize> UF2UpdateHandler<FLASH_SIZE> {
}
pub fn mark_updated(self: &mut Self) -> Result<(), ModbusErrorCode> {
+ self.bootloader_state_erased = false;
+
let mut updater = FirmwareUpdater::default();
match updater.mark_updated_blocking(&mut self.flash, &mut self.buf.0[..1]) {
Ok(()) => Ok(()),
@@ -339,6 +380,8 @@ impl<const FLASH_SIZE: usize> UF2UpdateHandler<FLASH_SIZE> {
}
pub fn mark_booted(self: &mut Self) -> Result<(), ModbusErrorCode> {
+ self.bootloader_state_erased = false;
+
let mut updater = FirmwareUpdater::default();
match updater.mark_booted_blocking(&mut self.flash, &mut self.buf.0[..1]) {
Ok(()) => Ok(()),
diff --git a/firmware/rust1/uf2.py b/firmware/rust1/uf2.py
index d6e71b9..f62d825 100644
--- a/firmware/rust1/uf2.py
+++ b/firmware/rust1/uf2.py
@@ -1,8 +1,5 @@
import sys, struct
-with open(sys.argv[1], "rb") as f:
- contents = f.read()
-
# see https://github.com/JoNil/elf2uf2-rs/blob/master/src/uf2.rs
UF2_MAGIC_START0 = 0x0A324655
UF2_MAGIC_START1 = 0x9E5D5157
@@ -15,42 +12,110 @@ UF2_FLAG_MD5_PRESENT = 0x00004000
RP2040_FAMILY_ID = 0xe48bff56
-offset = 0
-while offset < len(contents):
- frame = contents[offset:offset+512]
- magic_start0, magic_start1, flags, target_addr, payload_size, block_no, num_blocks, file_size \
- = struct.unpack_from("<IIIIIIII", frame, offset=0)
- magic_end, = struct.unpack_from("<I", frame, offset=512-4)
- if magic_start0 != UF2_MAGIC_START0 or magic_start1 != UF2_MAGIC_START1 or magic_end != UF2_MAGIC_END:
- print("not UF2 at offset %r" % offset)
- break
-
- flags_str = []
- flags2 = flags
- if (flags & UF2_FLAG_NOT_MAIN_FLASH) != 0:
- flags2 &= ~UF2_FLAG_NOT_MAIN_FLASH
- flags_str.append("not_main")
- if (flags & UF2_FLAG_FILE_CONTAINER) != 0:
- flags2 &= ~UF2_FLAG_FILE_CONTAINER
- flags_str.append("file")
- if (flags & UF2_FLAG_FAMILY_ID_PRESENT) != 0:
- flags2 &= ~UF2_FLAG_FAMILY_ID_PRESENT
- flags_str.append("family")
- if (flags & UF2_FLAG_MD5_PRESENT) != 0:
- flags2 &= ~UF2_FLAG_MD5_PRESENT
- flags_str.append("md5")
- if flags2 != 0:
- flags_str.append("0x08x" % flags2)
-
- file_size_or_family = ""
- if (flags & UF2_FLAG_FAMILY_ID_PRESENT) != 0:
- if file_size == RP2040_FAMILY_ID:
- file_size_or_family = "rp2040"
+class UF2Block(object):
+ __slots__ = ("bytes", "offset")
+
+ def __init__(self, bytes, offset=0):
+ self.bytes = bytes
+ self.offset = offset
+
+ @property
+ def magic_start0(self):
+ return struct.unpack_from("<I", self.bytes, 0)[0]
+ @property
+ def magic_start1(self):
+ return struct.unpack_from("<I", self.bytes, 4)[0]
+ @property
+ def flags_int(self):
+ return struct.unpack_from("<I", self.bytes, 8)[0]
+ @property
+ def target_addr(self):
+ return struct.unpack_from("<I", self.bytes, 12)[0]
+ @property
+ def payload_size(self):
+ return struct.unpack_from("<I", self.bytes, 16)[0]
+ @property
+ def block_no(self):
+ return struct.unpack_from("<I", self.bytes, 20)[0]
+ @property
+ def num_blocks(self):
+ return struct.unpack_from("<I", self.bytes, 24)[0]
+ @property
+ def file_size_or_family(self):
+ return struct.unpack_from("<I", self.bytes, 28)[0]
+ @property
+ def magic_end(self):
+ return struct.unpack_from("<I", self.bytes, 512-4)[0]
+ @property
+ def data(self):
+ return self.bytes[32:512-4]
+ @property
+ def payload(self):
+ return self.data[0..self.payload_size]
+
+ @property
+ def is_uf2(self):
+ return self.magic_start0 == UF2_MAGIC_START0 and self.magic_start1 == UF2_MAGIC_START1 and self.magic_end == UF2_MAGIC_END
+
+ @property
+ def not_valid_reason(self):
+ if not self.is_uf2:
+ return "no UF2 magic"
+ if self.block_no >= self.num_blocks or self.num_blocks == 0:
+ return "invalid block number"
+ if (self.flags_int & ~(UF2_FLAG_NOT_MAIN_FLASH | UF2_FLAG_FILE_CONTAINER | UF2_FLAG_FAMILY_ID_PRESENT | UF2_FLAG_MD5_PRESENT)) != 0:
+ return "unsupported flags"
+ return True
+ @property
+ def valid(self):
+ return self.not_valid_reason == True
+
+ @property
+ def flags_str(self):
+ flags_str = []
+ flags2 = self.flags_int
+ if (flags2 & UF2_FLAG_NOT_MAIN_FLASH) != 0:
+ flags2 &= ~UF2_FLAG_NOT_MAIN_FLASH
+ flags_str.append("not_main")
+ if (flags2 & UF2_FLAG_FILE_CONTAINER) != 0:
+ flags2 &= ~UF2_FLAG_FILE_CONTAINER
+ flags_str.append("file")
+ if (flags2 & UF2_FLAG_FAMILY_ID_PRESENT) != 0:
+ flags2 &= ~UF2_FLAG_FAMILY_ID_PRESENT
+ flags_str.append("family")
+ if (flags2 & UF2_FLAG_MD5_PRESENT) != 0:
+ flags2 &= ~UF2_FLAG_MD5_PRESENT
+ flags_str.append("md5")
+ if flags2 != 0:
+ flags_str.append("0x08x" % flags2)
+ return tuple(flags_str)
+
+ @property
+ def family(self):
+ if (self.flags_int & UF2_FLAG_FAMILY_ID_PRESENT) != 0:
+ if self.file_size_or_family == RP2040_FAMILY_ID:
+ return "rp2040"
+ else:
+ return "0x%08x" % self.file_size_or_family
else:
- file_size_or_family = "family 0x%08x" % file_size
- else:
- file_size_or_family = "file size %s" % file_size
+ return "none"
+
+def read_blocks(filename):
+ with open(filename, "rb") as f:
+ contents = f.read()
+
+ blocks = []
+ offset = 0
+ while offset < len(contents):
+ block = contents[offset:offset+512]
+ block = UF2Block(block, offset=offset)
+ if not block.valid:
+ raise Exception("no valid UF2 at offset %r: %r" % (offset, block.not_valid_reason))
+ blocks.append(block)
+ offset += 512
+ return blocks
- print("block %-6s: target 0x%08x, size %3s, block %3s/%3s, %s, %s" % (
- offset, target_addr, payload_size, block_no, num_blocks, file_size_or_family, " ".join(flags_str)))
- offset += 512
+if __name__ == "__main__":
+ for block in read_blocks(sys.argv[1]):
+ print("block %-6s: target 0x%08x, size %3s, block %3s/%3s, %s, %s" % (
+ block.offset, block.target_addr, block.payload_size, block.block_no, block.num_blocks, block.family, " ".join(block.flags_str)))
--
GitLab