pureboot: drive the serial port on Windows too
The host tool was standard-library-only but POSIX-only with it: termios and select() bound the port layer, and importing termios failed outright on Windows, so the module could not even load there. Split Port into PosixPort (unchanged) and a WindowsPort over the Win32 serial API through ctypes, picked by os.name; every call site keeps the Port name. kernel32 only, so the standard-library constraint holds. Windows has no select() for a COM handle, so the read deadlines move into the driver as COMMTIMEOUTS, re-armed per read: read_available() ends on a gap longer than a USB-serial latency timer coalesces (16 ms on FTDI parts), read_exact() on the count or its deadline. Opening asserts DTR and RTS as a POSIX open does, so a board wiring DTR to reset still pulses it. A failed configuration closes the handle before raising - a COM handle is exclusive, and the leak met the next open as "Access is denied". Win32 takes any integer baud and a driver may accept one its hardware cannot produce (an FT232R reports back a baud of 3 and keeps the old divisor), so obvious nonsense is refused where termios' table would have. Tested against an ATmega328P on COM6: info, fuses, both memories programmed and verified, session reconnect, hand-over, the loader self-update, and the write guard on its own slot. test_planner runs on Windows now as well. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -150,8 +150,13 @@ reflashing the application. The mega needs its fuses for the preflight
|
||||
|
||||
## Host tool
|
||||
|
||||
`pureboot.py` — Python 3, standard library only (termios drives any tty,
|
||||
a USB adapter as well as a simavr pty):
|
||||
`pureboot.py` — Python 3, standard library only. The port layer is the one
|
||||
platform-specific part: termios drives any tty on POSIX (a USB adapter as
|
||||
well as a simavr pty), the Win32 serial API through `ctypes` drives a COM
|
||||
port on Windows (`--port COM6`; the `\\.\` form for two-digit ports is
|
||||
supplied by the tool). Opening the port asserts DTR and RTS on both, so a
|
||||
board that wires DTR to reset gets its reset pulse and opens the activation
|
||||
window by itself.
|
||||
|
||||
pureboot.py --port /dev/ttyUSB0 --baud 57600 \
|
||||
--info --fuses --flash app.hex
|
||||
@@ -192,3 +197,7 @@ Per chip preset, `ctest` runs:
|
||||
then every power-fail phase: the device is killed mid-write, restarted
|
||||
from its flash dump, and a re-run must complete the update with the
|
||||
application intact throughout.
|
||||
|
||||
`size`, `pi`, and `planner` are host logic and run anywhere; the three
|
||||
simulator-driven targets need simavr and a pty, so they are POSIX-only —
|
||||
on Windows the tool is exercised against real hardware.
|
||||
|
||||
@@ -16,18 +16,24 @@ the resident slot, and restores what the staging slot held — resumable at
|
||||
every phase from the flash state plus a host-side state file carrying the
|
||||
saved bytes.
|
||||
|
||||
Python standard library only; the serial port is driven with termios, so any
|
||||
tty works — a USB adapter as well as a simavr pty.
|
||||
Python standard library only; the serial port is driven with termios on POSIX
|
||||
and the Win32 serial API (through ctypes) on Windows, so any tty or COM port
|
||||
works — a USB adapter as well as a simavr pty.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import select
|
||||
import sys
|
||||
import termios
|
||||
import time
|
||||
|
||||
if os.name == "nt":
|
||||
import ctypes
|
||||
from ctypes import wintypes
|
||||
else:
|
||||
import select
|
||||
import termios
|
||||
|
||||
PROMPT = b"+"
|
||||
PROTOCOL_VERSION = 1
|
||||
SLOT = 512 # the loader slot size; also the self-update staging distance
|
||||
@@ -40,8 +46,8 @@ class Error(Exception):
|
||||
# ---------------------------------------------------------------- serial ---
|
||||
|
||||
|
||||
class Port:
|
||||
"""A raw serial port with deadline-based reads."""
|
||||
class PosixPort:
|
||||
"""A raw serial port with deadline-based reads, over termios."""
|
||||
|
||||
def __init__(self, path, baud):
|
||||
self.fd = os.open(path, os.O_RDWR | os.O_NOCTTY)
|
||||
@@ -86,6 +92,171 @@ class Port:
|
||||
return data
|
||||
|
||||
|
||||
if os.name == "nt":
|
||||
# The same port, over the Win32 serial API — kernel32 through ctypes, so
|
||||
# the tool stays standard-library only. Timeouts live in the driver
|
||||
# (COMMTIMEOUTS) rather than in a readiness call: Windows has no select()
|
||||
# for a COM handle, so each read asks the driver for its own deadline.
|
||||
|
||||
_GENERIC_READ, _GENERIC_WRITE = 0x80000000, 0x40000000
|
||||
_OPEN_EXISTING, _PURGE_RXCLEAR = 3, 0x0008
|
||||
_INVALID_HANDLE = wintypes.HANDLE(-1).value
|
||||
# A gap this long ends a read_available(): longer than the coalescing a
|
||||
# USB-serial adapter's latency timer imposes (16 ms on FTDI parts), so a
|
||||
# burst is not split, short enough to stay responsive.
|
||||
_GAP_MS = 30
|
||||
|
||||
class _DCB(ctypes.Structure):
|
||||
_fields_ = [
|
||||
("DCBlength", wintypes.DWORD),
|
||||
("BaudRate", wintypes.DWORD),
|
||||
("fBits", wintypes.DWORD), # the packed flag bitfield, set below
|
||||
("wReserved", wintypes.WORD),
|
||||
("XonLim", wintypes.WORD),
|
||||
("XoffLim", wintypes.WORD),
|
||||
("ByteSize", wintypes.BYTE),
|
||||
("Parity", wintypes.BYTE),
|
||||
("StopBits", wintypes.BYTE),
|
||||
("XonChar", ctypes.c_char),
|
||||
("XoffChar", ctypes.c_char),
|
||||
("ErrorChar", ctypes.c_char),
|
||||
("EofChar", ctypes.c_char),
|
||||
("EvtChar", ctypes.c_char),
|
||||
("wReserved1", wintypes.WORD),
|
||||
]
|
||||
|
||||
class _COMMTIMEOUTS(ctypes.Structure):
|
||||
_fields_ = [
|
||||
("ReadIntervalTimeout", wintypes.DWORD),
|
||||
("ReadTotalTimeoutMultiplier", wintypes.DWORD),
|
||||
("ReadTotalTimeoutConstant", wintypes.DWORD),
|
||||
("WriteTotalTimeoutMultiplier", wintypes.DWORD),
|
||||
("WriteTotalTimeoutConstant", wintypes.DWORD),
|
||||
]
|
||||
|
||||
_k32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
||||
_LPDWORD = ctypes.POINTER(wintypes.DWORD)
|
||||
# Declared, not inferred: a HANDLE is a pointer, and a defaulted int
|
||||
# return would truncate it on 64-bit.
|
||||
_k32.CreateFileW.restype = wintypes.HANDLE
|
||||
_k32.CreateFileW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD,
|
||||
wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE]
|
||||
_k32.ReadFile.argtypes = [wintypes.HANDLE, wintypes.LPVOID, wintypes.DWORD, _LPDWORD, wintypes.LPVOID]
|
||||
_k32.WriteFile.argtypes = [wintypes.HANDLE, wintypes.LPCVOID, wintypes.DWORD, _LPDWORD, wintypes.LPVOID]
|
||||
_k32.GetCommState.argtypes = [wintypes.HANDLE, ctypes.POINTER(_DCB)]
|
||||
_k32.SetCommState.argtypes = [wintypes.HANDLE, ctypes.POINTER(_DCB)]
|
||||
_k32.SetCommTimeouts.argtypes = [wintypes.HANDLE, ctypes.POINTER(_COMMTIMEOUTS)]
|
||||
_k32.PurgeComm.argtypes = [wintypes.HANDLE, wintypes.DWORD]
|
||||
_k32.CloseHandle.argtypes = [wintypes.HANDLE]
|
||||
|
||||
def _fail(what):
|
||||
code = ctypes.get_last_error()
|
||||
raise Error(f"{what}: {ctypes.FormatError(code).strip()} (Windows error {code})")
|
||||
|
||||
class WindowsPort:
|
||||
"""A raw serial port with deadline-based reads, over Win32."""
|
||||
|
||||
def __init__(self, path, baud):
|
||||
# Win32 takes the rate as a plain integer, so unlike termios any
|
||||
# rate the hardware can divide down to is available — but a driver
|
||||
# may also accept one it cannot produce (an FT232R takes a baud of
|
||||
# 3, reports it back, and goes on using the previous divisor).
|
||||
# Only obvious nonsense is refusable; the rest is the driver's word.
|
||||
if baud < 50:
|
||||
raise Error(f"unsupported baud rate {baud}")
|
||||
# \\.\COM6: the device-namespace form. A bare COMn resolves only
|
||||
# for n < 10, and double-digit ports are routine on Windows.
|
||||
if path.lower().startswith("com") and path[3:].isdigit():
|
||||
path = rf"\\.\{path}"
|
||||
self.handle = _k32.CreateFileW(
|
||||
path, _GENERIC_READ | _GENERIC_WRITE, 0, None, _OPEN_EXISTING, 0, None
|
||||
)
|
||||
if self.handle == _INVALID_HANDLE:
|
||||
_fail(f"cannot open {path}")
|
||||
self.timeouts = None
|
||||
try:
|
||||
dcb = _DCB()
|
||||
dcb.DCBlength = ctypes.sizeof(_DCB)
|
||||
if not _k32.GetCommState(self.handle, ctypes.byref(dcb)):
|
||||
_fail(f"cannot read the state of {path}")
|
||||
dcb.BaudRate, dcb.ByteSize, dcb.Parity, dcb.StopBits = baud, 8, 0, 0 # 8N1
|
||||
# fBinary, and DTR/RTS asserted (fDtrControl and fRtsControl,
|
||||
# two bits each, = _ENABLE); every other flag clear, so no
|
||||
# parity and no flow control. Raising both matches what opening
|
||||
# a POSIX tty does — including the reset pulse on the boards
|
||||
# that wire DTR to it.
|
||||
dcb.fBits = 0x1 | (1 << 4) | (1 << 12)
|
||||
if not _k32.SetCommState(self.handle, ctypes.byref(dcb)):
|
||||
_fail(f"cannot configure {path} for {baud} baud 8N1")
|
||||
# Arm them once here too: reads re-arm per call, but the write
|
||||
# timeout would otherwise stay at the driver's default — which
|
||||
# may be "wait forever" — until the first read.
|
||||
self._deadline(_GAP_MS, 1000)
|
||||
except Error:
|
||||
# An open port outlives the exception otherwise, and a COM
|
||||
# handle is exclusive: the next attempt would meet its own
|
||||
# leftover as "Access is denied".
|
||||
self.close()
|
||||
raise
|
||||
|
||||
def close(self):
|
||||
_k32.CloseHandle(self.handle)
|
||||
|
||||
def _deadline(self, interval, total):
|
||||
"""Arm the driver's read timeouts: `interval` ms of quiet ends a
|
||||
read once bytes have arrived, `total` ms ends it regardless."""
|
||||
if self.timeouts == (interval, total):
|
||||
return
|
||||
spec = _COMMTIMEOUTS()
|
||||
spec.ReadIntervalTimeout = interval
|
||||
spec.ReadTotalTimeoutConstant = total
|
||||
spec.WriteTotalTimeoutConstant = 5000
|
||||
if not _k32.SetCommTimeouts(self.handle, ctypes.byref(spec)):
|
||||
_fail("cannot set the port timeouts")
|
||||
self.timeouts = (interval, total)
|
||||
|
||||
def _read(self, count):
|
||||
buffer = ctypes.create_string_buffer(count)
|
||||
got = wintypes.DWORD()
|
||||
if not _k32.ReadFile(self.handle, buffer, count, ctypes.byref(got), None):
|
||||
_fail("read failed")
|
||||
return buffer.raw[: got.value]
|
||||
|
||||
def write(self, data):
|
||||
written = wintypes.DWORD()
|
||||
if not _k32.WriteFile(self.handle, data, len(data), ctypes.byref(written), None):
|
||||
_fail("write failed")
|
||||
if written.value != len(data):
|
||||
raise Error(f"short write: {written.value} of {len(data)} bytes")
|
||||
|
||||
def flush_input(self):
|
||||
if not _k32.PurgeComm(self.handle, _PURGE_RXCLEAR):
|
||||
_fail("cannot flush the input buffer")
|
||||
|
||||
def read_available(self, wait):
|
||||
"""Everything that arrives within `wait` seconds of quiet start."""
|
||||
# A zero total means *no* timeout to the driver, so never round
|
||||
# down to it — the same trap on the deadline below.
|
||||
self._deadline(_GAP_MS, max(1, round(wait * 1000)))
|
||||
return self._read(4096)
|
||||
|
||||
def read_exact(self, count, timeout):
|
||||
data = b""
|
||||
deadline = time.monotonic() + timeout
|
||||
while len(data) < count:
|
||||
remaining = deadline - time.monotonic()
|
||||
if remaining <= 0:
|
||||
raise Error(f"timeout: got {len(data)} of {count} bytes")
|
||||
# No interval timeout here: only the count or the deadline
|
||||
# ends the read, so a gap mid-reply is simply waited out.
|
||||
self._deadline(0, max(1, round(remaining * 1000)))
|
||||
data += self._read(count - len(data))
|
||||
return data
|
||||
|
||||
|
||||
Port = WindowsPort if os.name == "nt" else PosixPort
|
||||
|
||||
|
||||
# -------------------------------------------------------------- protocol ---
|
||||
|
||||
|
||||
@@ -637,7 +808,7 @@ def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="pureboot host tool", epilog="operations run in the order listed above"
|
||||
)
|
||||
parser.add_argument("--port", required=True, help="serial device (or simavr pty)")
|
||||
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("--info", action="store_true", help="print the device info block")
|
||||
|
||||
Reference in New Issue
Block a user