Flipdot-display/tools/nano_net_upload.py

#!/usr/bin/env python3
"""Flash the Arduino Nano over WiFi, through the ESP32's flash bridge.

Usage:  python nano_net_upload.py firmware.hex HOST[:PORT] [--baud auto|57600|115200]

--baud is the Nano bootloader's baud rate:
  57600   old Nano bootloader (ATmegaBOOT)
  115200  Optiboot ("new bootloader"); needs the ESP32 reset wire
  auto    try 57600 first, then 115200 (default)

The ESP32 restarts the Nano when we connect (so its bootloader starts) and then passes
bytes between this TCP connection and the Nano's serial port. This script talks the
STK500v1 protocol of the Arduino bootloaders directly. avrdude can't be used here: its
network ports are unreliable on Windows.

Normally run by PlatformIO: pio run -e nano_ota -t upload
"""

import argparse
import socket
import sys
import time

DEFAULT_PORT = 2323
PAGE_SIZE = 128  # ATmega328P flash page in bytes
SIGNATURE = b"\x1e\x95\x0f"  # ATmega328P

# Bootloaders: baud rate -> (name, flash space left for the firmware)
BOOTLOADERS = {
    57600: ("old bootloader", 32768 - 2048),
    115200: ("Optiboot", 32768 - 512),
}
AUTO_ORDER = [57600, 115200]

# STK500v1
STK_OK = 0x10
STK_INSYNC = 0x14
CRC_EOP = 0x20
GET_SYNC = 0x30
ENTER_PROGMODE = 0x50
LEAVE_PROGMODE = 0x51
LOAD_ADDRESS = 0x55
PROG_PAGE = 0x64
READ_PAGE = 0x74
READ_SIGN = 0x75

BRIDGE_TIMEOUT = 5.0  # wait for the ESP32's OK (it power cycles the Nano first)
SYNC_TIMEOUT = 4.0  # wait for the bootloader to answer after the restart
REPLY_TIMEOUT = 1.0


class UploadError(Exception):
    pass


class NoBootloader(UploadError):
    """The bootloader didn't answer (wrong baud rate, missing wire, ...)."""


def read_hex(path):
    """Parse an Intel HEX file into a bytearray (gaps filled with 0xFF), padded to whole pages."""
    image = bytearray()
    base = 0
    with open(path) as f:
        for line_no, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
            if not line.startswith(":"):
                raise UploadError(f"{path}:{line_no}: not an Intel HEX line")
            raw = bytes.fromhex(line[1:])
            if sum(raw) & 0xFF:
                raise UploadError(f"{path}:{line_no}: checksum error")
            count, addr, rtype, data = raw[0], (raw[1] << 8) | raw[2], raw[3], raw[4:4 + raw[0]]
            if len(data) != count:
                raise UploadError(f"{path}:{line_no}: length mismatch")
            if rtype == 0x00:  # data
                start = base + addr
                if start + count > len(image):
                    image.extend(b"\xff" * (start + count - len(image)))
                image[start:start + count] = data
            elif rtype == 0x01:  # end of file
                break
            elif rtype == 0x02:  # extended segment address
                base = ((data[0] << 8) | data[1]) << 4
            elif rtype == 0x04:  # extended linear address
                base = ((data[0] << 8) | data[1]) << 16
    if len(image) % PAGE_SIZE:
        image.extend(b"\xff" * (PAGE_SIZE - len(image) % PAGE_SIZE))
    return image


class Bootloader:
    def __init__(self, sock):
        self.sock = sock

    def _recv(self, n, timeout=REPLY_TIMEOUT):
        data = b""
        deadline = time.monotonic() + timeout
        while len(data) < n:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                break
            self.sock.settimeout(remaining)
            try:
                chunk = self.sock.recv(n - len(data))
            except socket.timeout:
                break
            if not chunk:
                raise UploadError("connection closed by the ESP32")
            data += chunk
        return data

    def _drain(self):
        """Throw away everything that is still arriving (e.g. answers to earlier syncs)."""
        while self._recv(256, timeout=0.1):
            pass

    def command(self, payload, reply_len=0):
        """Send a command and return its reply data (between INSYNC and OK)."""
        self.sock.sendall(bytes(payload) + bytes([CRC_EOP]))
        reply = self._recv(reply_len + 2)
        if len(reply) != reply_len + 2 or reply[0] != STK_INSYNC or reply[-1] != STK_OK:
            raise UploadError(f"bad reply to command 0x{payload[0]:02x}: {reply.hex() or 'nothing'}")
        return reply[1:-1]

    def sync(self):
        """Keep sending GET_SYNC until the bootloader answers (it starts after the restart)."""
        deadline = time.monotonic() + SYNC_TIMEOUT
        while time.monotonic() < deadline:
            self.sock.sendall(bytes([GET_SYNC, CRC_EOP]))
            if self._recv(2, timeout=0.2) == bytes([STK_INSYNC, STK_OK]):
                # Answers to earlier syncs may still be on their way: clear them, then
                # sync once more so the replies line up with our commands.
                self._drain()
                self.command([GET_SYNC])
                return
        raise NoBootloader(f"no answer from the Nano bootloader within {SYNC_TIMEOUT:.0f} s")

    def load_address(self, byte_addr):
        word = byte_addr // 2
        self.command([LOAD_ADDRESS, word & 0xFF, word >> 8])

    def write_page(self, byte_addr, data):
        self.load_address(byte_addr)
        self.command([PROG_PAGE, len(data) >> 8, len(data) & 0xFF, ord("F")] + list(data))

    def read_page(self, byte_addr, length):
        self.load_address(byte_addr)
        return self.command([READ_PAGE, length >> 8, length & 0xFF, ord("F")], reply_len=length)


