Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
117 changes: 117 additions & 0 deletions cli/src/mpftp/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -2313,6 +2313,102 @@ def format_tree(tree: dict, port: str, board: str, variant: str, interpreter: st
return None


# MicroPython's own USB CDC on a native-USB ESP32 (TinyUSB, Espressif's VID),
# and the ports the ROM shows in download mode: USB-Serial/JTAG on the S3 and
# later chips, USB-OTG on the S2. The CDC ignores DTR and RTS, so esptool's
# reset can't reach the ROM through it.
_ESP_TINYUSB_CDC = (0x303A, 0x4001)
_ESP_ROM_PORTS = {(0x303A, 0x1001), (0x303A, 0x0002)}
_ROM_PORT_WAIT_S = 20.0


def _usb_id(port: dict) -> tuple[Any, Any]:
return port.get("vid"), port.get("pid")


def _chip_mac(serial: Any) -> str:
"""The MAC in a USB serial number, as 12 hex digits ("" when there is none).

MicroPython's CDC reports it as ``C04E3003F6300000`` and the ROM as
``C0:4E:30:03:F6:30``; both start with the chip's MAC.
"""
digits = re.sub(r"[^0-9A-Fa-f]", "", str(serial or "")).upper()
return digits[:12] if len(digits) >= 12 else ""


def _wants_rom_port(ns: argparse.Namespace) -> bool:
"""True when this flash should first send the board to its ROM loader."""
if getattr(ns, "port", "") != "esp32" or getattr(ns, "uf2", False):
return False
if (getattr(ns, "before", "") or "") == "no-reset":
return False # the caller says the board is already in its ROM
device = ns.device or ""
return "://" not in device


def _find_rom_port(ports: list[dict], mac: str, before: set[str]) -> Optional[str]:
"""The ROM download port for the chip with ``mac``.

A port whose serial number carries no MAC (the S2's ROM) can only be
told by being new.
"""
for p in ports:
if _usb_id(p) not in _ESP_ROM_PORTS:
continue
rom_mac = _chip_mac(p.get("serial_number"))
if mac and rom_mac == mac:
return p["device"]
if not (mac and rom_mac) and p["device"] not in before:
return p["device"]
return None


def _enter_esp_rom(device: str, baud: int) -> str:
"""Return the port to flash: ``device``, or the ROM port it reboots onto.

A board on a UART bridge, or already in its ROM, flashes where it is. One
running MicroPython on its native USB is sent to its bootloader the way
``mpftp bootloader`` does it, and then flashed on the ROM's own port,
picked by the MAC in its USB serial number.
"""
client, mode = get_client()
try:
ports = client.call("list_ports") or []
cur = next((p for p in ports if p.get("device") == device), None)
if cur is None or _usb_id(cur) != _ESP_TINYUSB_CDC:
return device
mac = _chip_mac(cur.get("serial_number"))
before = {p.get("device") for p in ports}
print(
f"[mpftp] {device} is MicroPython's native USB, which esptool can't "
"reset: sending the board to its bootloader",
file=sys.stderr,
)
ensure_device(client, device, baud)
res = client.call("bootloader")
if not res.get("ok"):
_die(
f"could not send {device} to its bootloader: {res.get('error')}. "
"Hold BOOT while tapping RESET, then flash the port it comes back on."
)
deadline = time.monotonic() + _ROM_PORT_WAIT_S
while time.monotonic() < deadline:
time.sleep(0.7)
rom = _find_rom_port(client.call("list_ports") or [], mac, before)
if rom:
print(f"[mpftp] flashing on the ROM download port {rom}", file=sys.stderr)
return rom
_die(
f"{device} went to its bootloader, but no ROM download port "
f"appeared within {_ROM_PORT_WAIT_S:.0f} s. Hold BOOT "
"while tapping RESET, then flash the port it comes back on."
)
finally:
if mode.startswith("sidecar"):
client.close()
return device # not reached: _die raises


def cmd_firmware(ns: argparse.Namespace) -> None:
sub = ns.fw_cmd
if sub == "list":
Expand Down Expand Up @@ -2392,6 +2488,12 @@ def cmd_firmware(ns: argparse.Namespace) -> None:
extra.append("--uf2")
if getattr(ns, "uf2_timeout", 0):
extra += ["--uf2-timeout", str(ns.uf2_timeout)]
for name in ("before", "after"):
if getattr(ns, name, ""):
extra += [f"--{name}", getattr(ns, name)]
if ns.device and _wants_rom_port(ns):
rom = _enter_esp_rom(ns.device, ns.baud)
extra[extra.index("--device") + 1] = rom
res = _engine_stream("flash", extra)
out(res)
if not res.get("ok"):
Expand Down Expand Up @@ -3111,6 +3213,21 @@ def build_parser() -> argparse.ArgumentParser:
help="Copy a .uf2 to a bootloader volume instead of flashing over serial")
fwf.add_argument("--uf2-timeout", dest="uf2_timeout", type=float, default=0.0,
help="Seconds to wait for the volume to unmount (default 30)")
fwf.add_argument(
"--before",
default="",
type=lambda s: str(s).replace("_", "-"),
choices=["", "default-reset", "no-reset", "usb-reset"],
help="esp32: esptool's reset before flashing (default default-reset; no-reset "
"for a board already in its ROM loader)",
)
fwf.add_argument(
"--after",
default="",
type=lambda s: str(s).replace("_", "-"),
choices=["", "hard-reset", "soft-reset", "no-reset", "watchdog-reset"],
help="esp32: esptool's reset after flashing (default hard-reset)",
)
fwf.set_defaults(func=cmd_firmware)

