diff --git a/components/ota/README.md b/components/ota/README.md index 2eff367d29..52af2e6553 100644 --- a/components/ota/README.md +++ b/components/ota/README.md @@ -94,15 +94,38 @@ is the routing id (OTA is **module 0**). OTA layers its message types on it. rejects and resynchronizes past oversized or corrupt frames, so buffering stays bounded - host → device (requests): `0x01 BEGIN(u32 image_size)`, `0x02 DATA(bytes)`, - `0x03 FINISH`, `0x04 ABORT`; device → host (replies, reply flag set): + `0x03 FINISH`, `0x04 ABORT`, `0x08 GET_STATUS`, `0x09 MARK_VALID`, + `0x0A MARK_INVALID`; device → host (replies, reply flag set): `0x05 OK(u32 bytes_received)`, `0x06 ERROR(u32 code + utf8 message)`, - `0x07 PROGRESS(u32 written, u32 total)` -- transactions are serialized: the host waits for `OK` / `ERROR` before the - next frame + `0x07 PROGRESS(u32 written, u32 total)`, `0x0B STATUS(u8 flags + running app + version + project name)` (flags bit0 = pending-verify, bit1 = rollback-supported; + each string is u8-length-prefixed) +- transactions are serialized: the host waits for `OK` / `ERROR` (or `STATUS`) + before the next frame +- **rollback is host-driven** (see below): after an OTA the new image boots + *pending verify*, and the **host** confirms it with `MARK_VALID` once it has + checked the device is healthy — the running app must not confirm itself, or a + broken build could mark itself valid before failing. `MARK_INVALID` rolls back + to the previous image and reboots; `GET_STATUS` reports whether the running + image is still pending verify. The [espp OTA Console](https://esp-cpp.github.io/espp/apps/ota_console.html) (`web/ota_console.html`) implements this protocol over WebUSB in the browser. +### Command line: build → OTA + +The [`python/espp_ota`](python/) tool speaks the same protocol from a terminal. +Because this component ships a `project_include.cmake`, any project using it gets +a build-and-flash-over-USB target — the OTA counterpart to `idf.py flash`: + +```sh +pip install pyusb # once (needs a libusb backend) +idf.py ota-usb # builds the app, then OTAs it over USB +``` + +Or drive it directly: `python -m espp_ota flash build/.bin` (see +[`python/README.md`](python/README.md)). + ## Rollback With `CONFIG_BOOTLOADER_APP_ROLLBACK_ENABLE=y`, a freshly-installed app boots diff --git a/components/ota/example/main/ota_example.cpp b/components/ota/example/main/ota_example.cpp index fea3bae584..922f56fc02 100644 --- a/components/ota/example/main/ota_example.cpp +++ b/components/ota/example/main/ota_example.cpp @@ -348,6 +348,34 @@ extern "C" void app_main(void) { reply_error(ec, "abort failed"); break; } + case proto::MessageType::GetStatus: { + // Report rollback status + the running firmware (so the host can show what + // is now running before confirming it). Session-independent (no BEGIN). + uint8_t flags = 0; +#if defined(CONFIG_BOOTLOADER_APP_ROLLBACK_ENABLE) + flags |= proto::kStatusRollbackSupported; + if (ota.is_pending_verify()) + flags |= proto::kStatusPendingVerify; +#endif + const auto desc = ota.running_app_description(); + usb.write_vendor(proto::make_status(flags, desc.version, desc.project_name)); + break; + } + case proto::MessageType::MarkValid: + // The HOST confirms the running image after its own health checks — the app + // must not confirm itself. Cancels the pending rollback. + if (ota.mark_app_valid(ec)) + usb.write_vendor(proto::make_ok(0)); + else + reply_error(ec, "mark valid failed"); + break; + case proto::MessageType::MarkInvalid: + // Reject the running image: roll back to the previous app and reboot. This + // does not return on success (the device reboots), so reply first. + usb.write_vendor(proto::make_ok(0)); + if (!ota.mark_app_invalid_and_rollback(ec)) + reply_error(ec, "rollback failed"); // only reached if rollback failed + break; default: reply_error(std::make_error_code(std::errc::not_supported), "unknown message type"); break; diff --git a/components/ota/include/detail/ota_stream_protocol.hpp b/components/ota/include/detail/ota_stream_protocol.hpp index 6cbfa505ab..116b0d5353 100644 --- a/components/ota/include/detail/ota_stream_protocol.hpp +++ b/components/ota/include/detail/ota_stream_protocol.hpp @@ -25,15 +25,26 @@ // message. // // Message types & payloads (host -> device, requests): -// 0x01 BEGIN — payload: u32 image_size (0 = unknown / streaming). -// 0x02 DATA — payload: raw image bytes (1..kMaxPayloadSize per frame). -// 0x03 FINISH — no payload. Validates + activates the received image. -// 0x04 ABORT — no payload. Discards the in-progress session. +// 0x01 BEGIN — payload: u32 image_size (0 = unknown / streaming). +// 0x02 DATA — payload: raw image bytes (1..kMaxPayloadSize per frame). +// 0x03 FINISH — no payload. Validates + activates the received image. +// 0x04 ABORT — no payload. Discards the in-progress session. +// 0x08 GET_STATUS — no payload. Asks for a STATUS reply (rollback state). +// 0x09 MARK_VALID — no payload. Confirms the running image (cancel rollback). +// 0x0A MARK_INVALID— no payload. Rolls back to the previous image + reboots. // // Message types & payloads (device -> host, replies; reply flag set): // 0x05 OK — payload: u32 bytes_received so far. // 0x06 ERROR — payload: u32 code followed by a UTF-8 message. // 0x07 PROGRESS — payload: u32 written, u32 total (0 if unknown). Optional. +// 0x0B STATUS — payload: u8 flags (bit0 pending_verify, bit1 rollback_supported), +// then the running app's version and project name (each a +// u8-length-prefixed UTF-8 string). +// +// Rollback (bootloader rollback support): a freshly flashed image boots "pending +// verify" and rolls back on the next reset unless confirmed. The device app must +// NOT confirm itself; the HOST confirms it (MARK_VALID) after its own health +// checks — a broken build could otherwise mark itself valid before failing. // // Flow control: the host serializes transactions — it sends one frame and waits // for the matching OK / ERROR reply before sending the next — so the device @@ -86,11 +97,26 @@ enum class MessageType : uint8_t { Ok = 0x05, ///< device -> host: success reply (payload: u32 bytes_received so far) Error = 0x06, ///< device -> host: failure reply (payload: u32 code + utf8 message) Progress = 0x07, ///< device -> host: optional progress (payload: u32 written, u32 total) + // Rollback control (bootloader rollback support). After an OTA the new image + // boots "pending verify" and rolls back on the next reset unless confirmed. + // The device app must NOT confirm itself (a broken app could still do so before + // failing); the HOST confirms it once it has verified the device is healthy. + GetStatus = 0x08, ///< host -> device: query rollback status (no payload) -> Status reply + MarkValid = 0x09, ///< host -> device: confirm the running image (cancel rollback), no payload + MarkInvalid = 0x0A, ///< host -> device: reject the running image (roll back + reboot), no payload + Status = 0x0B, ///< device -> host reply: u8 flags (bit0 pending_verify, bit1 rollback_supported) +}; + +/// Status-reply flag bits (MessageType::Status payload byte 0). +enum StatusFlags : uint8_t { + kStatusPendingVerify = 0x01, ///< running image awaits confirmation (will roll back if not) + kStatusRollbackSupported = 0x02, ///< bootloader rollback support is compiled in }; /// Whether a message type is a device->host reply (sets the frame reply flag). inline bool is_reply(MessageType type) { - return type == MessageType::Ok || type == MessageType::Error || type == MessageType::Progress; + return type == MessageType::Ok || type == MessageType::Error || type == MessageType::Progress || + type == MessageType::Status; } /// @brief Build an encoded OTA frame (typed overload of stream_frame::build_frame). @@ -120,6 +146,34 @@ inline std::vector make_finish() { return build_frame(MessageType::Fini /// Build an ABORT frame (no payload). inline std::vector make_abort() { return build_frame(MessageType::Abort); } +/// Build a GET_STATUS frame (no payload). The device replies with STATUS. +inline std::vector make_get_status() { return build_frame(MessageType::GetStatus); } + +/// Build a MARK_VALID frame (no payload). Confirms the running image. +inline std::vector make_mark_valid() { return build_frame(MessageType::MarkValid); } + +/// Build a MARK_INVALID frame (no payload). Rolls back + reboots the device. +inline std::vector make_mark_invalid() { return build_frame(MessageType::MarkInvalid); } + +/// Append a length-prefixed (u8 length) UTF-8 string, truncated to 255 bytes. +inline void put_str(std::vector &out, std::string_view s) { + const uint8_t len = static_cast(std::min(s.size(), 255)); + out.push_back(len); + out.insert(out.end(), s.begin(), s.begin() + len); +} + +/// Build a STATUS reply: flags (OR of StatusFlags) plus the running app's version +/// and project name (each a u8-length-prefixed string), so the host can report +/// what firmware is now running before confirming it. +inline std::vector make_status(uint8_t flags, std::string_view version = {}, + std::string_view project = {}) { + std::vector p; + p.push_back(flags); + put_str(p, version); + put_str(p, project); + return build_frame(MessageType::Status, p); +} + /// Build an OK reply (bytes_received so far). inline std::vector make_ok(uint32_t bytes_received) { std::vector payload; @@ -187,6 +241,38 @@ inline std::optional parse_progress(const Frame &frame) { return info; } +/// Decoded STATUS reply payload. +struct StatusInfo { + uint8_t flags{}; ///< OR of StatusFlags (pending_verify / rollback_supported) + std::string version; ///< Running app version (may be empty) + std::string project_name; ///< Running app project name (may be empty) + + bool pending_verify() const { return (flags & kStatusPendingVerify) != 0; } + bool rollback_supported() const { return (flags & kStatusRollbackSupported) != 0; } +}; + +/// Parse a STATUS frame payload: [flags u8][version u8-len+bytes][project +/// u8-len+bytes]. Trailing strings are optional (older devices sent flags only); +/// returns std::nullopt only if the payload is empty. +inline std::optional parse_status(const Frame &frame) { + if (frame.payload.empty()) + return std::nullopt; + StatusInfo info{}; + info.flags = frame.payload[0]; + size_t i = 1; + auto read_str = [&](std::string &out) { + if (i >= frame.payload.size()) + return; + const size_t len = frame.payload[i++]; + const size_t n = std::min(len, frame.payload.size() - i); + out.assign(frame.payload.begin() + i, frame.payload.begin() + i + n); + i += n; + }; + read_str(info.version); + read_str(info.project_name); + return info; +} + } // namespace ota_stream } // namespace detail } // namespace espp diff --git a/components/ota/project_include.cmake b/components/ota/project_include.cmake new file mode 100644 index 0000000000..b13e3dfbf9 --- /dev/null +++ b/components/ota/project_include.cmake @@ -0,0 +1,63 @@ +# espp `ota` component — build-system integration for OTA-over-USB. +# +# Included automatically by ESP-IDF (in project scope) for any project that uses +# the `ota` component. It registers an `ota-usb` build target so you can build +# and OTA-flash your app over USB in one step, the same way `idf.py flash` works +# for the serial bootloader: +# +# idf.py ota-usb # builds the app, then OTAs it over USB +# idf.py build ota-usb # equivalent explicit form (also works pre-CMake 3.19) +# +# Device/port overrides are read from the environment by the tool, e.g.: +# ESPP_OTA_PID=0x1234 idf.py ota-usb +# +# The work is done by the pure-Python `espp_ota` tool shipped alongside this file +# (components/ota/python/). It needs `pyusb` at flash time (not at build time): +# pip install pyusb +# +# For full control (a specific serial, chunk size, discovery probe, ...) run the +# tool directly: python -m espp_ota flash build/.bin --help + +if(NOT TARGET ota-usb) + idf_build_get_property(python PYTHON) + set(__espp_ota_pkg_dir "${CMAKE_CURRENT_LIST_DIR}/python") + # CMAKE_PROJECT_NAME is already set here (the real project() runs before + # idf_build_process includes this file); the app .bin lands in the build dir. + set(__espp_ota_bin "${CMAKE_BINARY_DIR}/${CMAKE_PROJECT_NAME}.bin") + + # Prepend our package dir to PYTHONPATH rather than replacing it, so a + # PYTHONPATH the environment already relies on is preserved. Use the host's + # path separator. ($ENV{PYTHONPATH} is the value at configure time, which is + # the same environment `idf.py ota-usb` runs in.) + if(WIN32) + set(__espp_ota_pathsep ";") + else() + set(__espp_ota_pathsep ":") + endif() + set(__espp_ota_pythonpath "${__espp_ota_pkg_dir}") + if(DEFINED ENV{PYTHONPATH} AND NOT "$ENV{PYTHONPATH}" STREQUAL "") + set(__espp_ota_pythonpath "${__espp_ota_pkg_dir}${__espp_ota_pathsep}$ENV{PYTHONPATH}") + endif() + + add_custom_target(ota-usb + COMMAND ${CMAKE_COMMAND} -E env "PYTHONPATH=${__espp_ota_pythonpath}" + ${python} -m espp_ota flash "${__espp_ota_bin}" + WORKING_DIRECTORY "${CMAKE_BINARY_DIR}" + VERBATIM + USES_TERMINAL + COMMENT "OTA-flashing ${__espp_ota_bin} over USB (espp_ota)") + + # `gen_project_binary` (the target that produces the app .bin) is defined + # later in project.cmake, so add the build dependency once this directory + # scope has finished processing. On CMake < 3.19 (no cmake_language(DEFER)) + # the target still works via the explicit `idf.py build ota-usb` form. + function(__espp_ota_link_build_dependency) + if(TARGET gen_project_binary) + add_dependencies(ota-usb gen_project_binary) + endif() + endfunction() + if(CMAKE_VERSION VERSION_GREATER_EQUAL "3.19") + cmake_language(DEFER DIRECTORY "${CMAKE_SOURCE_DIR}" + CALL __espp_ota_link_build_dependency) + endif() +endif() diff --git a/components/ota/python/README.md b/components/ota/python/README.md new file mode 100644 index 0000000000..d2c8c04b9f --- /dev/null +++ b/components/ota/python/README.md @@ -0,0 +1,133 @@ +# espp_ota — OTA over USB from the command line + +A small, pure-Python host tool that updates an espp device over USB using the +espp `stream_frame` framing + OTA stream protocol (dispatcher **module 0**) — the +same protocol the on-device [`ota` example](../example/) serves and +[`ota_console.html`](../web/ota_console.html) drives from the browser. + +It talks to the device's USB **vendor (WebUSB)** interface (`bInterfaceClass +0xFF`, one bulk IN + one bulk OUT endpoint). The frame codec and OTA protocol are +standard-library only; the USB transport uses [`pyusb`](https://pypi.org/project/pyusb/), +imported lazily. + +## Seamless: build → OTA with `idf.py` + +If your project uses the espp `ota` component, its `project_include.cmake` +registers an `ota-usb` build target, so you can build and flash over USB in one +step (just like `idf.py flash` does over the serial bootloader): + +```sh +pip install pyusb # once (libusb backend: `brew install libusb`, `apt install libusb-1.0-0`) +idf.py ota-usb # builds the app, then OTAs it over USB +# or, equivalently / on CMake < 3.19: +idf.py build ota-usb +``` + +Override the target device without editing anything (the tool reads these): + +```sh +ESPP_OTA_PID=0x1234 idf.py ota-usb +``` + +## Standalone CLI + +Run it directly for full control (or when you already have a `.bin`): + +```sh +python -m espp_ota flash build/my_app.bin # BEGIN -> stream -> FINISH +python -m espp_ota flash build/my_app.bin --pid 0x1234 --chunk-size 2048 +python -m espp_ota list # list matching USB devices +python -m espp_ota discover # probe the device's dispatcher +python -m espp_ota status # is the running image pending verify? +python -m espp_ota mark-valid # confirm the running image (cancel rollback) +python -m espp_ota rollback # reject it: roll back + reboot +``` + +Installed with the espp wheel it's also available as the `espp-ota` command +(`pip install "espp[usb]"`, or `"espp[usb-ui]"` to also get the `rich` UI). + +### Rollback: confirm the image from the host + +With bootloader rollback enabled, a freshly flashed image boots **pending verify** +and rolls back on the next reset unless it is confirmed. Confirmation is +deliberately **host-driven** — the running app must not mark *itself* valid, since +a broken build could do so right before crashing. + +`flash` **auto-verifies by default**: after streaming the image it waits for the +device to reboot, reconnects (the device re-enumerates with the same VID/PID), +and reads the running firmware + rollback status. If the new image is *responding* +to OTA commands and still *pending verify*, it has booted — so the host marks it +valid. It prints the before → after firmware, e.g.: + +``` +● Currently running: ota_example 1.0.0 + ... flashing ... +● ota_example 1.0.0 → ota_example 1.1.0 +The new image booted and responded, so it has been marked valid (rollback cancelled). +``` + +Pass `--no-verify` to skip that step (then confirm later with `mark-valid`), or +`--verify-timeout ` to change how long it waits for the device to +reappear. The `status`, `mark-valid`, and `rollback` commands remain available to +do it manually; `rollback` rejects the running image (roll back + reboot). + +## Library use + +```python +from espp_ota import OtaClient, UsbVendorTransport + +with open("build/my_app.bin", "rb") as f: + image = f.read() + +with UsbVendorTransport() as t: # default VID/PID 0x1209:0x0d32 + OtaClient(t, progress=lambda w, tot: print(w, "/", tot)).flash(image) +``` + +## Protocol + +`module = 0`; requests are host→device, replies device→host (reply flag set). +Flow control is one request in flight — each request waits for its OK/ERROR +reply before the next is sent. + +| type | name | dir | payload | +|------|------|-----|---------| +| 0x01 | BEGIN | host→dev | u32 image_size (0 = unknown/streaming) | +| 0x02 | DATA | host→dev | image bytes (1..4096) | +| 0x03 | FINISH | host→dev | — (validate + activate) | +| 0x04 | ABORT | host→dev | — | +| 0x05 | OK | dev→host | u32 bytes_received | +| 0x06 | ERROR | dev→host | u32 code + utf-8 message | +| 0x07 | PROGRESS | dev→host | u32 written, u32 total | + +The wire framing is `espp::stream_frame` v2 (magic `0x4F54`, CRC-32); see +`espp_ota/frame.py`. Host tests (codec + a full OTA against a mock device) live +in `tests/test_ota_host.py` and run with plain `python3`. + +## Output + +The tool draws a [`rich`](https://pypi.org/project/rich/) progress bar (spinner, +bar, %, bytes, transfer speed, ETA) and colorizes status / error lines. Under +`idf.py ota-usb` the tool's stdout/stderr are captured pipes, so the bar is drawn +straight to the controlling terminal (`/dev/tty`, `CONOUT$` on Windows) and still +animates in place. Without a terminal (CI / redirected output) it prints periodic +plain-text lines instead. `rich` is optional — the output degrades to a plain +`\r` bar or text without it. It ships in the ESP-IDF Python environment (so +`idf.py ota-usb` already has it) and is pulled in by `pip install "espp[usb-ui]"`. + +`idf.py ota-usb` mid-flash — the rich bar (%, size, transfer speed, ETA) animates +in place even though idf.py captures the tool's output: + +![espp_ota flashing over USB](https://github.com/user-attachments/assets/a042481a-2964-4b06-9109-bb2dcb4e355b) + +…and on completion: + +![espp_ota OTA complete](https://github.com/user-attachments/assets/da8e1b71-c65f-4ecc-9223-e232f8591ceb) + +## Requirements + +- Python 3.8+ +- `pyusb` + a libusb backend (only for the actual USB transport): + - macOS: `brew install libusb` + - Linux: `apt install libusb-1.0-0` (add a udev rule for non-root access) + - Windows: the device advertises WebUSB + MS-OS-2.0, so WinUSB binds + automatically; otherwise bind it once with [Zadig](https://zadig.akeo.ie/). diff --git a/components/ota/python/espp_ota/__init__.py b/components/ota/python/espp_ota/__init__.py new file mode 100644 index 0000000000..9e8aae9c28 --- /dev/null +++ b/components/ota/python/espp_ota/__init__.py @@ -0,0 +1,35 @@ +"""espp_ota — pure-Python host tool to OTA-update an espp device over USB. + +Speaks the espp ``stream_frame`` framing + OTA stream protocol (dispatcher +module 0) over the device's USB vendor (WebUSB) interface — the same protocol +``components/ota/web/ota_console.html`` implements in the browser and the +``ota`` example serves on-device. + +The codec (:mod:`espp_ota.frame`) and protocol (:mod:`espp_ota.protocol`) are +standard-library only; the USB transport (:mod:`espp_ota.transport`) needs +`pyusb`, imported lazily. + +Typical use:: + + from espp_ota import OtaClient, UsbVendorTransport + with UsbVendorTransport() as t: + OtaClient(t, progress=lambda w, tot: ...).flash(open("app.bin", "rb").read()) +""" + +from .client import OtaClient +from .protocol import ErrorInfo, MessageType, OtaError, ProgressInfo +from .transport import DEFAULT_PID, DEFAULT_VID, TransportError, UsbVendorTransport + +__all__ = [ + "OtaClient", + "UsbVendorTransport", + "TransportError", + "OtaError", + "MessageType", + "ErrorInfo", + "ProgressInfo", + "DEFAULT_VID", + "DEFAULT_PID", +] + +__version__ = "0.1.0" diff --git a/components/ota/python/espp_ota/__main__.py b/components/ota/python/espp_ota/__main__.py new file mode 100644 index 0000000000..dbdd066172 --- /dev/null +++ b/components/ota/python/espp_ota/__main__.py @@ -0,0 +1,6 @@ +import sys + +from .cli import main + +if __name__ == "__main__": + sys.exit(main()) diff --git a/components/ota/python/espp_ota/cli.py b/components/ota/python/espp_ota/cli.py new file mode 100644 index 0000000000..1caf8bcd0f --- /dev/null +++ b/components/ota/python/espp_ota/cli.py @@ -0,0 +1,286 @@ +"""Command-line interface: ``python -m espp_ota ``. + +Commands: + flash BEGIN -> stream DATA -> FINISH an image over USB. + list List matching USB devices. + discover Probe the device (dispatcher ListModules) and report reply. + +VID/PID default to the espp UsbDevice default (0x1209:0x0d32) but can be +overridden (also via the ESPP_OTA_VID / ESPP_OTA_PID env vars, which the CMake +``ota-usb`` target forwards). +""" + +from __future__ import annotations + +import argparse +import os +import sys +import time +from typing import Optional + +from . import __version__, ui +from .client import OtaClient +from .protocol import OtaError +from .transport import DEFAULT_PID, DEFAULT_VID, TransportError, UsbVendorTransport, list_devices + +CON = ui.Console() + + +def _auto_int(text: str) -> int: + return int(text, 0) # accepts 0x1209, 4617, etc. + + +def _human_size(n: int) -> str: + size = float(n) + for unit in ("B", "KiB", "MiB", "GiB"): + if size < 1024 or unit == "GiB": + return f"{int(size)} {unit}" if unit == "B" else f"{size:.1f} {unit}" + size /= 1024 + return f"{n} B" + + +def _env_int(name: str, default: int) -> int: + val = os.environ.get(name) + return _auto_int(val) if val else default + + +def _add_device_args(p: argparse.ArgumentParser) -> None: + p.add_argument("--vid", type=_auto_int, default=_env_int("ESPP_OTA_VID", DEFAULT_VID), + help="USB vendor id (default 0x%04x)" % DEFAULT_VID) + p.add_argument("--pid", type=_auto_int, default=_env_int("ESPP_OTA_PID", DEFAULT_PID), + help="USB product id (default 0x%04x; pass -1 to match any)" % DEFAULT_PID) + p.add_argument("--serial", default=os.environ.get("ESPP_OTA_SERIAL"), + help="match a specific device serial number") + p.add_argument("--interface", type=_auto_int, default=None, + help="force a specific vendor interface number") + + +def _make_transport(args) -> UsbVendorTransport: + """Build an (unopened) transport; use it as a context manager (`with`).""" + pid = None if args.pid is not None and args.pid < 0 else args.pid + return UsbVendorTransport(vid=args.vid, pid=pid, serial=args.serial, + interface=args.interface) + + +def _reconnect(args, timeout_s: float = 20.0): + """Reopen the device after it reboots (same VID/PID, re-enumerates). Retries + until it appears or the timeout elapses; returns the opened transport or None.""" + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + try: + return _make_transport(args) + except TransportError: + time.sleep(0.5) + return None + + +def _cmd_flash(args) -> int: + with open(args.binary, "rb") as fh: + image = fh.read() + if not image: + CON.error("image is empty") + return 2 + size = 0 if args.unknown_size else len(image) + + before = None # firmware running before the flash + with _make_transport(args) as t: + if not args.quiet: + CON.note(f"● Connected to {t.description}") + client = OtaClient(t) + try: + before = client.get_status() + if not args.quiet: + CON.info(f" Currently running: {before.firmware_str()}") + except OtaError: + pass # older device without GET_STATUS; carry on + if not args.quiet: + CON.info(f" Flashing {args.binary} ({_human_size(len(image))})") + start = time.monotonic() + with ui.Progress(len(image), label="Flashing", quiet=args.quiet) as prog: + client = OtaClient( + t, + chunk_size=args.chunk_size, + progress=prog.update, + begin_timeout_ms=args.begin_timeout, + data_timeout_ms=args.data_timeout, + finish_timeout_ms=args.finish_timeout, + ) + client.flash(image, image_size=size) + if not args.quiet: + dt = time.monotonic() - start + rate = len(image) / dt / 1024 if dt else 0 + CON.success(f"OTA complete in {dt:.1f}s ({rate:.0f} KiB/s). " + f"The device is activating the new image and rebooting.") + + if args.no_verify: + CON.info("Auto-verify disabled. Once you have checked the device, confirm the " + "image with `espp-ota mark-valid` (or it rolls back on the next reset).") + return 0 + return _auto_verify(args, before) + + +def _auto_verify(args, before) -> int: + """Reconnect after the reboot and confirm the running image can respond before + marking it valid. This is the safety check: a pending-verify image that can + answer OTA commands has booted, so the host confirms it.""" + CON.info("Waiting for the device to reboot and reconnect…") + time.sleep(2.0) # give it a moment to drop off the bus before we scan + t = _reconnect(args, timeout_s=args.verify_timeout) + if t is None: + CON.warn("device did not reappear after the reboot. If it booted correctly, " + "confirm the image with `espp-ota mark-valid`; otherwise it will roll " + "back on the next reset.") + return 1 + with t: + client = OtaClient(t) + try: + st = client.get_status() + except OtaError as exc: + CON.error(f"reconnected but the device did not report status: {exc}. " + "Not confirming — it will roll back on the next reset.") + return 1 + # Report the transition. + if before is not None: + CON.note(f"● {before.firmware_str()} → {st.firmware_str()}") + else: + CON.note(f"● Now running: {st.firmware_str()}") + if not st.rollback_supported: + CON.success("Update complete (device has no rollback; nothing to confirm).") + return 0 + if not st.pending_verify: + CON.success("Update complete; the running image is already confirmed.") + return 0 + # It responded to OTA commands AND is pending verify -> it booted; confirm. + client.mark_valid() + CON.success("The new image booted and responded, so it has been marked valid " + "(rollback cancelled).") + return 0 + + +def _cmd_list(args) -> int: + pid = None if args.pid is not None and args.pid < 0 else args.pid + found = list_devices(vid=args.vid, pid=pid) + if not found: + CON.warn("no matching USB devices found") + return 1 + for vid, pid_, desc in found: + print(f"0x{vid:04x}:0x{pid_:04x} {desc}") + return 0 + + +def _cmd_discover(args) -> int: + with _make_transport(args) as t: + frames = OtaClient(t).discover(timeout_ms=args.timeout) + if not frames: + CON.warn("no discovery reply (device may not run a Dispatcher on the " + "vendor interface)") + return 1 + for fr in frames: + CON.info(f"reply module=0x{fr.module:02x} type=0x{fr.type:02x} " + f"reply={fr.is_reply} payload={len(fr.payload)} bytes") + return 0 + + +def _cmd_status(args) -> int: + with _make_transport(args) as t: + st = OtaClient(t).get_status() + CON.note(f"● Running: {st.firmware_str()}") + if not st.rollback_supported: + CON.info("rollback: not supported (CONFIG_BOOTLOADER_APP_ROLLBACK_ENABLE off)") + elif st.pending_verify: + CON.warn("running image is PENDING VERIFY — confirm it with `mark-valid` " + "(or it rolls back on the next reset)") + else: + CON.success("running image is confirmed (not pending verify)") + return 0 + + +def _cmd_mark_valid(args) -> int: + with _make_transport(args) as t: + client = OtaClient(t) + try: + fw = client.get_status().firmware_str() + except OtaError: + fw = None + client.mark_valid() + CON.success("running image marked valid; rollback cancelled" + + (f" ({fw})" if fw else "")) + return 0 + + +def _cmd_rollback(args) -> int: + with _make_transport(args) as t: + OtaClient(t).mark_invalid() + CON.success("device rolling back to the previous image and rebooting") + return 0 + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser(prog="espp_ota", description=__doc__.split("\n")[0]) + p.add_argument("--version", action="version", version=f"espp_ota {__version__}") + sub = p.add_subparsers(dest="command", required=True) + + f = sub.add_parser("flash", help="OTA-update a binary over USB") + f.add_argument("binary", help="path to the app .bin to flash") + _add_device_args(f) + f.add_argument("--chunk-size", type=_auto_int, default=4096, + help="DATA payload bytes per frame (1..4096, default 4096)") + f.add_argument("--unknown-size", action="store_true", + help="stream with size 0 (device erases the whole partition)") + f.add_argument("--begin-timeout", type=int, default=60000, help="ms (default 60000)") + f.add_argument("--data-timeout", type=int, default=5000, help="ms (default 5000)") + f.add_argument("--finish-timeout", type=int, default=60000, help="ms (default 60000)") + f.add_argument("-q", "--quiet", action="store_true", help="suppress progress output") + f.add_argument("--no-verify", action="store_true", + help="don't auto-verify: skip the reconnect + mark-valid after reboot") + f.add_argument("--verify-timeout", type=float, default=20.0, + help="seconds to wait for the device to reappear after reboot (default 20)") + f.set_defaults(func=_cmd_flash) + + lst = sub.add_parser("list", help="list matching USB devices") + _add_device_args(lst) + lst.set_defaults(func=_cmd_list) + + d = sub.add_parser("discover", help="probe the device's dispatcher (ListModules)") + _add_device_args(d) + d.add_argument("--timeout", type=int, default=2000, help="ms (default 2000)") + d.set_defaults(func=_cmd_discover) + + st = sub.add_parser("status", help="query rollback status (is the image pending verify?)") + _add_device_args(st) + st.set_defaults(func=_cmd_status) + + mv = sub.add_parser("mark-valid", + help="confirm the running image (cancel rollback) after verifying it") + _add_device_args(mv) + mv.set_defaults(func=_cmd_mark_valid) + + rb = sub.add_parser("rollback", help="reject the running image: roll back + reboot") + _add_device_args(rb) + rb.set_defaults(func=_cmd_rollback) + return p + + +def main(argv: Optional[list] = None) -> int: + args = build_parser().parse_args(argv) + try: + return args.func(args) + except (OtaError, TransportError) as exc: + CON.error(exc) + return 1 + except FileNotFoundError as exc: + CON.error(exc) + return 2 + except OSError as exc: + # pyusb's USBError derives from OSError/IOError, so routine USB failures + # (unplug mid-flash, permission denied, missing libusb backend) land here + # instead of raising an ugly traceback. Report them cleanly. + CON.error(exc) + return 1 + except KeyboardInterrupt: + CON.warn("interrupted") + return 130 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/components/ota/python/espp_ota/client.py b/components/ota/python/espp_ota/client.py new file mode 100644 index 0000000000..b6ee534cde --- /dev/null +++ b/components/ota/python/espp_ota/client.py @@ -0,0 +1,184 @@ +"""The OTA session driver: BEGIN -> DATA* -> FINISH over a byte transport. + +Transport-agnostic: it needs an object with ``write(bytes, timeout_ms)`` and +``read(max_len, timeout_ms) -> bytes`` (``b""`` on timeout), e.g. +:class:`espp_ota.transport.UsbVendorTransport`. Flow control is one request in +flight — each request waits for its OK/ERROR reply before the next is sent — +matching the device and ``ota_console.html``. +""" + +from __future__ import annotations + +import errno +import time +from collections import deque +from typing import Callable, Deque, List, Optional + +from . import frame as _f +from . import protocol as _p +from .protocol import MessageType, OtaError + +ProgressFn = Callable[[int, int], None] # (written, total) -> None + + +class OtaClient: + def __init__( + self, + transport, + chunk_size: int = _f.MAX_PAYLOAD_SIZE, + progress: Optional[ProgressFn] = None, + begin_timeout_ms: int = 60000, + data_timeout_ms: int = 5000, + finish_timeout_ms: int = 60000, + ) -> None: + if not (1 <= chunk_size <= _f.MAX_PAYLOAD_SIZE): + raise ValueError(f"chunk_size must be 1..{_f.MAX_PAYLOAD_SIZE}") + self._t = transport + self._chunk = chunk_size + self._progress = progress + self._begin_to = begin_timeout_ms + self._data_to = data_timeout_ms + self._finish_to = finish_timeout_ms + self._parser = _f.StreamParser() + self._pending: Deque[_f.Frame] = deque() + + # -- reply plumbing ------------------------------------------------------- + def _next_frame(self, deadline: float) -> _f.Frame: + while True: + if self._pending: + return self._pending.popleft() + remaining = deadline - time.monotonic() + if remaining <= 0: + raise OtaError("timed out waiting for a device reply") + data = self._t.read(_f.MAX_FRAME_SIZE, timeout_ms=max(1, int(remaining * 1000))) + if data: + self._pending.extend(self._parser.feed(data)) + + def _transact(self, request: bytes, timeout_ms: int, + want: MessageType = MessageType.OK) -> _f.Frame: + """Send one request and return the matching reply (module 0). + + ``want`` is the success reply type expected (OK by default; STATUS for a + status query). PROGRESS frames are surfaced to the callback and skipped; + an ERROR reply raises :class:`OtaError`; frames for other modules are + ignored.""" + self._t.write(request, timeout_ms=timeout_ms) + deadline = time.monotonic() + timeout_ms / 1000.0 + while True: + fr = self._next_frame(deadline) + # Only device->host replies on module 0 are ours. Skip other modules' + # traffic and any device-originated *request* on module 0 (reply flag + # clear) so a request whose type collides with a reply type can never + # be mistaken for a reply (as the browser probe also enforces). + if fr.module != _p.MODULE or not fr.is_reply: + continue + if fr.type == MessageType.PROGRESS: + info = _p.parse_progress(fr) + if info and self._progress: + self._progress(info.written, info.total) + continue + if fr.type == MessageType.ERROR: + info = _p.parse_error(fr) + if info: + raise OtaError(f"device error: {info.message}", info.code) + raise OtaError("device error (unparseable ERROR reply)") + if fr.type == want: + return fr + raise OtaError(f"unexpected reply type 0x{fr.type:02x}") + + # -- public API ----------------------------------------------------------- + def flash(self, image: bytes, image_size: Optional[int] = None) -> None: + """Run a full OTA: BEGIN(size) -> DATA chunks -> FINISH. + + ``image_size`` defaults to ``len(image)``; pass 0 for an unknown-size + (streaming) session (the device erases the whole partition).""" + if not image: + raise OtaError("empty image") + size = len(image) if image_size is None else image_size + + # The device keeps its OTA session across a host disconnect, so a + # previously interrupted flash can leave it "busy" and reject this run's + # BEGIN. Recover from THAT case only: if BEGIN is rejected specifically + # with device_or_resource_busy (EBUSY), send an ABORT to release the stale + # session and retry BEGIN once. Any other failure (a timeout, a transport + # error) is NOT retried — retrying on the same uncorrelated stream could + # pair a delayed reply with the wrong request and desync the protocol. + try: + self._transact(_p.make_begin(size), self._begin_to) + except OtaError as exc: + if exc.code != errno.EBUSY: + raise + self.abort() # clear a stale session left by a prior interrupted run + self._transact(_p.make_begin(size), self._begin_to) + + # From here on, on any failure (device ERROR, timeout, Ctrl-C, transport + # error) send a best-effort ABORT to release the session before propagating. + try: + total = len(image) + sent = 0 + for off in range(0, total, self._chunk): + chunk = image[off : off + self._chunk] + ok = self._transact(_p.make_data(chunk), self._data_to) + sent += len(chunk) + # OK carries bytes_received; prefer it, fall back to our own count. + received = _p.parse_u32(ok) + if self._progress: + self._progress(received if received is not None else sent, total) + + self._transact(_p.make_finish(), self._finish_to) + except BaseException: + self.abort() # best-effort; swallows its own errors + raise + + def abort(self) -> None: + # Best-effort: called from flash()'s failure path where the transport may + # already be gone, so swallow every error (OtaError, transport, etc.). + try: + self._transact(_p.make_abort(), self._data_to) + except Exception: + pass # best-effort cleanup; the link may already be gone + + # -- rollback control ----------------------------------------------------- + def get_status(self) -> "_p.StatusInfo": + """Query the device's rollback status (a STATUS reply).""" + fr = self._transact(_p.make_get_status(), self._data_to, want=MessageType.STATUS) + info = _p.parse_status(fr) + if info is None: + raise OtaError("unparseable STATUS reply") + return info + + def mark_valid(self) -> None: + """Confirm the running image (cancel the pending rollback). The host does + this after verifying the device is healthy — the app must not confirm + itself.""" + self._transact(_p.make_mark_valid(), self._data_to) + + def mark_invalid(self) -> None: + """Reject the running image: the device rolls back to the previous app and + reboots. The device may reboot before/without replying, so a missing reply + is treated as success.""" + try: + self._transact(_p.make_mark_invalid(), self._data_to) + except OtaError as exc: + # A timeout (no code) is expected — the device rebooted. Re-raise a + # real device ERROR (rollback refused, e.g. no previous app). + if exc.code is not None: + raise + + def discover(self, timeout_ms: int = 2000) -> List[_f.Frame]: + """Send a dispatcher ListModules request; return the matching reply. + + Useful as a connectivity probe before flashing. Reads until the actual + discovery reply arrives (module 0xFF, reply flag set, ListModules type), + ignoring unrelated / device-initiated frames; returns [] on timeout. The + discovery TLV payload is not decoded here.""" + self._t.write(_p.make_discovery_request(), timeout_ms=self._data_to) + deadline = time.monotonic() + timeout_ms / 1000.0 + while True: + try: + fr = self._next_frame(deadline) + except OtaError: + return [] # timed out + if (fr.module == _p.DISCOVERY_MODULE and fr.is_reply + and fr.type == _p.DISCOVERY_LIST_MODULES): + return [fr] diff --git a/components/ota/python/espp_ota/frame.py b/components/ota/python/espp_ota/frame.py new file mode 100644 index 0000000000..6a407e32df --- /dev/null +++ b/components/ota/python/espp_ota/frame.py @@ -0,0 +1,181 @@ +"""espp ``stream_frame`` v2 codec — pure Python, standard library only. + +This mirrors ``components/stream_frame/include/stream_frame.hpp`` so a host tool +can speak the exact same wire protocol the device does, without building the +espp Python bindings. + +Wire format (all multi-byte fields little-endian):: + + [magic u16 = 0x4F54 "OT"][flags u8][module u8][type u8] + {[correlation u16] iff flags bit1}[len u32][payload][crc32 u32] + +``crc32`` is the standard zlib CRC-32 over every byte from the magic through the +payload (i.e. the whole frame except the trailing CRC field). Python's +``zlib.crc32`` is that exact algorithm, so ``crc32(b"123456789") == 0xCBF43926`` +matches the C++ golden value. +""" + +from __future__ import annotations + +import struct +import zlib +from dataclasses import dataclass +from typing import List, Optional, Tuple + +# ---- constants (kept in step with stream_frame.hpp) ------------------------- +MAGIC = 0x4F54 +MAGIC_BYTES = struct.pack(" int: + """Standard zlib/IEEE CRC-32 (matches ``espp::stream_frame::crc32``).""" + return zlib.crc32(data) & 0xFFFFFFFF + + +def make_flags(reply: bool, version: int = VERSION) -> int: + return ((version & 0x0F) << 4) | (FLAG_REPLY if reply else 0) + + +def flags_is_reply(flags: int) -> bool: + return bool(flags & FLAG_REPLY) + + +def flags_version(flags: int) -> int: + return (flags >> 4) & 0x0F + + +def flags_has_correlation(flags: int) -> bool: + return bool(flags & FLAG_CORRELATION) + + +def build_frame( + module: int, + type_: int, + payload: bytes = b"", + reply: bool = False, + correlation: Optional[int] = None, +) -> bytes: + """Encode one frame. Raises ``ValueError`` if the payload is too large.""" + if len(payload) > MAX_PAYLOAD_SIZE: + raise ValueError( + f"payload {len(payload)} bytes exceeds MAX_PAYLOAD_SIZE ({MAX_PAYLOAD_SIZE})" + ) + flags = make_flags(reply) + if correlation is not None: + flags |= FLAG_CORRELATION + out = bytearray(MAGIC_BYTES) + out += bytes((flags & 0xFF, module & 0xFF, type_ & 0xFF)) + if correlation is not None: + out += struct.pack(" bool: + return flags_is_reply(self.flags) + + @property + def version(self) -> int: + return flags_version(self.flags) + + +class StreamParser: + """Incremental, resynchronizing frame parser. + + Mirrors ``espp::stream_frame::StreamParser``: feed arbitrary chunks (USB bulk + transfers may split or coalesce frames) and it yields the complete, CRC-valid + frames they contain, resynchronizing past a bad magic, an oversized length, + or a CRC mismatch by dropping one byte and retrying. + """ + + def __init__(self) -> None: + self._buf = bytearray() + self.dropped_bytes = 0 + + def reset(self) -> None: + self._buf.clear() + self.dropped_bytes = 0 + + def buffered(self) -> int: + return len(self._buf) + + def feed(self, data: bytes) -> List[Frame]: + self._buf += data + frames: List[Frame] = [] + while True: + frame, consumed = self._try_one() + if consumed == 0: + break # need more bytes + del self._buf[:consumed] + if frame is not None: + frames.append(frame) + return frames + + def _try_one(self) -> Tuple[Optional[Frame], int]: + """Return (frame|None, bytes_to_consume). consumed==0 means "need more". + + A non-None frame with consumed>0 is a good frame; a None frame with + consumed==1 is a resync (drop one byte and keep scanning).""" + buf = self._buf + n = len(buf) + # Find the magic. Need at least 2 bytes to check it. + if n < 2: + # Could a single trailing byte be the start of the magic? Keep it. + if n == 1 and buf[0] == MAGIC_BYTES[0]: + return None, 0 + if n == 1: + self.dropped_bytes += 1 + return None, 1 + return None, 0 + if buf[0] != MAGIC_BYTES[0] or buf[1] != MAGIC_BYTES[1]: + self.dropped_bytes += 1 + return None, 1 # resync: drop one byte + # Enough for the fixed header? + if n < HEADER_SIZE: + return None, 0 + flags = buf[2] + module = buf[3] + type_ = buf[4] + offset = 5 + correlation: Optional[int] = None + if flags_has_correlation(flags): + if n < HEADER_SIZE + CORRELATION_SIZE: + return None, 0 + correlation = struct.unpack_from(" MAX_PAYLOAD_SIZE: + self.dropped_bytes += 1 + return None, 1 # bogus length -> resync + total = offset + length + CRC_SIZE + if n < total: + return None, 0 # wait for the rest of the frame + want_crc = struct.unpack_from(" resync + payload = bytes(buf[offset : offset + length]) + return Frame(flags, module, type_, payload, correlation), total diff --git a/components/ota/python/espp_ota/protocol.py b/components/ota/python/espp_ota/protocol.py new file mode 100644 index 0000000000..17d6d9a763 --- /dev/null +++ b/components/ota/python/espp_ota/protocol.py @@ -0,0 +1,171 @@ +"""espp OTA stream protocol (dispatcher module 0). + +Mirrors ``components/ota/include/detail/ota_stream_protocol.hpp``: the OTA +message-type enum, frame builders (``make_*``) and reply parsers (``parse_*``) +layered on the :mod:`espp_ota.frame` codec. + +Requests are host->device (reply flag = 0); replies are device->host +(reply flag = 1). Flow control is one-frame-in-flight: the host sends a request +and waits for the matching OK/ERROR reply before sending the next. +""" + +from __future__ import annotations + +import struct +from dataclasses import dataclass +from enum import IntEnum +from typing import Optional + +from . import frame as _f + +#: OTA occupies dispatcher module id 0. +MODULE = 0 + +#: Discovery meta-module (see components/dispatcher). ListModules == 0x00. +DISCOVERY_MODULE = 0xFF +DISCOVERY_LIST_MODULES = 0x00 + + +class MessageType(IntEnum): + BEGIN = 0x01 # host->device: u32 image_size (0 = unknown / streaming) + DATA = 0x02 # host->device: raw image bytes (1..MAX_PAYLOAD_SIZE) + FINISH = 0x03 # host->device: validate + activate (no payload) + ABORT = 0x04 # host->device: discard the session (no payload) + OK = 0x05 # device->host: u32 bytes_received so far + ERROR = 0x06 # device->host: u32 code + utf-8 message + PROGRESS = 0x07 # device->host: u32 written, u32 total (0 if unknown) + GET_STATUS = 0x08 # host->device: query rollback status (no payload) -> STATUS + MARK_VALID = 0x09 # host->device: confirm the running image (cancel rollback) + MARK_INVALID = 0x0A # host->device: reject the running image (roll back + reboot) + STATUS = 0x0B # device->host: u8 flags (see StatusFlags) + + +class StatusFlags(IntEnum): + PENDING_VERIFY = 0x01 # running image awaits confirmation (rolls back if not) + ROLLBACK_SUPPORTED = 0x02 # bootloader rollback support is compiled in + + +_REPLY_TYPES = {MessageType.OK, MessageType.ERROR, MessageType.PROGRESS, MessageType.STATUS} + + +def _build(type_: MessageType, payload: bytes = b"") -> bytes: + return _f.build_frame(MODULE, int(type_), payload, reply=type_ in _REPLY_TYPES) + + +# ---- request builders (host -> device) -------------------------------------- +def make_begin(image_size: int) -> bytes: + return _build(MessageType.BEGIN, struct.pack(" bytes: + return _build(MessageType.DATA, chunk) + + +def make_finish() -> bytes: + return _build(MessageType.FINISH) + + +def make_abort() -> bytes: + return _build(MessageType.ABORT) + + +def make_get_status() -> bytes: + return _build(MessageType.GET_STATUS) + + +def make_mark_valid() -> bytes: + return _build(MessageType.MARK_VALID) + + +def make_mark_invalid() -> bytes: + return _build(MessageType.MARK_INVALID) + + +def make_discovery_request() -> bytes: + """A dispatcher discovery (ListModules) request on module 0xFF.""" + return _f.build_frame(DISCOVERY_MODULE, DISCOVERY_LIST_MODULES, b"", reply=False) + + +# ---- reply parsers (device -> host) ----------------------------------------- +@dataclass +class ErrorInfo: + code: int + message: str + + +@dataclass +class ProgressInfo: + written: int + total: int # 0 if unknown + + +def parse_u32(fr: _f.Frame) -> Optional[int]: + """The single-u32 payload of a BEGIN echo or an OK (bytes_received).""" + if len(fr.payload) != 4: + return None + return struct.unpack(" Optional[ErrorInfo]: + if len(fr.payload) < 4: + return None + code = struct.unpack_from(" Optional[ProgressInfo]: + if len(fr.payload) != 8: + return None + written, total = struct.unpack(" str: + """A short 'project vX.Y' label for the running firmware.""" + if self.project_name and self.version: + return f"{self.project_name} {self.version}" + return self.project_name or self.version or "(unknown)" + + +def parse_status(fr: _f.Frame) -> Optional[StatusInfo]: + # payload: [flags u8][version u8-len+bytes][project u8-len+bytes]; the strings + # are optional (older devices sent flags only). + if not fr.payload: + return None + flags = fr.payload[0] + i = 1 + + def read_str() -> str: + nonlocal i + if i >= len(fr.payload): + return "" + n = fr.payload[i] + i += 1 + s = fr.payload[i:i + n].decode("utf-8", errors="replace") + i += n + return s + + version = read_str() + project = read_str() + return StatusInfo( + pending_verify=bool(flags & StatusFlags.PENDING_VERIFY), + rollback_supported=bool(flags & StatusFlags.ROLLBACK_SUPPORTED), + version=version, + project_name=project, + ) + + +class OtaError(RuntimeError): + """An ERROR reply, a protocol violation, or a transport failure.""" + + def __init__(self, message: str, code: Optional[int] = None) -> None: + super().__init__(message if code is None else f"{message} (code {code})") + self.code = code diff --git a/components/ota/python/espp_ota/transport.py b/components/ota/python/espp_ota/transport.py new file mode 100644 index 0000000000..9ed003f145 --- /dev/null +++ b/components/ota/python/espp_ota/transport.py @@ -0,0 +1,198 @@ +"""USB vendor-interface transport for the OTA host tool. + +Talks to the device's WebUSB / vendor interface (``bInterfaceClass == 0xFF``, +one bulk IN + one bulk OUT endpoint) — the same interface ``ota_console.html`` +uses from the browser. Uses `pyusb` (libusb); it is imported lazily so the +:mod:`espp_ota.frame` / :mod:`espp_ota.protocol` layers stay stdlib-only. + +Default device id is the espp ``UsbDevice`` default, ``0x1209:0x0d32``; both are +overridable (a project may set its own VID/PID). +""" + +from __future__ import annotations + +from typing import List, Optional, Tuple + +DEFAULT_VID = 0x1209 +DEFAULT_PID = 0x0D32 +VENDOR_CLASS = 0xFF + + +class TransportError(RuntimeError): + pass + + +def _import_usb(): + try: + import usb.core # noqa: F401 + import usb.util # noqa: F401 + except ImportError as exc: # pragma: no cover - environment dependent + raise TransportError( + "pyusb is required for the USB transport. Install it with " + "`pip install pyusb` (and a libusb backend: `brew install libusb` on " + "macOS, `apt install libusb-1.0-0` on Linux; on Windows bind WinUSB " + "with Zadig if the device is not already driverless)." + ) from exc + import usb.core as core + import usb.util as util + + return core, util + + +def list_devices(vid: int = DEFAULT_VID, pid: Optional[int] = None) -> List[Tuple[int, int, str]]: + """Return (vid, pid, description) for candidate devices matching the filter.""" + core, util = _import_usb() + kwargs = {"find_all": True, "idVendor": vid} + if pid is not None: + kwargs["idProduct"] = pid + out: List[Tuple[int, int, str]] = [] + for dev in core.find(**kwargs): + try: + desc = util.get_string(dev, dev.iProduct) or "" + except Exception: + desc = "" + out.append((dev.idVendor, dev.idProduct, desc)) + return out + + +class UsbVendorTransport: + """Open the vendor bulk pipe of an espp device and read/write raw frames. + + Use as a context manager:: + + with UsbVendorTransport() as t: + t.write(frame_bytes) + reply = t.read(4111, timeout_ms=5000) + """ + + def __init__( + self, + vid: int = DEFAULT_VID, + pid: Optional[int] = DEFAULT_PID, + serial: Optional[str] = None, + interface: Optional[int] = None, + ) -> None: + self._vid = vid + self._pid = pid + self._serial = serial + self._want_itf = interface + self._core, self._util = _import_usb() + self._dev = None + self._itf_num: Optional[int] = None + self._ep_in = None + self._ep_out = None + self._claimed = False + + # -- lifecycle ------------------------------------------------------------ + def open(self) -> "UsbVendorTransport": + core, util = self._core, self._util + + def _match(dev): + if self._serial is None: + return True + try: + return util.get_string(dev, dev.iSerialNumber) == self._serial + except Exception: + return False + + kwargs = {"idVendor": self._vid} + if self._pid is not None: + kwargs["idProduct"] = self._pid + dev = core.find(custom_match=_match, **kwargs) + if dev is None: + raise TransportError( + f"no device found matching vid=0x{self._vid:04x}" + + (f" pid=0x{self._pid:04x}" if self._pid is not None else "") + + (f" serial={self._serial!r}" if self._serial else "") + ) + self._dev = dev + + cfg = dev.get_active_configuration() + itf, ep_in, ep_out = self._find_vendor_interface(cfg) + self._itf_num = itf.bInterfaceNumber + self._ep_in, self._ep_out = ep_in, ep_out + + # Detach a kernel driver if one grabbed the interface (rare for a pure + # vendor class, but be safe on Linux). + try: + if dev.is_kernel_driver_active(self._itf_num): + dev.detach_kernel_driver(self._itf_num) + except (NotImplementedError, self._core.USBError): + pass # no kernel driver bound (or the platform can't detach) -> nothing to do + + self._util.claim_interface(dev, self._itf_num) + self._claimed = True + return self + + def _find_vendor_interface(self, cfg): + util = self._util + for itf in cfg: + if self._want_itf is not None and itf.bInterfaceNumber != self._want_itf: + continue + if itf.bInterfaceClass != VENDOR_CLASS: + continue + ep_in = ep_out = None + for ep in itf: + is_bulk = util.endpoint_type(ep.bmAttributes) == util.ENDPOINT_TYPE_BULK + if not is_bulk: + continue + if util.endpoint_direction(ep.bEndpointAddress) == util.ENDPOINT_IN: + ep_in = ep + else: + ep_out = ep + if ep_in is not None and ep_out is not None: + return itf, ep_in, ep_out + raise TransportError( + "no vendor interface (class 0xFF with a bulk IN+OUT endpoint pair) found; " + "is the device running the OTA example over its WebUSB/vendor interface?" + ) + + def close(self) -> None: + if self._dev is None: + return + try: + if self._claimed and self._itf_num is not None: + self._util.release_interface(self._dev, self._itf_num) + except Exception: + pass # teardown is best-effort (device may already be gone/unplugged) + try: + self._util.dispose_resources(self._dev) + except Exception: + pass # ditto: free libusb handles best-effort, never raise from close() + self._dev = None + self._claimed = False + + def __enter__(self) -> "UsbVendorTransport": + return self.open() + + def __exit__(self, *exc) -> None: + self.close() + + # -- I/O ------------------------------------------------------------------ + def write(self, data: bytes, timeout_ms: int = 5000) -> None: + n = self._ep_out.write(data, timeout_ms) + if n != len(data): + raise TransportError(f"short write: {n}/{len(data)} bytes") + + def read(self, max_len: int, timeout_ms: int = 5000) -> bytes: + """Read up to ``max_len`` bytes; return ``b""`` on timeout (not an error).""" + try: + arr = self._ep_in.read(max_len, timeout_ms) + except self._core.USBError as exc: + # A genuine timeout is expected (poll again), but a real I/O error must + # propagate. pyusb>=1.1 raises the USBTimeoutError subclass; older pyusb + # raises USBError with errno 110 (ETIMEDOUT). Do NOT treat an unknown + # errno as a timeout — that would silently swallow backend failures. + timeout_cls = getattr(self._core, "USBTimeoutError", None) + is_timeout = (timeout_cls is not None and isinstance(exc, timeout_cls)) or ( + getattr(exc, "errno", None) == 110) + if is_timeout: + return b"" + raise + return bytes(arr) + + @property + def description(self) -> str: + if self._dev is None: + return "" + return f"0x{self._dev.idVendor:04x}:0x{self._dev.idProduct:04x} (interface {self._itf_num})" diff --git a/components/ota/python/espp_ota/ui.py b/components/ota/python/espp_ota/ui.py new file mode 100644 index 0000000000..d8d5681019 --- /dev/null +++ b/components/ota/python/espp_ota/ui.py @@ -0,0 +1,210 @@ +"""Terminal UI: a nice progress bar + colorized messages, with graceful fallback. + +The progress bar is drawn on the controlling terminal, so it animates in place +both standalone and under ``idf.py ota-usb`` (where the tool's stdout/stderr are +captured pipes — see ``_open_progress_stream``, which opens ``/dev/tty`` / +``CONOUT$`` to bypass the capture). If `rich` is available it draws a rich bar +(spinner, bar, %, bytes, transfer speed, ETA) and colorizes status/error lines; +otherwise it falls back to a manual ``\r`` bar, and to periodic plain-text lines +when there is no terminal at all (CI / redirected output). + +`rich` is optional. It ships in the ESP-IDF Python environment (so +``idf.py ota-usb`` already has it) and is pulled in by ``pip install "espp[usb-ui]"``. +""" + +from __future__ import annotations + +import os +import sys +import time +from typing import Optional + + +def _isatty() -> bool: + try: + return sys.stderr.isatty() + except Exception: + return False + + +def _open_progress_stream(): + """A writable stream connected to the real terminal, plus an 'owned' flag. + + A live progress bar needs a terminal to animate on. Under ``idf.py ota-usb`` + the tool's stdout/stderr are captured pipes (so it forwards our lines one at a + time, scrolling), but the process still has a *controlling terminal* — so we + open ``/dev/tty`` (``CONOUT$`` on Windows) and draw the bar straight to it, + bypassing the capture. Returns ``(stream, owned)`` or ``(None, False)`` when + there is no terminal at all (CI, fully redirected).""" + try: + if sys.stderr.isatty(): + return sys.stderr, False + except Exception: + pass # stderr may not support isatty() (e.g. a wrapped stream); fall through + # Try only the terminal device for THIS platform. Using the wrong name (e.g. + # "CONOUT$" on POSIX) would create a stray regular file in the cwd and write + # progress there instead of falling back to stderr. + term_name = "CONOUT$" if os.name == "nt" else "/dev/tty" + try: + return open(term_name, "w"), True + except Exception: + return None, False # no controlling terminal (CI / redirected) -> stderr fallback + + +def _have_rich() -> bool: + try: + import rich # noqa: F401 + return True + except Exception: + return False + + +def _ansi_enabled() -> bool: + if os.environ.get("NO_COLOR") is not None: + return False + if os.environ.get("CLICOLOR_FORCE") or os.environ.get("FORCE_COLOR"): + return True + return _isatty() + + +class Console: + """Styled status/error output (stderr). Uses rich when available.""" + + def __init__(self) -> None: + self._rich = None + if _have_rich(): + try: + from rich.console import Console as RichConsole + self._rich = RichConsole(file=sys.stderr, highlight=False) + except Exception: + self._rich = None + + def _emit(self, text: str, rich_style: Optional[str], ansi: Optional[str]) -> None: + if self._rich is not None: + self._rich.print(text, style=rich_style, soft_wrap=True) + return + if ansi and _ansi_enabled(): + text = f"\033[{ansi}m{text}\033[0m" + sys.stderr.write(text + "\n") + sys.stderr.flush() + + def info(self, text: str) -> None: + self._emit(text, None, None) + + def note(self, text: str) -> None: + """A prominent status line (e.g. the connected device) — bold cyan.""" + self._emit(text, "bold cyan", "1;36") + + def success(self, text: str) -> None: + self._emit(text, "bold green", "1;32") + + def warn(self, text: str) -> None: + self._emit(text, "yellow", "33") + + def error(self, text: str) -> None: + # Distinct prefix so it stands out even when color is stripped (idf.py). + self._emit(f"espp_ota ERROR: {text}", "bold red", "1;31") + + +class Progress: + """OTA progress reporter; use as a context manager, feed :meth:`update`. + + with Progress(total, "Flashing", quiet) as p: + client.flash(image) # progress=p.update + """ + + def __init__(self, total: int, label: str = "Flashing", quiet: bool = False) -> None: + self._total = total or 0 + self._label = label + self._quiet = quiet + self._rich = None # rich Progress, when available + self._task = None + self._term = None # a real-terminal stream for an in-place bar + self._own_term = False + self._plain = False # draw a manual \r bar on self._term + self._last_pct = -1000 + self._last_t = 0.0 + self._newline_done = False + + def __enter__(self) -> "Progress": + if self._quiet: + return self + self._term, self._own_term = _open_progress_stream() + if self._term is not None and _have_rich(): + try: + from rich.console import Console as RichConsole + from rich.progress import (BarColumn, DownloadColumn, Progress as RichProgress, + SpinnerColumn, TaskProgressColumn, TextColumn, + TimeRemainingColumn, TransferSpeedColumn) + # force_terminal: the stream is a real tty (possibly /dev/tty) even + # though our stdout/stderr were captured by idf.py. + self._rich = RichProgress( + SpinnerColumn(), + TextColumn("[bold blue]{task.description}"), + BarColumn(), + TaskProgressColumn(), + DownloadColumn(), + TransferSpeedColumn(), + TimeRemainingColumn(), + console=RichConsole(file=self._term, force_terminal=True), + ) + self._rich.start() + self._task = self._rich.add_task(self._label, total=self._total or None) + except Exception: + self._rich = None + if self._rich is None and self._term is not None: + self._plain = True # manual in-place bar on the terminal + return self + + def update(self, written: int, total: int) -> None: + if self._quiet: + return + if self._rich is not None: + self._rich.update(self._task, completed=written, total=total or None) + return + now = time.monotonic() + done = bool(total) and written >= total + if self._plain: + # in-place carriage-return bar on the real terminal (throttled ~10 Hz) + if now - self._last_t < 0.1 and not done: + return + self._last_t = now + if total: + pct = min(100, int(100 * written / total)) + self._term.write(f"\r {self._label} {self._text_bar(pct)} " + f"{written // 1024}/{total // 1024} KB {pct:3d}%") + else: + self._term.write(f"\r {self._label} {written // 1024} KB") + if done: + self._term.write("\n") + self._term.flush() + return + # No terminal at all (CI / fully redirected): throttled newline lines. + if total: + pct = int(100 * written / total) + if done or pct >= self._last_pct + 5: + self._last_pct = 100 if done else pct + sys.stderr.write(f" {self._label} {written // 1024}/{total // 1024} KB " + f"({self._last_pct} %)\n") + sys.stderr.flush() + elif done or now - self._last_t >= 1.0: + self._last_t = now + sys.stderr.write(f" {self._label} {written // 1024} KB\n") + sys.stderr.flush() + + @staticmethod + def _text_bar(pct: int, width: int = 28) -> str: + filled = min(width, max(0, pct * width // 100)) + return "[" + "#" * filled + "-" * (width - filled) + "]" + + def __exit__(self, *exc) -> None: + if self._rich is not None: + try: + self._rich.stop() + except Exception: + pass # tearing down the display must never raise + try: + if self._own_term and self._term is not None: + self._term.close() + except Exception: + pass # closing the borrowed /dev/tty handle is best-effort diff --git a/components/ota/python/tests/test_ota_host.py b/components/ota/python/tests/test_ota_host.py new file mode 100644 index 0000000000..751aaa76df --- /dev/null +++ b/components/ota/python/tests/test_ota_host.py @@ -0,0 +1,197 @@ +"""Host tests for espp_ota: codec round-trips + a full OTA against a mock device. + +Runs with plain ``python3`` (no hardware, no pyusb). Also importable by pytest. +""" + +import os +import struct +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from espp_ota import frame as F # noqa: E402 +from espp_ota import protocol as P # noqa: E402 +from espp_ota.client import OtaClient # noqa: E402 +from espp_ota.protocol import MessageType, OtaError # noqa: E402 + + +class MockDevice: + """Loopback transport implementing the OTA device side in memory. + + The host writes request frames; ``read()`` returns the device's replies. + Set ``fail_after`` to make the device answer a DATA frame with ERROR.""" + + def __init__(self, fail_after=None, emit_progress=False): + self._parser = F.StreamParser() + self._out = bytearray() + self.image = bytearray() + self.received = 0 + self.data_frames = 0 + self.finished = False + self._fail_after = fail_after + self._emit_progress = emit_progress + + def write(self, data, timeout_ms=0): + for fr in self._parser.feed(data): + self._handle(fr) + + def read(self, max_len, timeout_ms=0): + if not self._out: + return b"" + chunk = bytes(self._out[:max_len]) + del self._out[: len(chunk)] + return chunk + + def _reply(self, b): + self._out += b + + def _handle(self, fr): + if fr.module != P.MODULE: + return + t = fr.type + if t == MessageType.BEGIN: + # Emulate a stale session left by a prior interrupted flash: reject the + # first BEGIN as busy until an ABORT clears it. + if getattr(self, "_busy", False): + self._reply(P._build(MessageType.ERROR, struct.pack(" self._fail_after: + self._reply(P._build(MessageType.ERROR, struct.pack(" only the good one survives + bad = bytearray(good) + bad[-1] ^= 0xFF + p2 = F.StreamParser() + frames2 = p2.feed(bytes(bad) + good) + _ok("skip bad CRC, keep good", + len(frames2) == 1 and frames2[0].type == 1 and p2.dropped_bytes >= 1) + # a frame split across two feeds + p3 = F.StreamParser() + half = len(good) // 2 + _ok("partial holds", p3.feed(good[:half]) == []) + _ok("completes on rest", len(p3.feed(good[half:])) == 1) + + +def test_full_flash(): + image = bytes(bytearray((i * 7) & 0xFF for i in range(4096 * 2 + 123))) # 2+ chunks + dev = MockDevice(emit_progress=True) + seen = [] + OtaClient(dev, progress=lambda w, t: seen.append((w, t))).flash(image) + _ok("device received full image", bytes(dev.image) == image) + _ok("device saw FINISH", dev.finished) + _ok("chunking (3 DATA frames)", dev.data_frames == 3) + _ok("progress reported", len(seen) > 0 and seen[-1][0] == len(image)) + + +def test_error_reply(): + dev = MockDevice(fail_after=1) + raised = False + try: + OtaClient(dev).flash(bytes(4096 * 3)) + except OtaError as exc: + raised = True + _ok("error carries device code", exc.code == 22) + _ok("error path raises OtaError", raised) + + +def test_small_image_one_chunk(): + dev = MockDevice() + OtaClient(dev).flash(b"\xe9tiny firmware") + _ok("single-chunk image", bytes(dev.image) == b"\xe9tiny firmware" and dev.data_frames == 1) + + +def test_rollback_control(): + """get_status / mark_valid over the loopback mock.""" + dev = MockDevice() + dev._pending = True + st = OtaClient(dev).get_status() + _ok("status pending+supported", st.pending_verify and st.rollback_supported) + _ok("status reports firmware", st.version == "1.0.0" and st.project_name == "ota_example") + OtaClient(dev).mark_valid() + _ok("mark_valid confirmed", getattr(dev, "marked_valid", False) and not dev._pending) + st2 = OtaClient(dev).get_status() + _ok("status confirmed after mark_valid", not st2.pending_verify) + OtaClient(dev).mark_invalid() + _ok("rollback requested", getattr(dev, "rolled_back", False)) + + +def test_begin_busy_recovers(): + """A stale session (BEGIN rejected as busy) is cleared by an ABORT + retry.""" + dev = MockDevice() + dev._busy = True # device thinks a prior session is still open + OtaClient(dev).flash(b"\xe9hello world payload") + _ok("recovered from busy BEGIN", bytes(dev.image) == b"\xe9hello world payload" + and dev.finished) + + +if __name__ == "__main__": + test_frame_golden() + test_parser_resync() + test_full_flash() + test_error_reply() + test_small_image_one_chunk() + test_begin_busy_recovers() + test_rollback_control() + print("all host tests passed") diff --git a/components/ota/web/ota_console.html b/components/ota/web/ota_console.html index b3f0b34dc2..e50ce2aec6 100644 --- a/components/ota/web/ota_console.html +++ b/components/ota/web/ota_console.html @@ -23,9 +23,11 @@ crc32("123456789") == 0xCBF43926. - payload length is capped at 4096 bytes per frame. - host -> device requests (flags reply bit 0): 0x01 BEGIN(u32 image_size), - 0x02 DATA(bytes), 0x03 FINISH, 0x04 ABORT; device -> host replies - (flags reply bit 1): 0x05 OK(u32 bytes_received), - 0x06 ERROR(u32 code + utf8 message), 0x07 PROGRESS(u32 written, u32 total). + 0x02 DATA(bytes), 0x03 FINISH, 0x04 ABORT, 0x08 GET_STATUS, 0x09 MARK_VALID, + 0x0A MARK_INVALID; device -> host replies (flags reply bit 1): + 0x05 OK(u32 bytes_received), 0x06 ERROR(u32 code + utf8 message), + 0x07 PROGRESS(u32 written, u32 total), 0x0B STATUS(u8 flags: bit0 pending + verify, bit1 rollback supported). - transactions are serialized: exactly one command frame is in flight and the host waits for its OK / ERROR reply (PROGRESS frames are informational and may arrive before the reply). @@ -218,6 +220,14 @@

Firmware update

+
+ + + +
+

After an OTA the new image boots + pending verify: confirm it here (the app must not confirm itself) + or it rolls back on the next reset.

0 / 0 bytes @@ -268,8 +278,12 @@

Log

const FLAGS_REQUEST = (FLAGS_VERSION << 4) | 0; // host->device request: 0x10 (reply bit 0) const FLAGS_REPLY_BIT = 0x01; // bit0 set on device->host replies const FLAG_CORRELATION = 0x02; // bit1: optional u16 correlation id present after `type` - const TYPE = { BEGIN: 0x01, DATA: 0x02, FINISH: 0x03, ABORT: 0x04, OK: 0x05, ERROR: 0x06, PROGRESS: 0x07 }; - const TYPE_NAME = { 0x01: "BEGIN", 0x02: "DATA", 0x03: "FINISH", 0x04: "ABORT", 0x05: "OK", 0x06: "ERROR", 0x07: "PROGRESS" }; + const TYPE = { BEGIN: 0x01, DATA: 0x02, FINISH: 0x03, ABORT: 0x04, OK: 0x05, ERROR: 0x06, PROGRESS: 0x07, + GET_STATUS: 0x08, MARK_VALID: 0x09, MARK_INVALID: 0x0A, STATUS: 0x0B }; + const TYPE_NAME = { 0x01: "BEGIN", 0x02: "DATA", 0x03: "FINISH", 0x04: "ABORT", 0x05: "OK", 0x06: "ERROR", 0x07: "PROGRESS", + 0x08: "GET_STATUS", 0x09: "MARK_VALID", 0x0A: "MARK_INVALID", 0x0B: "STATUS" }; + // STATUS reply flag bits. + const STATUS_PENDING_VERIFY = 0x01, STATUS_ROLLBACK_SUPPORTED = 0x02; // esp_ota_begin erases flash (seconds for a large / unknown-size image) and // esp_ota_end re-reads + hashes the whole image, so give BEGIN / FINISH a // much longer deadline than the per-4KiB DATA writes. @@ -282,7 +296,8 @@

Log

// =================================================================== const els = {}; for (const id of ["status", "connectBtn", "anyDevice", "devInfo", "fileInput", "fileInfo", - "uploadBtn", "abortBtn", "progressFill", "statBytes", "statPercent", + "uploadBtn", "abortBtn", "statusBtn", "markValidBtn", "rollbackBtn", + "progressFill", "statBytes", "statPercent", "statRate", "statNote", "logFrames", "clearLogBtn", "log", "unsupported"]) { els[id] = document.getElementById(id); } @@ -426,6 +441,25 @@

Log

return { written: view.getUint32(0, true), total: view.getUint32(4, true) }; } + // STATUS payload: [flags u8][version u8-len+bytes][project u8-len+bytes]. + function parseStatus(payload) { + if (!payload || payload.length < 1) return null; + const dec = new TextDecoder(); + let i = 1; + const readStr = () => { + if (i >= payload.length) return ""; + const n = payload[i++]; const s = dec.decode(payload.subarray(i, i + n)); i += n; return s; + }; + const version = readStr(), project = readStr(); + return { + flags: payload[0], + pendingVerify: !!(payload[0] & STATUS_PENDING_VERIFY), + rollbackSupported: !!(payload[0] & STATUS_ROLLBACK_SUPPORTED), + version, project, + firmware: (project && version) ? (project + " " + version) : (project || version || "(unknown)"), + }; + } + // =================================================================== // WebUSB connect / disconnect (vendor 0xFF interface discovery) // =================================================================== @@ -434,6 +468,10 @@

