avr/bootloader
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity
Files
735ffab7dc3d51a64a49c53d87e7d9b03e71c2e1
bootloader/pureboot/pureboot.py
BlackMark 735ffab7dc fix: the four tiers stop describing features they do not have, and three gates start failing 
The reading pass over this repo found the tiers disagreeing with themselves,
and every fix here was measured.

**The turn-around guard is real code.** `tsb_asm` and `tsb_tricks` wrote
`for (std::uint8_t guard = 46; guard; --guard) ;` between taking the one-wire
line and the first UDR0 store, under a comment naming it a turn-around guard.
It has no side effect, so GCC deleted it - `sts UCSR0B` went straight to
`sts UDR0` - while the hand-written oracle spends six bytes on that wait and
libavr's own half-duplex spends them through `delay::cycles`. Two of four
tiers described a feature they did not have, which made the size gradient a
comparison between different loaders. `avr::delay::cycles<one bit time>()`
bottoms out in asm and cannot be deleted.

**The entry belongs to the library, and hand-rolling it was expensive.** Three
tiers wrote their own naked `.vectors` stub with `asm volatile("clr
__zero_reg__")` - which design.md fences to libavr and never a port, and which
`tsb_tricks` denied having in its own title line. `avr::startup::entry` also
keeps the body `noinline` for a stated reason: avr-ld must not shrink a
`.vectors` section, so a loader inlined into one forfeits call relaxation
everywhere. `tsb_pure` came out **836 -> 734** bytes for that alone.
`stack::hardware` - the reset value this part guarantees, with the write kept
where a part does not - saved another four, which is what let `tsb_asm` afford
the guard it had been four bytes short of. It fills its 512-byte section
exactly now, with the whole feature set.

**`tsb_pure` had no receive timeout.** Its `rx()` was `read_blocking()`, so a
silent host wedged the password gate and the command loop forever - the one
fix the oracle's own header lists by name, and one the other three tiers
implement. It is bounded now, and 0-on-silence falls through every compare as
theirs does.

Three gates could pass without proving anything. `sizes.py check-readme`
reported a match when every row's lookup missed; `check_size.cmake` used
`CMAKE_MATCH_1` without checking the match succeeded, which is the guard its
sibling `check_unit.cmake` has and it is the size gate; `check_pi.py` raised
IndexError instead of reporting a position-independence break that changed the
image's length. And `check.sh` spelled the 37-chip list a second time beside
make_presets.py, where a chip added to one and missed in the other is a
silently unbuilt chip - it reads the presets now, and produces the same 37 and
12.

tsbtest.py gains the scenario nothing covered: a wrong password byte must
neither activate the loader nor reach the emergency erase behind it. Red-green
on a tier with the refusal removed.

Smaller, all measured or checked: the signature is `hw::db.signature` in every
tier as the page size and EEPROM end beside it already were; `act_min` derives
from the clock; pureboot.py's `rjmp` helpers refuse a part past rjmp's
4096-word reach rather than silently folding an offset (unreachable today, the
ATtiny85 sits exactly on it); the host tool calls space 2 `data` as the wire
and the loader do; `.clangd` strips the fifth GCC-only flag the build passes;
pbrig's bitclock guard reads its own ladder; pbreloc's unexplained retry is
gone, the write being reliable on five runs without it; and the four tier
sizes live in oracle/README.md's table instead of four file headers and a
CMake comment.

`--poke` before `--peek` turned out to be right - pbtest.py round-trips a poke
through the peek behind it - so the parser order and README say so now.

Every chip green, the README size table matching every image.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-12 16:41:50 +02:00

1883 lines
87 KiB
Python
Raw Blame History

