Merge branch 'rp2040-upload-improvements' into integration

esphome/__main__.py

@@ -58,7 +58,9 @@ from esphome.helpers import get_bool_env, indent, is_ip_address
 from esphome.log import AnsiFore, color, setup_log
 from esphome.types import ConfigType
 from esphome.util import (
-    get_rp2040_mass_storage_volumes,
+    PICOTOOL_PACKAGE,
+    detect_rp2040_bootsel,
+    get_picotool_path,
     get_serial_ports,
     list_yaml_files,
     run_external_command,
@@ -175,7 +177,7 @@ class PortType(StrEnum):
     NETWORK = "NETWORK"
     MQTT = "MQTT"
     MQTTIP = "MQTTIP"
-    MASS_STORAGE = "MASS_STORAGE"
+    BOOTSEL = "BOOTSEL"
 
 
 # Magic MQTT port types that require special handling
@@ -254,14 +256,14 @@ def choose_upload_log_host(
         (f"{port.path} ({port.description})", port.path) for port in get_serial_ports()
     ]
 
-    # Add RP2040 mass storage volumes when uploading
+    # Add RP2040 BOOTSEL device option when uploading
     if (
         purpose == Purpose.UPLOADING
         and CORE.data.get(KEY_CORE, {}).get(KEY_TARGET_PLATFORM) == PLATFORM_RP2040
     ):
-        for vol in get_rp2040_mass_storage_volumes():
-            # Use MS: prefix so get_port_type() identifies as MASS_STORAGE
-            options.append((f"{vol.path} ({vol.description})", f"MS:{vol.path}"))
+        picotool = _find_picotool()
+        if picotool is not None and detect_rp2040_bootsel(picotool) > 0:
+            options.append(("RP2040 BOOTSEL (via picotool)", "BOOTSEL"))
 
     if purpose == Purpose.LOGGING:
         if has_mqtt_logging():
@@ -280,14 +282,11 @@ def choose_upload_log_host(
         if has_mqtt_ip_lookup():
             options.append(("Over The Air (MQTT IP lookup)", "MQTTIP"))
 
-    # Show helpful BOOTSEL instructions for RP2040 when no USB device is found
+    # Show helpful BOOTSEL instructions for RP2040 when no BOOTSEL device is found
     if (
         purpose == Purpose.UPLOADING
         and CORE.data.get(KEY_CORE, {}).get(KEY_TARGET_PLATFORM) == PLATFORM_RP2040
-        and not any(
-            get_port_type(opt[1]) in (PortType.SERIAL, PortType.MASS_STORAGE)
-            for opt in options
-        )
+        and not any(get_port_type(opt[1]) == PortType.BOOTSEL for opt in options)
     ):
         if not options:
             raise EsphomeError(
@@ -441,13 +440,13 @@ def get_port_type(port: str) -> PortType:
 
     Returns:
         PortType.SERIAL for serial ports (/dev/ttyUSB0, COM1, etc.)
-        PortType.MASS_STORAGE for RP2040 BOOTSEL mass storage volumes
+        PortType.BOOTSEL for RP2040 BOOTSEL upload via picotool
         PortType.MQTT for MQTT logging
         PortType.MQTTIP for MQTT IP lookup
         PortType.NETWORK for IP addresses, hostnames, or mDNS names
     """
-    if port.startswith("MS:"):
-        return PortType.MASS_STORAGE
+    if port == "BOOTSEL":
+        return PortType.BOOTSEL
     if port.startswith("/") or port.startswith("COM"):
         return PortType.SERIAL
     if port == "MQTT":
@@ -757,67 +756,83 @@ def upload_using_platformio(config: ConfigType, port: str) -> int:
     return platformio_api.run_platformio_cli_run(config, CORE.verbose, *upload_args)
 
 
-def upload_using_uf2_copy(config: ConfigType, mount_path: str) -> int:
-    """Upload firmware to RP2040 by copying UF2 file to mass storage volume.
+def _find_picotool() -> Path | None:
+    """Find the picotool binary from PlatformIO packages."""
+    from esphome import platformio_api
 
-    When an RP2040 is in BOOTSEL mode, it appears as a USB mass storage device.
-    Firmware can be uploaded by simply copying the .uf2 file to the volume.
+    try:
+        idedata = platformio_api.get_idedata(CORE.config)
+    except Exception:  # noqa: BLE001
+        return None
+    return get_picotool_path(idedata.cc_path)
+
+
+def upload_using_picotool(config: ConfigType) -> int:
+    """Upload firmware to RP2040 in BOOTSEL mode using picotool.
+
+    Uses picotool to load the ELF firmware directly via USB, avoiding
+    the mass storage copy approach that causes "disk not ejected properly"
+    warnings on macOS.
     """
+    import subprocess
+
     from esphome import platformio_api
-    from esphome.helpers import ProgressBar
 
     idedata = platformio_api.get_idedata(config)
-    build_dir = Path(idedata.firmware_elf_path).parent
-    uf2_file = build_dir / "firmware.uf2"
+    firmware_elf = Path(idedata.firmware_elf_path)
 
-    if not uf2_file.exists():
+    if not firmware_elf.is_file():
         _LOGGER.error(
-            "UF2 firmware file not found at %s. Make sure the project has been compiled first.",
-            uf2_file,
+            "Firmware ELF file not found at %s. "
+            "Make sure the project has been compiled first.",
+            firmware_elf,
         )
         return 1
 
-    dest_dir = Path(mount_path)
-    if not dest_dir.is_dir():
+    picotool = get_picotool_path(idedata.cc_path)
+    if picotool is None:
         _LOGGER.error(
-            "Mass storage volume %s is no longer available. "
-            "Is the RP2040 still in BOOTSEL mode?",
-            mount_path,
+            "picotool not found. Ensure the RP2040 PlatformIO platform "
+            "is installed (%s).",
+            PICOTOOL_PACKAGE,
         )
         return 1
 
-    dest_file = dest_dir / uf2_file.name
-    file_size = uf2_file.stat().st_size
-    if file_size == 0:
-        _LOGGER.error("UF2 firmware file is empty: %s", uf2_file)
-        return 1
-    _LOGGER.info("Uploading UF2 firmware to %s (%s bytes)", mount_path, file_size)
-
-    progress = ProgressBar()
+    _LOGGER.info("Uploading firmware to RP2040 via picotool...")
     try:
-        chunk_size = 65536
-        bytes_written = 0
-        with open(uf2_file, "rb") as src, open(dest_file, "wb") as dst:
-            while True:
-                chunk = src.read(chunk_size)
-                if not chunk:
-                    break
-                dst.write(chunk)
-                dst.flush()
-                os.fsync(dst.fileno())
-                bytes_written += len(chunk)
-                progress.update(bytes_written / file_size)
-        progress.done()
+        # Don't capture stdout — let picotool write directly to the terminal
+        # so progress bars display in real-time with \r updates.
+        # Capture stderr only so we can detect permission errors.
+        result = subprocess.run(
+            [str(picotool), "load", "-v", "-x", str(firmware_elf)],
+            stderr=subprocess.PIPE,
+            timeout=60,
+            check=False,
+        )
+    except subprocess.TimeoutExpired:
+        _LOGGER.error("picotool upload timed out after 60 seconds.")
+        return 1
     except OSError as err:
-        progress.done()
-        _LOGGER.error("Failed to copy UF2 file to %s: %s", mount_path, err)
+        _LOGGER.error("Failed to run picotool: %s", err)
+        return 1
+
+    if result.returncode != 0:
+        stderr = result.stderr.decode("utf-8", errors="replace").strip()
+        if stderr:
+            for line in stderr.splitlines():
+                safe_print(line)
+        if "LIBUSB_ERROR_ACCESS" in stderr or "Permission denied" in stderr:
+            msg = "Permission denied accessing USB device."
+            if sys.platform.startswith("linux"):
+                msg += (
+                    " You may need to add udev rules for RP2040 devices."
+                    " See: https://github.com/raspberrypi/picotool#linux-permissions"
+                )
+            _LOGGER.error(msg)
+        else:
+            _LOGGER.error("picotool upload failed (exit code %d).", result.returncode)
         return 1
 
-    _LOGGER.info(
-        "Successfully copied firmware to %s. "
-        "The device will automatically reset and run the new firmware.",
-        mount_path,
-    )
     return 0
 
 
@@ -889,11 +904,9 @@ def upload_program(
 
     port_type = get_port_type(host)
 
-    if port_type == PortType.MASS_STORAGE:
-        # Strip the MS: prefix to get the actual mount path
-        mount_path = host[3:]
-        exit_code = upload_using_uf2_copy(config, mount_path)
-        # Return None for device - mass storage can't be used for logging,
+    if port_type == PortType.BOOTSEL:
+        exit_code = upload_using_picotool(config)
+        # Return None for device - BOOTSEL can't be used for logging,
         # so command_run will show the interactive chooser for log source
         return exit_code, None
 
@@ -1103,7 +1116,7 @@ def command_run(args: ArgsProtocol, config: ConfigType) -> int | None:
     if args.no_logs:
         return 0
 
-    # After mass storage upload, wait for a new serial port to appear
+    # After BOOTSEL upload, wait for a new serial port to appear
     # so it shows up in the log chooser
     if (
         successful_device is None

esphome/util.py

@@ -355,66 +355,43 @@ def get_serial_ports() -> list[SerialPort]:
     return result
 
 
-class MassStorageVolume:
-    """Represents a mass storage volume for RP2040 BOOTSEL upload."""
-
-    def __init__(self, path: Path, description: str) -> None:
-        self.path = path
-        self.description = description
+PICOTOOL_PACKAGE = "tool-picotool-rp2040-earlephilhower"
 
 
-def get_rp2040_mass_storage_volumes() -> list[MassStorageVolume]:
-    """Detect mounted RP2040 BOOTSEL mass storage volumes.
+def get_picotool_path(cc_path: str) -> Path | None:
+    """Derive the picotool binary path from the PlatformIO toolchain cc_path.
 
-    When an RP2040 is in BOOTSEL mode, it appears as a USB mass storage
-    device named 'RPI-RP2'. This function finds those mount points.
+    The cc_path from IDEData points to the toolchain package, e.g.:
+    ~/.platformio/packages/toolchain-rp2040-earlephilhower/bin/arm-none-eabi-gcc
+    Picotool is in a sibling package:
+    ~/.platformio/packages/tool-picotool-rp2040-earlephilhower/picotool
     """
-    result: list[MassStorageVolume] = []
+    cc = Path(cc_path)
+    # Go from .../packages/toolchain-.../bin/gcc up to .../packages/
+    packages_dir = cc.parent.parent.parent
+    binary_name = "picotool.exe" if sys.platform == "win32" else "picotool"
+    picotool = packages_dir / PICOTOOL_PACKAGE / binary_name
+    if picotool.is_file():
+        return picotool
+    return None
 
-    if sys.platform == "darwin":
-        # macOS: /Volumes/RPI-RP2
-        result.extend(
-            MassStorageVolume(path, "RP2040 BOOTSEL")
-            for path in Path("/Volumes").glob("RPI-RP2*")
-            if path.is_dir()
+
+def detect_rp2040_bootsel(picotool_path: str | Path) -> int:
+    """Detect RP2040/RP2350 devices in BOOTSEL mode using picotool.
+
+    Returns the number of devices found (by counting 'type:' lines in output),
+    matching PlatformIO's detection approach.
+    """
+    try:
+        result = subprocess.run(
+            [str(picotool_path), "info", "-d"],
+            capture_output=True,
+            timeout=10,
+            check=False,
         )
-
-    elif sys.platform.startswith("linux"):
-        # Linux: /media/<user>/RPI-RP2, /run/media/<user>/RPI-RP2, /mnt/RPI-RP2
-        search_patterns = [
-            Path("/media").glob("*/RPI-RP2*"),
-            Path("/run/media").glob("*/RPI-RP2*"),
-            Path("/mnt").glob("RPI-RP2*"),
-        ]
-        for pattern in search_patterns:
-            try:
-                result.extend(
-                    MassStorageVolume(path, "RP2040 BOOTSEL")
-                    for path in pattern
-                    if path.is_dir()
-                )
-            except OSError:
-                continue
-
-    elif sys.platform == "win32":
-        # Windows: Check drive letters for RPI-RP2 volume label
-        import ctypes
-
-        for letter in "DEFGHIJKLMNOPQRSTUVWXYZ":
-            drive = f"{letter}:\\"
-            if not Path(drive).exists():
-                continue
-            try:
-                volume_name = ctypes.create_unicode_buffer(1024)
-                ctypes.windll.kernel32.GetVolumeInformationW(
-                    drive, volume_name, 1024, None, None, None, None, 0
-                )
-                if volume_name.value.startswith("RPI-RP2"):
-                    result.append(MassStorageVolume(Path(drive), "RP2040 BOOTSEL"))
-            except OSError:
-                continue
-
-    return result
+        return result.stdout.count(b"type:")
+    except (OSError, subprocess.TimeoutExpired):
+        return 0
 
 
 def get_esp32_arduino_flash_error_help() -> str | None:

tests/unit_tests/test_main.py

@@ -40,8 +40,8 @@ from esphome.__main__ import (
     show_logs,
     upload_program,
     upload_using_esptool,
+    upload_using_picotool,
     upload_using_platformio,
-    upload_using_uf2_copy,
 )
 from esphome.components.esp32 import KEY_ESP32, KEY_VARIANT, VARIANT_ESP32
 from esphome.const import (
@@ -177,9 +177,9 @@ def mock_upload_using_platformio() -> Generator[Mock]:
 
 
 @pytest.fixture
-def mock_upload_using_uf2_copy() -> Generator[Mock]:
-    """Mock upload_using_uf2_copy for testing."""
-    with patch("esphome.__main__.upload_using_uf2_copy") as mock:
+def mock_upload_using_picotool() -> Generator[Mock]:
+    """Mock upload_using_picotool for testing."""
+    with patch("esphome.__main__.upload_using_picotool") as mock:
         yield mock
 
 
@@ -861,18 +861,17 @@ def test_choose_upload_log_host_no_address_with_ota_config() -> None:
 
 
 @pytest.mark.usefixtures("mock_no_serial_ports")
-def test_choose_upload_log_host_no_defaults_with_rp2040_mass_storage(
+def test_choose_upload_log_host_no_defaults_with_rp2040_bootsel(
     mock_choose_prompt: Mock,
 ) -> None:
-    """Test interactive mode shows RP2040 mass storage volumes."""
+    """Test interactive mode shows RP2040 BOOTSEL option via picotool."""
     setup_core(platform=PLATFORM_RP2040)
 
-    mock_volumes = [
-        MagicMock(path=Path("/Volumes/RPI-RP2"), description="RP2040 BOOTSEL"),
-    ]
-    with patch(
-        "esphome.__main__.get_rp2040_mass_storage_volumes",
-        return_value=mock_volumes,
+    with (
+        patch(
+            "esphome.__main__._find_picotool", return_value=Path("/usr/bin/picotool")
+        ),
+        patch("esphome.__main__.detect_rp2040_bootsel", return_value=1),
     ):
         result = choose_upload_log_host(
             default=None,
@@ -880,9 +879,8 @@ def test_choose_upload_log_host_no_defaults_with_rp2040_mass_storage(
             purpose=Purpose.UPLOADING,
         )
         assert result == ["/dev/ttyUSB0"]  # mock_choose_prompt default
-        vol_path = str(Path("/Volumes/RPI-RP2"))
         mock_choose_prompt.assert_called_once_with(
-            [(f"{vol_path} (RP2040 BOOTSEL)", f"MS:{vol_path}")],
+            [("RP2040 BOOTSEL (via picotool)", "BOOTSEL")],
             purpose=Purpose.UPLOADING,
         )
 
@@ -894,9 +892,9 @@ def test_choose_upload_log_host_rp2040_no_device_shows_bootsel_help() -> None:
 
     with (
         patch(
-            "esphome.__main__.get_rp2040_mass_storage_volumes",
-            return_value=[],
+            "esphome.__main__._find_picotool", return_value=Path("/usr/bin/picotool")
         ),
+        patch("esphome.__main__.detect_rp2040_bootsel", return_value=0),
         pytest.raises(EsphomeError, match="BOOTSEL"),
     ):
         choose_upload_log_host(
@@ -919,9 +917,9 @@ def test_choose_upload_log_host_rp2040_bootsel_tip_with_ota(
 
     with (
         patch(
-            "esphome.__main__.get_rp2040_mass_storage_volumes",
-            return_value=[],
+            "esphome.__main__._find_picotool", return_value=Path("/usr/bin/picotool")
         ),
+        patch("esphome.__main__.detect_rp2040_bootsel", return_value=0),
         patch(
             "esphome.__main__.choose_prompt",
             return_value="192.168.1.100",
@@ -936,10 +934,35 @@ def test_choose_upload_log_host_rp2040_bootsel_tip_with_ota(
         assert "BOOTSEL" in caplog.text
 
 
-def test_choose_upload_log_host_no_mass_storage_for_non_rp2040(
+def test_choose_upload_log_host_rp2040_bootsel_tip_with_serial_ports(
+    caplog: pytest.LogCaptureFixture,
+    mock_choose_prompt: Mock,
+) -> None:
+    """Test BOOTSEL tip shown when serial ports exist but no BOOTSEL device."""
+    setup_core(platform=PLATFORM_RP2040)
+
+    mock_ports = [MockSerialPort("/dev/ttyACM0", "RP2040 Serial")]
+    with (
+        patch("esphome.__main__.get_serial_ports", return_value=mock_ports),
+        patch(
+            "esphome.__main__._find_picotool",
+            return_value=Path("/usr/bin/picotool"),
+        ),
+        patch("esphome.__main__.detect_rp2040_bootsel", return_value=0),
+        caplog.at_level(logging.INFO, logger="esphome.__main__"),
+    ):
+        choose_upload_log_host(
+            default=None,
+            check_default=None,
+            purpose=Purpose.UPLOADING,
+        )
+        assert "BOOTSEL" in caplog.text
+
+
+def test_choose_upload_log_host_no_bootsel_for_non_rp2040(
     mock_no_serial_ports: Mock,
 ) -> None:
-    """Test that mass storage detection is not run for non-RP2040 platforms."""
+    """Test that BOOTSEL detection is not run for non-RP2040 platforms."""
     setup_core(
         platform=PLATFORM_ESP32,
         config={CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]},
@@ -947,9 +970,7 @@ def test_choose_upload_log_host_no_mass_storage_for_non_rp2040(
     )
 
     with (
-        patch(
-            "esphome.__main__.get_rp2040_mass_storage_volumes",
-        ) as mock_get_volumes,
+        patch("esphome.__main__._find_picotool") as mock_find_picotool,
         patch(
             "esphome.__main__.choose_prompt",
             return_value="192.168.1.100",
@@ -960,36 +981,32 @@ def test_choose_upload_log_host_no_mass_storage_for_non_rp2040(
             check_default=None,
             purpose=Purpose.UPLOADING,
         )
-        mock_get_volumes.assert_not_called()
+        mock_find_picotool.assert_not_called()
 
 
-def test_choose_upload_log_host_rp2040_serial_and_mass_storage(
+def test_choose_upload_log_host_rp2040_serial_and_bootsel(
     mock_choose_prompt: Mock,
 ) -> None:
-    """Test both serial ports and mass storage volumes shown for RP2040."""
+    """Test both serial ports and BOOTSEL option shown for RP2040."""
     setup_core(platform=PLATFORM_RP2040)
 
     mock_ports = [MockSerialPort("/dev/ttyACM0", "RP2040 Serial")]
-    mock_volumes = [
-        MagicMock(path=Path("/Volumes/RPI-RP2"), description="RP2040 BOOTSEL"),
-    ]
     with (
         patch("esphome.__main__.get_serial_ports", return_value=mock_ports),
         patch(
-            "esphome.__main__.get_rp2040_mass_storage_volumes",
-            return_value=mock_volumes,
+            "esphome.__main__._find_picotool", return_value=Path("/usr/bin/picotool")
         ),
+        patch("esphome.__main__.detect_rp2040_bootsel", return_value=1),
     ):
         choose_upload_log_host(
             default=None,
             check_default=None,
             purpose=Purpose.UPLOADING,
         )
-        vol_path = str(Path("/Volumes/RPI-RP2"))
         mock_choose_prompt.assert_called_once_with(
             [
                 ("/dev/ttyACM0 (RP2040 Serial)", "/dev/ttyACM0"),
-                (f"{vol_path} (RP2040 BOOTSEL)", f"MS:{vol_path}"),
+                ("RP2040 BOOTSEL (via picotool)", "BOOTSEL"),
             ],
             purpose=Purpose.UPLOADING,
         )
@@ -1266,108 +1283,152 @@ def test_upload_program_serial_upload_failed(
     mock_upload_using_esptool.assert_called_once()
 
 
-def test_upload_program_mass_storage(
-    mock_upload_using_uf2_copy: Mock,
+def test_upload_program_bootsel(
+    mock_upload_using_picotool: Mock,
     mock_get_port_type: Mock,
 ) -> None:
-    """Test upload_program with mass storage for RP2040."""
+    """Test upload_program with BOOTSEL for RP2040."""
     setup_core(platform=PLATFORM_RP2040)
-    mock_get_port_type.return_value = "MASS_STORAGE"
-    mock_upload_using_uf2_copy.return_value = 0
+    mock_get_port_type.return_value = "BOOTSEL"
+    mock_upload_using_picotool.return_value = 0
 
     config = {}
     args = MockArgs()
-    devices = ["MS:/Volumes/RPI-RP2"]
+    devices = ["BOOTSEL"]
 
     exit_code, host = upload_program(config, args, devices)
 
     assert exit_code == 0
-    # Mass storage device can't be used for logging, so host should be None
+    # BOOTSEL device can't be used for logging, so host should be None
     assert host is None
-    mock_upload_using_uf2_copy.assert_called_once_with(config, "/Volumes/RPI-RP2")
+    mock_upload_using_picotool.assert_called_once_with(config)
 
 
-def test_upload_program_mass_storage_failed(
-    mock_upload_using_uf2_copy: Mock,
+def test_upload_program_bootsel_failed(
+    mock_upload_using_picotool: Mock,
     mock_get_port_type: Mock,
 ) -> None:
-    """Test upload_program when mass storage upload fails."""
+    """Test upload_program when BOOTSEL upload fails."""
     setup_core(platform=PLATFORM_RP2040)
-    mock_get_port_type.return_value = "MASS_STORAGE"
-    mock_upload_using_uf2_copy.return_value = 1
+    mock_get_port_type.return_value = "BOOTSEL"
+    mock_upload_using_picotool.return_value = 1
 
     config = {}
     args = MockArgs()
-    devices = ["MS:/Volumes/RPI-RP2"]
+    devices = ["BOOTSEL"]
 
     exit_code, host = upload_program(config, args, devices)
 
     assert exit_code == 1
     assert host is None
-    mock_upload_using_uf2_copy.assert_called_once_with(config, "/Volumes/RPI-RP2")
+    mock_upload_using_picotool.assert_called_once_with(config)
 
 
-def test_upload_using_uf2_copy_success(tmp_path: Path) -> None:
-    """Test upload_using_uf2_copy copies UF2 file with progress."""
+def test_upload_using_picotool_success(tmp_path: Path) -> None:
+    """Test upload_using_picotool succeeds."""
     setup_core(platform=PLATFORM_RP2040, tmp_path=tmp_path)
 
-    # Create a mock UF2 file
     build_dir = tmp_path / "build"
     build_dir.mkdir()
-    uf2_file = build_dir / "firmware.uf2"
-    uf2_file.write_bytes(b"\x00" * 1024)
+    firmware_elf = build_dir / "firmware.elf"
+    firmware_elf.write_bytes(b"\x00" * 1024)
 
-    # Create a mock mount point
-    mount_dir = tmp_path / "RPI-RP2"
-    mount_dir.mkdir()
+    # Create picotool binary
+    packages_dir = tmp_path / "packages"
+    toolchain_bin = packages_dir / "toolchain-rp2040-earlephilhower" / "bin"
+    toolchain_bin.mkdir(parents=True)
+    picotool_dir = packages_dir / "tool-picotool-rp2040-earlephilhower"
+    picotool_dir.mkdir(parents=True)
+    picotool = picotool_dir / "picotool"
+    picotool.touch()
 
     mock_idedata = MagicMock()
-    mock_idedata.firmware_elf_path = str(build_dir / "firmware.elf")
+    mock_idedata.firmware_elf_path = str(firmware_elf)
+    mock_idedata.cc_path = str(toolchain_bin / "arm-none-eabi-gcc")
+
+    mock_result = MagicMock()
+    mock_result.returncode = 0
+    mock_result.stderr = b""
 
     config = {}
-    with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata):
-        exit_code = upload_using_uf2_copy(config, str(mount_dir))
+    with (
+        patch("esphome.platformio_api.get_idedata", return_value=mock_idedata),
+        patch("subprocess.run", return_value=mock_result),
+    ):
+        exit_code = upload_using_picotool(config)
 
     assert exit_code == 0
-    assert (mount_dir / "firmware.uf2").exists()
-    assert (mount_dir / "firmware.uf2").read_bytes() == b"\x00" * 1024
 
 
-def test_upload_using_uf2_copy_no_uf2_file(tmp_path: Path) -> None:
-    """Test upload_using_uf2_copy when UF2 file is missing."""
+def test_upload_using_picotool_no_elf(tmp_path: Path) -> None:
+    """Test upload_using_picotool when ELF file is missing."""
     setup_core(platform=PLATFORM_RP2040, tmp_path=tmp_path)
 
     build_dir = tmp_path / "build"
     build_dir.mkdir()
 
-    mount_dir = tmp_path / "RPI-RP2"
-    mount_dir.mkdir()
-
     mock_idedata = MagicMock()
     mock_idedata.firmware_elf_path = str(build_dir / "firmware.elf")
+    mock_idedata.cc_path = "/fake/path/gcc"
 
     config = {}
     with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata):
-        exit_code = upload_using_uf2_copy(config, str(mount_dir))
+        exit_code = upload_using_picotool(config)
 
     assert exit_code == 1
 
 
-def test_upload_using_uf2_copy_mount_gone(tmp_path: Path) -> None:
-    """Test upload_using_uf2_copy when mount point disappeared."""
+def test_upload_using_picotool_not_found(tmp_path: Path) -> None:
+    """Test upload_using_picotool when picotool binary not found."""
     setup_core(platform=PLATFORM_RP2040, tmp_path=tmp_path)
 
     build_dir = tmp_path / "build"
     build_dir.mkdir()
-    uf2_file = build_dir / "firmware.uf2"
-    uf2_file.write_bytes(b"\x00" * 512)
+    firmware_elf = build_dir / "firmware.elf"
+    firmware_elf.write_bytes(b"\x00" * 512)
 
     mock_idedata = MagicMock()
-    mock_idedata.firmware_elf_path = str(build_dir / "firmware.elf")
+    mock_idedata.firmware_elf_path = str(firmware_elf)
+    mock_idedata.cc_path = "/fake/path/gcc"
 
     config = {}
     with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata):
-        exit_code = upload_using_uf2_copy(config, str(tmp_path / "nonexistent"))
+        exit_code = upload_using_picotool(config)
+
+    assert exit_code == 1
+
+
+def test_upload_using_picotool_permission_error(tmp_path: Path) -> None:
+    """Test upload_using_picotool shows helpful message on permission error."""
+    setup_core(platform=PLATFORM_RP2040, tmp_path=tmp_path)
+
+    build_dir = tmp_path / "build"
+    build_dir.mkdir()
+    firmware_elf = build_dir / "firmware.elf"
+    firmware_elf.write_bytes(b"\x00" * 512)
+
+    packages_dir = tmp_path / "packages"
+    toolchain_bin = packages_dir / "toolchain-rp2040-earlephilhower" / "bin"
+    toolchain_bin.mkdir(parents=True)
+    picotool_dir = packages_dir / "tool-picotool-rp2040-earlephilhower"
+    picotool_dir.mkdir(parents=True)
+    picotool = picotool_dir / "picotool"
+    picotool.touch()
+
+    mock_idedata = MagicMock()
+    mock_idedata.firmware_elf_path = str(firmware_elf)
+    mock_idedata.cc_path = str(toolchain_bin / "arm-none-eabi-gcc")
+
+    mock_result = MagicMock()
+    mock_result.returncode = 1
+    mock_result.stderr = b"LIBUSB_ERROR_ACCESS"
+
+    config = {}
+    with (
+        patch("esphome.platformio_api.get_idedata", return_value=mock_idedata),
+        patch("subprocess.run", return_value=mock_result),
+    ):
+        exit_code = upload_using_picotool(config)
 
     assert exit_code == 1
 
@@ -1896,9 +1957,7 @@ def test_get_port_type() -> None:
     assert get_port_type("esphome-device.local") == "NETWORK"
     assert get_port_type("10.0.0.1") == "NETWORK"
 
-    assert get_port_type("MS:/Volumes/RPI-RP2") == "MASS_STORAGE"
-    assert get_port_type("MS:/media/user/RPI-RP2") == "MASS_STORAGE"
-    assert get_port_type("MS:D:\\") == "MASS_STORAGE"
+    assert get_port_type("BOOTSEL") == "BOOTSEL"
 
 
 def test_has_mqtt_ip_lookup() -> None:

tests/unit_tests/test_util.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from pathlib import Path
+import subprocess
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -405,137 +406,95 @@ def test_shlex_quote_edge_cases() -> None:
     assert util.shlex_quote("   ") == "'   '"
 
 
-def test_get_rp2040_mass_storage_volumes_macos(tmp_path: Path) -> None:
-    """Test RP2040 mass storage detection on macOS."""
-    volumes_dir = tmp_path / "Volumes"
-    volumes_dir.mkdir()
-    rpi_vol = volumes_dir / "RPI-RP2"
-    rpi_vol.mkdir()
+def test_get_picotool_path_found(tmp_path: Path) -> None:
+    """Test picotool path derivation from cc_path."""
+    # Create the expected directory structure
+    packages_dir = tmp_path / "packages"
+    toolchain_dir = packages_dir / "toolchain-rp2040-earlephilhower" / "bin"
+    toolchain_dir.mkdir(parents=True)
+    gcc = toolchain_dir / "arm-none-eabi-gcc"
+    gcc.touch()
 
-    with (
-        patch("esphome.util.sys") as mock_sys,
-        patch("esphome.util.Path") as mock_path_cls,
+    picotool_dir = packages_dir / "tool-picotool-rp2040-earlephilhower"
+    picotool_dir.mkdir(parents=True)
+    picotool = picotool_dir / "picotool"
+    picotool.touch()
+
+    result = util.get_picotool_path(str(gcc))
+    assert result == picotool
+
+
+def test_get_picotool_path_not_found(tmp_path: Path) -> None:
+    """Test picotool path returns None when not installed."""
+    packages_dir = tmp_path / "packages"
+    toolchain_dir = packages_dir / "toolchain-rp2040-earlephilhower" / "bin"
+    toolchain_dir.mkdir(parents=True)
+    gcc = toolchain_dir / "arm-none-eabi-gcc"
+    gcc.touch()
+
+    result = util.get_picotool_path(str(gcc))
+    assert result is None
+
+
+def test_get_picotool_path_windows(tmp_path: Path) -> None:
+    """Test picotool path uses .exe on Windows."""
+    packages_dir = tmp_path / "packages"
+    toolchain_dir = packages_dir / "toolchain-rp2040-earlephilhower" / "bin"
+    toolchain_dir.mkdir(parents=True)
+    gcc = toolchain_dir / "arm-none-eabi-gcc.exe"
+    gcc.touch()
+
+    picotool_dir = packages_dir / "tool-picotool-rp2040-earlephilhower"
+    picotool_dir.mkdir(parents=True)
+    picotool = picotool_dir / "picotool.exe"
+    picotool.touch()
+
+    with patch("esphome.util.sys.platform", "win32"):
+        result = util.get_picotool_path(str(gcc))
+    assert result == picotool
+
+
+def test_detect_rp2040_bootsel_found() -> None:
+    """Test BOOTSEL device detection when device is present."""
+    mock_result = MagicMock()
+    mock_result.stdout = b"Device Information\n type: RP2040\n"
+    with patch("esphome.util.subprocess.run", return_value=mock_result):
+        count = util.detect_rp2040_bootsel("/usr/bin/picotool")
+    assert count == 1
+
+
+def test_detect_rp2040_bootsel_multiple() -> None:
+    """Test BOOTSEL detection with multiple devices."""
+    mock_result = MagicMock()
+    mock_result.stdout = b"type: RP2040\ntype: RP2350\n"
+    with patch("esphome.util.subprocess.run", return_value=mock_result):
+        count = util.detect_rp2040_bootsel("/usr/bin/picotool")
+    assert count == 2
+
+
+def test_detect_rp2040_bootsel_none() -> None:
+    """Test BOOTSEL detection when no device found."""
+    mock_result = MagicMock()
+    mock_result.stdout = (
+        b"No accessible RP2040/RP2350 devices in BOOTSEL mode were found.\n"
+    )
+    with patch("esphome.util.subprocess.run", return_value=mock_result):
+        count = util.detect_rp2040_bootsel("/usr/bin/picotool")
+    assert count == 0
+
+
+def test_detect_rp2040_bootsel_oserror() -> None:
+    """Test BOOTSEL detection handles OSError."""
+    with patch("esphome.util.subprocess.run", side_effect=OSError("not found")):
+        count = util.detect_rp2040_bootsel("/usr/bin/picotool")
+    assert count == 0
+
+
+def test_detect_rp2040_bootsel_timeout() -> None:
+    """Test BOOTSEL detection handles timeout."""
+    with patch(
+        "esphome.util.subprocess.run",
+        side_effect=subprocess.TimeoutExpired("picotool", 10),
     ):
-        mock_sys.platform = "darwin"
-        # Make Path("/Volumes") return our tmp_path version
-        mock_path_cls.side_effect = lambda p: (
-            volumes_dir if p == "/Volumes" else Path(p)
-        )
-
-        result = util.get_rp2040_mass_storage_volumes()
-
-    assert len(result) == 1
-    assert result[0].description == "RP2040 BOOTSEL"
-
-
-def test_get_rp2040_mass_storage_volumes_none_found(tmp_path: Path) -> None:
-    """Test RP2040 mass storage detection when no volumes found."""
-    # Point at an empty directory so no RPI-RP2* matches
-    empty_dir = tmp_path / "Volumes"
-    empty_dir.mkdir()
-
-    with (
-        patch("esphome.util.sys.platform", "darwin"),
-        patch(
-            "esphome.util.Path",
-            side_effect=lambda p: empty_dir if p == "/Volumes" else Path(p),
-        ),
-    ):
-        result = util.get_rp2040_mass_storage_volumes()
-
-    assert result == []
-
-
-def test_get_rp2040_mass_storage_volumes_linux(tmp_path: Path) -> None:
-    """Test RP2040 mass storage detection on Linux."""
-    # Create /media/<user>/RPI-RP2 structure
-    media_dir = tmp_path / "media"
-    media_dir.mkdir()
-    user_dir = media_dir / "testuser"
-    user_dir.mkdir()
-    rp2_dir = user_dir / "RPI-RP2"
-    rp2_dir.mkdir()
-
-    # Create /run/media and /mnt as empty dirs
-    run_media_dir = tmp_path / "run_media"
-    run_media_dir.mkdir()
-    mnt_dir = tmp_path / "mnt"
-    mnt_dir.mkdir()
-
-    def mock_path_side_effect(p: str) -> Path:
-        if p == "/media":
-            return media_dir
-        if p == "/run/media":
-            return run_media_dir
-        if p == "/mnt":
-            return mnt_dir
-        return Path(p)
-
-    with (
-        patch("esphome.util.sys.platform", "linux"),
-        patch("esphome.util.Path", side_effect=mock_path_side_effect),
-    ):
-        result = util.get_rp2040_mass_storage_volumes()
-
-    assert len(result) == 1
-    assert result[0].description == "RP2040 BOOTSEL"
-
-
-def test_get_rp2040_mass_storage_volumes_linux_oserror(tmp_path: Path) -> None:
-    """Test RP2040 mass storage detection on Linux handles OSError."""
-    media_dir = tmp_path / "media"
-    media_dir.mkdir()
-
-    def mock_path_side_effect(p: str) -> Path:
-        if p == "/media":
-            return media_dir
-        if p in ("/run/media", "/mnt"):
-            # Return a path that will raise OSError when globbed
-            return tmp_path / "nonexistent"
-        return Path(p)
-
-    with (
-        patch("esphome.util.sys.platform", "linux"),
-        patch("esphome.util.Path", side_effect=mock_path_side_effect),
-    ):
-        result = util.get_rp2040_mass_storage_volumes()
-
-    assert result == []
-
-
-def test_get_rp2040_mass_storage_volumes_windows() -> None:
-    """Test RP2040 mass storage detection on Windows."""
-    mock_ctypes = MagicMock()
-    mock_volume_name = MagicMock()
-    mock_volume_name.value = "RPI-RP2"
-    mock_ctypes.create_unicode_buffer.return_value = mock_volume_name
-
-    def path_side_effect(p: str) -> MagicMock:
-        inst = MagicMock()
-        inst.exists.return_value = p == "D:\\"
-        return inst
-
-    with (
-        patch("esphome.util.sys.platform", "win32"),
-        patch.dict("sys.modules", {"ctypes": mock_ctypes}),
-        patch("esphome.util.Path", side_effect=path_side_effect),
-    ):
-        result = util.get_rp2040_mass_storage_volumes()
-
-    assert len(result) >= 1
-    assert result[0].description == "RP2040 BOOTSEL"
-
-
-def test_get_rp2040_mass_storage_volumes_unsupported_platform() -> None:
-    """Test RP2040 mass storage detection on unsupported platform returns empty."""
-    with patch("esphome.util.sys.platform", "freebsd"):
-        result = util.get_rp2040_mass_storage_volumes()
-
-    assert result == []
-
-
-def test_mass_storage_volume_attributes() -> None:
-    """Test MassStorageVolume class attributes."""
-    vol = util.MassStorageVolume(Path("/Volumes/RPI-RP2"), "RP2040 BOOTSEL")
-    assert vol.path == Path("/Volumes/RPI-RP2")
-    assert vol.description == "RP2040 BOOTSEL"
+        count = util.detect_rp2040_bootsel("/usr/bin/picotool")
+    assert count == 0