Log

let manualDisconnect = false; let uploading = false; let expectRestart = false; + // After a successful upload the device reboots (dropping WebUSB). When the + // user reconnects we auto-check the new firmware and offer to confirm it. + // Holds { before: "" } or null. + let awaitingVerify = null; const parser = new StreamParser(); if (!navigator.usb) { @@ -487,6 +525,32 @@

Log

logLine("sys", "Connected to " + name + ". Vendor iface " + ifaceNumber + ", bulk IN 0x" + epIn.toString(16) + " / OUT 0x" + epOut.toString(16) + "."); checkModulePresent().catch(() => {}); // best-effort: warn if this device lacks the OTA module + maybePromptVerify().catch(() => {}); // if we just flashed, verify + confirm the new image + } + + // After a reboot-triggering upload the user reconnects here; auto-check the + // running image and, if it is pending verify (so it booted and can answer), + // ask the user to confirm it — the firmware must not confirm itself. + async function maybePromptVerify() { + if (!awaitingVerify) return; + const before = awaitingVerify.before; + awaitingVerify = null; + const st = await queryStatus(); + if (!st) return; + if (!st.rollbackSupported) { logLine("ok", "Reconnected; running " + st.firmware + " (no rollback to confirm)."); return; } + if (!st.pendingVerify) { logLine("ok", "Reconnected; running " + st.firmware + " (already confirmed)."); return; } + const msg = "The device rebooted into new firmware and is responding.\n\n" + + (before ? "Previous: " + before + "\n" : "") + + "Now running: " + st.firmware + "\n\n" + + "Mark this image VALID? It rolls back on the next reset if you don't."; + if (confirm(msg)) { + try { + await transact(TYPE.MARK_VALID, null, DATA_TIMEOUT_MS); + logLine("ok", "New image (" + st.firmware + ") marked valid; rollback cancelled."); + } catch (e) { logLine("err", "Mark valid failed: " + e.message); } + } else { + logLine("err", "Left unconfirmed (" + st.firmware + "); it rolls back on the next reset. Use “Mark valid” to confirm later."); + } } async function openAndClaim() { @@ -571,6 +635,11 @@