#!/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 array
import fcntl
import select
import termios
PROMPT = b"+"
# The refusal, from pureboot 9: the prompt inverted, so no single flipped bit
# turns "nothing happened" into "it did".
NAK = bytes((~PROMPT[0] & 0xFF,))
VERSION = 10 # this tool's own version - free to drift from a loader's
# The loader versions this tool can drive. A pureboot version implies its wire
# protocol, which carries no number of its own, so this window is where that
# map lives: the tool keeps a decoder for every generation in it (1-4 speak
# the per-memory commands, 5 the unified pair; 6 marks the OSCCAL-carrying
# builds and changes nothing on the wire; 8 the one-wire deployments, whose
# only host-side trace is the --one-wire echo discard; 9 seals every command
# and answers each seal before acting), and a version it has no decoder for
# moves the floor.
OLDEST_LOADER = 1
NEWEST_LOADER = 9
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.
# 6 marks the builds that may carry a baked OSCCAL trim, nothing on the wire;
# 7 gives 'J' a selector byte (older loaders take the bare address - jump()
# sends each form to the version that speaks it) and re-homes the autobaud
# unit into the GPIOR pair where the chip has one; 8 marks the builds whose
# deployment may be one-wire (hardware half-duplex, or a software link folded
# onto one pin) - nothing on the wire either, but a shared line makes the
# host read its own bytes back, which is what --one-wire consumes.
UNIFIED_LOADER = 5
SP_FLASH, SP_EEPROM, SP_DATA, SP_FUSE, SP_SPM = 0, 1, 2, 3, 4
# pureboot 9 replaces the command letters with bits and seals every command.
# The header is one shape for all of them - opcode, selector, address, count,
# seal - and the loader answers the seal *before* it acts: PROMPT accepts, NAK
# refuses and nothing happened. That verdict is what lets a refusal stay local
# to its own command: the payload of a fill or a write burst only goes out
# after the header has been accepted, so a rejected header never leaves the
# host pushing bytes into a loader that has gone back to reading commands.
#
# A transfer is the absence of the other three opcodes, and its direction is
# the low bit. Identify is bit 5 because both knock bytes carry it: 'b' has to
# still ask the version (the host cannot know which generation it is talking to
# until something answers), and 'p' has to stay harmless against a loader
# already in session - pureboot 9 reserves no invalid opcode, so a knock that
# meant nothing before would otherwise eat the five bytes behind it.
SEALED_LOADER = 9
OP_WRITE, OP_FILL, OP_JUMP, OP_IDENTIFY = 1, 4, 8, 0x20
# What a sealed header's fields must fold to. Non-zero, so a run of one
# repeated byte - a stuck line, a page of erased flash read as a header -
# cannot satisfy it.
SEAL = 0x5A
# An SPM command carries its SPMCSR byte in the count field, where the seal
# covers it. There is no data phase: a byte after the header would arrive after
# the seal had already been checked, which is the hole the seal exists to close.
# An autobaud loader keeps its measured bit period readable, encoded as
# delay-loop counts: (bit cycles - UNIT_DISCOUNT) / UNIT_LOOP_CYCLES,
# floored - the spin granule and per-bit overhead of libavr's software UART.
# v5/v6 keep it at ram_start; v7 moves it into GPIOR2:GPIOR1 on the chips
# that have the pair (their data addresses are in the geometry) and keeps
# ram_start only where they do not exist. --info undoes the encoding to
# report the true clock, which therefore sits within one granule below it.
UNIT_LOOP_CYCLES, UNIT_DISCOUNT = 4, 8
# 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, ram_start, gpior1)
# ram_start is where SRAM begins in data space: the classic megas and the
# tinies keep it right after the plain I/O registers (0x60), the x8/x4
# generations past their extended I/O file (0x100). gpior1 is GPIOR1's
# data address - 0x32 on the t25/45/85, 0x4A from the x8 generation on,
# None where the chip has no pair (t13, classic megas). A v7 autobaud
# loader's measured bit period lives in GPIOR2:GPIOR1 where they exist
# and at exactly ram_start elsewhere (its only RAM object; the loader's
# own build pins the layout); v5/v6 always used ram_start. --info reads
# whichever home the answering version implies.
(0x1E, 0x90, 0x07): (1024, 32, 64, True, 0x60, None), # ATtiny13/13A
(0x1E, 0x91, 0x08): (2048, 32, 128, True, 0x60, 0x32), # ATtiny25
(0x1E, 0x92, 0x06): (4096, 64, 256, True, 0x60, 0x32), # ATtiny45
(0x1E, 0x93, 0x0B): (8192, 64, 512, True, 0x60, 0x32), # ATtiny85
(0x1E, 0x92, 0x05): (4096, 64, 256, True, 0x100, 0x4A), # ATmega48/48A
(0x1E, 0x92, 0x0A): (4096, 64, 256, True, 0x100, 0x4A), # ATmega48P/48PA
(0x1E, 0x93, 0x07): (8192, 64, 512, False, 0x60, None), # ATmega8/8A
(0x1E, 0x93, 0x0A): (8192, 64, 512, False, 0x100, 0x4A), # ATmega88/88A
(0x1E, 0x93, 0x0F): (8192, 64, 512, False, 0x100, 0x4A), # ATmega88P/88PA
(0x1E, 0x94, 0x03): (16384, 128, 512, False, 0x60, None), # ATmega16/16A
(0x1E, 0x94, 0x06): (16384, 128, 512, False, 0x100, 0x4A), # ATmega168/168A
(0x1E, 0x94, 0x0B): (16384, 128, 512, False, 0x100, 0x4A), # ATmega168P/168PA
(0x1E, 0x94, 0x0A): (16384, 128, 512, False, 0x100, 0x4A), # ATmega164P/164PA
(0x1E, 0x94, 0x0F): (16384, 128, 512, False, 0x100, 0x4A), # ATmega164A
(0x1E, 0x95, 0x02): (32768, 128, 1024, False, 0x60, None), # ATmega32/32A
(0x1E, 0x95, 0x0F): (32768, 128, 1024, False, 0x100, 0x4A), # ATmega328P
(0x1E, 0x95, 0x14): (32768, 128, 1024, False, 0x100, 0x4A), # ATmega328
(0x1E, 0x95, 0x08): (32768, 128, 1024, False, 0x100, 0x4A), # ATmega324P
(0x1E, 0x95, 0x11): (32768, 128, 1024, False, 0x100, 0x4A), # ATmega324PA
(0x1E, 0x95, 0x15): (32768, 128, 1024, False, 0x100, 0x4A), # ATmega324A
(0x1E, 0x96, 0x09): (65536, 256, 2048, False, 0x100, 0x4A), # ATmega644/644A
(0x1E, 0x96, 0x0A): (65536, 256, 2048, False, 0x100, 0x4A), # ATmega644P/644PA
(0x1E, 0x97, 0x05): (131072, 256, 4096, False, 0x100, 0x4A),# ATmega1284P
(0x1E, 0x97, 0x06): (131072, 256, 4096, False, 0x100, 0x4A),# 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. A rate with
no B-constant - the off-nominal probes `--scan` walks - goes through
Linux's termios2 BOTHER; a platform without that ioctl refuses the rate
by name."""
# The termios2 ioctl pair and cflag bits, and the struct's ispeed/ospeed
# word offsets: four flag words, then a line-discipline byte and 19
# control chars padded to word 9 (include/uapi/asm-generic/termbits.h).
_TCGETS2, _TCSETS2 = 0x802C542A, 0x402C542B
_BOTHER, _CBAUD = 0o010000, 0o010017
_ISPEED, _OSPEED = 9, 10
@staticmethod
def _speed(baud):
return getattr(termios, f"B{baud}", None)
def _set_arbitrary(self, baud):
buf = array.array("i", [0] * (self._OSPEED + 1))
try:
fcntl.ioctl(self.fd, self._TCGETS2, buf, True)
buf[2] = (buf[2] & ~self._CBAUD) | self._BOTHER
buf[self._ISPEED] = buf[self._OSPEED] = baud
fcntl.ioctl(self.fd, self._TCSETS2, buf)
except OSError:
raise Error(f"this platform cannot set {baud} Bd (no termios2)") from None
def _apply_baud(self, attrs, baud):
speed = self._speed(baud)
attrs[4] = attrs[5] = speed if speed is not None else termios.B38400
termios.tcsetattr(self.fd, termios.TCSANOW, attrs)
if speed is None:
self._set_arbitrary(baud)
self.baud = baud
def __init__(self, path, baud):
self.fd = os.open(path, os.O_RDWR | os.O_NOCTTY)
try:
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
attrs[6][termios.VMIN] = 0
attrs[6][termios.VTIME] = 0
self._apply_baud(attrs, baud)
except BaseException:
os.close(self.fd)
raise
def set_baud(self, baud):
"""Retune the port without closing it - the fd stays open, so no DTR
pulse and no reset. That matters: the only caller is mid-session with a
loader copy that a reset would throw away."""
self._apply_baud(termios.tcgetattr(self.fd), baud)
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)
self.baud = baud
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 set_baud(self, baud):
"""Retune the port on its live handle - SetCommState only, so the
handle is never reopened and DTR never drops. That matters: the only
caller is mid-session with a loader copy a reset would throw away."""
if baud < 50:
raise Error(f"unsupported baud rate {baud}")
dcb = _DCB()
dcb.DCBlength = ctypes.sizeof(_DCB)
if not _k32.GetCommState(self.handle, ctypes.byref(dcb)):
_fail("cannot read the port state")
dcb.BaudRate = baud
if not _k32.SetCommState(self.handle, ctypes.byref(dcb)):
_fail(f"cannot retune the port to {baud} baud")
self.baud = baud
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
class OneWirePort:
"""The host side of a shared line (--one-wire): an FTDI-style adapter on
a one-wire link reads back every byte it transmits - its RX is tied to
its own TX through the line. Consume that echo at each write and verify
it, which doubles as a wiring check: an echo that never comes is an RX
not on the line, and is reported as itself instead of decoding as a
device reply.
The device's reply may interleave with the echo of a multi-byte write -
a loader already in session re-prompts after the knock's first byte
while the second is still queued behind that reply - so the echo is
matched byte for byte and anything else arriving in between is device
traffic, held for the next read."""
def __init__(self, port):
self._port = port
self._pending = b""
self.lost_echoes = 0
def __getattr__(self, name):
return getattr(self._port, name)
def write(self, data, blind=False):
"""Put `data` on the line and consume its echo.
`blind` marks the protocol's one multi-byte write with no ack between
its bytes - the knock. Aimed at a loader already in session, its first
byte draws a prompt while the second is still going out, and on real
wiring the device's push-pull ack **wins the line** against the host's
1 k series resistor: that second byte is *destroyed, not delayed*, and
its echo never comes. Measured on an ATtiny13A at 57600 - the loader
answers a single byte perfectly and loses the knock's second every
time. So on a blind write a missing echo is a property of the wiring
rather than a fault in it, and the caller's retry is what deals with
it. Every other write is ack-paced and cannot collide, so a missing
echo there really is an RX that is not on the line.
"""
data = bytes(data)
self._port.write(data)
# The echo arrives at line rate - 10 bits a byte - plus adapter
# latency; a generous floor keeps slow rates and USB scheduling out
# of the error path.
deadline = time.monotonic() + 10 * len(data) / self._port.baud + 0.5
remaining = data
while remaining and time.monotonic() < deadline:
# Speculative, so it cannot be read_exact, whose contract is to
# raise: doing that made the diagnosis below unreachable on every
# quiet line and surfaced a bare "timeout: got 0 of 1 bytes" in
# its place - the one message this class exists to replace.
for byte in self._port.read_available(0.02):
if remaining and byte == remaining[0]:
remaining = remaining[1:]
else:
self._pending += bytes((byte,))
if not remaining:
return
if not blind:
raise Error(f"one-wire echo missing after {len(data) - len(remaining)} of "
f"{len(data)} byte(s) - is the adapter's RX tied to the line?")
self.lost_echoes += len(remaining)
# Which loss this is matters, and the count says it. *Some* bytes lost is
# the device's ack winning the line against the host's series resistor -
# ordinary, and what the retry absorbs. *Every* byte lost is nothing
# coming back at all, which is a line that is not free: an application
# holding the shared pin low (this rig's LED demo ends that way), a
# wedge, or an RX that is not on the line. Same retry either way, but
# blaming an ack that never happened sends the reader to the wrong place.
if len(remaining) == len(data):
verbose(f"one-wire: none of {len(data)} byte(s) echoed - the line is not "
f"coming back. Held low by something? (a pin driven low, a wedge, "
f"or an RX not on the line)")
else:
verbose(f"one-wire: {len(remaining)} of {len(data)} knock byte(s) lost to the "
f"device's ack; retrying")
def write_blind(self, data):
self.write(data, blind=True)
def read_exact(self, count, timeout):
taken, self._pending = self._pending[:count], self._pending[count:]
if len(taken) == count:
return taken
return taken + self._port.read_exact(count - len(taken), timeout)
def read_available(self, wait):
taken, self._pending = self._pending, b""
return taken + self._port.read_available(0 if taken else wait)
def flush_input(self):
self._pending = b""
self._port.flush_input()
# -------------------------------------------------------------- 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 is not something the
identity reports, and from pureboot 9 nothing on the device cares."""
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
# Where SRAM begins, from the signature - None only for a chip this
# tool has no geometry row for, which the wire-block path (v1-4)
# permits where from_identity refuses.
geometry = CHIP_GEOMETRY.get(tuple(self.signature))
self.ram = geometry[4] if geometry else None
# Where this loader keeps the measured bit period (None when a fixed
# signature row is missing): the GPIOR pair from v7 where the chip
# has one, ram_start before that and everywhere without the pair.
gpior1 = geometry[5] if geometry else None
self.unit_home = gpior1 if self.version >= 7 and gpior1 is not None else self.ram
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
# The pre-knock drain runs once per port: the bytes it exists for are
# leftovers from before this process opened the port. Re-knocks later
# in the same session must not pay it - a fresh activation window is
# already burning while they wait.
self._line_drained = False
# The link this session is speaking. It moves when the host follows a
# staging copy built for another one (enter_copy).
self.baud = getattr(port, "baud", None)
self._link_declared = 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.
The timeout is short on purpose: a real answer follows the prompt
within a frame time or two, so half a second is dozens of times the
worst case - while a *false* prompt match (a stale byte, reset
garbage) makes this read collect noise, and every second spent on it
comes out of the activation window the retry needs."""
head = self.port.read_exact(4, 0.5)
if head[0:2] == b"PB":
return Info(head + self.port.read_exact(8, 0.5))
return Info.from_identity(head)
def identity(self):
"""Ask a live session who it is.
The identity command takes no argument and changes nothing, which makes
it the one question whose answer is known in advance - so it doubles as
the host's check that the stream is still in step, and as the byte a
lost host resynchronises on."""
self.port.write(b"b")
answer = self._read_identity()
self._expect_prompt()
return answer
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.
Before the port's first knock ever, the line is drained until quiet: a
prompt from a previous session (`--stay`) can still be in the USB
pipeline when the port opens, where a flush cannot clear what has not
arrived yet - and on a board that resets when its port opens, trusting
that stale byte would spend the fresh activation window reading noise
from a device that never heard the knock. Once only, and bounded:
later re-knocks in this session face no foreign leftovers, and their
own window is already burning."""
deadline = time.monotonic() + wait
if not self._line_drained:
self._line_drained = True
drain = time.monotonic() + 0.25
while self.port.read_available(0.05):
if time.monotonic() > drain:
break
knocks = 0
refusal = None
# The knock is the only write in the protocol with no ack between its
# bytes, so on a shared line it is the only one whose echo may
# legitimately not come back - the device's ack collides with it and
# wins (OneWirePort.write). Losing a byte here is what the retry below
# is for; raising instead aborted the loop before it ever ran, which on
# real wiring made every reconnect into a live session fail.
knock_out = getattr(self.port, "write_blind", self.port.write)
while True:
self.port.flush_input()
knock_out(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
# A malformed or unknown identity is retried as noise, but
# it was an answer: if nothing better ever arrives, naming
# it beats reporting silence.
refusal = failed
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:
if refusal is not None:
raise Error(f"no usable answer - the last identity reply failed: {refusal}")
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()}")
@property
def sealed(self):
"""pureboot 9 and later: bit opcodes, a sealed header, and a verdict
on that seal before the command runs."""
return self.info is not None and self.info.version >= SEALED_LOADER
def _header(self, op, space, address, count, timeout=2.0):
"""Send a sealed header and take the loader's verdict on it.
Returning normally means the loader has accepted the command and not
yet done it - which is the whole point of the verdict, and why the
payload of a fill or a write burst is sent only after this returns."""
head = bytes((op, selector(space, address), address & 0xFF, (address >> 8) & 0xFF, count & 0xFF))
seal = SEAL
for byte in head:
seal ^= byte
self.port.write(head + bytes((seal,)))
answer = self.port.read_exact(1, timeout)
if answer == NAK:
raise Error(
f"the loader refused the command (opcode {op:#04x}, {address:#06x}): the "
f"seal did not match, so nothing was done - the link mangled the header"
)
if answer != PROMPT:
raise Error(f"expected a verdict on the seal, got {answer.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))
if self.sealed:
self._header(0, space, address, chunk, 5.0)
data += self.port.read_exact(chunk, 5.0)
self._expect_prompt(5.0)
else:
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))]
if self.sealed:
self._header(OP_WRITE, space, address, len(chunk))
else:
self.port.write(bytes((ord("g"), selector(space, address), address & 0xFF,
(address >> 8) & 0xFF, len(chunk) & 0xFF)))
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.
From pureboot 9 the operation rides the header's count field instead of
arriving as data behind it, which is what puts it inside the seal: the
loader will not hand a byte to SPMCSR that the host did not seal, and
the verdict says whether it did."""
if self.sealed:
self._header(OP_WRITE, SP_SPM, address, operation, 5.0)
self._expect_prompt(5.0)
return
self._write_space(SP_SPM, address, bytes((operation,)))
def read_data(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_DATA, address, count)
def write_data(self, address, data):
self._write_space(SP_DATA, 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:
# The fill loads 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.
if self.sealed:
# The page goes out only once the header is accepted. It is the
# protocol's one unacked burst, so a header refused after the
# host had already started sending it would leave the page
# being read as commands - which is exactly what the verdict
# is placed here to prevent.
self._header(OP_FILL, SP_FLASH, address, len(data) & 0xFF)
self.port.write(data)
self._expect_prompt(2.0)
else:
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.
From v7 the jump rides the unified decode, so it carries a selector
byte the loader ignores; older loaders take the bare address. From v9
the ack is the verdict on its seal - a jump to a mangled address is a
jump into arbitrary code, so it is sealed like everything else."""
if self.sealed:
self._header(OP_JUMP, 0, word_address, 0)
return
if self.info.version >= 7:
self.port.write(bytes((ord("J"), 0, word_address & 0xFF, word_address >> 8)))
else:
self.port.write(bytes((ord("J"), word_address & 0xFF, word_address >> 8)))
self._expect_prompt()
def enter_copy(self, byte_address, wait, link=None):
"""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.
`link` is that copy's own `(baud, autobaud)`, for when it is not this
session's. A staging copy *is* the new image, so it speaks the rate and
backend it was built for; the host has to be told which, because 512
bytes of position-independent code carry no header to read it from.
Retuning goes through the open port, so no DTR pulse resets the copy that
is now running - and the session keeps the new link afterwards, since
every later jump lands in the same image.
"""
baud, autobaud = link if link is not None else (self.baud, self.autobaud)
if link is not None:
self._link_declared = True
self.jump(byte_address // 2)
if baud is not None and baud != self.baud:
self.port.set_baud(baud)
self.baud = baud
self.autobaud = autobaud
try:
return self.connect_autobaud(wait) if autobaud else self.connect(wait)
except Error as unheard:
if self._link_declared:
raise
# The bare activation timeout sends the operator to look at wiring,
# while on a patched-vector part the application region is already
# gone. Name the one cause that fits: the copy answers on its own
# link, not the resident's.
raise Error(
f"the copy at {byte_address:#06x} did not answer on this session's "
f"link ({baud} Bd, {'autobaud' if autobaud else 'fixed baud'}). An "
f"image built for another baud or backend speaks that one instead - "
f"say which with --staged-baud / --staged-autobaud"
) from unheard
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 ---
# rjmp's displacement is 12 bits, so a part whose flash is wider than 4096
# words cannot be walked this way: an offset the hardware wraps modulo the
# flash size and one it wraps modulo 4096 are then different addresses, and
# nothing in the opcode says which was meant. Every chip that needs the
# reset-vector surgery is at or under that today - the ATtiny85 sits exactly on
# it - and this is what fails loudly if one is ever added that is not.
RJMP_REACH_WORDS = 1 << 12
def rjmp_wraps_cleanly(flash_words):
return flash_words <= RJMP_REACH_WORDS
def rjmp_target(word_address, opcode, flash_words):
if not rjmp_wraps_cleanly(flash_words):
raise Error(f"{flash_words} words of flash is past rjmp's {RJMP_REACH_WORDS}-word reach - "
f"a relocated reset vector cannot be read back from the opcode alone")
return (word_address + 1 + (opcode & 0x0FFF)) % flash_words
def rjmp_to(word_address, destination, flash_words):
if not rjmp_wraps_cleanly(flash_words):
raise Error(f"{flash_words} words of flash is past rjmp's {RJMP_REACH_WORDS}-word reach - "
f"a relocated reset vector cannot be spelled as one rjmp")
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, staged_link=None):
"""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.
`staged_link` is the new image's own `(baud, autobaud)` where it differs from
this session's - the copies the host enters *are* that image, so they answer
on its link and not the resident's. Note what this does to the idempotence
above: once the staging copy is installed, the resumable state is only
reachable on the new link, so a re-run has to name it too."""
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 in place would be a copy overwriting itself as it runs. 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, link=staged_link)
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_data(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_data(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
def scan_ratios():
"""The probe walk, in percent of the built rate: the built rate itself
first, then +/-10 % in 2 % steps nearest-first - a drifted oscillator near
its trim is the common case, and each probe costs a reset."""
return [0] + [sign * step for step in (2, 4, 6, 8, 10) for sign in (-1, 1)]
def scan_rate(baud, pct):
return round(baud * (100 + pct) / 100)
def scan_report(baud, pct, version, clock=None):
"""The findings, one per line: the found rate is the session workaround,
its ratio to the built rate is the oscillator's offset, and the fixes are
the OSCCAL bake (~1 %/step, opposing the drift) or the autobaud build."""
rate = scan_rate(baud, pct)
lines = [f"scan: answered at {rate} Bd ({pct:+d} % of the built rate) - pureboot {version}",
f" session --baud {rate}"]
if clock:
lines.append(f" clock ~{clock * (100 + pct) // 100} Hz (built for {clock})")
if pct:
direction = "lower" if pct > 0 else "higher"
lines.append(f" fix rebuild with OSCCAL ~{abs(pct)} steps {direction} (~1 %/step), "
"or the autobaud build")
else:
lines.append(" fix none - the built rate answers; check the earlier wiring instead")
return lines
def op_scan(port_path, baud, wait, clock=None, one_wire=False):
"""A fixed-baud loader whose oscillator drifted still answers - at the
drifted ratio, since its rate scales with its clock. One probe per
activation window, and with an application resident the window opens
exactly once per reset, so each probe announces itself and expects a
fresh reset before knocking. On a shared line the probes echo back like
everything else; undiscarded they would answer every rate."""
for pct in scan_ratios():
rate = scan_rate(baud, pct)
print(f"scan: {rate} Bd ({pct:+d} %) - reset the target", flush=True)
try:
port = Port(port_path, rate)
except Error as unmakeable:
print(f"scan: {rate} Bd skipped - {unmakeable}")
continue
if one_wire:
port = OneWirePort(port)
try:
info = Loader(port).connect(wait)
except Error:
continue
finally:
port.close()
for line in scan_report(baud, pct, info.version, clock):
print(line)
return
raise Error("no answer within +/-10 % of the built rate - check the wiring, or deploy the "
"autobaud build, which has no rate to miss (README.md)")
# -------------------------------------------------------------------- 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("--one-wire", action="store_true",
help="the link is a shared line: read back and discard this tool's own "
"echoed bytes (any backend of a one-wire deployment)")
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("--scan", action="store_true",
help="walk +/-10%% around --baud for a fixed-baud loader gone silent - one "
"reset per probe, standalone (README.md: deployment)")
parser.add_argument("--clock", type=int, metavar="HZ",
help="the clock the loader was built for - lets --scan and an autobaud "
"--info state drift in absolute terms")
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)")
# The update enters the staging copy, which is the new image and so speaks
# the link *it* was built for. Nothing in the image says which, so where it
# differs from this session's these name it and the host follows.
parser.add_argument("--staged-baud", metavar="BD", type=int,
help="the baud the --update-loader image was built for, where it "
"differs from --baud")
parser.add_argument("--staged-autobaud", action=argparse.BooleanOptionalAction, default=None,
help="whether that image is an autobaud build, where it differs "
"from --autobaud")
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("--poke", metavar="ADDR:HEX", help="write hex bytes into data space - "
"pureboot 5 and later")
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("--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")
if args.scan:
if args.autobaud:
parser.error("--scan probes fixed rates; an autobaud loader has none to miss")
op_scan(args.port, args.baud, args.wait, args.clock, args.one_wire)
return
port = Port(args.port, args.baud)
if args.one_wire:
port = OneWirePort(port)
verbose(f"{args.port}: {args.baud} Bd 8N1, DTR/RTS asserted"
+ (", one-wire echo discarded" if args.one_wire else ""))
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}")
if args.autobaud and info.unit_home is not None:
# The measured bit period, from wherever this version keeps it
# (unit_home); decoded and times the rate this session drives,
# that is the true clock - the number to hold an OSCCAL bake
# or a fixed-baud build against (README.md: deployment). The
# autobaud identity path refuses unknown signatures, so the
# home is always known here; the guard states that dependency.
unit = int.from_bytes(loader.read_data(info.unit_home, 2), "little")
cycles = unit * UNIT_LOOP_CYCLES + UNIT_DISCOUNT
clock = cycles * args.baud
offset = f", {(clock / args.clock - 1) * 100:+.1f} % of {args.clock}" if args.clock else ""
print(f" measured {clock} Hz ({cycles} cycles/bit x {args.baud} Bd{offset})")
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"
staged_link = None
if args.staged_baud is not None or args.staged_autobaud is not None:
staged_link = (
args.staged_baud if args.staged_baud is not None else args.baud,
args.staged_autobaud if args.staged_autobaud is not None else args.autobaud,
)
op_update_loader(loader, args.wait, args.update_loader, state, fuse_bytes,
staged_link)
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, OSError) as error:
print(f"error: {error}", file=sys.stderr)
sys.exit(1)
except KeyboardInterrupt:
sys.exit(130)
Reference in New Issue View Git Blame Copy Permalink