BlackMark
579ca81b27
Measured on an ATtiny13A with the link folded onto PB3 and the FTDI's TX reaching it through 1 k: a knock aimed at a loader already in session loses its second byte every time, 8 runs of 8, never intermittently. The first byte draws a prompt while the second is still going out and the device's push-pull ack wins the line against the resistor, so that byte is destroyed rather than delayed — which is what the README predicted and the sim bridge cannot show, since it arbitrates the line by queueing. The recovery for it existed and could not run. Two defects: OneWirePort.write read its echo with read_exact, whose contract is to raise, so the "one-wire echo missing — is the adapter's RX tied to the line?" message was unreachable on any line that simply fell quiet, and a bare "timeout: got 0 of 1 bytes" surfaced in its place. The one message the class exists to produce could never be produced. The read is speculative and is now read_available. And any raise from write aborted _handshake before the retry loop that exists to absorb exactly this, whose docstring already claimed it "converges into an already-live session" — true on a pty, impossible on real wiring. The knock is now the one write marked blind: a missing echo there is a property of the shared line, counted and reported under -v rather than raised. Every other write is ack-paced and cannot collide, so a missing echo there still means an RX that is not on the line, and still raises. Both gates green on Windows (31/31 m328p, 16/16 t13a); on hardware the reconnect now converges on the first knock, the surviving prompt being all the handshake needs. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1755 lines
80 KiB
Python
1755 lines
80 KiB
Python
#!/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"+"
|
||
VERSION = 9 # 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), and a version it has
|
||
# no decoder for moves the floor.
|
||
OLDEST_LOADER = 1
|
||
NEWEST_LOADER = 8
|
||
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_RAM, SP_FUSE, SP_SPM = 0, 1, 2, 3, 4
|
||
|
||
# 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)
|
||
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 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
|
||
# 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 _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()}")
|
||
|
||
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.
|
||
From v7 'J' rides the unified decode, so it carries a selector byte
|
||
the loader ignores; older loaders take the bare address."""
|
||
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 ---
|
||
|
||
|
||
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, 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 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, 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_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
|
||
|
||
|
||
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("--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")
|
||
|
||
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_ram(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 × {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)
|