diff --git a/pureboot/pureboot.py b/pureboot/pureboot.py index 2693cbe..59835b6 100644 --- a/pureboot/pureboot.py +++ b/pureboot/pureboot.py @@ -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)