fwdt = fwsub.add_parser(
Expand Down
147 changes: 147 additions & 0 deletions cli/tests/test_flash_native_usb.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,147 @@
# SPDX-FileCopyrightText: 2026 Brad Barnett
#
# SPDX-License-Identifier: MIT

"""`firmware flash` reaches a native-USB ESP32 running MicroPython (mpftp#208),
and passes --before and --after through to the engine (mpftp#209).

MicroPython's TinyUSB CDC (303A:4001) ignores DTR and RTS, so esptool's reset
keeps talking to the running interpreter. The flash has to send the board to
its bootloader and flash the ROM's port (303A:1001) it comes back on.
"""

from __future__ import annotations

import io
import unittest
from contextlib import redirect_stderr
from unittest import mock

from mpftp import cli

CDC = {"device": "COM104", "vid": 0x303A, "pid": 0x4001, "serial_number": "C04E3003F6300000"}
OTHER_CDC = {"device": "COM105", "vid": 0x303A, "pid": 0x4001, "serial_number": "70041DAD86C40000"}
ROM = {"device": "COM103", "vid": 0x303A, "pid": 0x1001, "serial_number": "C0:4E:30:03:F6:30"}
OTHER_ROM = {"device": "COM102", "vid": 0x303A, "pid": 0x1001, "serial_number": "70:04:1D:AD:86:C4"}
S2_ROM = {"device": "COM29", "vid": 0x303A, "pid": 0x0002, "serial_number": "0"}
UART = {"device": "COM7", "vid": 0x10C4, "pid": 0xEA60, "serial_number": "0001"}


class FakeClient:
"""Answers list_ports from a script of port lists, and records calls."""

def __init__(self, port_lists, bootloader_ok=True):
self.port_lists = list(port_lists)
self.calls = []
self.bootloader_ok = bootloader_ok

def call(self, method, params=None):
self.calls.append(method)
if method == "list_ports":
return self.port_lists.pop(0) if len(self.port_lists) > 1 else self.port_lists[0]
if method == "connect":
return {"ok": True}
if method == "bootloader":
return {"ok": self.bootloader_ok, "error": None if self.bootloader_ok else "no REPL"}
raise AssertionError(method)

def close(self):
pass


def run_flash(argv, client, engine_result=None):
"""Parse ``argv`` with the real parser, run it, return the engine's args."""
ns = cli.build_parser().parse_args(argv)
seen = {}

def engine(cmd, extra):
seen["cmd"], seen["extra"] = cmd, extra
return engine_result or {"ok": True}

with mock.patch.object(cli, "get_client", return_value=(client, "sidecar")), \
mock.patch.object(cli, "_engine_stream", side_effect=engine), \
mock.patch.object(cli, "_sel_args", return_value=["--port", ns.port]), \
mock.patch.object(cli, "boards"), \
mock.patch.object(cli.time, "sleep"), \
mock.patch.object(cli, "out"), \
redirect_stderr(io.StringIO()):
cli.cmd_firmware(ns)
return seen["extra"]


BASE = ["firmware", "flash", "--port", "esp32", "--artifact", "fw.bin", "--offset", "0x10000"]


class NativeUsbFlash(unittest.TestCase):
def test_native_usb_board_is_flashed_on_its_rom_port(self):
client = FakeClient([[CDC, OTHER_CDC, OTHER_ROM], [OTHER_CDC, OTHER_ROM], [OTHER_CDC, OTHER_ROM, ROM]])
extra = run_flash(BASE + ["-d", "COM104"], client)
self.assertIn("bootloader", client.calls)
self.assertEqual(extra[extra.index("--device") + 1], "COM103")

def test_rom_port_is_matched_by_mac_not_by_being_new(self):
# Another board's ROM port appears first; it is not this board.
client = FakeClient([[CDC], [OTHER_ROM], [OTHER_ROM, ROM]])
extra = run_flash(BASE + ["-d", "COM104"], client)
self.assertEqual(extra[extra.index("--device") + 1], "COM103")

def test_an_s2_rom_port_with_no_mac_is_taken_when_new(self):
client = FakeClient([[CDC, OTHER_ROM], [OTHER_ROM, S2_ROM]])
extra = run_flash(BASE + ["-d", "COM104"], client)
self.assertEqual(extra[extra.index("--device") + 1], "COM29")

def test_uart_bridge_board_is_flashed_where_it_is(self):
client = FakeClient([[UART]])
extra = run_flash(BASE + ["-d", "COM7"], client)
self.assertNotIn("bootloader", client.calls)
self.assertEqual(extra[extra.index("--device") + 1], "COM7")

def test_board_already_in_rom_is_flashed_where_it_is(self):
client = FakeClient([[ROM]])
extra = run_flash(BASE + ["-d", "COM103"], client)
self.assertNotIn("bootloader", client.calls)
self.assertEqual(extra[extra.index("--device") + 1], "COM103")

def test_no_reset_skips_the_bootloader_step(self):
client = FakeClient([[CDC]])
extra = run_flash(BASE + ["-d", "COM104", "--before", "no-reset"], client)
self.assertEqual(client.calls, [])
self.assertEqual(extra[extra.index("--device") + 1], "COM104")

def test_no_rom_port_fails_without_flashing_the_dead_port(self):
client = FakeClient([[CDC], [OTHER_ROM]])
with mock.patch.object(cli, "_ROM_PORT_WAIT_S", 0.0), \
mock.patch.object(cli.time, "monotonic", side_effect=[0.0, 1.0, 2.0]), \
self.assertRaises(SystemExit):
run_flash(BASE + ["-d", "COM104"], client)

def test_refused_bootloader_fails(self):
client = FakeClient([[CDC]], bootloader_ok=False)
with self.assertRaises(SystemExit):
run_flash(BASE + ["-d", "COM104"], client)


class ResetModes(unittest.TestCase):
def test_before_and_after_reach_the_engine(self):
client = FakeClient([[ROM]])
extra = run_flash(BASE + ["-d", "COM103", "--before", "no_reset", "--after", "watchdog-reset"], client)
self.assertEqual(extra[extra.index("--before") + 1], "no-reset")
self.assertEqual(extra[extra.index("--after") + 1], "watchdog-reset")

def test_engine_accepts_what_the_cli_passes(self):
from mpftp import firmware

ns = firmware.build_parser().parse_args(
["flash", "--port", "esp32", "--before", "no-reset", "--after", "watchdog-reset"]
)
self.assertEqual((ns.before, ns.after), ("no-reset", "watchdog-reset"))


class ChipMac(unittest.TestCase):
def test_both_serial_forms_give_the_same_mac(self):
self.assertEqual(cli._chip_mac(CDC["serial_number"]), cli._chip_mac(ROM["serial_number"]))
self.assertEqual(cli._chip_mac(None), "")


if __name__ == "__main__":
unittest.main()
11 changes: 8 additions & 3 deletions docs/guides/download-and-flash-firmware.md
Original file line number Diff line number Diff line change
Expand Up @@ -44,9 +44,14 @@ ESP32 and S2, 0x2000 on the P4 and C5, 0x0 on the rest. mpftp lets go of the
board first, shows progress, and reconnects when the board restarts. Tick
**Erase all of flash first** (`--erase`) for a clean board; it deletes the
files on it. A board whose REPL is its own USB (an S2 or S3 on native USB) is
put into its bootloader first. If no download port appears, hold BOOT, tap
RESET, choose the new port and flash again; [Recover a board](recover-a-board.md)
covers the stubborn cases.
put into its bootloader first, and flashed on the ROM download port it comes
back on, found by the MAC in its USB serial number. If no download port
appears, hold BOOT, tap RESET, choose the new port and flash again;
[Recover a board](recover-a-board.md) covers the stubborn cases.

`--before` and `--after` choose esptool's reset on either side of the write:
`--before no-reset` for a board already in its ROM loader, or `--after
watchdog-reset` for one that a plain reset leaves in download mode.

Flashing without a rebuild to another board is `mpftp firmware flash -d OTHER_PORT`
after a build.
Expand Down
Loading