Log

const file = els.fileInput.files && els.fileInput.files[0]; els.uploadBtn.disabled = !(device && file && !uploading); els.abortBtn.disabled = !uploading; + // Rollback controls need a connection and no upload in flight. + const rbReady = !!device && !uploading; + els.statusBtn.disabled = !rbReady; + els.markValidBtn.disabled = !rbReady; + els.rollbackBtn.disabled = !rbReady; if (!device) els.devInfo.textContent = "Not connected. Default filter: any espp device (VID 0x1209); the vendor (0xFF) interface is discovered from the descriptors at runtime."; } @@ -674,6 +743,8 @@

Log

if (progress) updateProgress(progress.written, progress.total || null); continue; // informational; keep waiting for OK / ERROR } + // STATUS is the reply to GET_STATUS; return its raw payload (flags byte). + if (reply.type === TYPE.STATUS) return reply.payload; if (reply.type === TYPE.OK) return parseU32(reply.payload); if (reply.type === TYPE.ERROR) { const info = parseError(reply.payload); @@ -808,6 +879,14 @@

Log

const startedAt = Date.now(); let sent = 0; + // Record the firmware we're replacing so the post-reboot prompt can show + // the before -> after transition (best-effort; older devices lack STATUS). + let beforeFirmware = ""; + try { + const st0 = parseStatus(await transact(TYPE.GET_STATUS, null, DATA_TIMEOUT_MS)); + if (st0) beforeFirmware = st0.firmware; + } catch (_) {} + try { logLine("sys", "BEGIN: image size " + image.length.toLocaleString() + " bytes (device erases the update partition — this can take a while)..."); await transact(TYPE.BEGIN, u32Payload(image.length), BEGIN_TIMEOUT_MS); @@ -829,9 +908,14 @@

