avr/bootloader
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

host: one fact per line, progress bars, --verbose

Browse Source 
--info prints the decoded info block field by field and --fuses each
byte on its own line plus the BOOTSZ/BOOTRST meaning on boot-sectioned
megas. Transfers that take wire time draw a transient progress bar on
stderr when it is a tty — logs, pipes and the tests see only the
summary lines. -v/--verbose narrates decisions: knock counts, the
programming plan, update state handling and per-phase page counts.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
BlackMark
2026-07-21 23:42:57 +02:00
parent a977e507f3
commit 9a528f9411
1 changed files with 161 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

195
pureboot/pureboot.py
View File

@@ -38,11 +38,57 @@ PROMPT = b"+"
PROTOCOL_VERSION = 1
SLOT = 512 # the loader slot on byte-addressed chips; word-addressed ones (>64 KiB) use 1 KiB — their own smallest boot sector
VERBOSE = False
def verbose(message):
"""Detail printed only under --verbose: decisions and derived facts, not
per-byte chatter — the progress bar carries the bulk transfers."""
if VERBOSE:
print(f" {message}")
class Error(Exception):
pass
class Progress:
"""A transient in-place bar on stderr for the operations that take wire
time. Drawn only when stderr is a tty — logs, pipes and the test harness
see nothing — and erased once done; the summary line each operation
prints afterwards is the persistent record. A zero total (or no label)
disables it, so callers can pass one through unconditionally."""
def __init__(self, label, total, unit="pages"):
self.label, self.total, self.unit = label, total, unit
self.done = 0
self.width = 0
self.live = bool(label) and total > 0 and sys.stderr.isatty()
self._draw()
def __enter__(self):
return self
def __exit__(self, *exc):
if self.live:
sys.stderr.write("\r" + " " * self.width + "\r")
sys.stderr.flush()
def step(self, n=1):
self.done += n
self._draw()
def _draw(self):
if not self.live:
return
bar = 24 * self.done // self.total
line = (f"{self.label:<16} [{'#' * bar}{'-' * (24 - bar)}] "
f"{100 * self.done // self.total:3d}% {self.done}/{self.total} {self.unit}")
self.width = max(self.width, len(line))
sys.stderr.write("\r" + line)
sys.stderr.flush()
# ---------------------------------------------------------------- serial ---
@@ -295,6 +341,23 @@ class Info:
f"EEPROM {self.eeprom_size} B, {vector}"
)
def lines(self):
"""The info block as one fact per line — what --info prints."""
if self.patch_vector:
hand_over = f"host-patched reset vector, trampoline at {self.base - 2:#06x}"
else:
hand_over = "hardware boot section, jump to word 0"
return (
f"signature {' '.join(f'{b:02x}' for b in self.signature)}",
f"flash {self.flash_size} B, {self.page} B pages"
+ (", word-addressed wire" if self.word_flash else ""),
f"application 0x0000..{self.base - 1:#06x} ({self.base} B)",
f"loader {self.base:#06x} ({self.slot} B slot)",
f"staging {self.stage:#06x}",
f"EEPROM {self.eeprom_size} B",
f"hand-over {hand_over}",
)
class Loader:
"""A pureboot session. Between commands the loader has prompted `+` and
@@ -312,8 +375,10 @@ class Loader:
absorbs whatever they produced."""
self.port.flush_input()
deadline = time.monotonic() + wait
knocks = 0
while True:
self.port.write(b"pb")
knocks += 1
if PROMPT in self.port.read_available(0.4):
break
if time.monotonic() > deadline:
@@ -323,6 +388,7 @@ class Loader:
self.port.write(b"b")
self.info = Info(self.port.read_exact(12, 2.0))
self._expect_prompt()
verbose(f"loader answered knock {knocks}; info block read")
return self.info
def _expect_prompt(self, timeout=2.0):
@@ -373,7 +439,7 @@ class Loader:
head = bytes((ord("W"), wire & 0xFF, wire >> 8))
self._command(head + data, 0, 2.0)
def write_eeprom(self, address, data):
def write_eeprom(self, address, data, progress=None):
offset = 0
while offset < len(data):
chunk = data[offset : offset + 256]
@@ -382,6 +448,8 @@ class Loader:
for byte in chunk:
self.port.write(bytes((byte,)))
self._expect_prompt() # per-byte ack: the write has begun
if progress:
progress.step()
self._expect_prompt() # the next command prompt
address += len(chunk)
offset += len(chunk)
@@ -487,6 +555,8 @@ def plan_flash(image, info):
final += bytearray([0xFF] * (trampoline_page + page - len(final)))
jump = rjmp_to(trampoline_word, entry, flash_words)
final[info.base - 2], final[info.base - 1] = jump & 0xFF, jump >> 8
verbose(f"vector surgery: word 0 -> loader {info.base:#06x}, "
f"trampoline {info.base - 2:#06x} -> entry word {entry:#06x}")
pages = {a: bytes(final[a : a + page]) for a in range(0, len(final), page)}
return pages
@@ -681,17 +751,22 @@ class UpdateState:
os.unlink(self.path)
def write_differing(loader, base, content, order=None):
def write_differing(loader, base, content, order=None, label=None):
"""Program the pages of `content` at `base` that differ from flash —
idempotent, so a resumed phase redoes only what an interruption left."""
idempotent, so a resumed phase redoes only what an interruption left.
A label puts the compare-and-program loop on the progress bar."""
page = loader.info.page
offsets = order if order is not None else range(0, len(content), page)
offsets = list(order) if order is not None else list(range(0, len(content), page))
written = 0
for offset in offsets:
want = content[offset : offset + page]
if loader.read_flash(base + offset, page) != want:
loader.write_page(base + offset, want)
written += 1
with Progress(label, len(offsets)) as bar:
for offset in offsets:
want = content[offset : offset + page]
if loader.read_flash(base + offset, page) != want:
loader.write_page(base + offset, want)
written += 1
bar.step()
if label:
verbose(f"{label}: {written} of {len(offsets)} pages differed")
for at in range(0, len(content), 256):
if loader.read_flash(base + at, min(256, len(content) - at)) != content[at : at + 256]:
raise Error(f"verify failed at {base + at:#06x} after programming")
@@ -723,6 +798,10 @@ def op_update_loader(loader, wait, path, state_path, fuse_bytes):
page = info.page
state = UpdateState(state_path)
if os.path.exists(state_path):
verbose(f"resuming the update recorded in {state_path}")
else:
verbose(f"saving the staging slot to {state_path}")
state.load_or_save(loader)
# Install the staging copy — unless a loader already sits whole in the
@@ -741,7 +820,7 @@ def op_update_loader(loader, wait, path, state_path, fuse_bytes):
current = loader.read_flash(info.stage, info.slot)
staged_loader = image_info(current[:268])
if staged_loader is not None and staged_loader.raw == info.raw and current == state.staging:
print(f"staging slot already holds a loader — left in place")
print("staging slot already holds a loader — left in place")
else:
# On a chip whose staging slot starts at address 0 (the 1 KB
# tiny13s), its first page carries the reset vector: written last,
@@ -750,30 +829,34 @@ def op_update_loader(loader, wait, path, state_path, fuse_bytes):
order = list(range(0, info.slot, page))
if info.stage == 0:
order = order[1:] + [0]
if write_differing(loader, info.stage, staged, order):
if write_differing(loader, info.stage, staged, order, label="staging copy"):
print(f"staging copy installed at {info.stage:#06x}")
# Enter it and let it rewrite the resident slot. Where a patched reset
# vector routes through the resident (a tiny with the staging slot away
# from page 0), word 0 is re-aimed at the staging copy around the
# rewrite, so a power failure mid-rewrite still resets into a loader.
verbose(f"entering the staging copy at {info.stage:#06x}")
loader.enter_copy(info.stage, wait)
redirect = info.patch_vector and info.stage != 0
if redirect:
verbose("word 0 re-aimed at the staging copy for the rewrite")
patch_word0(loader, state.page0, info.stage)
if write_differing(loader, info.base, resident):
if write_differing(loader, info.base, resident, label="resident"):
print(f"resident loader rewritten at {info.base:#06x}")
# Enter the new resident and put the staging region back: page 0 first
# where it lives in that region (word 0 then points at the new resident
# for the rest of the restore), the saved trampoline with the rest.
verbose(f"entering the new resident at {info.base:#06x}")
loader.enter_copy(info.base, wait)
if redirect:
verbose("word 0 restored")
write_differing(loader, 0, state.page0)
order = list(range(0, info.slot, page))
if info.stage == 0:
order = [0] + order[1:]
write_differing(loader, info.stage, state.staging, order)
write_differing(loader, info.stage, state.staging, order, label="staging restore")
state.discard()
print(f"loader updated: {len(image)} B at {info.base:#06x}, staging region restored")
@@ -808,39 +891,49 @@ def op_erase_flash(loader):
is erased and the reset walk reaches the loader anyway."""
blank = bytes([0xFF] * loader.info.page)
addresses = range(0, loader.info.base, loader.info.page)
for address in reversed(addresses) if loader.info.patch_vector else addresses:
loader.write_page(address, blank)
with Progress("erase", len(addresses)) as bar:
for address in reversed(addresses) if loader.info.patch_vector else addresses:
loader.write_page(address, blank)
bar.step()
print(f"erase: {loader.info.base // loader.info.page} pages")
def op_erase_eeprom(loader):
loader.write_eeprom(0, bytes([0xFF] * loader.info.eeprom_size))
with Progress("erase EEPROM", loader.info.eeprom_size, "B") as bar:
loader.write_eeprom(0, bytes([0xFF] * loader.info.eeprom_size), progress=bar)
print(f"erase: {loader.info.eeprom_size} B of EEPROM")
def op_flash(loader, path, erase, verify, fuse_bytes=None, force=False):
image = load_image(path)
verbose(f"{path}: {len(image)} B image")
pages = plan_flash(image, loader.info)
check_walk_region(pages, loader.info, fuse_bytes, force)
if erase:
op_erase_flash(loader)
order = covered(pages, loader.info, skip_blank=erase)
for address in order:
loader.write_page(address, pages[address])
if len(order) != len(pages):
verbose(f"{len(pages) - len(order)} blank pages skipped (erased flash underneath)")
with Progress("flash", len(order)) as bar:
for address in order:
loader.write_page(address, pages[address])
bar.step()
print(f"flash: {path}: {len(order)} pages")
if verify:
verify_pages(loader, pages)
def verify_pages(loader, pages):
for address in sorted(pages):
got = loader.read_flash(address, loader.info.page)
if got != pages[address]:
first = next(i for i in range(len(got)) if got[i] != pages[address][i])
raise Error(
f"verify failed at {address + first:#06x}: "
f"wrote {pages[address][first]:02x}, read {got[first]:02x}"
)
with Progress("verify", len(pages)) as bar:
for address in sorted(pages):
got = loader.read_flash(address, loader.info.page)
if got != pages[address]:
first = next(i for i in range(len(got)) if got[i] != pages[address][i])
raise Error(
f"verify failed at {address + first:#06x}: "
f"wrote {pages[address][first]:02x}, read {got[first]:02x}"
)
bar.step()
print(f"verify: {len(pages)} pages ok")
@@ -848,8 +941,19 @@ def op_verify_flash(loader, path):
verify_pages(loader, plan_flash(load_image(path), loader.info))
def read_progress(reader, total, label):
"""A bulk read in 256-byte wire chunks under a progress bar."""
data = b""
with Progress(label, total, "B") as bar:
while len(data) < total:
chunk = min(256, total - len(data))
data += reader(len(data), chunk)
bar.step(chunk)
return data
def op_read_flash(loader, path):
data = loader.read_flash(0, loader.info.base)
data = read_progress(loader.read_flash, loader.info.base, "read flash")
open(path, "wb").write(data)
print(f"read flash: {len(data)} B -> {path}")
@@ -860,10 +964,11 @@ def op_eeprom(loader, path, erase, verify):
raise Error(f"EEPROM image is {len(image)} B, device has {loader.info.eeprom_size}")
if erase:
op_erase_eeprom(loader)
loader.write_eeprom(0, image)
with Progress("eeprom", len(image), "B") as bar:
loader.write_eeprom(0, image, progress=bar)
print(f"eeprom: {path}: {len(image)} B")
if verify:
got = loader.read_eeprom(0, len(image))
got = read_progress(loader.read_eeprom, len(image), "verify EEPROM")
if got != image:
first = next(i for i in range(len(got)) if got[i] != image[i])
raise Error(f"verify failed at EEPROM {first:#06x}: wrote {image[first]:02x}, read {got[first]:02x}")
@@ -872,7 +977,7 @@ def op_eeprom(loader, path, erase, verify):
def op_verify_eeprom(loader, path):
image = load_image(path)
got = loader.read_eeprom(0, len(image))
got = read_progress(loader.read_eeprom, len(image), "verify EEPROM")
if got != image:
first = next(i for i in range(len(got)) if got[i] != image[i])
raise Error(f"verify failed at EEPROM {first:#06x}: expected {image[first]:02x}, read {got[first]:02x}")
@@ -880,15 +985,30 @@ def op_verify_eeprom(loader, path):
def op_read_eeprom(loader, path):
data = loader.read_eeprom(0, loader.info.eeprom_size)
data = read_progress(loader.read_eeprom, loader.info.eeprom_size, "read EEPROM")
open(path, "wb").write(data)
print(f"read EEPROM: {len(data)} B -> {path}")
def op_fuses(loader):
low, lock, extended, high = loader.read_fuses()
print(f"fuses: low {low:02x} high {high:02x} extended {extended:02x} lock {lock:02x}")
return bytes((low, lock, extended, high))
print("fuses:")
print(f" low 0x{low:02x}")
print(f" high 0x{high:02x}")
print(f" extended 0x{extended:02x}")
print(f" lock 0x{lock:02x}")
fuse_bytes = bytes((low, lock, extended, high))
# On a boot-sectioned mega the BOOTSZ/BOOTRST decode is the fuse fact the
# loader's whole deployment hangs on — say it in words.
if not loader.info.patch_vector:
try:
bootrst, bls_start = mega_boot(loader.info, fuse_bytes)
reset = "reset enters it" if bootrst else "reset boots the application"
print(f" boot section at {bls_start:#06x} ({loader.info.flash_size - bls_start} B), "
f"BOOTRST {'programmed' if bootrst else 'unprogrammed'} — {reset}")
except Error:
pass # unknown signature: the raw bytes above still stand
return fuse_bytes
# -------------------------------------------------------------------- cli ---
@@ -918,7 +1038,11 @@ def main():
parser.add_argument("--verify-eeprom", metavar="FILE", help="compare EEPROM against an image")
parser.add_argument("--force", action="store_true", help="override refusable safety checks")
parser.add_argument("--stay", action="store_true", help="leave the loader in its session")
parser.add_argument("-v", "--verbose", action="store_true",
help="print decisions and derived facts as operations run")
args = parser.parse_args()
global VERBOSE
VERBOSE = args.verbose
if args.update_loader and (args.flash or args.erase_flash):
parser.error("--update-loader does not combine with application flash operations")
@@ -931,11 +1055,14 @@ def main():
parser.error("--assume-fuses takes 8 hex digits: low,lock,extended,high")
port = Port(args.port, args.baud)
verbose(f"{args.port}: {args.baud} Bd 8N1, DTR/RTS asserted")
try:
loader = Loader(port)
info = loader.connect(args.wait)
if args.info:
print(f"device: {info.describe()}")
print("device:")
for line in info.lines():
print(f" {line}")
fuse_bytes = fuse_override
if args.fuses or (args.update_loader and not info.patch_vector and fuse_bytes is None):
read = op_fuses(loader)