pureboot.py: an update follows the staging copy onto its own link
--update-loader works by entering copies of the *new* image and letting them rewrite the resident. Those copies speak the link they were built for, but the host went on knocking with the session's baud and backend — the resident's. Where the image changed either, the staging copy was installed and then never answered: resident untouched, and on a 1 KiB tiny the staging slot is the whole application region, so the application was already gone. The wire cannot be probed for it. 512 bytes of position-independent code carry no header saying what rate they were built for, so the operator declares it: --staged-baud and --staged-autobaud, applied from the jump into the staging copy onward. Retuning goes through the open port — SetCommState or tcsetattr on the live handle, never a reopen — because a DTR pulse would reset the copy being talked to. Undeclared against a changed link it still cannot work, but the error now names that as the cause instead of reporting the bare activation timeout that sent the operator looking at wiring. The README's idempotence claim needed the same qualification: from step 2 a re-run must reach the new image, and after step 3 word 0 points at the staging copy, so on a patched-vector part the resident's link reaches nothing at all. Found on an ATtiny13A, where two controls differing only in the activation window updated cleanly and so isolated the link as the variable. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -162,6 +162,8 @@ if(PROJECT_IS_TOP_LEVEL)
|
|||||||
${CMAKE_CURRENT_SOURCE_DIR}/pureboot/pureboot.py)
|
${CMAKE_CURRENT_SOURCE_DIR}/pureboot/pureboot.py)
|
||||||
add_test(NAME pureboot.handshake
|
add_test(NAME pureboot.handshake
|
||||||
COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/test/test_handshake.py)
|
COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/test/test_handshake.py)
|
||||||
|
add_test(NAME pureboot.updatelink
|
||||||
|
COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/test/test_update_link.py)
|
||||||
endif()
|
endif()
|
||||||
|
|
||||||
# The protocol test flashes this fixture through the loader with the real
|
# The protocol test flashes this fixture through the loader with the real
|
||||||
|
|||||||
@@ -328,6 +328,23 @@ with any pureboot build — a re-timed window, a newer version — using the
|
|||||||
loader itself as its own staging loader. The image is the loader's own 512
|
loader itself as its own staging loader. The image is the loader's own 512
|
||||||
bytes as a raw binary, or the Intel HEX the build emits beside it.
|
bytes as a raw binary, or the Intel HEX the build emits beside it.
|
||||||
|
|
||||||
|
One thing the image cannot tell the host: **which link it speaks.** The update
|
||||||
|
works by entering copies of the *new* image (steps 3 and 4 below), so a build
|
||||||
|
made for another baud or another backend answers on that one and not on the
|
||||||
|
session's — and 512 bytes of position-independent code carry no header to read
|
||||||
|
it from. Where the new image's link differs, name it:
|
||||||
|
|
||||||
|
```sh
|
||||||
|
# a 57600 fixed-baud resident, replaced by an autobaud build
|
||||||
|
pureboot.py --port … --baud 57600 --update-loader ab.bin --staged-autobaud
|
||||||
|
# …or by a 38400 build of the same backend
|
||||||
|
pureboot.py --port … --baud 57600 --update-loader sw38400.bin --staged-baud 38400
|
||||||
|
```
|
||||||
|
|
||||||
|
The host retunes on the open port, so no DTR pulse resets the copy it is talking
|
||||||
|
to. Omit them against a changed link and the update stops after installing the
|
||||||
|
staging copy, saying so and naming this as the cause.
|
||||||
|
|
||||||
The preflight refuses an image built for another chip: the stamp every pureboot
|
The preflight refuses an image built for another chip: the stamp every pureboot
|
||||||
binary carries must resolve to the device's own geometry, and the error names
|
binary carries must resolve to the device's own geometry, and the error names
|
||||||
both. Die revisions share their base signature and geometry, so their images
|
both. Die revisions share their base signature and geometry, so their images
|
||||||
@@ -348,10 +365,15 @@ are interchangeable — as the silicon is.
|
|||||||
content, and the state file is discarded.
|
content, and the state file is discarded.
|
||||||
|
|
||||||
Every phase is idempotent and keyed off the actual flash state, so re-running
|
Every phase is idempotent and keyed off the actual flash state, so re-running
|
||||||
the same command after any interruption resumes and completes. The state file
|
the same command after any interruption resumes and completes — with one
|
||||||
carries the only bytes not recoverable from the device; losing it mid-update
|
qualification, which is the link again: from step 2 on, the copy the re-run has
|
||||||
still completes the update, and the staging region comes back by reflashing
|
to reach is the *new* image, so a resumed run needs the same `--staged-*` as the
|
||||||
the application. A boot-sectioned mega needs its fuses for the preflight — read
|
first one. On a patched-vector part step 3 also re-aims word 0 at the staging
|
||||||
|
copy, so after that point a reset reaches the new image's link and **only** that
|
||||||
|
one; a re-run on the resident's link finds nothing at all. The state file carries
|
||||||
|
the only bytes not recoverable from the device; losing it mid-update still
|
||||||
|
completes the update, and the staging region comes back by reflashing the
|
||||||
|
application. A boot-sectioned mega needs its fuses for the preflight — read
|
||||||
from the device, or supplied with `--assume-fuses` where reading is impossible
|
from the device, or supplied with `--assume-fuses` where reading is impossible
|
||||||
(simulators).
|
(simulators).
|
||||||
|
|
||||||
@@ -425,6 +447,11 @@ Per chip preset, `ctest` runs:
|
|||||||
- `pureboot.handshake` — the host tool's activation must not hang on a target
|
- `pureboot.handshake` — the host tool's activation must not hang on a target
|
||||||
that never falls quiet: the drain after a prompt is bounded by the handshake
|
that never falls quiet: the drain after a prompt is bounded by the handshake
|
||||||
deadline, and a well-behaved loader still connects;
|
deadline, and a well-behaved loader still connects;
|
||||||
|
- `pureboot.updatelink` — an update whose image changes the baud or the backend
|
||||||
|
must follow the staging copy onto *its* link, since that copy is the new image;
|
||||||
|
and where nothing was declared, the failure must name the link rather than
|
||||||
|
report a bare activation timeout, because by then the staging slot is written
|
||||||
|
and on a 1 KiB tiny that was the application;
|
||||||
- `pureboot.planner` — the host tool's pure logic: programming orders and their
|
- `pureboot.planner` — the host tool's pure logic: programming orders and their
|
||||||
recovery properties, the surgery, the staging composition, the boot-fuse
|
recovery properties, the surgery, the staging composition, the boot-fuse
|
||||||
decode, the update preflight over synthetic fuse bytes, and the repairing
|
decode, the update preflight over synthetic fuse bytes, and the repairing
|
||||||
|
|||||||
@@ -146,6 +146,13 @@ class Progress:
|
|||||||
class PosixPort:
|
class PosixPort:
|
||||||
"""A raw serial port with deadline-based reads, over termios."""
|
"""A raw serial port with deadline-based reads, over termios."""
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _speed(baud):
|
||||||
|
try:
|
||||||
|
return getattr(termios, f"B{baud}")
|
||||||
|
except AttributeError:
|
||||||
|
raise Error(f"unsupported baud rate {baud}") from None
|
||||||
|
|
||||||
def __init__(self, path, baud):
|
def __init__(self, path, baud):
|
||||||
self.fd = os.open(path, os.O_RDWR | os.O_NOCTTY)
|
self.fd = os.open(path, os.O_RDWR | os.O_NOCTTY)
|
||||||
attrs = termios.tcgetattr(self.fd)
|
attrs = termios.tcgetattr(self.fd)
|
||||||
@@ -153,14 +160,20 @@ class PosixPort:
|
|||||||
attrs[1] = 0 # oflag
|
attrs[1] = 0 # oflag
|
||||||
attrs[2] = termios.CREAD | termios.CLOCAL | termios.CS8 # cflag
|
attrs[2] = termios.CREAD | termios.CLOCAL | termios.CS8 # cflag
|
||||||
attrs[3] = 0 # lflag
|
attrs[3] = 0 # lflag
|
||||||
try:
|
attrs[4] = attrs[5] = self._speed(baud)
|
||||||
speed = getattr(termios, f"B{baud}")
|
|
||||||
except AttributeError:
|
|
||||||
raise Error(f"unsupported baud rate {baud}") from None
|
|
||||||
attrs[4] = attrs[5] = speed
|
|
||||||
attrs[6][termios.VMIN] = 0
|
attrs[6][termios.VMIN] = 0
|
||||||
attrs[6][termios.VTIME] = 0
|
attrs[6][termios.VTIME] = 0
|
||||||
termios.tcsetattr(self.fd, termios.TCSANOW, attrs)
|
termios.tcsetattr(self.fd, termios.TCSANOW, attrs)
|
||||||
|
self.baud = baud
|
||||||
|
|
||||||
|
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."""
|
||||||
|
attrs = termios.tcgetattr(self.fd)
|
||||||
|
attrs[4] = attrs[5] = self._speed(baud)
|
||||||
|
termios.tcsetattr(self.fd, termios.TCSANOW, attrs)
|
||||||
|
self.baud = baud
|
||||||
|
|
||||||
def close(self):
|
def close(self):
|
||||||
os.close(self.fd)
|
os.close(self.fd)
|
||||||
@@ -289,6 +302,7 @@ if os.name == "nt":
|
|||||||
# timeout would otherwise stay at the driver's default — which
|
# timeout would otherwise stay at the driver's default — which
|
||||||
# may be "wait forever" — until the first read.
|
# may be "wait forever" — until the first read.
|
||||||
self._deadline(_GAP_MS, 1000)
|
self._deadline(_GAP_MS, 1000)
|
||||||
|
self.baud = baud
|
||||||
except Error:
|
except Error:
|
||||||
# An open port outlives the exception otherwise, and a COM
|
# An open port outlives the exception otherwise, and a COM
|
||||||
# handle is exclusive: the next attempt would meet its own
|
# handle is exclusive: the next attempt would meet its own
|
||||||
@@ -296,6 +310,21 @@ if os.name == "nt":
|
|||||||
self.close()
|
self.close()
|
||||||
raise
|
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):
|
def close(self):
|
||||||
_k32.CloseHandle(self.handle)
|
_k32.CloseHandle(self.handle)
|
||||||
|
|
||||||
@@ -457,6 +486,10 @@ class Loader:
|
|||||||
# Set once a session is established over an autobaud link, so a
|
# Set once a session is established over an autobaud link, so a
|
||||||
# re-entry after 'J' repeats the handshake that worked.
|
# re-entry after 'J' repeats the handshake that worked.
|
||||||
self.autobaud = False
|
self.autobaud = 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):
|
def _read_identity(self):
|
||||||
"""The 'b' reply, in either of the two layouts a loader may send.
|
"""The 'b' reply, in either of the two layouts a loader may send.
|
||||||
@@ -664,12 +697,41 @@ class Loader:
|
|||||||
self.port.write(bytes((ord("J"), word_address & 0xFF, word_address >> 8)))
|
self.port.write(bytes((ord("J"), word_address & 0xFF, word_address >> 8)))
|
||||||
self._expect_prompt()
|
self._expect_prompt()
|
||||||
|
|
||||||
def enter_copy(self, byte_address, wait):
|
def enter_copy(self, byte_address, wait, link=None):
|
||||||
"""Jump into the loader copy at `byte_address` and knock it — a slot
|
"""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."""
|
base is that copy's entry stub, so it can only land there.
|
||||||
autobaud = self.autobaud
|
|
||||||
|
`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)
|
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)
|
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):
|
def run_application(self):
|
||||||
self.jump(self.info.app_entry_word)
|
self.jump(self.info.app_entry_word)
|
||||||
@@ -1017,10 +1079,16 @@ def patch_word0(loader, page0, target_base):
|
|||||||
return bytes(patched)
|
return bytes(patched)
|
||||||
|
|
||||||
|
|
||||||
def op_update_loader(loader, wait, path, state_path, fuse_bytes):
|
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
|
"""Replace the resident loader with `path`, using the loader as its own
|
||||||
staging loader. Every phase is idempotent and keyed off the flash state,
|
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."""
|
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
|
info = loader.info
|
||||||
image = loader_image(path)
|
image = loader_image(path)
|
||||||
for warning in update_preflight(image, info, fuse_bytes):
|
for warning in update_preflight(image, info, fuse_bytes):
|
||||||
@@ -1065,7 +1133,7 @@ def op_update_loader(loader, wait, path, state_path, fuse_bytes):
|
|||||||
# routes through the resident, word 0 is re-aimed at the staging copy for
|
# 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.
|
# the rewrite, so a power loss mid-rewrite still resets into a loader.
|
||||||
verbose(f"entering the staging copy at {info.stage:#06x}")
|
verbose(f"entering the staging copy at {info.stage:#06x}")
|
||||||
loader.enter_copy(info.stage, wait)
|
loader.enter_copy(info.stage, wait, link=staged_link)
|
||||||
redirect = info.patch_vector and info.stage != 0
|
redirect = info.patch_vector and info.stage != 0
|
||||||
if redirect:
|
if redirect:
|
||||||
verbose("word 0 re-aimed at the staging copy for the rewrite")
|
verbose("word 0 re-aimed at the staging copy for the rewrite")
|
||||||
@@ -1302,6 +1370,15 @@ def main():
|
|||||||
parser.add_argument("--fuses", action="store_true", help="read the fuse and lock bytes")
|
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("--update-loader", metavar="FILE", help="replace the loader with this pureboot binary")
|
||||||
parser.add_argument("--state", metavar="FILE", help="update state file (default: FILE.pbstate)")
|
parser.add_argument("--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 "
|
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)")
|
"(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("--erase-flash", action="store_true", help="0xff over the application flash")
|
||||||
@@ -1351,7 +1428,14 @@ def main():
|
|||||||
fuse_bytes = read
|
fuse_bytes = read
|
||||||
if args.update_loader:
|
if args.update_loader:
|
||||||
state = args.state or args.update_loader + ".pbstate"
|
state = args.state or args.update_loader + ".pbstate"
|
||||||
op_update_loader(loader, args.wait, args.update_loader, state, fuse_bytes)
|
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:
|
if args.flash:
|
||||||
op_flash(loader, args.flash, args.erase_flash, not args.no_verify, fuse_bytes, args.force)
|
op_flash(loader, args.flash, args.erase_flash, not args.no_verify, fuse_bytes, args.force)
|
||||||
elif args.erase_flash:
|
elif args.erase_flash:
|
||||||
|
|||||||
167
test/test_update_link.py
Executable file
167
test/test_update_link.py
Executable file
@@ -0,0 +1,167 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Self-update across a link change: the host must follow the staging copy.
|
||||||
|
|
||||||
|
`--update-loader` installs the new image in the staging slot and then *enters
|
||||||
|
it* to have it rewrite the resident. That copy is the new image, so it speaks the
|
||||||
|
new image's baud and backend — but the host was talking to the *resident*. Where
|
||||||
|
the two differ, the host kept knocking at the old rate in the old mode, the
|
||||||
|
staging copy never answered, and the update stranded: staging installed, resident
|
||||||
|
untouched, and on a 1 KiB tiny the application region (which *is* the staging
|
||||||
|
slot there) already gone.
|
||||||
|
|
||||||
|
The wire cannot be probed for this — 512 bytes of position-independent code carry
|
||||||
|
no header saying what rate they were built for — so the operator declares it, and
|
||||||
|
a mismatch with nothing declared has to say so instead of reporting a bare
|
||||||
|
timeout.
|
||||||
|
|
||||||
|
Stdlib only, no device: host-tool logic, so it runs on every chip's preset beside
|
||||||
|
pureboot.planner.
|
||||||
|
"""
|
||||||
|
import importlib.util
|
||||||
|
import pathlib
|
||||||
|
|
||||||
|
PB = pathlib.Path(__file__).resolve().parents[1] / "pureboot" / "pureboot.py"
|
||||||
|
_spec = importlib.util.spec_from_file_location("pureboot", PB)
|
||||||
|
pb = importlib.util.module_from_spec(_spec)
|
||||||
|
_spec.loader.exec_module(pb)
|
||||||
|
|
||||||
|
IDENTITY = b"\x05\x1e\x95\x0f" # pureboot 5 + m328p signature
|
||||||
|
P = F = 0
|
||||||
|
|
||||||
|
|
||||||
|
def check(name, ok, detail=""):
|
||||||
|
global P, F
|
||||||
|
P, F = P + (1 if ok else 0), F + (0 if ok else 1)
|
||||||
|
print(f" [{'PASS' if ok else 'FAIL'}] {name}" + (f" — {detail}" if detail else ""))
|
||||||
|
|
||||||
|
|
||||||
|
class TwoLinkPort:
|
||||||
|
"""A board whose resident and staging copy answer on different links.
|
||||||
|
|
||||||
|
Only the rate currently set decides who can be heard, which is the physical
|
||||||
|
truth: a loader's bit timing is a cycle count, so a copy built for another
|
||||||
|
rate is unreadable until the host retunes. The knock bytes carry the mode, so
|
||||||
|
a backend mismatch is caught the same way.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, resident=(57600, False), staged=(38400, False)):
|
||||||
|
self.resident, self.staged = resident, staged
|
||||||
|
self.baud = resident[0]
|
||||||
|
self.entered = False # a 'J' has handed control to the staging copy
|
||||||
|
self.switches = [] # every retune the host asked for
|
||||||
|
self.pending = bytearray() # what the device has queued to send
|
||||||
|
|
||||||
|
# --- the part under test needs this to exist at all
|
||||||
|
def set_baud(self, baud):
|
||||||
|
self.baud = baud
|
||||||
|
self.switches.append(baud)
|
||||||
|
|
||||||
|
def flush_input(self):
|
||||||
|
self.pending.clear()
|
||||||
|
|
||||||
|
def _audible(self, knock=None):
|
||||||
|
baud, autobaud = self.staged if self.entered else self.resident
|
||||||
|
if self.baud != baud:
|
||||||
|
return False
|
||||||
|
if knock is None:
|
||||||
|
return True
|
||||||
|
return knock == (bytes((pb.CALIBRATE, ord("p"))) if autobaud else b"pb")
|
||||||
|
|
||||||
|
def write(self, data):
|
||||||
|
data = bytes(data)
|
||||||
|
if data[:1] == b"J" and len(data) == 3:
|
||||||
|
# The resident acks the jump, then control moves to the copy.
|
||||||
|
if self._audible():
|
||||||
|
self.pending += pb.PROMPT
|
||||||
|
self.entered = True
|
||||||
|
elif data in (b"pb", bytes((pb.CALIBRATE, ord("p")))):
|
||||||
|
if self._audible(data):
|
||||||
|
self.pending += pb.PROMPT
|
||||||
|
elif data == b"b":
|
||||||
|
if self._audible():
|
||||||
|
self.pending += IDENTITY + pb.PROMPT
|
||||||
|
|
||||||
|
def read_available(self, wait):
|
||||||
|
out, self.pending = bytes(self.pending), bytearray()
|
||||||
|
return out
|
||||||
|
|
||||||
|
def read_exact(self, count, timeout):
|
||||||
|
if len(self.pending) < count:
|
||||||
|
raise pb.Error(f"timeout: got {len(self.pending)} of {count} bytes")
|
||||||
|
out, self.pending = bytes(self.pending[:count]), self.pending[count:]
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def connected(port):
|
||||||
|
"""A Loader already in session with the resident."""
|
||||||
|
loader = pb.Loader(port)
|
||||||
|
loader.connect(2.0)
|
||||||
|
return loader
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
# The control first: where the staged image keeps the resident's link, the
|
||||||
|
# flow works and needs no retune. This is the case that always passed, and
|
||||||
|
# it is what made the bug look like "self-update is broken" rather than
|
||||||
|
# "self-update cannot change the link".
|
||||||
|
port = TwoLinkPort(resident=(57600, False), staged=(57600, False))
|
||||||
|
loader = connected(port)
|
||||||
|
try:
|
||||||
|
loader.enter_copy(0x7C00, 2.0)
|
||||||
|
check("same link: staging copy entered", True)
|
||||||
|
except pb.Error as error:
|
||||||
|
check("same link: staging copy entered", False, str(error))
|
||||||
|
|
||||||
|
# A baud change, declared. The host must retune before knocking.
|
||||||
|
port = TwoLinkPort(resident=(57600, False), staged=(38400, False))
|
||||||
|
loader = connected(port)
|
||||||
|
try:
|
||||||
|
loader.enter_copy(0x7C00, 2.0, link=(38400, False))
|
||||||
|
check("baud change declared: entered after retuning", 38400 in port.switches,
|
||||||
|
f"switches={port.switches}")
|
||||||
|
except (pb.Error, TypeError) as error:
|
||||||
|
check("baud change declared: entered after retuning", False, repr(error))
|
||||||
|
|
||||||
|
# A backend change, declared: the knock itself has to become the calibration
|
||||||
|
# pulse, or an autobaud staging copy never hears a thing.
|
||||||
|
port = TwoLinkPort(resident=(57600, False), staged=(57600, True))
|
||||||
|
loader = connected(port)
|
||||||
|
try:
|
||||||
|
loader.enter_copy(0x7C00, 2.0, link=(57600, True))
|
||||||
|
check("backend change declared: entered as autobaud", True)
|
||||||
|
except (pb.Error, TypeError) as error:
|
||||||
|
check("backend change declared: entered as autobaud", False, repr(error))
|
||||||
|
|
||||||
|
# Nothing declared against a changed link: it still cannot work, but the
|
||||||
|
# error has to name the cause. A bare "no answer" sent the operator looking
|
||||||
|
# at the wiring while the application region sat erased.
|
||||||
|
port = TwoLinkPort(resident=(57600, False), staged=(38400, False))
|
||||||
|
loader = connected(port)
|
||||||
|
try:
|
||||||
|
loader.enter_copy(0x7C00, 0.3)
|
||||||
|
check("undeclared mismatch: reported", False, "unexpectedly succeeded")
|
||||||
|
except pb.Error as error:
|
||||||
|
text = str(error).lower()
|
||||||
|
check("undeclared mismatch: error names the link, not just a timeout",
|
||||||
|
"link" in text or "baud" in text or "backend" in text, str(error))
|
||||||
|
except TypeError as error:
|
||||||
|
check("undeclared mismatch: error names the link, not just a timeout",
|
||||||
|
False, repr(error))
|
||||||
|
|
||||||
|
# The resident's own link must be restored for the caller: a declared
|
||||||
|
# staging link is for the copy, and the tool talks to the new resident after.
|
||||||
|
port = TwoLinkPort(resident=(57600, False), staged=(38400, False))
|
||||||
|
loader = connected(port)
|
||||||
|
try:
|
||||||
|
loader.enter_copy(0x7C00, 2.0, link=(38400, False))
|
||||||
|
check("session records the link it is now speaking", loader.baud == 38400,
|
||||||
|
f"loader.baud={getattr(loader, 'baud', None)}")
|
||||||
|
except (pb.Error, TypeError, AttributeError) as error:
|
||||||
|
check("session records the link it is now speaking", False, repr(error))
|
||||||
|
|
||||||
|
print(f"\n {P} passed, {F} failed")
|
||||||
|
return 1 if F else 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
raise SystemExit(main())
|
||||||
Reference in New Issue
Block a user