Log

els.progressFill.className = "progress-fill done"; els.progressFill.style.width = "100%"; setStatus("connected", "Update complete"); - setNote("Update complete — device is restarting into the new firmware. " + - "With rollback enabled it must mark itself valid on first boot."); + // Arm the post-reboot verification: when the user reconnects, we check the + // new image can respond and prompt to confirm it (the app must not confirm + // itself). The reboot drops WebUSB, so this needs a manual reconnect. + awaitingVerify = { before: beforeFirmware }; + setNote("Update complete — device is restarting into the new firmware. It " + + "boots pending-verify; reconnect to confirm it (it must not confirm itself)."); logLine("sys", "Update complete (" + sent.toLocaleString() + " bytes). Device restarting — expect a USB disconnect."); + logLine("sys", "Reconnect with the Connect button to verify + confirm the new image."); } catch (e) { els.progressFill.className = "progress-fill failed"; if (e && e.userAbort) { @@ -864,6 +948,46 @@

Log

logLine("sys", "Abort requested; stopping after the in-flight chunk..."); }); + // ---- rollback controls (host confirms the image; the app must not) -------- + // Query STATUS and return the parsed info (logs + returns null on failure). + async function queryStatus() { + try { + const st = parseStatus(await transact(TYPE.GET_STATUS, null, DATA_TIMEOUT_MS)); + return st; + } catch (e) { logLine("err", "Status failed: " + e.message); return null; } + } + els.statusBtn.addEventListener("click", async () => { + const st = await queryStatus(); + if (!st) return; + logLine("sys", "Running firmware: " + st.firmware); + if (!st.rollbackSupported) { + logLine("sys", "Rollback not supported on the device (CONFIG_BOOTLOADER_APP_ROLLBACK_ENABLE off)."); + } else if (st.pendingVerify) { + logLine("err", "Running image is PENDING VERIFY — click “Mark valid” to confirm it, or it rolls back on the next reset."); + } else { + logLine("ok", "Running image is confirmed (not pending verify)."); + } + }); + + els.markValidBtn.addEventListener("click", async () => { + try { + await transact(TYPE.MARK_VALID, null, DATA_TIMEOUT_MS); + logLine("ok", "Running image marked valid; rollback cancelled."); + } catch (e) { logLine("err", "Mark valid failed: " + e.message); } + }); + + els.rollbackBtn.addEventListener("click", async () => { + if (!confirm("Roll back to the previous image and reboot the device?")) return; + // The device reboots on success and may not reply, so a deadline/link loss + // here is the expected outcome, not an error. + try { + await transact(TYPE.MARK_INVALID, null, DATA_TIMEOUT_MS); + logLine("ok", "Rollback acknowledged; device rebooting into the previous image."); + } catch (e) { + logLine("sys", "Rollback sent; device is rebooting (no reply expected). [" + e.message + "]"); + } + }); + logLine("sys", "espp OTA Console ready. Connect a device running the espp ota example (vendor/WebUSB interface)."); diff --git a/doc/en/ota/ota.rst b/doc/en/ota/ota.rst index c00662bafa..ca52751a9d 100644 --- a/doc/en/ota/ota.rst +++ b/doc/en/ota/ota.rst @@ -34,6 +34,32 @@ message types on top); to run OTA alongside other protocols (crash-dump, CAN, :doc:`../dispatcher/index` — the ``ota`` example does exactly this (OTA is module id 0). +Command line: build → OTA +------------------------- + +The ``ota`` component ships a ``project_include.cmake`` and a pure-Python host +tool (``components/ota/python/espp_ota``), so any project using it can build and +OTA-flash over USB in one step — the OTA counterpart to ``idf.py flash``:: + + pip install pyusb # once (needs a libusb backend) + idf.py ota-usb # builds the app, then OTAs it over USB + +The tool draws a live progress bar (percent, size, transfer speed, ETA) and +colorizes its output. Because ``idf.py`` captures the target's output, the bar is +drawn straight to the controlling terminal so it still animates in place: + +.. image:: https://github.com/user-attachments/assets/a042481a-2964-4b06-9109-bb2dcb4e355b + :alt: espp_ota flashing an image over USB via idf.py ota-usb + :width: 100% + +.. image:: https://github.com/user-attachments/assets/da8e1b71-c65f-4ecc-9223-e232f8591ceb + :alt: espp_ota reporting a completed OTA over USB + :width: 100% + +For full control (a specific serial, chunk size, discovery probe) run it directly +with ``python -m espp_ota flash build/.bin`` — see +``components/ota/python/README.md``. + .. ------------------------------- Example ------------------------------------- .. toctree:: diff --git a/pyproject.toml b/pyproject.toml index bb454efef3..2ebb0710b3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,6 +35,18 @@ Issues = "https://github.com/esp-cpp/espp/issues" # the shipped packages is stdlib-only. `pip install espp[serial]` enables the # espp_odrive serial/UART transports. serial = ["pyserial"] +# espp_ota's USB transport imports pyusb lazily; the frame codec + OTA protocol +# are stdlib-only. `pip install espp[usb]` enables the OTA-over-USB host tool. +usb = ["pyusb"] +# `rich` is optional — it gives the nicer progress bar / colorized output (it also +# ships in the ESP-IDF Python env, so `idf.py ota-usb` already has it); the tool +# degrades to plain text without it. `pip install espp[usb-ui]` adds it. +usb-ui = ["pyusb", "rich"] + +[project.scripts] +# Standalone CLI (also runnable as `python -m espp_ota`). The `ota` component's +# CMake integration wires `idf.py ota-usb` to the same tool for a build->OTA flow. +espp-ota = "espp_ota.cli:main" [tool.scikit-build] # The CMake project for the host (PC) build lives in lib/; the repo root is the @@ -48,6 +60,10 @@ cmake.source-dir = "lib" wheel.packages = [ "lib/python_bindings/espp", "components/odrive_native/python/espp_odrive", + # espp_ota: pure-python OTA-over-USB host tool (stdlib-only except a + # lazily-imported pyusb; see the `usb` extra). Single source of truth lives in + # the ota component. + "components/ota/python/espp_ota", ] wheel.exclude = ["**/.mypy_cache", "**/__pycache__"] # Persistent CMake build dir (gitignored) so repeated local builds - notably