def progress(label, done, total, step=PAGE_SIZE):
    """Print a line every 10% (PlatformIO shows each update on its own line)."""
    percent = 100 * done // total
    previous = 100 * (done - step) // total
    if done == step or done == total or percent // 10 != previous // 10:
        print(f"{label} {percent:3d}%  ({done}/{total} bytes)", flush=True)


def connect(host, port, baud):
    """Connect to the ESP32 bridge and have it restart the Nano into its bootloader."""
    try:
        sock = socket.create_connection((host, port), timeout=10)
    except OSError as e:
        raise UploadError(f"can't connect to {host}:{port}: {e}")
    sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

    sock.sendall(f"FLASH {baud}\n".encode())
    reply = b""
    deadline = time.monotonic() + BRIDGE_TIMEOUT
    while not reply.endswith(b"\n") and time.monotonic() < deadline:
        sock.settimeout(max(deadline - time.monotonic(), 0.01))
        try:
            chunk = sock.recv(64)
        except socket.timeout:
            break
        if not chunk:
            break
        reply += chunk
    if reply != b"OK\n":
        sock.close()
        raise UploadError(
            f"the ESP32 flash bridge didn't accept the request (reply: {reply!r}). "
            "Is the ESP32 running up-to-date firmware?")
    return sock


def flash(sock, image, baud):
    name, flash_size = BOOTLOADERS[baud]
    if len(image) > flash_size:
        raise UploadError(f"firmware is {len(image)} bytes, but with the {name} the Nano has room for {flash_size}")

    boot = Bootloader(sock)
    signature = boot.command([READ_SIGN], reply_len=3)
    if signature != SIGNATURE:
        raise UploadError(f"unexpected device signature {signature.hex()} (expected ATmega328P)")
    print(f"Found ATmega328P, bootloader: {name} ({baud} baud)")

    boot.command([ENTER_PROGMODE])

    for addr in range(0, len(image), PAGE_SIZE):
        boot.write_page(addr, image[addr:addr + PAGE_SIZE])
        progress("Writing  ", addr + PAGE_SIZE, len(image))

    for addr in range(0, len(image), PAGE_SIZE):
        expected = image[addr:addr + PAGE_SIZE]
        actual = boot.read_page(addr, PAGE_SIZE)
        if actual != expected:
            for i, (a, b) in enumerate(zip(actual, expected)):
                if a != b:
                    raise UploadError(f"verify failed at 0x{addr + i:04x}: read 0x{a:02x}, expected 0x{b:02x}")
        progress("Verifying", addr + PAGE_SIZE, len(image))

    boot.command([LEAVE_PROGMODE])


def upload(hex_path, host, port, bauds):
    image = read_hex(hex_path)
    print(f"Firmware: {hex_path} ({len(image)} bytes)")

    for i, baud in enumerate(bauds):
        print(f"Connecting to the ESP32 flash bridge at {host}:{port}, "
              f"restarting the Nano into its bootloader at {baud} baud ...", flush=True)
        with connect(host, port, baud) as sock:
            try:
                Bootloader(sock).sync()
            except NoBootloader:
                if i + 1 < len(bauds):
                    print(f"No answer at {baud} baud, trying {bauds[i + 1]} ...", flush=True)
                    continue
                raise NoBootloader(
                    f"no answer from the Nano bootloader at {', '.join(map(str, bauds))} baud. "
                    "Check the Nano TX -> ESP32 GPIO 16 wire. A Nano with Optiboot also needs "
                    "the reset wire (NANO_RESET_PIN in src/esp32/config.h).")
            flash(sock, image, baud)
            print("Done. The ESP32 restarts the Nano with the new firmware.")
            return


def main():
    parser = argparse.ArgumentParser(description="Flash the Nano over WiFi through the ESP32.")
    parser.add_argument("hex", help="firmware .hex file")
    parser.add_argument("target", help=f"ESP32 address, HOST[:PORT] (default port {DEFAULT_PORT})")
    parser.add_argument("--baud", default="auto", choices=["auto"] + [str(b) for b in BOOTLOADERS],
                        help="bootloader baud rate (default: auto)")
    args = parser.parse_args()

    host, _, port = args.target.partition(":")
    bauds = AUTO_ORDER if args.baud == "auto" else [int(args.baud)]
    try:
        upload(args.hex, host, int(port) if port else DEFAULT_PORT, bauds)
    except UploadError as e:
        sys.stdout.flush()
        print(f"ERROR: {e}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())