#!/usr/bin/env python3 """pureboot host tool — the smart half of the protocol (README.md). The device exposes primitives; everything composite is here: HEX/raw images, programming with repairing read-back verification, the reset-vector surgery the boot-section-less chips need, and the self-update that stages the loader one slot lower and lets it rewrite the resident. Standard library only. The port is termios on POSIX and the Win32 serial API through ctypes on Windows, so any tty or COM port works. """ import argparse import json import os import sys import time if os.name == "nt": import ctypes from ctypes import wintypes else: import select import termios PROMPT = b"+" VERSION = 5 # this tool's own version — free to drift from a loader's # The loader versions this tool speaks. A pureboot version implies its wire # protocol, which carries no number of its own, so this window is where that # map lives: every version so far speaks the same protocol, and one that # changes it becomes the new floor here. OLDEST_LOADER = 1 NEWEST_LOADER = 5 SLOT = 512 # the loader slot, on every chip RETRIES = 3 # rewrites of a page that reads back wrong, before the run stops # pureboot 5 replaced the four per-memory commands with one pair: 'G' reads and # 'g' writes, each taking a selector byte, a 16-bit address and a count, over # the spaces below. The loader carries one transfer loop instead of four bodies # — which is what buys the data space and the host-issued SPM operations. UNIFIED_LOADER = 5 SP_FLASH, SP_EEPROM, SP_RAM, SP_FUSE, SP_SPM = 0, 1, 2, 3, 4 # A selector's high nibble is the flash bank — the address bits above the 16-bit # wire address — so a transfer names a byte address within one 64 KiB bank and # no command has to speak word addresses. No single transfer may cross a bank # boundary; the host chunks to keep that true. def selector(space, address): return space | ((address >> 16) << 4) # The SPM operations pureboot 5 leaves to the host: a write to SP_SPM hands its # byte to SPMCSR and fires the instruction at the selected flash address. Every # part pureboot targets agrees on these encodings. SPM_ERASE, SPM_WRITE, SPM_RWWSRE = 0x03, 0x05, 0x11 # Calibration byte for an autobaud loader: 0xC0 is a start bit plus six zero # data bits — one low pulse of seven bit-times, which the loader times into its # per-bit unit. Sent at whatever baud the host chose; the loader locks to it. CALIBRATE = 0xC0 # pureboot 5 answers 'b' with its version and the chip signature; the host # derives the rest of the geometry from the signature rather than reading a # table off the device. flash, page, eeprom, patch-vector per distinct # signature, over every chip pureboot targets (the loader computes the same # from its chip database at build time). Die revisions that share a signature # share this row, as they share the silicon. CHIP_GEOMETRY = { # signature : (flash, page, eeprom, patch_vector) (0x1E, 0x90, 0x07): (1024, 32, 64, True), # ATtiny13/13A (0x1E, 0x91, 0x08): (2048, 32, 128, True), # ATtiny25 (0x1E, 0x92, 0x06): (4096, 64, 256, True), # ATtiny45 (0x1E, 0x93, 0x0B): (8192, 64, 512, True), # ATtiny85 (0x1E, 0x92, 0x05): (4096, 64, 256, True), # ATmega48/48A (0x1E, 0x92, 0x0A): (4096, 64, 256, True), # ATmega48P/48PA (0x1E, 0x93, 0x07): (8192, 64, 512, False), # ATmega8/8A (0x1E, 0x93, 0x0A): (8192, 64, 512, False), # ATmega88/88A (0x1E, 0x93, 0x0F): (8192, 64, 512, False), # ATmega88P/88PA (0x1E, 0x94, 0x03): (16384, 128, 512, False), # ATmega16/16A (0x1E, 0x94, 0x06): (16384, 128, 512, False), # ATmega168/168A (0x1E, 0x94, 0x0B): (16384, 128, 512, False), # ATmega168P/168PA (0x1E, 0x94, 0x0A): (16384, 128, 512, False), # ATmega164P/164PA (0x1E, 0x94, 0x0F): (16384, 128, 512, False), # ATmega164A (0x1E, 0x95, 0x02): (32768, 128, 1024, False), # ATmega32/32A (0x1E, 0x95, 0x0F): (32768, 128, 1024, False), # ATmega328P (0x1E, 0x95, 0x14): (32768, 128, 1024, False), # ATmega328 (0x1E, 0x95, 0x08): (32768, 128, 1024, False), # ATmega324P (0x1E, 0x95, 0x11): (32768, 128, 1024, False), # ATmega324PA (0x1E, 0x95, 0x15): (32768, 128, 1024, False), # ATmega324A (0x1E, 0x96, 0x09): (65536, 256, 2048, False), # ATmega644/644A (0x1E, 0x96, 0x0A): (65536, 256, 2048, False), # ATmega644P/644PA (0x1E, 0x97, 0x05): (131072, 256, 4096, False),# ATmega1284P (0x1E, 0x97, 0x06): (131072, 256, 4096, False),# ATmega1284 } VERBOSE = False def verbose(message): if VERBOSE: print(f" {message}") class Error(Exception): pass class Progress: """A transient bar on stderr, drawn only for a tty and erased when done — logs and pipes see only the summary line each operation prints. No label or a zero total disables it, so callers can pass one 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 --- class PosixPort: """A raw serial port with deadline-based reads, over termios.""" def __init__(self, path, baud): self.fd = os.open(path, os.O_RDWR | os.O_NOCTTY) attrs = termios.tcgetattr(self.fd) attrs[0] = 0 # iflag attrs[1] = 0 # oflag attrs[2] = termios.CREAD | termios.CLOCAL | termios.CS8 # cflag attrs[3] = 0 # lflag try: speed = getattr(termios, f"B{baud}") except AttributeError: raise Error(f"unsupported baud rate {baud}") from None attrs[4] = attrs[5] = speed attrs[6][termios.VMIN] = 0 attrs[6][termios.VTIME] = 0 termios.tcsetattr(self.fd, termios.TCSANOW, attrs) def close(self): os.close(self.fd) def write(self, data): os.write(self.fd, data) def flush_input(self): termios.tcflush(self.fd, termios.TCIFLUSH) def read_available(self, wait): """Everything that arrives within `wait` seconds of quiet start.""" ready, _, _ = select.select([self.fd], [], [], wait) return os.read(self.fd, 4096) if ready else b"" def read_exact(self, count, timeout): data = b"" deadline = time.monotonic() + timeout while len(data) < count: remaining = deadline - time.monotonic() if remaining <= 0: raise Error(f"timeout: got {len(data)} of {count} bytes") ready, _, _ = select.select([self.fd], [], [], remaining) if ready: data += os.read(self.fd, count - len(data)) return data if os.name == "nt": # The same port, over the Win32 serial API — kernel32 through ctypes, so # the tool stays standard-library only. Timeouts live in the driver # (COMMTIMEOUTS) rather than in a readiness call: Windows has no select() # for a COM handle, so each read asks the driver for its own deadline. _GENERIC_READ, _GENERIC_WRITE = 0x80000000, 0x40000000 _OPEN_EXISTING, _PURGE_RXCLEAR = 3, 0x0008 _INVALID_HANDLE = wintypes.HANDLE(-1).value # A gap this long ends a read_available(): longer than the coalescing a # USB-serial adapter's latency timer imposes (16 ms on FTDI parts), so a # burst is not split, short enough to stay responsive. _GAP_MS = 30 class _DCB(ctypes.Structure): _fields_ = [ ("DCBlength", wintypes.DWORD), ("BaudRate", wintypes.DWORD), ("fBits", wintypes.DWORD), # the packed flag bitfield, set below ("wReserved", wintypes.WORD), ("XonLim", wintypes.WORD), ("XoffLim", wintypes.WORD), ("ByteSize", wintypes.BYTE), ("Parity", wintypes.BYTE), ("StopBits", wintypes.BYTE), ("XonChar", ctypes.c_char), ("XoffChar", ctypes.c_char), ("ErrorChar", ctypes.c_char), ("EofChar", ctypes.c_char), ("EvtChar", ctypes.c_char), ("wReserved1", wintypes.WORD), ] class _COMMTIMEOUTS(ctypes.Structure): _fields_ = [ ("ReadIntervalTimeout", wintypes.DWORD), ("ReadTotalTimeoutMultiplier", wintypes.DWORD), ("ReadTotalTimeoutConstant", wintypes.DWORD), ("WriteTotalTimeoutMultiplier", wintypes.DWORD), ("WriteTotalTimeoutConstant", wintypes.DWORD), ] _k32 = ctypes.WinDLL("kernel32", use_last_error=True) _LPDWORD = ctypes.POINTER(wintypes.DWORD) # Declared, not inferred: a HANDLE is a pointer, and a defaulted int # return would truncate it on 64-bit. _k32.CreateFileW.restype = wintypes.HANDLE _k32.CreateFileW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE] _k32.ReadFile.argtypes = [wintypes.HANDLE, wintypes.LPVOID, wintypes.DWORD, _LPDWORD, wintypes.LPVOID] _k32.WriteFile.argtypes = [wintypes.HANDLE, wintypes.LPCVOID, wintypes.DWORD, _LPDWORD, wintypes.LPVOID] _k32.GetCommState.argtypes = [wintypes.HANDLE, ctypes.POINTER(_DCB)] _k32.SetCommState.argtypes = [wintypes.HANDLE, ctypes.POINTER(_DCB)] _k32.SetCommTimeouts.argtypes = [wintypes.HANDLE, ctypes.POINTER(_COMMTIMEOUTS)] _k32.PurgeComm.argtypes = [wintypes.HANDLE, wintypes.DWORD] _k32.CloseHandle.argtypes = [wintypes.HANDLE] def _fail(what): code = ctypes.get_last_error() raise Error(f"{what}: {ctypes.FormatError(code).strip()} (Windows error {code})") class WindowsPort: """A raw serial port with deadline-based reads, over Win32.""" def __init__(self, path, baud): # Win32 takes the rate as a plain integer, so unlike termios any # rate the hardware can divide down to is available — but a driver # may also accept one it cannot produce (an FT232R takes a baud of # 3, reports it back, and goes on using the previous divisor). # Only obvious nonsense is refusable; the rest is the driver's word. if baud < 50: raise Error(f"unsupported baud rate {baud}") # \\.\COM6: the device-namespace form. A bare COMn resolves only # for n < 10, and double-digit ports are routine on Windows. if path.lower().startswith("com") and path[3:].isdigit(): path = rf"\\.\{path}" self.handle = _k32.CreateFileW( path, _GENERIC_READ | _GENERIC_WRITE, 0, None, _OPEN_EXISTING, 0, None ) if self.handle == _INVALID_HANDLE: _fail(f"cannot open {path}") self.timeouts = None try: dcb = _DCB() dcb.DCBlength = ctypes.sizeof(_DCB) if not _k32.GetCommState(self.handle, ctypes.byref(dcb)): _fail(f"cannot read the state of {path}") dcb.BaudRate, dcb.ByteSize, dcb.Parity, dcb.StopBits = baud, 8, 0, 0 # 8N1 # fBinary, and DTR/RTS asserted (fDtrControl and fRtsControl, # two bits each, = _ENABLE); every other flag clear, so no # parity and no flow control. Raising both matches what opening # a POSIX tty does — including the reset pulse on the boards # that wire DTR to it. dcb.fBits = 0x1 | (1 << 4) | (1 << 12) if not _k32.SetCommState(self.handle, ctypes.byref(dcb)): _fail(f"cannot configure {path} for {baud} baud 8N1") # Arm them once here too: reads re-arm per call, but the write # timeout would otherwise stay at the driver's default — which # may be "wait forever" — until the first read. self._deadline(_GAP_MS, 1000) except Error: # An open port outlives the exception otherwise, and a COM # handle is exclusive: the next attempt would meet its own # leftover as "Access is denied". self.close() raise def close(self): _k32.CloseHandle(self.handle) def _deadline(self, interval, total): """Arm the driver's read timeouts: `interval` ms of quiet ends a read once bytes have arrived, `total` ms ends it regardless.""" if self.timeouts == (interval, total): return spec = _COMMTIMEOUTS() spec.ReadIntervalTimeout = interval spec.ReadTotalTimeoutConstant = total spec.WriteTotalTimeoutConstant = 5000 if not _k32.SetCommTimeouts(self.handle, ctypes.byref(spec)): _fail("cannot set the port timeouts") self.timeouts = (interval, total) def _read(self, count): buffer = ctypes.create_string_buffer(count) got = wintypes.DWORD() if not _k32.ReadFile(self.handle, buffer, count, ctypes.byref(got), None): _fail("read failed") return buffer.raw[: got.value] def write(self, data): written = wintypes.DWORD() if not _k32.WriteFile(self.handle, data, len(data), ctypes.byref(written), None): _fail("write failed") if written.value != len(data): raise Error(f"short write: {written.value} of {len(data)} bytes") def flush_input(self): if not _k32.PurgeComm(self.handle, _PURGE_RXCLEAR): _fail("cannot flush the input buffer") def read_available(self, wait): """Everything that arrives within `wait` seconds of quiet start.""" # A zero total means *no* timeout to the driver, so never round # down to it — the same trap on the deadline below. self._deadline(_GAP_MS, max(1, round(wait * 1000))) return self._read(4096) def read_exact(self, count, timeout): data = b"" deadline = time.monotonic() + timeout while len(data) < count: remaining = deadline - time.monotonic() if remaining <= 0: raise Error(f"timeout: got {len(data)} of {count} bytes") # No interval timeout here: only the count or the deadline # ends the read, so a gap mid-reply is simply waited out. self._deadline(0, max(1, round(remaining * 1000))) data += self._read(count - len(data)) return data Port = WindowsPort if os.name == "nt" else PosixPort # -------------------------------------------------------------- protocol --- class Info: """The 12-byte info block.""" @classmethod def from_identity(cls, raw): """pureboot 5's reply: the version and the chip signature. The rest of the geometry is looked up from the signature — the loader derived the same facts from its chip database at build time, so nothing is guessed, it is simply not sent. Reconstructs a block in the older layout, so every derived attribute below is shared with the loaders that do send one. The base is where application flash ends, which is a property of the chip and not of the copy answering: a loader staged one slot lower reports the same geometry the resident one does, exactly as the loaders that send a block do. Which slot a copy runs in matters only to its own write guard, which is the loader's business.""" if len(raw) != 4: raise Error(f"bad identity reply: {raw.hex()}") version, signature = raw[0], tuple(raw[1:4]) geometry = CHIP_GEOMETRY.get(signature) if geometry is None: sig = " ".join(f"{b:02x}" for b in signature) raise Error(f"unknown signature {sig} — this tool has no geometry for it") flash, page, eeprom, patch = geometry base = flash - SLOT word_flash = flash > 0x10000 wire_base = base // 2 if word_flash else base flags = (1 if patch else 0) | (2 if word_flash else 0) raw12 = bytes((ord("P"), ord("B"), version, *signature, page & 0xFF, wire_base & 0xFF, wire_base >> 8, eeprom & 0xFF, eeprom >> 8, flags)) return cls(raw12) def __init__(self, raw): if len(raw) != 12 or raw[0:2] != b"PB": raise Error(f"bad info block: {raw.hex()}") self.version = raw[2] if not OLDEST_LOADER <= self.version <= NEWEST_LOADER: raise Error( f"pureboot {self.version}: this tool (version {VERSION}) speaks pureboot " f"{OLDEST_LOADER}..{NEWEST_LOADER} — a newer loader needs a newer tool" ) self.raw = bytes(raw) self.signature = raw[3:6] self.page = raw[6] or 256 # the wire count convention: 0 means 256 self.patch_vector = bool(raw[11] & 1) # Bit 1: the flash runs past what one 16-bit address covers. Through # pureboot 4 that made flash addresses words on the wire; pureboot 5 # keeps them bytes and carries the bank in the selector instead. Every # address in this tool stays a byte address either way and converts at # the wire. self.word_flash = bool(raw[11] & 2) scale = 2 if self.word_flash else 1 self.base = (raw[7] | (raw[8] << 8)) * scale self.eeprom_size = raw[9] | (raw[10] << 8) self.flash_size = self.base + SLOT self.stage = self.base - SLOT # where a staging copy of the loader goes # The hand-over target as 'J' takes it: the trampoline below the # loader, or word 0 where BOOTRST re-vectors reset in hardware. self.app_entry_word = (self.base - 2) // 2 if self.patch_vector else 0 def describe(self): sig = " ".join(f"{b:02x}" for b in self.signature) vector = "host-patched reset vector" if self.patch_vector else "hardware boot section" return ( f"signature {sig}, page {self.page} B, " f"app flash {self.base} B (loader at {self.base:#06x}), " f"EEPROM {self.eeprom_size} B, {vector}" ) def lines(self): """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"version pureboot {self.version}", f"signature {' '.join(f'{b:02x}' for b in self.signature)}", f"flash {self.flash_size} B, {self.page} B pages" + (", past one 16-bit bank" if self.word_flash else ""), f"application 0x0000..{self.base - 1:#06x} ({self.base} B)", f"loader {self.base:#06x} ({SLOT} B slot)", f"staging {self.stage:#06x}", f"EEPROM {self.eeprom_size} B", f"hand-over {hand_over}", ) class Loader: """A session. Between commands the loader has prompted and awaits a command byte; every method restores that, except jump() — after which the target must be knocked afresh.""" def __init__(self, port): self.port = port self.info = None # Set once a session is established over an autobaud link, so a # re-entry after 'J' repeats the handshake that worked. self.autobaud = False def _read_identity(self): """The 'b' reply, in either of the two layouts a loader may send. pureboot 5 answers with its version and the signature; older loaders answer with a 12-byte block. The version byte cannot be mistaken for the older block's 'P', so four bytes are enough to tell them apart.""" head = self.port.read_exact(4, 2.0) if head[0:2] == b"PB": return Info(head + self.port.read_exact(8, 2.0)) return Info.from_identity(head) def _handshake(self, wait, knock, what): """One activation, retried until the loader answers or the window closes. The identity reply is what proves the loader is listening — a prompt byte alone does not, since one left over from a previous session can still be in the pipeline while the port opening resets the device into a fresh window, where a command without its knock is discarded. Each attempt is therefore the whole handshake. This also converges into an already-live session: the knock bytes are ignored there and the drain absorbs whatever they produced.""" deadline = time.monotonic() + wait knocks = 0 while True: self.port.flush_input() self.port.write(knock) knocks += 1 if PROMPT in self.port.read_available(0.4): # Settle: absorb a real loader's trailing bytes before asking # for the identity. Bounded by the deadline so a target that # never falls quiet — a board stuck in a reset loop, whose # garbage carries a stray prompt — cannot spin here forever. while self.port.read_available(0.3): if time.monotonic() > deadline: break self.port.write(b"b") try: # A version the tool cannot speak is the loader's own # answer, not a failed knock: Info reports it rather than # sending the tool round the loop again. self.info = self._read_identity() except Error as failed: if "pureboot" in str(failed): raise self.info = None if self.info is not None: self._expect_prompt() verbose(f"loader answered {what} {knocks}; identity read") return self.info if time.monotonic() > deadline: raise Error("no answer — reset the device within its activation window") def connect(self, wait): """Knock 'p' then 'b' and read the identity.""" return self._handshake(wait, b"pb", "knock") def connect_autobaud(self, wait): """The autobaud handshake. In place of the p+b knock the host sends the calibration pulse — one seven-bit-time low pulse at the host's chosen baud, which the loader times into its per-bit unit — then a single 'p' the loader decodes at the rate it just measured. A lost pulse, or a knock landing while the loader is mid-frame, simply fails to answer and leaves the measurement loop waiting for the next pulse, so the retry in _handshake covers it.""" self.autobaud = True return self._handshake(wait, bytes((CALIBRATE, ord("p"))), "calibration") def _expect_prompt(self, timeout=2.0): byte = self.port.read_exact(1, timeout) if byte != PROMPT: raise Error(f"expected prompt, got {byte.hex()}") def _command(self, tx, reply_len=0, timeout=2.0): self.port.write(tx) reply = self.port.read_exact(reply_len, timeout) if reply_len else b"" self._expect_prompt(timeout) return reply def _stream_read(self, command, address, count, address_scale=1): data = b"" while count: chunk = min(count, 256) wire = address // address_scale head = bytes((ord(command), wire & 0xFF, wire >> 8, chunk & 0xFF)) data += self._command(head, chunk, 5.0) address += chunk count -= chunk return data @property def unified(self): """pureboot 5 and later: one 'G'/'g' pair over selector-named spaces.""" return self.info is not None and self.info.version >= UNIFIED_LOADER def _read_space(self, space, address, count): """A run out of any space, chunked to 256 bytes and to bank bounds.""" data = b"" while count: chunk = min(count, 256, 0x10000 - (address & 0xFFFF)) head = bytes((ord("G"), selector(space, address), address & 0xFF, (address >> 8) & 0xFF, chunk & 0xFF)) data += self._command(head, chunk, 5.0) address += chunk count -= chunk return data def _write_space(self, space, address, data, progress=None): """A run into any space. Each byte is acked as its write begins — an EEPROM cell and an SPM operation both need that pacing, and the ack is what the loader sends in place of a completion status.""" offset = 0 while offset < len(data): chunk = data[offset : offset + min(256, 0x10000 - (address & 0xFFFF))] head = bytes((ord("g"), selector(space, address), address & 0xFF, (address >> 8) & 0xFF, len(chunk) & 0xFF)) self.port.write(head) for byte in chunk: self.port.write(bytes((byte,))) self._expect_prompt() if progress: progress.step() self._expect_prompt() # the next command prompt address += len(chunk) offset += len(chunk) def spm(self, operation, address): """One SPM operation at a flash address — the erase, write and RWW re-enable that pureboot 4 ran inside 'W' and pureboot 5 leaves here.""" self._write_space(SP_SPM, address, bytes((operation,))) def read_ram(self, address, count): """Data space: SRAM, and with it the register file and every I/O register, which share the address space on AVR. New in pureboot 5.""" return self._read_space(SP_RAM, address, count) def write_ram(self, address, data): self._write_space(SP_RAM, address, data) def read_flash(self, address, count): if self.unified: return self._read_space(SP_FLASH, address, count) if not self.info.word_flash: return self._stream_read("R", address, count) # Word-addressed wire: widen to even bounds and never let one read # cross a 64 KiB boundary (the device holds RAMPZ for a whole run). start = address & ~1 span = (address + count + 1 & ~1) - start data = b"" at = start remaining = span while remaining: chunk = min(remaining, 0x10000 - (at & 0xFFFF)) data += self._stream_read("R", at, chunk, address_scale=2) at += chunk remaining -= chunk return data[address - start : address - start + count] def read_eeprom(self, address, count): if self.unified: return self._read_space(SP_EEPROM, address, count) return self._stream_read("r", address, count) def write_page(self, address, data): assert len(data) == self.info.page and address % self.info.page == 0 if self.unified: # 'W' fills the page buffer and stops there; the erase and the write # are host-issued SPM operations. Only a chip with a boot section # has RWW to re-enable — on the others bit 4 of SPMCSR means # something else entirely, so it must not be sent. head = bytes((ord("W"), selector(SP_FLASH, address), address & 0xFF, (address >> 8) & 0xFF)) self._command(head + data, 0, 2.0) self.spm(SPM_ERASE, address) self.spm(SPM_WRITE, address) if not self.info.patch_vector: self.spm(SPM_RWWSRE, address) return wire = address // (2 if self.info.word_flash else 1) head = bytes((ord("W"), wire & 0xFF, wire >> 8)) self._command(head + data, 0, 2.0) def write_eeprom(self, address, data, progress=None): if self.unified: self._write_space(SP_EEPROM, address, data, progress) return offset = 0 while offset < len(data): chunk = data[offset : offset + 256] head = bytes((ord("w"), address & 0xFF, address >> 8, len(chunk) & 0xFF)) self.port.write(head) 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) def read_fuses(self): if self.unified: return self._read_space(SP_FUSE, 0, 4) return self._command(b"F", 4, 2.0) def jump(self, word_address): """The device acks, then execution continues at the word address.""" self.port.write(bytes((ord("J"), word_address & 0xFF, word_address >> 8))) self._expect_prompt() def enter_copy(self, byte_address, wait): """Jump into the loader copy at `byte_address` and knock it — a slot base is that copy's entry stub, so it can only land there.""" autobaud = self.autobaud self.jump(byte_address // 2) return self.connect_autobaud(wait) if autobaud else self.connect(wait) def run_application(self): self.jump(self.info.app_entry_word) # ---------------------------------------------------------------- images --- def load_image(path): """Raw binary, or Intel HEX by extension (.hex/.ihx/.ihex).""" data = open(path, "rb").read() if not path.lower().endswith((".hex", ".ihx", ".ihex")): if not data: raise Error(f"{path}: empty image") return data memory = {} for number, line in enumerate(data.decode("ascii", "replace").splitlines(), 1): line = line.strip() if not line: continue if not line.startswith(":"): raise Error(f"{path}:{number}: not an Intel HEX record") record = bytes.fromhex(line[1:]) if sum(record) & 0xFF: raise Error(f"{path}:{number}: checksum mismatch") count, address, kind = record[0], (record[1] << 8) | record[2], record[3] payload = record[4 : 4 + count] if kind == 0: for i, byte in enumerate(payload): memory[address + i] = byte elif kind == 1: break elif kind in (2, 4) and not any(payload): continue # a zero base extends nothing elif kind in (3, 5): continue # start address: irrelevant, reset is the entry else: raise Error(f"{path}:{number}: record type {kind} reaches beyond the 16-bit space") if not memory: raise Error(f"{path}: empty image") return bytes(memory.get(i, 0xFF) for i in range(max(memory) + 1)) # --------------------------------------------------------------- surgery --- def rjmp_target(word_address, opcode, flash_words): return (word_address + 1 + (opcode & 0x0FFF)) % flash_words def rjmp_to(word_address, destination, flash_words): return 0xC000 | ((destination - word_address - 1) % flash_words % 0x1000) def plan_flash(image, info): """The pages to program, as {page_address: bytes}, already carrying the reset-vector surgery where the chip needs it.""" page = info.page limit = info.base - (2 if info.patch_vector else 0) if len(image) > limit: raise Error(f"image is {len(image)} B, application flash ends at {limit}") final = bytearray(image) + bytearray([0xFF] * (-len(image) % page)) if info.patch_vector: flash_words = info.flash_size // 2 word0 = final[0] | (final[1] << 8) if word0 & 0xF000 != 0xC000: raise Error( "the image's reset vector is not an rjmp — pureboot's vector " "surgery cannot re-home it (crt-less entry at address 0?)" ) entry = rjmp_target(0, word0, flash_words) if entry >= info.base // 2: raise Error( "the image's reset vector already targets the loader — this " "is a read-back of a patched image; flash the original" ) trampoline_word = (info.base - 2) // 2 patch = rjmp_to(0, info.base // 2, flash_words) final[0], final[1] = patch & 0xFF, patch >> 8 trampoline_page = info.base - page if len(final) < trampoline_page + page: 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 def covered(pages, info, skip_blank): """Pages in programming order, optionally dropping all-0xff ones (sound only over erased flash, and never a load-bearing page). A patched vector puts page 0 first and the trampoline page second, so from the first write on a reset lands in the loader and its fall-through on the application entry — every interruption point recoverable. A hardware boot section re-vectors reset regardless; page 0 goes last there, which maximizes what an interrupted image retains.""" trampoline_page = info.base - info.page if info.patch_vector else None first = [0, trampoline_page] if info.patch_vector else [] rest = [a for a in sorted(pages) if a not in first] if skip_blank: rest = [a for a in rest if pages[a].count(0xFF) != len(pages[a])] order = [a for a in first if a in pages] + rest if not info.patch_vector: order = [a for a in order if a != 0] + ([0] if 0 in pages else []) return order # ----------------------------------------------------------------- fuses --- # Per-chip boot fuse geometry, keyed by the signature's family/part bytes: # which byte of the 'F' reply (low, lock, extended, high) carries BOOTSZ/ # BOOTRST, and the BOOTSZ->words ladder. A die revision shares its base # signature, so one row covers it. The m48s have no boot section and no # row — their info block says patch-vector and this table is never # consulted. Sources: Atmel-2486/2466/2503 (HIGH fuse), Atmel-2545/8271/ # DS40002065 (x8: EXTENDED, except the m328s' HIGH), Atmel-8272/8011/2593/ # 42719 (x4: HIGH). _LADDER_128 = {0b11: 128, 0b10: 256, 0b01: 512, 0b00: 1024} _LADDER_256 = {0b11: 256, 0b10: 512, 0b01: 1024, 0b00: 2048} _LADDER_512 = {0b11: 512, 0b10: 1024, 0b01: 2048, 0b00: 4096} BOOT_FUSE = { bytes((0x93, 0x07)): (3, _LADDER_128), # m8/8A bytes((0x94, 0x03)): (3, _LADDER_128), # m16/16A bytes((0x95, 0x02)): (3, _LADDER_256), # m32/32A bytes((0x93, 0x0A)): (2, _LADDER_128), # m88/88A bytes((0x93, 0x0F)): (2, _LADDER_128), # m88P/88PA bytes((0x94, 0x06)): (2, _LADDER_128), # m168/168A bytes((0x94, 0x0B)): (2, _LADDER_128), # m168P/168PA bytes((0x95, 0x14)): (3, _LADDER_256), # m328 bytes((0x95, 0x0F)): (3, _LADDER_256), # m328P bytes((0x94, 0x0F)): (3, _LADDER_128), # m164A bytes((0x94, 0x0A)): (3, _LADDER_128), # m164P/164PA bytes((0x95, 0x15)): (3, _LADDER_256), # m324A bytes((0x95, 0x08)): (3, _LADDER_256), # m324P bytes((0x95, 0x11)): (3, _LADDER_256), # m324PA bytes((0x96, 0x09)): (3, _LADDER_512), # m644/644A bytes((0x96, 0x0A)): (3, _LADDER_512), # m644P/644PA bytes((0x97, 0x06)): (3, _LADDER_512), # m1284 bytes((0x97, 0x05)): (3, _LADDER_512), # m1284P } def mega_boot(info, fuse_bytes): """Decode a mega's boot configuration from its fuses (the byte and the BOOTSZ ladder are per chip): BOOTSZ1:0 in bits 2:1 select the boot-section words, BOOTRST in bit 0 (programmed = 0) re-vectors reset to its start. Returns (bootrst_programmed, boot_section_start_byte).""" entry = BOOT_FUSE.get(bytes(info.signature[1:3])) if entry is None: raise Error(f"unknown mega signature {info.signature.hex()} — no boot fuse map") which, ladder = entry fuse = fuse_bytes[which] words = ladder[(fuse >> 1) & 0x03] return (fuse & 1) == 0, info.flash_size - words * 2 # ---------------------------------------------------------- loader update --- def image_info(image): """What a pureboot binary says about itself, or None. An update image is a bare slot: nothing about it names the chip it was built for, and installing a foreign one bricks the target — so every loader carries a stamp for this. Through pureboot 4 the stamp is the 12-byte info block the device also serves; pureboot 5 serves its identity from immediates and carries a 6-byte stamp (magic, version, signature) that only this exists for, from which the geometry is looked up exactly as it is for a live device. Searched once per known version, so the magic stays three selective bytes rather than two that code could carry by chance.""" for version in range(OLDEST_LOADER, NEWEST_LOADER + 1): at = image.find(b"PB" + bytes((version,))) if at < 0: continue if version >= UNIFIED_LOADER: if at <= len(image) - 6: return Info.from_identity(image[at + 2 : at + 6]) elif at <= len(image) - 12: return Info(image[at : at + 12]) return None def loader_image(path): """An update image as the slot's own content: a raw binary already is, while a HEX carries the blank below the loader's base, which is peeled off here. The base comes from the image's own block, not the device's, so a foreign image survives intact for the preflight to reject by name.""" image = load_image(path) embedded = image_info(image) if embedded and len(image) > embedded.base: image = image[embedded.base :] return image def staging_content(image, info): """The staging slot's content: the image, padding, and — where the hand-over jumps through the word below the resident — that word, which for a staging copy is its own last one. Composed as an rjmp to the resident, so an abandoned staging copy still falls through into a loader.""" budget = SLOT - 2 if info.patch_vector else SLOT if len(image) > budget: raise Error(f"loader image is {len(image)} B, the slot holds {budget}") content = bytearray(image) + bytearray([0xFF] * (SLOT - len(image))) if info.patch_vector: through = rjmp_to((info.base - 2) // 2, info.base // 2, info.flash_size // 2) content[SLOT - 2], content[SLOT - 1] = through & 0xFF, through >> 8 return bytes(content) def update_preflight(image, info, fuse_bytes): """Errors and warnings before any flash is touched. Returns warnings.""" embedded = image_info(image) if embedded is None: raise Error( "no pureboot info block in the update image — not a pureboot binary, " f"or a version this tool ({VERSION}) does not know" ) if embedded.raw[3:] != info.raw[3:]: raise Error( f"update image is for another target: it declares " f"[{embedded.describe()}], the device says [{info.describe()}]" ) warnings = [] if not info.patch_vector: if fuse_bytes is None: raise Error("a loader update on this chip needs its fuses — unreadable? pass --assume-fuses") bootrst, bls_start = mega_boot(info, fuse_bytes) if info.stage < bls_start: raise Error( f"cannot self-update: the staging slot {info.stage:#06x} lies below the " f"boot section ({bls_start:#06x}) where SPM is disabled " f"— a boot section of at least two slots ({2 * SLOT} B, BOOTSZ) is " f"required, and only an external programmer can change fuses" ) if not bootrst: warnings.append( "BOOTRST unprogrammed: reset boots the application throughout the update; " "an interruption is recovered by re-running this update" ) elif bls_start == info.stage: warnings.append( "BOOTRST targets the staging slot: brief unrecoverable windows exist while " "the staging copy itself is being installed or retired (page-write scale)" ) else: warnings.append( f"BOOTRST targets {bls_start:#06x}, inside application flash: reset reaches a " f"loader only across erased flash from there" ) return warnings class UpdateState: """The host-side memory of an update in flight: what the staging slot held (and page 0, where the update repoints it). Losing this file after the staging slot was overwritten loses those saved bytes — the update still completes, but the staging region can then only be restored by reflashing the application.""" def __init__(self, path): self.path = path self.data = None def load_or_save(self, loader): info = loader.info if os.path.exists(self.path): self.data = json.load(open(self.path)) if bytes.fromhex(self.data["signature"]) != info.signature or self.data["base"] != info.base: raise Error(f"{self.path} belongs to a different device — remove it to start over") return self.data = { "signature": info.signature.hex(), "base": info.base, "staging": loader.read_flash(info.stage, SLOT).hex(), "page0": loader.read_flash(0, info.page).hex() if info.patch_vector else "", } with open(self.path, "w") as f: json.dump(self.data, f) @property def staging(self): return bytes.fromhex(self.data["staging"]) @property def page0(self): return bytes.fromhex(self.data["page0"]) def discard(self): os.unlink(self.path) def write_differing(loader, base, content, order=None, label=None): """Program the pages of `content` at `base` that differ from flash, so a resumed phase redoes only what an interruption left.""" page = loader.info.page offsets = list(order) if order is not None else list(range(0, len(content), page)) written = 0 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") # The same bounded repair as verify_pages: here a page left wrong is a # half-written loader slot. for retry in range(RETRIES + 1): bad = [ offset for offset in range(0, len(content), page) if loader.read_flash(base + offset, len(content[offset : offset + page])) != content[offset : offset + page] ] if not bad: break if retry == RETRIES: raise Error( f"verify failed at {base + bad[0]:#06x} after programming " f"(still wrong after {RETRIES} retries)" ) for offset in bad: verbose(f"rewriting page {base + offset:#06x} (retry {retry + 1})") loader.write_page(base + offset, content[offset : offset + page]) written += 1 return written def patch_word0(loader, page0, target_base): """Re-aim word 0 at `target_base` — the resume insurance around rewriting a loader slot the reset path goes through.""" info = loader.info patched = bytearray(page0) word = rjmp_to(0, target_base // 2, info.flash_size // 2) patched[0], patched[1] = word & 0xFF, word >> 8 write_differing(loader, 0, bytes(patched)) return bytes(patched) def op_update_loader(loader, wait, path, state_path, fuse_bytes): """Replace the resident loader with `path`, using the loader as its own staging loader. Every phase is idempotent and keyed off the flash state, so a re-run resumes; the state file carries what the staging slot held.""" info = loader.info image = loader_image(path) for warning in update_preflight(image, info, fuse_bytes): print(f"note: {warning}") update = image_info(image) # the preflight proved it is there verbose(f"installing pureboot {update.version} over pureboot {info.version}") staged = staging_content(image, info) resident = bytes(image) + bytes([0xFF] * (SLOT - len(image))) 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) # A loader already sitting whole in the staging slot IS the staging copy: # rewriting it would only meet its own running-slot guard. Any pureboot # with the device's info block serves, since a staged copy only streams # pages. "Whole" needs both checks — the block where every image carries # it and matching byte for byte, and the slot unchanged since this update # began, so a half-written install takes the path below instead. current = loader.read_flash(info.stage, SLOT) # The whole slot is searched: a loader's stamp sits wherever its image put # it, which is the end of the code on pureboot 5 and the front of it # before that. staged_loader = image_info(current) if staged_loader is not None and staged_loader.raw == info.raw and current == state.staging: print("staging slot already holds a loader — left in place") else: # Where the staging slot starts at address 0 (the 1 KB tiny13s) its # first page carries the reset vector, so it goes last: until then a # reset still reaches the old resident. order = list(range(0, SLOT, page)) if info.stage == 0: order = order[1:] + [0] 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. Where a patched reset vector # routes through the resident, word 0 is re-aimed at the staging copy for # the rewrite, so a power loss 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, 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, SLOT, page)) if info.stage == 0: order = [0] + order[1:] write_differing(loader, info.stage, state.staging, order, label="staging restore") state.discard() print(f"loader updated: pureboot {update.version}, {len(image)} B at {info.base:#06x}, staging region restored") def check_walk_region(pages, info, fuse_bytes, force): """BOOTRST programmed below the loader means reset reaches it only by walking across erased flash; application data in that span would divert reset into itself. Needs the fuses (--fuses or --assume-fuses).""" if info.patch_vector or fuse_bytes is None: return bootrst, bls_start = mega_boot(info, fuse_bytes) if not bootrst or bls_start >= info.base: return overlap = [a for a in sorted(pages) if a >= bls_start and pages[a].count(0xFF) != len(pages[a])] if overlap and not force: raise Error( f"the image writes {overlap[0]:#06x}.. inside the reset walk region " f"[{bls_start:#06x}, {info.base:#06x}) (BOOTRST programmed): reset could no " f"longer reach the loader — --force to flash it anyway" ) # ------------------------------------------------------------ operations --- def op_erase_flash(loader): """0xff over the application area, descending where the reset vector is patched: page 0 goes last, so an interrupted erase still resets into the loader — and once it is gone, the erased walk reaches it anyway.""" blank = bytes([0xFF] * loader.info.page) addresses = range(0, loader.info.base, loader.info.page) 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): 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) 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, repair=True) def verify_pages(loader, pages, repair=False): """Read every page back and compare. With `repair`, a mismatch is rewritten and re-read up to RETRIES times first: a page filled over a dirty SPM buffer takes stale words, and the write that took them cleared the buffer, so one rewrite settles it. Anything still wrong is not that.""" repaired = 0 with Progress("verify", len(pages)) as bar: for address in sorted(pages): for retry in range(RETRIES + 1): got = loader.read_flash(address, loader.info.page) if got == pages[address]: break first = next(i for i in range(len(got)) if got[i] != pages[address][i]) detail = ( f"verify failed at {address + first:#06x}: " f"wrote {pages[address][first]:02x}, read {got[first]:02x}" ) if not repair: raise Error(detail) if retry == RETRIES: raise Error(f"{detail} (still wrong after {RETRIES} retries)") verbose(f"{detail} — rewriting page {address:#06x} (retry {retry + 1})") loader.write_page(address, pages[address]) repaired += 1 bar.step() note = f", {repaired} page rewrite(s)" if repaired else "" print(f"verify: {len(pages)} pages ok{note}") 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 = read_progress(loader.read_flash, loader.info.base, "read flash") open(path, "wb").write(data) print(f"read flash: {len(data)} B -> {path}") def op_eeprom(loader, path, erase, verify): image = load_image(path) if len(image) > loader.info.eeprom_size: raise Error(f"EEPROM image is {len(image)} B, device has {loader.info.eeprom_size}") if erase: op_erase_eeprom(loader) 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 = 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}") print(f"verify: {len(image)} B ok") def op_verify_eeprom(loader, path): image = load_image(path) 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}") print(f"verify: {len(image)} B of EEPROM ok") def op_read_eeprom(loader, path): 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 _require_unified(loader, what): if not loader.unified: raise Error(f"{what} needs pureboot {UNIFIED_LOADER} or later; this loader is {loader.info.version}") def _peek_spec(spec): """ADDR[:N] — addresses and counts in any Python integer base.""" address, _, count = spec.partition(":") return int(address, 0), int(count, 0) if count else 1 def op_peek(loader, spec): _require_unified(loader, "--peek") address, count = _peek_spec(spec) data = loader.read_ram(address, count) for offset in range(0, len(data), 16): row = data[offset : offset + 16] text = "".join(chr(b) if 0x20 <= b < 0x7F else "." for b in row) print(f"{address + offset:#06x} {row.hex(' '):<47} {text}") def op_poke(loader, spec): _require_unified(loader, "--poke") address, _, payload = spec.partition(":") if not payload: raise Error("--poke needs ADDR:HEX, for example 0x200:deadbeef") data = bytes.fromhex(payload.replace(" ", "")) loader.write_ram(int(address, 0), data) print(f"poke: {len(data)} B at {int(address, 0):#06x}") def op_fuses(loader): low, lock, extended, high = loader.read_fuses() 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 --- def main(): parser = argparse.ArgumentParser( description="pureboot host tool", epilog="operations run in the order listed above" ) parser.add_argument("--version", action="version", version=f"%(prog)s {VERSION} " f"(speaks pureboot {OLDEST_LOADER}..{NEWEST_LOADER})") parser.add_argument("--port", required=True, help="serial device: COM6, /dev/ttyUSB0, or a simavr pty") parser.add_argument("--baud", type=int, default=115200, help="115200 mega, 57600 tinies") parser.add_argument("--wait", type=float, default=30.0, help="seconds to keep knocking") parser.add_argument("--autobaud", action="store_true", help="drive an autobaud loader: send the 0xC0 calibration pulse and a single " "knock, and take geometry from the signature (no clock/baud baked in)") parser.add_argument("--info", action="store_true", help="print the device info block") parser.add_argument("--fuses", action="store_true", help="read the fuse and lock bytes") parser.add_argument("--update-loader", metavar="FILE", help="replace the loader with this pureboot binary") parser.add_argument("--state", metavar="FILE", help="update state file (default: FILE.pbstate)") parser.add_argument("--assume-fuses", metavar="HEX8", help="fuse bytes low,lock,ext,high as 8 hex digits " "(overrides reading them — e.g. under a simulator that cannot)") parser.add_argument("--erase-flash", action="store_true", help="0xff over the application flash") parser.add_argument("--flash", metavar="FILE", help="program an application (bin or ihex)") parser.add_argument("--no-verify", action="store_true", help="skip read-back after writes") parser.add_argument("--read-flash", metavar="FILE", help="dump the application flash") parser.add_argument("--verify-flash", metavar="FILE", help="compare flash against an image") parser.add_argument("--erase-eeprom", action="store_true", help="0xff over the EEPROM") parser.add_argument("--eeprom", metavar="FILE", help="program the EEPROM (bin or ihex)") parser.add_argument("--read-eeprom", metavar="FILE", help="dump the EEPROM") parser.add_argument("--verify-eeprom", metavar="FILE", help="compare EEPROM against an image") parser.add_argument("--peek", metavar="ADDR[:N]", help="read N bytes of data space (SRAM, registers, " "I/O) — pureboot 5 and later") parser.add_argument("--poke", metavar="ADDR:HEX", help="write hex bytes into data space — " "pureboot 5 and later") 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") fuse_override = None if args.assume_fuses: try: fuse_override = bytes.fromhex(args.assume_fuses) assert len(fuse_override) == 4 except (ValueError, AssertionError): 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_autobaud(args.wait) if args.autobaud else loader.connect(args.wait) if args.info: 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) if fuse_bytes is None: fuse_bytes = read if args.update_loader: state = args.state or args.update_loader + ".pbstate" op_update_loader(loader, args.wait, args.update_loader, state, fuse_bytes) if args.flash: op_flash(loader, args.flash, args.erase_flash, not args.no_verify, fuse_bytes, args.force) elif args.erase_flash: op_erase_flash(loader) if args.read_flash: op_read_flash(loader, args.read_flash) if args.verify_flash: op_verify_flash(loader, args.verify_flash) if args.eeprom: op_eeprom(loader, args.eeprom, args.erase_eeprom, not args.no_verify) elif args.erase_eeprom: op_erase_eeprom(loader) if args.read_eeprom: op_read_eeprom(loader, args.read_eeprom) if args.verify_eeprom: op_verify_eeprom(loader, args.verify_eeprom) if args.poke: op_poke(loader, args.poke) if args.peek: op_peek(loader, args.peek) if args.stay: print("loader stays in its session (reset to leave)") else: loader.run_application() print("application running") finally: port.close() if __name__ == "__main__": try: main() except Error as error: print(f"error: {error}", file=sys.stderr) sys.exit(1) except KeyboardInterrupt: sys.exit(130)