diff --git a/Tools/px4_uploader.py b/Tools/px4_uploader.py new file mode 100755 index 00000000000..8da12e32baf --- /dev/null +++ b/Tools/px4_uploader.py @@ -0,0 +1,2110 @@ +#!/usr/bin/env python3 +############################################################################ +# +# Copyright (c) 2012-2026 PX4 Development Team. All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions +# are met: +# +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in +# the documentation and/or other materials provided with the +# distribution. +# 3. Neither the name PX4 nor the names of its contributors may be +# used to endorse or promote products derived from this software +# without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS +# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE +# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, +# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, +# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS +# OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED +# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT +# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN +# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# +############################################################################ + +""" +PX4 Firmware Uploader v2 - Rewritten with improved error handling and debugging. + +This script uploads firmware to PX4-based flight controllers via their bootloader. + +The PX4 firmware file is a JSON-encoded Python object containing metadata fields +and a zlib-compressed base64-encoded firmware image. + +Key improvements over px_uploader.py: +- Proper exception hierarchy with full context +- Verbose/debug logging support +- Self-contained port detection +- Clean separation of concerns +- Configurable timeouts +- Better progress reporting +""" + +import argparse +import base64 +import glob +import json +import logging +import os +import socket +import struct +import sys +import time +import zlib +from dataclasses import dataclass, field +from enum import IntEnum +from pathlib import Path +from typing import Optional + +# Check Python version early +if sys.version_info < (3, 7): + print("Python 3.7 or later is required.", file=sys.stderr) + sys.exit(1) + +try: + import serial + import serial.tools.list_ports +except ImportError as e: + print(f"Failed to import pyserial: {e}", file=sys.stderr) + print("\nInstall it with: python -m pip install pyserial", file=sys.stderr) + sys.exit(1) + + +# ============================================================================= +# Logging Configuration +# ============================================================================= + +logger = logging.getLogger("px4_uploader") + + +def setup_logging(verbose: bool = False, debug: bool = False) -> None: + """Configure logging based on verbosity level. + + Args: + verbose: Enable INFO level logging for operational details + debug: Enable DEBUG level logging for protocol-level details + """ + if debug: + level = logging.DEBUG + fmt = "%(asctime)s.%(msecs)03d [%(levelname)s] %(name)s: %(message)s" + elif verbose: + level = logging.INFO + fmt = "[%(levelname)s] %(message)s" + else: + level = logging.WARNING + fmt = "%(message)s" + + handler = logging.StreamHandler() + handler.setFormatter(logging.Formatter(fmt, datefmt="%H:%M:%S")) + logger.addHandler(handler) + logger.setLevel(level) + + # Also check environment variable + if os.environ.get("PX4_UPLOADER_DEBUG", "").lower() in ("1", "true", "yes"): + logger.setLevel(logging.DEBUG) + + +# ============================================================================= +# Exception Hierarchy +# ============================================================================= + + +class UploadError(Exception): + """Base exception for all upload-related errors.""" + + def __init__( + self, + message: str, + port: Optional[str] = None, + operation: Optional[str] = None, + details: Optional[str] = None, + ): + self.port = port + self.operation = operation + self.details = details + + parts = [message] + if port: + parts.append(f"port={port}") + if operation: + parts.append(f"during {operation}") + if details: + parts.append(f"({details})") + + super().__init__(" ".join(parts)) + + +class ProtocolError(UploadError): + """Error in bootloader protocol communication.""" + + pass + + +class ConnectionError(UploadError): + """Error establishing or maintaining serial connection.""" + + pass + + +class FirmwareError(UploadError): + """Error loading or validating firmware file.""" + + pass + + +class BoardMismatchError(UploadError): + """Firmware not suitable for the connected board.""" + + pass + + +class TimeoutError(UploadError): + """Operation timed out.""" + + pass + + +class SiliconErrataError(UploadError): + """Board has silicon errata that prevents safe operation.""" + + pass + + +# ============================================================================= +# Protocol Constants +# ============================================================================= + + +class BootloaderCommand(IntEnum): + """Bootloader protocol commands.""" + + NOP = 0x00 # Guaranteed to be discarded by the bootloader + GET_SYNC = 0x21 + GET_DEVICE = 0x22 + CHIP_ERASE = 0x23 + CHIP_VERIFY = 0x24 # rev2 only + PROG_MULTI = 0x27 + READ_MULTI = 0x28 # rev2 only + GET_CRC = 0x29 # rev3+ + GET_OTP = 0x2A # rev4+, get a word from OTP area + GET_SN = 0x2B # rev4+, get a word from SN area + GET_CHIP = 0x2C # rev5+, get chip version + SET_BOOT_DELAY = 0x2D # rev5+, set boot delay + GET_CHIP_DES = 0x2E # rev5+, get chip description in ASCII + GET_VERSION = 0x2F # rev5+, get bootloader version in ASCII + REBOOT = 0x30 + CHIP_FULL_ERASE = 0x40 # Full erase of flash, rev6+ + + +class BootloaderResponse(IntEnum): + """Bootloader response codes.""" + + INSYNC = 0x12 + EOC = 0x20 + OK = 0x10 + FAILED = 0x11 + INVALID = 0x13 # rev3+ + BAD_SILICON_REV = 0x14 # rev5+ + + +class DeviceInfo(IntEnum): + """Device information parameter codes.""" + + BL_REV = 0x01 # Bootloader protocol revision + BOARD_ID = 0x02 # Board type + BOARD_REV = 0x03 # Board revision + FLASH_SIZE = 0x04 # Max firmware size in bytes + + +@dataclass +class ProtocolConfig: + """Protocol configuration constants.""" + + BL_REV_MIN: int = 2 # Minimum supported bootloader protocol + BL_REV_MAX: int = 6 # Maximum supported bootloader protocol + PROG_MULTI_MAX: int = ( + 252 # Max bytes per PROG_MULTI (protocol max 255, must be multiple of 4) + ) + READ_MULTI_MAX: int = 252 # Max bytes per READ_MULTI + MAX_DES_LENGTH: int = 20 # Max chip description length + + +# ============================================================================= +# Known PX4 USB Vendor/Product IDs +# ============================================================================= + +# Known VID/PID combinations for PX4 bootloaders and devices +PX4_USB_IDS: list[tuple[int, int, str]] = [ + # (Vendor ID, Product ID, Description) + (0x26AC, 0x0010, "3D Robotics PX4 FMU"), + (0x26AC, 0x0011, "3D Robotics PX4 BL"), + (0x26AC, 0x0012, "3D Robotics PX4IO"), + (0x26AC, 0x0032, "3D Robotics PX4 FMU v5"), + (0x3185, 0x0035, "Holybro Durandal"), + (0x3185, 0x0036, "Holybro Kakute"), + (0x3162, 0x004B, "Holybro Pixhawk 4"), + (0x1FC9, 0x001C, "NXP FMUK66"), + (0x2DAE, 0x1058, "Cube Orange"), + (0x2DAE, 0x1016, "Cube Black"), + (0x2DAE, 0x1011, "Cube Yellow"), + (0x0483, 0x5740, "STMicroelectronics Virtual COM Port"), # Generic ST bootloader + (0x1209, 0x5740, "Generic STM32"), + (0x1209, 0x5741, "ArduPilot"), + (0x2341, 0x8036, "Arduino Leonardo"), # Some PX4 boards use this +] + + +# ============================================================================= +# Firmware Class +# ============================================================================= + + +@dataclass +class Firmware: + """Loads and validates a PX4 firmware file. + + The firmware file is JSON containing metadata and a zlib-compressed, + base64-encoded firmware image. + + Attributes: + path: Path to the firmware file + board_id: Target board ID from firmware metadata + board_revision: Board revision from metadata + image: Decompressed firmware binary (padded to 4-byte alignment) + image_size: Original image size before padding + image_maxsize: Maximum image size the firmware was built for + description: Full firmware metadata dictionary + """ + + path: Path + board_id: int = field(init=False) + board_revision: int = field(init=False) + image: bytes = field(init=False) + image_size: int = field(init=False) + image_maxsize: int = field(init=False) + description: dict = field(init=False) + + def __post_init__(self): + """Load and validate the firmware file.""" + self.path = Path(self.path) + self._load() + + def _load(self) -> None: + """Load firmware from JSON file.""" + logger.info(f"Loading firmware from {self.path}") + + if not self.path.exists(): + raise FirmwareError(f"Firmware file not found: {self.path}") + + try: + with open(self.path, "r") as f: + self.description = json.load(f) + except json.JSONDecodeError as e: + raise FirmwareError(f"Invalid firmware JSON: {e}", details=str(self.path)) + except IOError as e: + raise FirmwareError( + f"Cannot read firmware file: {e}", details=str(self.path) + ) + + # Extract required fields + required_fields = ["image", "board_id", "image_size", "image_maxsize"] + for field_name in required_fields: + if field_name not in self.description: + raise FirmwareError( + f"Firmware missing required field: {field_name}", + details=str(self.path), + ) + + self.board_id = self.description["board_id"] + self.board_revision = self.description.get("board_revision", 0) + self.image_size = self.description["image_size"] + self.image_maxsize = self.description["image_maxsize"] + + # Decompress image + try: + compressed = base64.b64decode(self.description["image"]) + image_data = bytearray(zlib.decompress(compressed)) + except (base64.binascii.Error, zlib.error) as e: + raise FirmwareError( + f"Cannot decompress firmware image: {e}", details=str(self.path) + ) + + # Pad to 4-byte alignment + while len(image_data) % 4 != 0: + image_data.append(0xFF) + + self.image = bytes(image_data) + + logger.info( + f"Loaded firmware: board_id={self.board_id}, " + f"size={self.image_size} bytes ({self.usage_percent:.1f}%)" + ) + + @property + def usage_percent(self) -> float: + """Percentage of maximum flash used.""" + return (self.image_size / self.image_maxsize) * 100.0 + + def crc(self, padlen: int) -> int: + """Calculate CRC32 of firmware image with padding. + + Args: + padlen: Total length to pad image to (typically flash size) + + Returns: + CRC32 value matching bootloader's calculation + """ + state = 0xFFFFFFFF + state = zlib.crc32(self.image, state) + + padding_length = padlen - len(self.image) + if padding_length > 0: + padding = b"\xff" * padding_length + state = zlib.crc32(padding, state) + + return (state ^ 0xFFFFFFFF) & 0xFFFFFFFF + + +# ============================================================================= +# Serial Transport +# ============================================================================= + + +class SerialTransport: + """Handles serial port communication with proper resource management. + + Provides context manager support for automatic cleanup and configurable + timeouts. + """ + + def __init__( + self, + port: str, + baudrate: int = 115200, + timeout: float = 0.5, + write_timeout: float = 2.0, + ): + """Initialize serial transport. + + Args: + port: Serial port path (e.g., /dev/ttyUSB0, COM3) + baudrate: Baud rate for communication + timeout: Read timeout in seconds + write_timeout: Write timeout in seconds (0 = no timeout) + """ + self.port_name = port + self.baudrate = baudrate + self.timeout = timeout + self.write_timeout = write_timeout + self._port: Optional[serial.Serial] = None + self._chartime = 10.0 / baudrate # 8N1 = 10 bits per byte + + def __enter__(self) -> "SerialTransport": + self.open() + return self + + def __exit__(self, exc_type, exc_val, exc_tb) -> None: + self.close() + + def open(self) -> None: + """Open the serial port.""" + if self._port is not None and self._port.is_open: + return + + logger.debug(f"Opening serial port {self.port_name} at {self.baudrate} baud") + + try: + self._port = serial.Serial( + self.port_name, + self.baudrate, + timeout=self.timeout, + write_timeout=self.write_timeout, + ) + except serial.SerialException as e: + raise ConnectionError( + f"Cannot open serial port: {e}", port=self.port_name, operation="open" + ) + + def close(self) -> None: + """Close the serial port.""" + if self._port is not None: + logger.debug(f"Closing serial port {self.port_name}") + try: + self._port.close() + except Exception as e: + logger.warning(f"Error closing port {self.port_name}: {e}") + self._port = None + + @property + def is_open(self) -> bool: + """Check if port is open.""" + return self._port is not None and self._port.is_open + + def send(self, data: bytes) -> None: + """Send data over serial port. + + Args: + data: Bytes to send + + Raises: + ConnectionError: If send fails + """ + if not self.is_open: + raise ConnectionError( + "Port not open", port=self.port_name, operation="send" + ) + + logger.debug(f"TX: {data.hex()}") + + try: + self._port.write(data) + except serial.SerialException as e: + raise ConnectionError( + f"Write failed: {e}", port=self.port_name, operation="send" + ) + + def recv(self, count: int = 1, timeout: Optional[float] = None) -> bytes: + """Receive data from serial port. + + Args: + count: Number of bytes to receive + timeout: Override default timeout + + Returns: + Received bytes + + Raises: + TimeoutError: If timeout expires before all bytes received + ConnectionError: If read fails + """ + if not self.is_open: + raise ConnectionError( + "Port not open", port=self.port_name, operation="recv" + ) + + old_timeout = self._port.timeout + if timeout is not None: + self._port.timeout = timeout + + try: + data = self._port.read(count) + except serial.SerialException as e: + raise ConnectionError( + f"Read failed: {e}", port=self.port_name, operation="recv" + ) + finally: + if timeout is not None: + self._port.timeout = old_timeout + + if len(data) < count: + raise TimeoutError( + f"Timeout waiting for {count} bytes, got {len(data)}", + port=self.port_name, + operation="recv", + ) + + logger.debug(f"RX: {data.hex()}") + return data + + def flush(self) -> None: + """Flush output buffer.""" + if self._port is not None: + self._port.flush() + + def reset_buffers(self) -> None: + """Reset input and output buffers.""" + if self._port is not None: + self._port.reset_input_buffer() + self._port.reset_output_buffer() + + def set_baudrate(self, baudrate: int) -> None: + """Change baud rate. + + Args: + baudrate: New baud rate + """ + logger.debug(f"Changing baudrate to {baudrate}") + self.baudrate = baudrate + self._chartime = 10.0 / baudrate + + if self._port is not None: + try: + self._port.baudrate = baudrate + except (serial.SerialException, NotImplementedError) as e: + logger.debug(f"Cannot change baudrate: {e}") + raise + + @property + def chartime(self) -> float: + """Time to transmit one character.""" + return self._chartime + + +# ============================================================================= +# Bootloader Protocol +# ============================================================================= + + +class BootloaderProtocol: + """Implements the PX4 bootloader protocol. + + Handles all communication with the bootloader including sync, + identification, programming, and verification. + """ + + # Reboot command sequences + NSH_INIT = b"\r\r\r" + NSH_REBOOT_BL = b"reboot -b\n" + NSH_REBOOT = b"reboot\n" + + # MAVLink reboot commands (MAVLink v1 COMMAND_LONG with MAV_CMD_PREFLIGHT_REBOOT_SHUTDOWN) + MAVLINK_REBOOT_ID1 = bytes.fromhex( + "fe2172ff004c00004040000000000000000000000000" + "000000000000000000000000f600010000536b" + ) + MAVLINK_REBOOT_ID0 = bytes.fromhex( + "fe2145ff004c00004040000000000000000000000000" + "000000000000000000000000f600000000cc37" + ) + + def __init__( + self, + transport: SerialTransport, + sync_timeout: float = 0.5, + erase_timeout: float = 30.0, + windowed: bool = False, + ): + """Initialize bootloader protocol handler. + + Args: + transport: Serial transport instance + sync_timeout: Timeout for sync operations + erase_timeout: Timeout for chip erase + windowed: Use windowed mode for faster uploads on real serial ports + """ + self.transport = transport + self.sync_timeout = sync_timeout + self.erase_timeout = erase_timeout + + # Board info (populated by identify()) + self.bl_rev: int = 0 + self.board_type: int = 0 + self.board_rev: int = 0 + self.fw_maxsize: int = 0 + self.version: str = "unknown" + self.otp: bytes = b"" + self.sn: bytes = b"" + self.chip_id: int = 0 + self.chip_family: str = "" + self.chip_revision: str = "" + + # Windowed mode for faster uploads on some interfaces + self._windowed_mode = windowed + self._window_size = 0 + self._window_max = 256 + self._window_per = 2 # SYNC + result per block + + def _send_command(self, cmd: int, *args: bytes) -> None: + """Send a command to the bootloader. + + Args: + cmd: Command byte + *args: Additional data bytes + """ + data = bytes([cmd]) + b"".join(args) + bytes([BootloaderResponse.EOC]) + self.transport.send(data) + + def _recv_int(self) -> int: + """Receive a 32-bit little-endian integer.""" + raw = self.transport.recv(4) + return struct.unpack(" None: + """Wait for and validate sync response. + + Args: + flush: Whether to flush output buffer first + + Raises: + ProtocolError: If response is not valid INSYNC + OK + """ + if flush: + self.transport.flush() + + insync = self.transport.recv(1) + if insync[0] != BootloaderResponse.INSYNC: + raise ProtocolError( + f"Expected INSYNC (0x{BootloaderResponse.INSYNC:02X}), " + f"got 0x{insync[0]:02X}", + port=self.transport.port_name, + operation="sync", + ) + + result = self.transport.recv(1) + if result[0] == BootloaderResponse.INVALID: + raise ProtocolError( + "Bootloader reports INVALID OPERATION", port=self.transport.port_name + ) + if result[0] == BootloaderResponse.FAILED: + raise ProtocolError( + "Bootloader reports OPERATION FAILED", port=self.transport.port_name + ) + if result[0] == BootloaderResponse.BAD_SILICON_REV: + raise SiliconErrataError( + "Chip has silicon errata, programming not supported.\n" + "See https://docs.px4.io/main/en/flight_controller/silicon_errata.html", + port=self.transport.port_name, + ) + if result[0] != BootloaderResponse.OK: + raise ProtocolError( + f"Expected OK (0x{BootloaderResponse.OK:02X}), got 0x{result[0]:02X}", + port=self.transport.port_name, + ) + + def _try_sync(self) -> bool: + """Attempt to get sync without raising exceptions. + + Returns: + True if sync successful, False otherwise + """ + try: + self.transport.flush() + insync = self.transport.recv(1, timeout=0.1) + if insync[0] != BootloaderResponse.INSYNC: + return False + result = self.transport.recv(1, timeout=0.1) + if result[0] == BootloaderResponse.BAD_SILICON_REV: + raise SiliconErrataError( + "Chip has silicon errata, programming not supported", + port=self.transport.port_name, + ) + return result[0] == BootloaderResponse.OK + except TimeoutError: + return False + except Exception as e: + logger.debug(f"Sync attempt failed: {e}") + return False + + def _validate_sync_window(self, count: int) -> None: + """Validate multiple sync responses for windowed mode. + + Args: + count: Number of sync responses to validate (each is 2 bytes) + """ + if count <= 0: + return + + data = self.transport.recv(count) + if len(data) != count: + raise ProtocolError( + f"Expected {count} bytes, got {len(data)}", + port=self.transport.port_name, + operation="ack_window", + ) + + for i in range(0, len(data), 2): + if data[i] != BootloaderResponse.INSYNC: + raise ProtocolError( + f"Expected INSYNC at byte {i}, got 0x{data[i]:02X}", + port=self.transport.port_name, + ) + if data[i + 1] == BootloaderResponse.INVALID: + raise ProtocolError( + "Bootloader reports INVALID OPERATION", + port=self.transport.port_name, + ) + if data[i + 1] == BootloaderResponse.FAILED: + raise ProtocolError( + "Bootloader reports OPERATION FAILED", port=self.transport.port_name + ) + if data[i + 1] != BootloaderResponse.OK: + raise ProtocolError( + f"Expected OK, got 0x{data[i + 1]:02X}", + port=self.transport.port_name, + ) + + def _detect_interface_type(self) -> None: + """Detect if connected via USB CDC or real serial port. + + Currently just resets buffers. Windowed mode can be enabled manually + with --windowed for real serial ports (FTDI, etc.). + """ + self.transport.reset_buffers() + + def sync(self) -> None: + """Synchronize with bootloader. + + Sends sync command and waits for valid response. + + Raises: + ProtocolError: If sync fails + """ + logger.debug("Syncing with bootloader") + self.transport.reset_buffers() + self._send_command(BootloaderCommand.GET_SYNC) + self._get_sync() + logger.debug("Sync successful") + + def _get_device_info(self, param: int) -> int: + """Get device information parameter. + + Args: + param: DeviceInfo parameter code + + Returns: + Parameter value + """ + self._send_command(BootloaderCommand.GET_DEVICE, bytes([param])) + value = self._recv_int() + self._get_sync() + return value + + def _get_otp(self, address: int) -> bytes: + """Read 4 bytes from OTP area. + + Args: + address: OTP address (byte offset) + + Returns: + 4 bytes of OTP data + """ + self._send_command(BootloaderCommand.GET_OTP, struct.pack(" bytes: + """Read 4 bytes from serial number area. + + Args: + address: SN address (byte offset) + + Returns: + 4 bytes of SN data + """ + self._send_command(BootloaderCommand.GET_SN, struct.pack(" int: + """Get chip ID. + + Returns: + Chip ID value + """ + self._send_command(BootloaderCommand.GET_CHIP) + value = self._recv_int() + self._get_sync() + return value + + def _get_chip_description(self) -> tuple[str, str]: + """Get chip family and revision. + + Returns: + Tuple of (family, revision) strings + """ + self._send_command(BootloaderCommand.GET_CHIP_DES) + length = self._recv_int() + value = self.transport.recv(length) + self._get_sync() + + pieces = value.split(b",") + if len(pieces) >= 2: + return pieces[0].decode("latin-1"), pieces[1].decode("latin-1") + return "unknown", "unknown" + + def _get_version(self) -> str: + """Get bootloader version string. + + Returns: + Version string or "unknown" if not supported + """ + self._send_command(BootloaderCommand.GET_VERSION) + try: + length = self._recv_int() + value = self.transport.recv(length) + self._get_sync() + return value.decode("utf-8", errors="replace") + except (TimeoutError, ProtocolError): + # Older bootloaders don't support this + return "unknown" + + def identify(self) -> None: + """Identify the connected board. + + Queries bootloader for board information and stores in instance + attributes. + + Raises: + ProtocolError: If identification fails or protocol version unsupported + """ + logger.info("Identifying board...") + + self._detect_interface_type() + self.sync() + + # Get bootloader protocol revision + self.bl_rev = self._get_device_info(DeviceInfo.BL_REV) + logger.info(f"Bootloader protocol: v{self.bl_rev}") + + if self.bl_rev < ProtocolConfig.BL_REV_MIN: + raise ProtocolError( + f"Bootloader protocol {self.bl_rev} too old " + f"(minimum {ProtocolConfig.BL_REV_MIN})", + port=self.transport.port_name, + ) + if self.bl_rev > ProtocolConfig.BL_REV_MAX: + logger.warning( + f"Bootloader protocol {self.bl_rev} newer than supported " + f"({ProtocolConfig.BL_REV_MAX}), proceeding with caution" + ) + + # Get board info + self.board_type = self._get_device_info(DeviceInfo.BOARD_ID) + self.board_rev = self._get_device_info(DeviceInfo.BOARD_REV) + self.fw_maxsize = self._get_device_info(DeviceInfo.FLASH_SIZE) + + logger.info(f"Board type: {self.board_type}, revision: {self.board_rev}") + logger.info(f"Flash size: {self.fw_maxsize} bytes") + + # Get version string (v5+) + if self.bl_rev >= 5: + self.version = self._get_version() + logger.info(f"Bootloader version: {self.version}") + + # Get OTP and serial number (v4+) + if self.bl_rev >= 4: + self._read_otp_and_sn() + + # Get chip info (v5+) + if self.bl_rev >= 5: + self._read_chip_info() + + def _read_otp_and_sn(self) -> None: + """Read OTP and serial number data.""" + # Read OTP (32*6 = 192 bytes) + otp_data = bytearray() + for addr in range(0, 32 * 6, 4): + otp_data.extend(self._get_otp(addr)) + self.otp = bytes(otp_data) + + # Read serial number (12 bytes) + sn_data = bytearray() + for addr in range(0, 12, 4): + sn_bytes = self._get_sn(addr) + sn_data.extend(sn_bytes[::-1]) # Reverse byte order + self.sn = bytes(sn_data) + + logger.debug(f"Serial number: {self.sn.hex()}") + + # Try to get chip ID + try: + self.chip_id = self._get_chip() + logger.debug(f"Chip ID: 0x{self.chip_id:08X}") + except (TimeoutError, ProtocolError) as e: + logger.debug(f"Could not read chip ID: {e}") + + def _read_chip_info(self) -> None: + """Read chip family and revision (v5+).""" + try: + self.chip_family, self.chip_revision = self._get_chip_description() + logger.info(f"Chip: {self.chip_family} rev {self.chip_revision}") + except (TimeoutError, ProtocolError) as e: + logger.debug(f"Could not read chip description: {e}") + + def erase( + self, force_full: bool = False, progress_callback: Optional[callable] = None + ) -> None: + """Erase the flash memory. + + Args: + force_full: Force full chip erase (v6+) + progress_callback: Optional callback(progress, total) for progress + + Raises: + TimeoutError: If erase times out + ProtocolError: If erase fails + """ + logger.debug("Erasing flash") + + if force_full and self.bl_rev >= 6: + logger.debug("Using full chip erase") + self._send_command(BootloaderCommand.CHIP_FULL_ERASE) + else: + self._send_command(BootloaderCommand.CHIP_ERASE) + + # Erase can take a long time, poll for completion + deadline = time.monotonic() + self.erase_timeout + usual_duration = 15.0 + + while time.monotonic() < deadline: + elapsed = time.monotonic() - (deadline - self.erase_timeout) + remaining = deadline - time.monotonic() + + if progress_callback: + if remaining >= usual_duration: + progress_callback(elapsed, usual_duration) + else: + progress_callback(usual_duration, usual_duration) + + if self._try_sync(): + logger.debug("Erase complete") + if progress_callback: + progress_callback(1.0, 1.0) + return + + raise TimeoutError( + f"Erase timed out after {self.erase_timeout}s", + port=self.transport.port_name, + operation="erase", + ) + + def program( + self, firmware: Firmware, progress_callback: Optional[callable] = None + ) -> None: + """Program firmware to flash. + + Args: + firmware: Firmware instance to program + progress_callback: Optional callback(bytes_written, total_bytes) + + Raises: + ProtocolError: If programming fails + """ + image = firmware.image + total = len(image) + written = 0 + + logger.debug(f"Programming {total} bytes") + + # Split image into chunks + chunk_size = ProtocolConfig.PROG_MULTI_MAX + chunks = [image[i : i + chunk_size] for i in range(0, total, chunk_size)] + + for i, chunk in enumerate(chunks): + self._program_multi(chunk) + + if self._windowed_mode: + self._window_size += self._window_per + + # Periodically validate window + if (i + 1) % 256 == 0: + self._validate_sync_window(self._window_size) + self._window_size = 0 + else: + self._get_sync(flush=False) + + written += len(chunk) + if progress_callback: + progress_callback(written, total) + + # Validate any remaining window + if self._windowed_mode and self._window_size > 0: + self._validate_sync_window(self._window_size) + self._window_size = 0 + + logger.debug("Programming complete") + + def _program_multi(self, data: bytes) -> None: + """Program a chunk of data. + + Args: + data: Bytes to program (max PROG_MULTI_MAX) + """ + length = len(data) + cmd = bytes([BootloaderCommand.PROG_MULTI, length]) + data + cmd += bytes([BootloaderResponse.EOC]) + self.transport.send(cmd) + + if self._windowed_mode: + # Delay based on transmission time plus flash programming time + time.sleep(length * self.transport.chartime + 0.001) + + def verify_crc( + self, firmware: Firmware, progress_callback: Optional[callable] = None + ) -> None: + """Verify programmed firmware using CRC (v3+). + + Args: + firmware: Firmware instance to verify against + progress_callback: Optional callback for progress + + Raises: + ProtocolError: If verification fails + """ + if self.bl_rev < 3: + raise ProtocolError( + "CRC verification requires bootloader v3+", + port=self.transport.port_name, + ) + + logger.debug("Verifying CRC") + + expected_crc = firmware.crc(self.fw_maxsize) + logger.debug(f"Expected CRC: 0x{expected_crc:08X}") + + self._send_command(BootloaderCommand.GET_CRC) + + # CRC calculation takes time, especially on larger flash + time.sleep(0.5) + + if progress_callback: + progress_callback(0.5, 1.0) + + reported_crc = self._recv_int() + self._get_sync() + + if progress_callback: + progress_callback(1.0, 1.0) + + logger.debug(f"Reported CRC: 0x{reported_crc:08X}") + + if reported_crc != expected_crc: + raise ProtocolError( + f"CRC mismatch: expected 0x{expected_crc:08X}, " + f"got 0x{reported_crc:08X}", + port=self.transport.port_name, + operation="verify", + ) + + logger.debug("CRC verification passed") + + def verify_read( + self, firmware: Firmware, progress_callback: Optional[callable] = None + ) -> None: + """Verify programmed firmware by reading back (v2). + + Args: + firmware: Firmware instance to verify against + progress_callback: Optional callback(bytes_verified, total_bytes) + + Raises: + ProtocolError: If verification fails + """ + logger.debug("Verifying by read-back") + + self._send_command(BootloaderCommand.CHIP_VERIFY) + self._get_sync() + + image = firmware.image + total = len(image) + verified = 0 + + chunk_size = ProtocolConfig.READ_MULTI_MAX + chunks = [image[i : i + chunk_size] for i in range(0, total, chunk_size)] + + for chunk in chunks: + length = len(chunk) + cmd = bytes([BootloaderCommand.READ_MULTI, length]) + cmd += bytes([BootloaderResponse.EOC]) + self.transport.send(cmd) + self.transport.flush() + + readback = self.transport.recv(length) + self._get_sync() + + if readback != chunk: + logger.error(f"Verify failed at offset {verified}") + logger.debug(f"Expected: {chunk.hex()}") + logger.debug(f"Got: {readback.hex()}") + raise ProtocolError( + "Verification failed", + port=self.transport.port_name, + operation="verify", + ) + + verified += length + if progress_callback: + progress_callback(verified, total) + + logger.debug("Read-back verification passed") + + def verify( + self, firmware: Firmware, progress_callback: Optional[callable] = None + ) -> None: + """Verify programmed firmware using appropriate method. + + Uses CRC for v3+ bootloaders, read-back for v2. + + Args: + firmware: Firmware to verify against + progress_callback: Optional progress callback + """ + if self.bl_rev >= 3: + self.verify_crc(firmware, progress_callback) + else: + self.verify_read(firmware, progress_callback) + + def set_boot_delay(self, delay_ms: int) -> None: + """Set boot delay in flash (v5+). + + Args: + delay_ms: Boot delay in milliseconds + """ + if self.bl_rev < 5: + logger.warning("Boot delay requires bootloader v5+") + return + + self._send_command(BootloaderCommand.SET_BOOT_DELAY, struct.pack("b", delay_ms)) + self._get_sync() + logger.info(f"Boot delay set to {delay_ms}ms") + + def reboot(self) -> None: + """Reboot into the application. + + Raises: + ProtocolError: If reboot fails (v3+ validates first flash word) + """ + logger.info("Rebooting to application") + self._send_command(BootloaderCommand.REBOOT) + self.transport.flush() + + # v3+ can report failure if first flash word is invalid + if self.bl_rev >= 3: + try: + self._get_sync() + except TimeoutError: + # Timeout is expected - board is rebooting + pass + + def send_reboot_commands( + self, baudrates: list[int], use_protocol_splitter: bool = False + ) -> bool: + """Send reboot commands to try to enter bootloader. + + Tries MAVLink and NSH reboot commands at various baud rates. + + Args: + baudrates: List of baud rates to try + use_protocol_splitter: Use protocol splitter framing + + Returns: + True if commands were sent, False if no more baud rates to try + """ + for baudrate in baudrates: + try: + self.transport.set_baudrate(baudrate) + except (serial.SerialException, NotImplementedError): + continue + + logger.info(f"Sending reboot command at {baudrate} baud") + + def send(data: bytes) -> None: + if use_protocol_splitter: + self._send_protocol_splitter_frame(data) + else: + self.transport.send(data) + + try: + self.transport.flush() + send(self.MAVLINK_REBOOT_ID0) + send(self.MAVLINK_REBOOT_ID1) + send(self.NSH_INIT) + send(self.NSH_REBOOT_BL) + send(self.NSH_INIT) + send(self.NSH_REBOOT) + self.transport.flush() + except Exception as e: + logger.debug(f"Error sending reboot: {e}") + continue + + return True + + return False + + def _send_protocol_splitter_frame(self, data: bytes) -> None: + """Send data with protocol splitter framing. + + Header format: + - Byte 0: Magic ('S' = 0x53) + - Byte 1: Type (0) | Length high bits (7 bits) + - Byte 2: Length low bits + - Byte 3: Checksum (XOR of bytes 0-2) + """ + magic = 0x53 + len_h = (len(data) >> 8) & 0x7F + len_l = len(data) & 0xFF + checksum = magic ^ len_h ^ len_l + + header = bytes([magic, len_h, len_l, checksum]) + self.transport.send(header + data) + + +# ============================================================================= +# Port Detection +# ============================================================================= + + +class PortDetector: + """Detects PX4-compatible serial ports.""" + + # Platform-specific port patterns + LINUX_PATTERNS = [ + "/dev/serial/by-id/*PX4*", + "/dev/serial/by-id/*px4*", + "/dev/serial/by-id/*3D_Robotics*", + "/dev/serial/by-id/*Autopilot*", + "/dev/serial/by-id/*Bitcraze*", + "/dev/serial/by-id/*Gumstix*", + "/dev/serial/by-id/*Hex*", + "/dev/serial/by-id/*Holybro*", + "/dev/serial/by-id/*Cube*", + "/dev/serial/by-id/*ArduPilot*", + "/dev/serial/by-id/*BL_FMU*", + "/dev/serial/by-id/*_BL*", + "/dev/ttyACM*", + "/dev/ttyUSB*", + ] + + MACOS_PATTERNS = [ + "/dev/tty.usbmodemPX*", + "/dev/tty.usbmodem*", + "/dev/cu.usbmodemPX*", + "/dev/cu.usbmodem*", + ] + + WINDOWS_PATTERNS = [ + "COM*", + ] + + def __init__(self): + self.platform = sys.platform + + def detect_ports(self) -> list[str]: + """Detect available PX4-compatible serial ports. + + Returns: + List of port paths, prioritized by likelihood of being PX4 + """ + ports = set() + + # First, try USB VID/PID detection + vid_pid_ports = self._detect_by_vid_pid() + ports.update(vid_pid_ports) + + # Then try platform-specific patterns + pattern_ports = self._detect_by_patterns() + ports.update(pattern_ports) + + # Sort by priority (VID/PID matches first) + result = [] + for port in vid_pid_ports: + if port in ports: + result.append(port) + ports.discard(port) + result.extend(sorted(ports)) + + logger.info(f"Detected {len(result)} potential ports: {result}") + return result + + def _detect_by_vid_pid(self) -> list[str]: + """Detect ports by USB Vendor/Product ID. + + Returns: + List of ports matching known PX4 VID/PIDs + """ + ports = [] + known_ids = {(vid, pid) for vid, pid, _ in PX4_USB_IDS} + + try: + for port_info in serial.tools.list_ports.comports(): + if (port_info.vid, port_info.pid) in known_ids: + logger.debug( + f"Found PX4 device: {port_info.device} " + f"(VID=0x{port_info.vid:04X}, PID=0x{port_info.pid:04X})" + ) + ports.append(port_info.device) + except Exception as e: + logger.debug(f"VID/PID detection failed: {e}") + + return ports + + def _detect_by_patterns(self) -> list[str]: + """Detect ports by platform-specific glob patterns. + + Returns: + List of ports matching patterns + """ + if self.platform.startswith("linux"): + patterns = self.LINUX_PATTERNS + elif self.platform == "darwin": + patterns = self.MACOS_PATTERNS + elif self.platform.startswith("win") or self.platform == "cygwin": + patterns = self.WINDOWS_PATTERNS + else: + patterns = [] + + ports = [] + for pattern in patterns: + matches = glob.glob(pattern) + ports.extend(matches) + + return list(set(ports)) + + def expand_patterns(self, patterns: list[str]) -> list[str]: + """Expand glob patterns to actual port paths. + + Args: + patterns: List of port paths or glob patterns + + Returns: + List of expanded port paths + """ + ports = [] + for pattern in patterns: + if "*" in pattern or "?" in pattern: + matches = glob.glob(pattern) + if matches: + ports.extend(matches) + else: + logger.debug(f"Pattern matched no ports: {pattern}") + else: + ports.append(pattern) + + return list(set(ports)) + + +# ============================================================================= +# Progress Display +# ============================================================================= + + +class UploadProgressBar: + """Unified progress bar for the entire upload process. + + Shows a single progress bar with phases: + - Erase: 0-49% + - Program: 50-99% + - Verify: 99-100% + """ + + ERASE_START = 0 + ERASE_END = 49 + PROGRAM_START = 50 + PROGRAM_END = 99 + VERIFY_START = 99 + VERIFY_END = 100 + + def __init__(self, noninteractive: bool = False, json_output: bool = False): + # Use noninteractive mode if flag is set OR if not a TTY + self._json = json_output + self._noninteractive = json_output or noninteractive or not sys.stdout.isatty() + self._start_time = time.monotonic() + self._last_percent = -1 + self._last_printed_percent = -1 + self._phase = "Erase" + + def _render(self, percent: int) -> None: + """Render the progress bar.""" + if not self._noninteractive and percent == self._last_percent and percent < 100: + return + + if self._noninteractive: + # Print at 5% increments (0, 5, 10, ...) plus 99% for verify + next_milestone = ((self._last_printed_percent // 5) + 1) * 5 + if self._last_printed_percent < 0: + next_milestone = 0 + # Special case: print 99% only when in Verify phase + is_verify_99 = ( + self._phase == "Verify" + and self._last_printed_percent < 99 + and percent >= 99 + ) + if is_verify_99: + next_milestone = 99 + should_print = percent >= next_milestone or ( + percent >= 100 and self._last_printed_percent < 100 + ) + # Don't print 99% unless we're in Verify phase + if ( + should_print + and percent >= 99 + and percent < 100 + and self._phase != "Verify" + ): + should_print = False + if should_print: + # Print at the milestone, not the actual percent + if percent >= 100: + print_percent = 100 + elif percent >= 99 and self._phase == "Verify": + print_percent = 99 + else: + print_percent = (percent // 5) * 5 + if print_percent > self._last_printed_percent: + if print_percent >= 100: + phase = "done" + elif self._phase == "Erase": + phase = "erasing" + elif self._phase == "Program": + phase = "programming" + elif self._phase == "Verify": + phase = "verifying" + else: + phase = self._phase.lower() + if self._json: + print( + json.dumps( + { + "type": "progress", + "phase": phase, + "percent": print_percent, + } + ) + ) + else: + print(f"{phase.capitalize()}, progress: {print_percent}%") + self._last_printed_percent = print_percent + self._last_percent = percent + return + + # Step through each percent for smooth animation + if self._last_percent >= 0 and percent > self._last_percent + 1: + for p in range(self._last_percent + 1, percent): + self._render_single(p) + time.sleep(0.02) + + self._render_single(percent) + self._last_percent = percent + + def _render_single(self, percent: int) -> None: + """Render a single frame of the progress bar.""" + bar_width = 30 + filled_exact = bar_width * percent / 100.0 + filled_full = int(filled_exact) + filled_partial = filled_exact - filled_full + + # Unicode block characters for smooth progress + blocks = " ▏▎▍▌▋▊▉█" + partial_idx = int(filled_partial * 8) + + bar = "█" * filled_full + if filled_full < bar_width: + bar += blocks[partial_idx] + bar += " " * (bar_width - filled_full - 1) + + line = f"{self._phase:8s} ▕{bar}▏ {percent:3d}%" + print(f"\r{line}", end="", flush=True) + + def update_erase(self, current: float, total: float) -> None: + """Update progress during erase phase (0-45%).""" + self._phase = "Erase" + if total <= 0: + return + phase_progress = min(current / total, 1.0) + percent = int( + self.ERASE_START + phase_progress * (self.ERASE_END - self.ERASE_START) + ) + self._render(percent) + + def update_program(self, current: float, total: float) -> None: + """Update progress during program phase (50-99%).""" + self._phase = "Program" + if total <= 0: + return + phase_progress = min(current / total, 1.0) + percent = int( + self.PROGRAM_START + + phase_progress * (self.PROGRAM_END - self.PROGRAM_START) + ) + self._render(percent) + + def update_verify(self, current: float, total: float) -> None: + """Update progress during verify phase (90-100%).""" + self._phase = "Verify" + if total <= 0: + return + phase_progress = min(current / total, 1.0) + percent = int( + self.VERIFY_START + phase_progress * (self.VERIFY_END - self.VERIFY_START) + ) + self._render(percent) + + def finish(self) -> None: + """Complete the progress bar and show summary.""" + # Show "Verify" at 100% briefly so user sees verification passed + self._phase = "Verify" + self._last_percent = -1 # Force render + self._render(100) + + elapsed = time.monotonic() - self._start_time + + if self._noninteractive: + if self._json: + print(json.dumps({"type": "complete", "elapsed_seconds": int(elapsed)})) + else: + print(f"\nUploaded in {int(elapsed)}s") + return + + # Interactive mode: show 100% briefly, then clear and print summary + time.sleep(0.5) + print("\r\033[K", end="") + print(f"Uploaded in {int(elapsed)}s") + + +# ============================================================================= +# Uploader +# ============================================================================= + + +@dataclass +class UploaderConfig: + """Configuration for uploader.""" + + port: Optional[str] = None + baud_bootloader: int = 115200 + baud_flightstack: list[int] = field(default_factory=lambda: [57600]) + force: bool = False + force_erase: bool = False + boot_delay: Optional[int] = None + use_protocol_splitter: bool = False + retry_count: int = 3 + windowed: bool = False + noninteractive: bool = False + json_output: bool = False + + +class Uploader: + """Orchestrates firmware upload to PX4 bootloader.""" + + def __init__(self, config: UploaderConfig): + self.config = config + self.port_detector = PortDetector() + + def _print_message(self, message_type: str, **kwargs) -> None: + """Print a message, either as JSON or plain text.""" + if self.config.json_output: + # Format chip_id as hex string for JSON + if "chip_id" in kwargs and kwargs["chip_id"] is not None: + kwargs["chip_id"] = f"0x{kwargs['chip_id']:08X}" + # Remove None values from JSON output + kwargs = {k: v for k, v in kwargs.items() if v is not None} + print(json.dumps({"type": message_type, **kwargs})) + else: + # Format as plain text based on message type + if message_type == "board": + print( + f"\nFound board {kwargs['board_type']},{kwargs['board_rev']} " + f"protocol v{kwargs['protocol_version']} on {kwargs['port']}" + ) + elif message_type == "firmware": + print( + f"\nFirmware: board_id={kwargs['board_id']}, " + f"revision={kwargs['board_revision']}" + ) + print( + f"Size: {kwargs['image_size']} bytes ({kwargs['usage_percent']:.1f}%)" + ) + print(f"Bootloader version: {kwargs['bootloader_version']}") + elif message_type == "board_info": + if kwargs.get("serial"): + print(f"Serial: {kwargs['serial']}") + if kwargs.get("chip_id"): + print(f"Chip: 0x{kwargs['chip_id']:08X}") + if kwargs.get("chip_family"): + print(f"Family: {kwargs['chip_family']}") + if kwargs.get("chip_revision"): + print(f"Revision: {kwargs['chip_revision']}") + print(f"Flash: {kwargs['flash_size']} bytes") + print(f"Windowed mode: {'yes' if kwargs.get('windowed') else 'no'}") + + def upload(self, firmware_paths: list[str]) -> bool: + """Upload firmware to connected board. + + Args: + firmware_paths: List of firmware file paths to try + + Returns: + True if upload successful + + Raises: + UploadError: If upload fails + """ + # Load all firmware files + firmwares = [] + for path in firmware_paths: + try: + fw = Firmware(path) + firmwares.append(fw) + except FirmwareError as e: + logger.error(f"Failed to load {path}: {e}") + if len(firmware_paths) == 1: + raise + + if not firmwares: + raise FirmwareError("No valid firmware files") + + # Determine ports to try + if self.config.port: + patterns = self.config.port.split(",") + ports = self.port_detector.expand_patterns(patterns) + else: + ports = self.port_detector.detect_ports() + + if not ports: + raise ConnectionError("No serial ports found") + + logger.info(f"Trying ports: {ports}") + + # Send MAVLink release command to GCS + self._send_gcs_release() + + # Try each port + last_error = None + for port in ports: + try: + return self._upload_to_port(port, firmwares) + except BoardMismatchError as e: + logger.warning(f"Board mismatch on {port}: {e}") + last_error = e + continue + except (ConnectionError, TimeoutError) as e: + logger.debug(f"Connection failed on {port}: {e}") + last_error = e + continue + except UploadError as e: + logger.error(f"Upload failed on {port}: {e}") + last_error = e + raise + + if last_error: + raise last_error + raise ConnectionError("No bootloader found on any port") + + def _upload_to_port(self, port: str, firmwares: list[Firmware]) -> bool: + """Attempt upload on a specific port. + + Args: + port: Serial port path + firmwares: List of firmware options + + Returns: + True if successful + + Raises: + Various UploadError subclasses on failure + """ + logger.info(f"Trying port {port}") + + transport = SerialTransport( + port, + baudrate=self.config.baud_bootloader, + ) + + try: + transport.open() + except ConnectionError: + return False + + protocol = BootloaderProtocol( + transport, + windowed=self.config.windowed, + ) + + try: + # Try to identify bootloader + if not self._try_identify(transport, protocol): + return False + + # Find matching firmware + firmware = self._select_firmware(firmwares, protocol) + + # Perform upload + self._do_upload(protocol, firmware) + return True + + finally: + transport.close() + + def _try_identify( + self, transport: SerialTransport, protocol: BootloaderProtocol + ) -> bool: + """Try to identify the bootloader, sending reboot if needed. + + Args: + transport: Serial transport + protocol: Bootloader protocol handler + + Returns: + True if bootloader identified + """ + # First try to identify without reboot + try: + protocol.identify() + self._print_message( + "board", + board_type=protocol.board_type, + board_rev=protocol.board_rev, + protocol_version=protocol.bl_rev, + port=transport.port_name, + ) + return True + except (ProtocolError, TimeoutError): + pass + + # Try rebooting at each baud rate + for baud in self.config.baud_flightstack: + if not self.config.json_output: + print( + f"Attempting reboot on {transport.port_name} at {baud} baud...", + file=sys.stderr, + ) + + try: + transport.set_baudrate(baud) + except Exception: + continue + + # Send reboot commands multiple times to increase reliability + # The board might be busy and miss the first command + for attempt in range(3): + try: + transport.reset_buffers() + + # Send MAVLink reboot-to-bootloader commands + # Send broadcast (0/0) first, then targeted (1/0) + transport.send(protocol.MAVLINK_REBOOT_ID0) + transport.send(protocol.MAVLINK_REBOOT_ID1) + transport.flush() + + # Give MAVLink stack time to process + time.sleep(0.1) + + # Send NSH reboot-to-bootloader command + transport.send(protocol.NSH_INIT) + time.sleep(0.05) + transport.send(protocol.NSH_REBOOT_BL) + transport.flush() + + time.sleep(0.2) + except Exception: + pass + + # Wait for reboot - give the board time to process and restart + time.sleep(0.5) + transport.close() + time.sleep(0.5) + + # Reopen at bootloader baud rate and try to identify + try: + transport.set_baudrate(self.config.baud_bootloader) + transport.open() + except Exception: + continue + + # Try to identify multiple times - board may take time to enter bootloader + for identify_attempt in range(5): + try: + protocol.identify() + self._print_message( + "board", + board_type=protocol.board_type, + board_rev=protocol.board_rev, + protocol_version=protocol.bl_rev, + port=transport.port_name, + ) + return True + except (ProtocolError, TimeoutError): + # Board may still be rebooting, wait a bit and retry + time.sleep(0.3) + + return False + + def _select_firmware( + self, firmwares: list[Firmware], protocol: BootloaderProtocol + ) -> Firmware: + """Select appropriate firmware for the board. + + Args: + firmwares: Available firmware options + protocol: Protocol with board info + + Returns: + Selected firmware + + Raises: + BoardMismatchError: If no suitable firmware + """ + for fw in firmwares: + if fw.board_id == protocol.board_type: + if len(firmwares) > 1: + print(f"Using firmware {fw.path}") + return fw + + if self.config.force and len(firmwares) == 1: + print( + f"WARNING: Firmware board_id={firmwares[0].board_id} " + f"does not match device board_id={protocol.board_type}" + ) + print("FORCED UPLOAD, FLASHING ANYWAY!") + return firmwares[0] + + raise BoardMismatchError( + f"No suitable firmware for board {protocol.board_type}", + details=f"available: {[fw.board_id for fw in firmwares]}", + ) + + def _do_upload(self, protocol: BootloaderProtocol, firmware: Firmware) -> None: + """Perform the actual upload sequence. + + Args: + protocol: Bootloader protocol handler + firmware: Firmware to upload + """ + # Print firmware info + self._print_message( + "firmware", + board_id=firmware.board_id, + board_revision=firmware.board_revision, + image_size=firmware.image_size, + usage_percent=firmware.usage_percent, + bootloader_version=protocol.version, + ) + + # Check for silicon errata (bootloader v4 on Pixhawk) + if protocol.bl_rev == 4 and firmware.board_id == 9: + if firmware.image_size > 1032192 and not self.config.force: + raise SiliconErrataError( + "Board uses bootloader v4 and cannot safely flash >1MB.\n" + "Use px4_fmu-v2_default or update the bootloader.\n" + "Use --force to override if you know the board is safe." + ) + + # Check flash size + if protocol.fw_maxsize < firmware.image_size: + raise FirmwareError( + f"Firmware too large ({firmware.image_size} bytes) " + f"for flash ({protocol.fw_maxsize} bytes)" + ) + + # Check for undersized config + if ( + protocol.bl_rev >= 5 + and protocol.fw_maxsize > firmware.image_maxsize + and not self.config.force + ): + print( + f"WARNING: Board flash ({protocol.fw_maxsize} bytes) " + f"larger than firmware config ({firmware.image_maxsize} bytes)" + ) + + # Print OTP/SN info + self._print_board_info(protocol) + + # Create unified progress bar + if not self.config.json_output: + print() + progress = UploadProgressBar( + noninteractive=self.config.noninteractive, + json_output=self.config.json_output, + ) + + # Erase + protocol.erase( + force_full=self.config.force_erase, + progress_callback=progress.update_erase, + ) + + # Program + protocol.program(firmware, progress_callback=progress.update_program) + + # Verify + protocol.verify(firmware, progress_callback=progress.update_verify) + + # Set boot delay if requested + if self.config.boot_delay is not None: + protocol.set_boot_delay(self.config.boot_delay) + + # Reboot and show summary + protocol.reboot() + progress.finish() + + def _print_board_info(self, protocol: BootloaderProtocol) -> None: + """Print board OTP and chip info.""" + self._print_message( + "board_info", + serial=protocol.sn.hex() if protocol.sn else None, + chip_id=protocol.chip_id if protocol.chip_id else None, + chip_family=protocol.chip_family if protocol.chip_family else None, + chip_revision=protocol.chip_revision if protocol.chip_revision else None, + flash_size=protocol.fw_maxsize, + windowed=protocol._windowed_mode, + ) + + def _send_gcs_release(self) -> None: + """Send UDP message to release serial port from GCS.""" + try: + heartbeat = bytes.fromhex("fe097001010000000100020c5103033c8a") + command = bytes.fromhex( + "fe210101014c0000000000000000000000000000000000" + "00000000000000803f00000000f6000000008459" + ) + + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.sendto(heartbeat, ("127.0.0.1", 14550)) + sock.sendto(command, ("127.0.0.1", 14550)) + sock.close() + except Exception: + pass # Non-critical + + +# ============================================================================= +# Main Entry Point +# ============================================================================= + + +def main() -> int: + """Main entry point.""" + parser = argparse.ArgumentParser( + description="PX4 Firmware Uploader v2", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + %(prog)s firmware.px4 + %(prog)s --port /dev/ttyACM0 firmware.px4 + %(prog)s --port /dev/serial/by-id/*PX4* firmware.px4 + %(prog)s -v --force firmware.px4 + """, + ) + + parser.add_argument("firmware", nargs="+", help="Firmware file(s) to upload") + parser.add_argument( + "--port", + "-p", + help="Serial port(s) to use (comma-separated, supports wildcards). " + "If not specified, auto-detects PX4 devices.", + ) + parser.add_argument( + "--baud-bootloader", + type=int, + default=115200, + help="Bootloader baud rate (default: 115200)", + ) + parser.add_argument( + "--baud-flightstack", + default="57600", + help="Flight stack baud rate(s) for reboot (comma-separated, default: 57600)", + ) + parser.add_argument( + "--force", + "-f", + action="store_true", + help="Force upload even if board ID doesn't match", + ) + parser.add_argument( + "--force-erase", + action="store_true", + help="Force full chip erase (v6+ bootloader)", + ) + parser.add_argument( + "--boot-delay", type=int, help="Boot delay in milliseconds to store in flash" + ) + parser.add_argument( + "--use-protocol-splitter-format", + action="store_true", + help="Use protocol splitter framing for reboot commands", + ) + parser.add_argument( + "--windowed", + action="store_true", + help="Use windowed mode for faster uploads on real serial ports (FTDI)", + ) + parser.add_argument( + "--verbose", "-v", action="store_true", help="Enable verbose output" + ) + parser.add_argument( + "--debug", + "-d", + action="store_true", + help="Enable debug output (includes protocol traces)", + ) + parser.add_argument( + "--noninteractive", + action="store_true", + help="Non-interactive mode: print progress every 5%% for tools to parse", + ) + parser.add_argument( + "--noninteractive-json", + action="store_true", + help="Non-interactive JSON mode: print progress as JSON lines", + ) + + args = parser.parse_args() + + # Setup logging + setup_logging(verbose=args.verbose, debug=args.debug) + + # Warn about ModemManager on Linux + if ( + not args.noninteractive_json + and sys.platform.startswith("linux") + and os.path.exists("/usr/sbin/ModemManager") + ): + print("=" * 80) + print("WARNING: ModemManager detected. It may interfere with PX4 devices.") + print("Consider: sudo systemctl disable ModemManager") + print("=" * 80) + + # Parse baud rates + baud_flightstack = [int(x) for x in args.baud_flightstack.split(",")] + + # Create config + config = UploaderConfig( + port=args.port, + baud_bootloader=args.baud_bootloader, + baud_flightstack=baud_flightstack, + force=args.force, + force_erase=args.force_erase, + boot_delay=args.boot_delay, + use_protocol_splitter=args.use_protocol_splitter_format, + windowed=args.windowed, + noninteractive=args.noninteractive or args.noninteractive_json, + json_output=args.noninteractive_json, + ) + + if not args.noninteractive_json: + if args.use_protocol_splitter_format: + print("Using protocol splitter format for reboot commands") + print("Waiting for bootloader...") + + uploader = Uploader(config) + + try: + # Keep trying until we find a board or user interrupts + while True: + try: + if uploader.upload(args.firmware): + return 0 + except BoardMismatchError: + # No suitable firmware for this board + return 2 + except (ConnectionError, TimeoutError): + # No device found yet, keep trying + time.sleep(0.05) + except UploadError as e: + if args.noninteractive_json: + print(json.dumps({"type": "error", "message": str(e)})) + else: + print(f"\nError: {e}", file=sys.stderr) + return 1 + + except KeyboardInterrupt: + if args.noninteractive_json: + print(json.dumps({"type": "error", "message": "Upload aborted by user"})) + else: + print("\nUpload aborted by user.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/Tools/px_uploader.py b/Tools/px_uploader.py deleted file mode 100755 index 24b6511b6dd..00000000000 --- a/Tools/px_uploader.py +++ /dev/null @@ -1,947 +0,0 @@ -#!/usr/bin/env python3 -############################################################################ -# -# Copyright (c) 2012-2024 PX4 Development Team. All rights reserved. -# -# Redistribution and use in source and binary forms, with or without -# modification, are permitted provided that the following conditions -# are met: -# -# 1. Redistributions of source code must retain the above copyright -# notice, this list of conditions and the following disclaimer. -# 2. Redistributions in binary form must reproduce the above copyright -# notice, this list of conditions and the following disclaimer in -# the documentation and/or other materials provided with the -# distribution. -# 3. Neither the name PX4 nor the names of its contributors may be -# used to endorse or promote products derived from this software -# without specific prior written permission. -# -# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS -# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT -# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS -# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE -# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, -# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, -# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS -# OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED -# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT -# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN -# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE -# POSSIBILITY OF SUCH DAMAGE. -# -############################################################################ - -# -# Serial firmware uploader for the PX4FMU bootloader -# -# The PX4 firmware file is a JSON-encoded Python object, containing -# metadata fields and a zlib-compressed base64-encoded firmware image. -# -# The uploader uses the following fields from the firmware file: -# -# image -# The firmware that will be uploaded. -# image_size -# The size of the firmware in bytes. -# board_id -# The board for which the firmware is intended. -# board_revision -# Currently only used for informational purposes. -# - -import sys -import argparse -import binascii -import socket -import struct -import json -import zlib -import base64 -import time -import array -import os - -from sys import platform as _platform - -try: - import serial -except ImportError as e: - print(f"Failed to import serial: {e}") - print("") - print("You may need to install it using:") - print(" python -m pip install pyserial") - print("") - sys.exit(1) - - -# Detect python version -if sys.version_info[0] < 3: - raise RuntimeError("Python 2 is not supported. Please try again using Python 3.") - sys.exit(1) - - -# Use monotonic time where available -def _time(): - try: - return time.monotonic() - except Exception: - return time.time() - -class FirmwareNotSuitableException(Exception): - def __init__(self, message): - super(FirmwareNotSuitableException, self).__init__(message) - - -class firmware(object): - '''Loads a firmware file''' - - desc = {} - image = bytes() - - def __init__(self, path): - - # read the file - f = open(path, "r") - self.desc = json.load(f) - f.close() - - self.image = bytearray(zlib.decompress(base64.b64decode(self.desc['image']))) - - # pad image to 4-byte length - while ((len(self.image) % 4) != 0): - self.image.extend(b'\xff') - - def property(self, propname): - return self.desc[propname] - - def crc(self, padlen): - state = 0xFFFFFFFF - state = zlib.crc32(self.image, state) - padding_length = padlen - len(self.image) - if padding_length > 0: - padding = b'\xff' * padding_length - state = zlib.crc32(padding, state) - - return (state ^ 0xFFFFFFFF) & 0xFFFFFFFF - - -class uploader: - '''Uploads a firmware file to the PX4 bootloader''' - - # protocol bytes - INSYNC = b'\x12' - EOC = b'\x20' - - # reply bytes - OK = b'\x10' - FAILED = b'\x11' - INVALID = b'\x13' # rev3+ - BAD_SILICON_REV = b'\x14' # rev5+ - - # command bytes - NOP = b'\x00' # guaranteed to be discarded by the bootloader - GET_SYNC = b'\x21' - GET_DEVICE = b'\x22' - CHIP_ERASE = b'\x23' - CHIP_VERIFY = b'\x24' # rev2 only - PROG_MULTI = b'\x27' - READ_MULTI = b'\x28' # rev2 only - GET_CRC = b'\x29' # rev3+ - GET_OTP = b'\x2a' # rev4+ , get a word from OTP area - GET_SN = b'\x2b' # rev4+ , get a word from SN area - GET_CHIP = b'\x2c' # rev5+ , get chip version - SET_BOOT_DELAY = b'\x2d' # rev5+ , set boot delay - GET_CHIP_DES = b'\x2e' # rev5+ , get chip description in ASCII - GET_VERSION = b'\x2f' # rev5+ , get bootloader version in ASCII - CHIP_FULL_ERASE = b'\x40' # full erase of flash, rev6+ - MAX_DES_LENGTH = 20 - - REBOOT = b'\x30' - - INFO_BL_REV = b'\x01' # bootloader protocol revision - BL_REV_MIN = 2 # minimum supported bootloader protocol - BL_REV_MAX = 5 # maximum supported bootloader protocol - INFO_BOARD_ID = b'\x02' # board type - INFO_BOARD_REV = b'\x03' # board revision - INFO_FLASH_SIZE = b'\x04' # max firmware size in bytes - - PROG_MULTI_MAX = 252 # protocol max is 255, must be multiple of 4 - READ_MULTI_MAX = 252 # protocol max is 255 - - NSH_INIT = bytearray(b'\x0d\x0d\x0d') - NSH_REBOOT_BL = b"reboot -b\n" - NSH_REBOOT = b"reboot\n" - MAVLINK_REBOOT_ID1 = bytearray(b'\xfe\x21\x72\xff\x00\x4c\x00\x00\x40\x40\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xf6\x00\x01\x00\x00\x53\x6b') - MAVLINK_REBOOT_ID0 = bytearray(b'\xfe\x21\x45\xff\x00\x4c\x00\x00\x40\x40\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xf6\x00\x00\x00\x00\xcc\x37') - - MAX_FLASH_PRGRAM_TIME = 0.001 # Time on an F7 to send SYNC, RESULT from last data in multi RXed - - def __init__(self, portname, baudrate_bootloader, baudrate_flightstack): - # Open the port, keep the default timeout short so we can poll quickly. - # On some systems writes can suddenly get stuck without having a - # write_timeout > 0 set. - # chartime 8n1 * bit rate is us - self.chartime = 10 * (1.0 / baudrate_bootloader) - - # we use a window approche to SYNC, gathring - self.window = 0 - self.window_max = 256 - self.window_per = 2 # Sync, - self.ackWindowedMode = False # Assume Non Widowed mode for all USB CDC - self.port = serial.Serial(portname, baudrate_bootloader, timeout=0.5, write_timeout=0) - self.otp = b'' - self.sn = b'' - self.baudrate_bootloader = baudrate_bootloader - self.baudrate_flightstack = baudrate_flightstack - self.baudrate_flightstack_idx = -1 - self.force_erase = False - - def close(self): - if self.port is not None: - self.port.close() - - def open(self): - # upload timeout - timeout = _time() + 0.2 - - # attempt to open the port while it exists and until timeout occurs - while self.port is not None: - portopen = True - try: - portopen = self.port.is_open - except AttributeError: - portopen = self.port.isOpen() - - if not portopen and _time() < timeout: - try: - self.port.open() - except OSError: - # wait for the port to be ready - time.sleep(0.04) - except serial.SerialException: - # if open fails, try again later - time.sleep(0.04) - else: - break - - # debugging code - def __probe(self, state): - # self.port.setRTS(state) - return - - def __send(self, c): - # print("send " + binascii.hexlify(c)) - self.port.write(c) - - def __recv(self, count=1): - c = self.port.read(count) - if len(c) < 1: - raise RuntimeError("timeout waiting for data (%u bytes)" % count) - # print("recv " + binascii.hexlify(c)) - return c - - def __recv_int(self): - raw = self.__recv(4) - val = struct.unpack(" for a window of programing - # in about 13.81 Ms for 256 blocks written - def __ackSyncWindow(self, count): - if (count > 0): - data = bytearray(bytes(self.__recv(count))) - if (len(data) != count): - raise RuntimeError("Ack Window %i not %i " % (len(data), count)) - for i in range(0, len(data), 2): - if bytes([data[i]]) != self.INSYNC: - raise RuntimeError("unexpected %s instead of INSYNC" % data[i]) - if bytes([data[i+1]]) == self.INVALID: - raise RuntimeError("bootloader reports INVALID OPERATION") - if bytes([data[i+1]]) == self.FAILED: - raise RuntimeError("bootloader reports OPERATION FAILED") - if bytes([data[i+1]]) != self.OK: - raise RuntimeError("unexpected response 0x%x instead of OK" % ord(data[i+1])) - - # attempt to get back into sync with the bootloader - def __sync(self): - # send a stream of ignored bytes longer than the longest possible conversation - # that we might still have in progress - # self.__send(uploader.NOP * (uploader.PROG_MULTI_MAX + 2)) - self.port.flushInput() - self.__send(uploader.GET_SYNC + - uploader.EOC) - self.__getSync() - - def __trySync(self): - try: - self.port.flush() - if (self.__recv() != self.INSYNC): - # print("unexpected 0x%x instead of INSYNC" % ord(c)) - return False - c = self.__recv() - if (c == self.BAD_SILICON_REV): - raise NotImplementedError() - if (c != self.OK): - # print("unexpected 0x%x instead of OK" % ord(c)) - return False - return True - - except NotImplementedError: - raise RuntimeError("Programing not supported for this version of silicon!\n" - "See https://docs.px4.io/main/en/flight_controller/silicon_errata.html") - except RuntimeError: - # timeout, no response yet - return False - - # attempt to determins if the device is CDCACM or A FTDI - def __determineInterface(self): - self.port.flushInput() - # Set a baudrate that can not work on a real serial port - # in that it is 233% off. - try: - self.port.baudrate = self.baudrate_bootloader * 2.33 - except NotImplementedError as e: - # This error can occur because pySerial on Windows does not support odd baudrates - print(f"{e} -> could not check for FTDI device, assuming USB connection") - return - - self.__send(uploader.GET_SYNC + - uploader.EOC) - try: - self.__getSync(False) - except RuntimeError: - # if it fails we are on a real serial port - only leave this enabled on Windows - if sys.platform.startswith('win'): - self.ackWindowedMode = True - finally: - try: - self.port.baudrate = self.baudrate_bootloader - except Exception: - pass - - # send the GET_DEVICE command and wait for an info parameter - def __getInfo(self, param): - self.__send(uploader.GET_DEVICE + param + uploader.EOC) - value = self.__recv_int() - self.__getSync() - return value - - # send the GET_OTP command and wait for an info parameter - def __getOTP(self, param): - t = struct.pack("I", param) # int param as 32bit ( 4 byte ) char array. - self.__send(uploader.GET_OTP + t + uploader.EOC) - value = self.__recv(4) - self.__getSync() - return value - - # send the GET_SN command and wait for an info parameter - def __getSN(self, param): - t = struct.pack("I", param) # int param as 32bit ( 4 byte ) char array. - self.__send(uploader.GET_SN + t + uploader.EOC) - value = self.__recv(4) - self.__getSync() - return value - - # send the GET_CHIP command - def __getCHIP(self): - self.__send(uploader.GET_CHIP + uploader.EOC) - value = self.__recv_int() - self.__getSync() - return value - - # send the GET_CHIP command - def __getCHIPDes(self): - self.__send(uploader.GET_CHIP_DES + uploader.EOC) - length = self.__recv_int() - value = self.__recv(length) - self.__getSync() - pieces = value.split(b",") - return pieces - - def __getVersion(self): - self.__send(uploader.GET_VERSION + uploader.EOC) - try: - length = self.__recv_int() - value = self.__recv(length) - self.__getSync() - except RuntimeError: - # Bootloader doesn't support version call - return "unknown" - return value.decode() - - def __drawProgressBar(self, label, progress, maxVal): - if maxVal < progress: - progress = maxVal - - percent = (float(progress) / float(maxVal)) * 100.0 - - redraw = "\r" if sys.stdout.isatty() else "\n" - sys.stdout.write("%s%s: [%-20s] %.1f%%" % (redraw, label, '='*int(percent/5.0), percent)) - sys.stdout.flush() - - # send the CHIP_ERASE command and wait for the bootloader to become ready - def __erase(self, label): - print(f"Windowed mode: {self.ackWindowedMode}") - print("\n", end='') - - if self.force_erase: - print("Trying force erase of full chip...\n") - self.__send(uploader.CHIP_FULL_ERASE + - uploader.EOC) - else: - self.__send(uploader.CHIP_ERASE + - uploader.EOC) - - # erase is very slow, give it 30s - deadline = _time() + 30.0 - while _time() < deadline: - - usualEraseDuration = 15.0 - estimatedTimeRemaining = deadline-_time() - if estimatedTimeRemaining >= usualEraseDuration: - self.__drawProgressBar(label, 30.0-estimatedTimeRemaining, usualEraseDuration) - else: - self.__drawProgressBar(label, 10.0, 10.0) - sys.stdout.write(" (timeout: %d seconds) " % int(deadline-_time())) - sys.stdout.flush() - - if self.__trySync(): - self.__drawProgressBar(label, 10.0, 10.0) - if self.force_erase: - print("\nForce erase done.\n") - return - - if self.force_erase: - raise RuntimeError("timed out waiting for erase, force erase is likely not supported by bootloader!") - else: - raise RuntimeError("timed out waiting for erase") - - # send a PROG_MULTI command to write a collection of bytes - def __program_multi(self, data, windowMode): - - length = len(data).to_bytes(1, byteorder='big') - - self.__send(uploader.PROG_MULTI) - self.__send(length) - self.__send(data) - self.__send(uploader.EOC) - if (not windowMode): - self.__getSync(False) - else: - # The following is done to have minimum delay on the transmission - # of the ne fw. The per block cost of __getSync was about 16 mS per. - # Passively wait on Sync and Result using board rates and - # N.B. attempts to activly wait on InWating still carried 8 mS of overhead - self.__probe(False) - self.__probe(True) - time.sleep((ord(length) * self.chartime) + uploader.MAX_FLASH_PRGRAM_TIME) - self.__probe(False) - - # verify multiple bytes in flash - def __verify_multi(self, data): - - length = len(data).to_bytes(1, byteorder='big') - - self.__send(uploader.READ_MULTI) - self.__send(length) - self.__send(uploader.EOC) - self.port.flush() - programmed = self.__recv(len(data)) - if programmed != data: - print("got " + binascii.hexlify(programmed)) - print("expect " + binascii.hexlify(data)) - return False - self.__getSync() - return True - - # send the reboot command - def __reboot(self): - self.__send(uploader.REBOOT + - uploader.EOC) - self.port.flush() - - # v3+ can report failure if the first word flash fails - if self.bl_rev >= 3: - self.__getSync() - - # split a sequence into a list of size-constrained pieces - def __split_len(self, seq, length): - return [seq[i:i+length] for i in range(0, len(seq), length)] - - # upload code - def __program(self, label, fw): - self.__probe(False) - print("\n", end='') - code = fw.image - groups = self.__split_len(code, uploader.PROG_MULTI_MAX) - # Give imedate feedback - self.__drawProgressBar(label, 0, len(groups)) - uploadProgress = 0 - for bytes in groups: - self.__program_multi(bytes, self.ackWindowedMode) - # If in Window mode, extend the window size for the __ackSyncWindow - if self.ackWindowedMode: - self.window += self.window_per - - # Print upload progress (throttled, so it does not delay upload progress) - uploadProgress += 1 - if uploadProgress % 256 == 0: - self.__probe(True) - self.__probe(False) - self.__probe(True) - self.__ackSyncWindow(self.window) - self.__probe(False) - self.window = 0 - self.__drawProgressBar(label, uploadProgress, len(groups)) - - # Do any remaining fragment - self.__ackSyncWindow(self.window) - self.window = 0 - self.__drawProgressBar(label, 100, 100) - - # verify code - def __verify_v2(self, label, fw): - print("\n", end='') - self.__send(uploader.CHIP_VERIFY + - uploader.EOC) - self.__getSync() - code = fw.image - groups = self.__split_len(code, uploader.READ_MULTI_MAX) - verifyProgress = 0 - for bytes in groups: - verifyProgress += 1 - if verifyProgress % 256 == 0: - self.__drawProgressBar(label, verifyProgress, len(groups)) - if (not self.__verify_multi(bytes)): - raise RuntimeError("Verification failed") - self.__drawProgressBar(label, 100, 100) - - def __verify_v3(self, label, fw): - print("\n", end='') - self.__drawProgressBar(label, 1, 100) - expect_crc = fw.crc(self.fw_maxsize) - self.__send(uploader.GET_CRC + uploader.EOC) - time.sleep(0.5) - report_crc = self.__recv_int() - self.__getSync() - if report_crc != expect_crc: - print("Expected 0x%x" % expect_crc) - print("Got 0x%x" % report_crc) - raise RuntimeError("Program CRC failed") - self.__drawProgressBar(label, 100, 100) - - def __set_boot_delay(self, boot_delay): - self.__send(uploader.SET_BOOT_DELAY + - struct.pack("b", boot_delay) + - uploader.EOC) - self.__getSync() - - # get basic data about the board - def identify(self): - self.__determineInterface() - # make sure we are in sync before starting - self.__sync() - - # get the bootloader protocol ID first - self.bl_rev = self.__getInfo(uploader.INFO_BL_REV) - if (self.bl_rev < uploader.BL_REV_MIN) or (self.bl_rev > uploader.BL_REV_MAX): - print("Unsupported bootloader protocol %d" % uploader.INFO_BL_REV) - raise RuntimeError("Bootloader protocol mismatch") - - self.board_type = self.__getInfo(uploader.INFO_BOARD_ID) - self.board_rev = self.__getInfo(uploader.INFO_BOARD_REV) - self.fw_maxsize = self.__getInfo(uploader.INFO_FLASH_SIZE) - - self.version = self.__getVersion() - - # upload the firmware - def upload(self, fw_list, force=False, boot_delay=None, boot_check=False, force_erase=False): - self.force_erase = force_erase - # select correct binary - found_suitable_firmware = False - for file in fw_list: - fw = firmware(file) - if self.board_type == fw.property('board_id'): - if len(fw_list) > 1: print("using firmware binary {}".format(file)) - found_suitable_firmware = True - break - - if not found_suitable_firmware: - msg = "Firmware not suitable for this board (Firmware board_type=%u board_id=%u)" % ( - self.board_type, fw.property('board_id')) - print("WARNING: %s" % msg) - if force: - if len(fw_list) > 1: - raise FirmwareNotSuitableException("force flashing failed, more than one file provided, none suitable") - print("FORCED WRITE, FLASHING ANYWAY!") - else: - raise FirmwareNotSuitableException(msg) - - percent = fw.property('image_size') / fw.property('image_maxsize') - binary_size = float(fw.property('image_size')) - binary_max_size = float(fw.property('image_maxsize')) - percent = (binary_size / binary_max_size) * 100 - - print("Loaded firmware for board id: %s,%s size: %d bytes (%.2f%%) " % (fw.property('board_id'), fw.property('board_revision'), fw.property('image_size'), percent)) - print() - - print(f"Bootloader version: {self.version}") - - # Make sure we are doing the right thing - start = _time() - if self.board_type != fw.property('board_id'): - msg = "Firmware not suitable for this board (Firmware board_type=%u board_id=%u)" % ( - self.board_type, fw.property('board_id')) - print("WARNING: %s" % msg) - if force: - print("FORCED WRITE, FLASHING ANYWAY!") - else: - raise FirmwareNotSuitableException(msg) - - # Prevent uploads where the image would overflow the flash - if self.fw_maxsize < fw.property('image_size'): - raise RuntimeError("Firmware image is too large for this board") - - # OTP added in v4: - if self.bl_rev >= 4: - for byte in range(0, 32*6, 4): - x = self.__getOTP(byte) - self.otp = self.otp + x - # print(binascii.hexlify(x).decode('Latin-1') + ' ', end='') - # see src/modules/systemlib/otp.h in px4 code: - self.otp_id = self.otp[0:4] - self.otp_idtype = self.otp[4:5] - self.otp_vid = self.otp[8:4:-1] - self.otp_pid = self.otp[12:8:-1] - self.otp_coa = self.otp[32:160] - # show user: - try: - print("Sn: ", end='') - for byte in range(0, 12, 4): - x = self.__getSN(byte) - x = x[::-1] # reverse the bytes - self.sn = self.sn + x - print(binascii.hexlify(x).decode('Latin-1'), end='') # show user - print('') - print("Chip: %08x" % self.__getCHIP()) - - otp_id = self.otp_id.decode('Latin-1') - if ("PX4" in otp_id): - print("OTP id: " + otp_id) - print("OTP idtype: " + binascii.b2a_qp(self.otp_idtype).decode('Latin-1')) - print("OTP vid: " + binascii.hexlify(self.otp_vid).decode('Latin-1')) - print("OTP pid: " + binascii.hexlify(self.otp_pid).decode('Latin-1')) - print("OTP coa: " + binascii.b2a_base64(self.otp_coa).decode('Latin-1')) - - except Exception as e: - # ignore bad character encodings - print(f"Exception ignored: {e}") - pass - - # Silicon errata check was added in v5 - if (self.bl_rev >= 5): - des = self.__getCHIPDes() - if (len(des) == 2): - family, revision = des - print(f"Family: {family.decode()}") - print(f"Revision: {revision.decode()}") - print(f"Flash: {self.fw_maxsize} bytes") - - # Prevent uploads where the maximum image size of the board config is smaller than the flash - # of the board. This is a hint the user chose the wrong config and will lack features - # for this particular board. - - # This check should also check if the revision is an unaffected revision - # and thus can support the full flash, see - # https://github.com/PX4/Firmware/blob/master/src/drivers/boards/common/stm32/board_mcu_version.c#L125-L144 - - if self.fw_maxsize > fw.property('image_maxsize') and not force: - print(f"WARNING: Board can accept larger flash images ({self.fw_maxsize} bytes) than board config ({fw.property('image_maxsize')} bytes)") - else: - # If we're still on bootloader v4 on a Pixhawk, we don't know if we - # have the silicon errata and therefore need to flash px4_fmu-v2 - # with 1MB flash or if it supports px4_fmu-v3 with 2MB flash. - if fw.property('board_id') == 9 \ - and fw.property('image_size') > 1032192 \ - and not force: - raise RuntimeError("\nThe Board uses bootloader revision 4 and can therefore not determine\n" - "if flashing more than 1 MB (px4_fmu-v3_default) is safe, chances are\n" - "high that it is not safe! If unsure, use px4_fmu-v2_default.\n" - "\n" - "If you know you that the board does not have the silicon errata, use\n" - "this script with --force, or update the bootloader. If you are invoking\n" - "upload using make, you can use force-upload target to force the upload.\n") - self.__erase("Erase ") - self.__program("Program", fw) - - if self.bl_rev == 2: - self.__verify_v2("Verify ", fw) - else: - self.__verify_v3("Verify ", fw) - - if boot_delay is not None: - self.__set_boot_delay(boot_delay) - - print("\nRebooting.", end='') - self.__reboot() - self.port.close() - print(" Elapsed Time %3.3f\n" % (_time() - start)) - - def __next_baud_flightstack(self): - if self.baudrate_flightstack_idx + 1 >= len(self.baudrate_flightstack): - return False - try: - self.port.baudrate = self.baudrate_flightstack[self.baudrate_flightstack_idx + 1] - self.baudrate_flightstack_idx = self.baudrate_flightstack_idx + 1 - except serial.SerialException: - # Sometimes _configure_port fails - time.sleep(0.04) - - return True - - def send_protocol_splitter_format(self, data): - # Header Structure: - # bits: 1 2 3 4 5 6 7 8 - # header[0] - | Magic | (='S') - # header[1] - |T| LenH | (T - 0: mavlink; 1: rtps) - # header[2] - | LenL | - # header[3] - | Checksum | - - MAGIC = 83 - - len_bytes = len(data).to_bytes(2, "big") - LEN_H = len_bytes[0] & 127 - LEN_L = len_bytes[1] & 255 - CHECKSUM = MAGIC ^ LEN_H ^ LEN_L - - header_ints = [MAGIC, LEN_H, LEN_L, CHECKSUM] - header_bytes = struct.pack("{}B".format(len(header_ints)), *header_ints) - - self.__send(header_bytes) - self.__send(data) - - def send_reboot(self, use_protocol_splitter_format=False): - if (not self.__next_baud_flightstack()): - return False - - print("Attempting reboot on %s with baudrate=%d..." % (self.port.port, self.port.baudrate), file=sys.stderr) - if "ttyS" in self.port.port: - print("If the board does not respond, check the connection to the Flight Controller") - else: - print("If the board does not respond, unplug and re-plug the USB connector.", file=sys.stderr) - - try: - send_fct = self.__send - if use_protocol_splitter_format: - send_fct = self.send_protocol_splitter_format - - # try MAVLINK command first - self.port.flush() - send_fct(uploader.MAVLINK_REBOOT_ID1) - send_fct(uploader.MAVLINK_REBOOT_ID0) - # then try reboot via NSH - send_fct(uploader.NSH_INIT) - send_fct(uploader.NSH_REBOOT_BL) - send_fct(uploader.NSH_INIT) - send_fct(uploader.NSH_REBOOT) - self.port.flush() - self.port.baudrate = self.baudrate_bootloader - except Exception: - try: - self.port.flush() - self.port.baudrate = self.baudrate_bootloader - except Exception: - pass - - return True - - -def main(): - # Parse commandline arguments - parser = argparse.ArgumentParser(description="Firmware uploader for the PX autopilot system.") - parser.add_argument('--port', action="store", required=True, help="Comma-separated list of serial port(s) to which the FMU may be attached") - parser.add_argument('--baud-bootloader', action="store", type=int, default=115200, help="Baud rate of the serial port (default is 115200) when communicating with bootloader, only required for true serial ports.") - parser.add_argument('--baud-flightstack', action="store", default="57600", help="Comma-separated list of baud rate of the serial port (default is 57600) when communicating with flight stack (Mavlink or NSH), only required for true serial ports.") - parser.add_argument('--force', action='store_true', default=False, help='Override board type check, or silicon errata checks and continue loading') - parser.add_argument('--force-erase', action="store_true", help="Do not perform the blank check, always erase every sector of the application space") - parser.add_argument('--boot-delay', type=int, default=None, help='minimum boot delay to store in flash') - parser.add_argument('--use-protocol-splitter-format', action='store_true', help='use protocol splitter format for reboot') - parser.add_argument('firmware', action="store", nargs='+', help="Firmware file(s)") - args = parser.parse_args() - - if args.use_protocol_splitter_format: - print("Using protocol splitter format to reboot pixhawk!") - - # warn people about ModemManager which interferes badly with Pixhawk - if os.path.exists("/usr/sbin/ModemManager"): - print("==========================================================================================================") - print("WARNING: You should uninstall ModemManager as it conflicts with any non-modem serial device (like Pixhawk)") - print("==========================================================================================================") - - print("Waiting for bootloader...") - # tell any GCS that might be connected to the autopilot to give up - # control of the serial port - - # send to localhost and default GCS port - ipaddr = '127.0.0.1' - portnum = 14550 - - # COMMAND_LONG in MAVLink 1 - heartbeatpacket = bytearray.fromhex('fe097001010000000100020c5103033c8a') - commandpacket = bytearray.fromhex('fe210101014c00000000000000000000000000000000000000000000803f00000000f6000000008459') - - # initialize an UDP socket - s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - - # send heartbeat to initialize connection and command to free the link - s.sendto(heartbeatpacket, (ipaddr, portnum)) - s.sendto(commandpacket, (ipaddr, portnum)) - - # close the socket - s.close() - - # Spin waiting for a device to show up - try: - while True: - portlist = [] - patterns = args.port.split(",") - # on unix-like platforms use glob to support wildcard ports. This allows - # the use of /dev/serial/by-id/usb-3D_Robotics on Linux, which prevents the upload from - # causing modem hangups etc - if "linux" in _platform or "darwin" in _platform or "cygwin" in _platform: - import glob - for pattern in patterns: - portlist += glob.glob(pattern) - else: - portlist = patterns - - baud_flightstack = [int(x) for x in args.baud_flightstack.split(',')] - - successful = False - unsuitable_board = False - for port in portlist: - - # print("Trying %s" % port) - - # create an uploader attached to the port - try: - if "linux" in _platform: - # Linux, don't open Mac OS and Win ports - if "COM" not in port and "tty.usb" not in port: - up = uploader(port, args.baud_bootloader, baud_flightstack) - elif "darwin" in _platform: - # OS X, don't open Windows and Linux ports - if "COM" not in port and "ACM" not in port: - up = uploader(port, args.baud_bootloader, baud_flightstack) - elif "cygwin" in _platform: - # Cygwin, don't open native Windows COM and Linux ports - if "COM" not in port and "ACM" not in port: - up = uploader(port, args.baud_bootloader, baud_flightstack) - elif "win" in _platform: - # Windows, don't open POSIX ports - if "/" not in port: - up = uploader(port, args.baud_bootloader, baud_flightstack) - except Exception as e: - # open failed, rate-limit our attempts - time.sleep(0.05) - print(f"Exception ignored: {e}") - - # and loop to the next port - continue - - found_bootloader = False - while True: - up.open() - - # port is open, try talking to it - try: - # identify the bootloader - up.identify() - found_bootloader = True - print() - print(f"Found board id: {up.board_type},{up.board_rev} bootloader protocol revision {up.bl_rev} on {port}") - break - - except (RuntimeError, serial.SerialException): - - if not up.send_reboot(args.use_protocol_splitter_format): - break - - # wait for the reboot, without we might run into Serial I/O Error 5 - time.sleep(0.25) - - # always close the port - up.close() - - # wait for the close, without we might run into Serial I/O Error 6 - time.sleep(0.3) - - if not found_bootloader: - # Go to the next port - continue - - try: - # ok, we have a bootloader, try flashing it - up.upload(args.firmware, force=args.force, boot_delay=args.boot_delay, force_erase=args.force_erase) - - # if we made this far without raising exceptions, the upload was successful - successful = True - - except RuntimeError as e: - # print the error - print(f"\n\nError: {e}") - - except FirmwareNotSuitableException: - unsuitable_board = True - up.close() - continue - - except IOError: - up.close() - continue - - finally: - # always close the port - up.close() - - # we could loop here if we wanted to wait for more boards... - if successful: - sys.exit(0) - else: - sys.exit(1) - - if unsuitable_board: - # If we land here, we went through all ports, did not flash any - # board and found at least one unsuitable board. - # Exit with 2, so a caller can distinguish from other errors - sys.exit(2) - - # Delay retries to < 20 Hz to prevent spin-lock from hogging the CPU - time.sleep(0.05) - - # CTRL+C aborts the upload/spin-lock by interrupt mechanics - except KeyboardInterrupt: - print("\n Upload aborted by user.") - sys.exit(0) - - -if __name__ == '__main__': - main() - -# vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4 diff --git a/platforms/nuttx/cmake/upload.cmake b/platforms/nuttx/cmake/upload.cmake index 83f623c090b..3bffd08535a 100644 --- a/platforms/nuttx/cmake/upload.cmake +++ b/platforms/nuttx/cmake/upload.cmake @@ -31,69 +31,32 @@ # ############################################################################ -# NuttX CDCACM vendor and product strings -set(vendorstr_underscore) -set(productstr_underscore) -string(REPLACE " " "_" vendorstr_underscore ${CONFIG_CDCACM_VENDORSTR}) -string(REPLACE "," "_" vendorstr_underscore "${vendorstr_underscore}") -string(REPLACE " " "_" productstr_underscore ${CONFIG_CDCACM_PRODUCTSTR}) - -set(serial_ports) -if(${CMAKE_HOST_SYSTEM_NAME} STREQUAL "Linux") - - set(px4_usb_path "${vendorstr_underscore}_${productstr_underscore}") - set(px4_bl_usb_path "${vendorstr_underscore}_BL") - - list(APPEND serial_ports - # NuttX vendor + product string - /dev/serial/by-id/*-${px4_usb_path}* - - # Bootloader - /dev/serial/by-id/*_${px4_bl_usb_path}* - /dev/serial/by-id/*PX4_BL* # typical bootloader USB device string - /dev/serial/by-id/*BL_FMU* - - # TODO: handle these per board - /dev/serial/by-id/usb-The_Autopilot* - /dev/serial/by-id/usb-Bitcraze* - /dev/serial/by-id/pci-Bitcraze* - /dev/serial/by-id/usb-Gumstix* - /dev/serial/by-id/usb-Hex_ProfiCNC* - /dev/serial/by-id/usb-UVify* - /dev/serial/by-id/usb-ArduPilot* - ) - -elseif(${CMAKE_HOST_SYSTEM_NAME} STREQUAL "Darwin") - list(APPEND serial_ports - /dev/tty.usbmodemPX*,/dev/tty.usbmodem* - ) -elseif(${CMAKE_HOST_SYSTEM_NAME} STREQUAL "CYGWIN") - list(APPEND serial_ports - /dev/ttyS* - ) -elseif(${CMAKE_HOST_SYSTEM_NAME} STREQUAL "Windows") - foreach(port RANGE 32 0) - list(APPEND serial_ports - "COM${port}") - endforeach() -endif() - -string(REPLACE ";" "," serial_ports "${serial_ports}") +# Uploader script auto-detects PX4 devices by USB VID/PID +set(PX4_UPLOADER_SCRIPT "${PX4_SOURCE_DIR}/Tools/px4_uploader.py") add_custom_target(upload - COMMAND ${PYTHON_EXECUTABLE} ${PX4_SOURCE_DIR}/Tools/px_uploader.py --port ${serial_ports} ${fw_package} + COMMAND ${PYTHON_EXECUTABLE} ${PX4_UPLOADER_SCRIPT} ${fw_package} DEPENDS ${fw_package} COMMENT "uploading px4" VERBATIM USES_TERMINAL WORKING_DIRECTORY ${PX4_BINARY_DIR} - ) +) add_custom_target(force-upload - COMMAND ${PYTHON_EXECUTABLE} ${PX4_SOURCE_DIR}/Tools/px_uploader.py --force --port ${serial_ports} ${fw_package} + COMMAND ${PYTHON_EXECUTABLE} ${PX4_UPLOADER_SCRIPT} --force ${fw_package} DEPENDS ${fw_package} COMMENT "uploading px4 with --force" VERBATIM USES_TERMINAL WORKING_DIRECTORY ${PX4_BINARY_DIR} - ) +) + +add_custom_target(upload-verbose + COMMAND ${PYTHON_EXECUTABLE} ${PX4_UPLOADER_SCRIPT} --verbose ${fw_package} + DEPENDS ${fw_package} + COMMENT "uploading px4 with verbose output" + VERBATIM + USES_TERMINAL + WORKING_DIRECTORY ${PX4_BINARY_DIR} +)