Support multiple --device arguments for address fallback

esphome/__main__.py

@@ -356,7 +356,7 @@ def upload_program(config, args, host):
             return upload_using_esptool(config, host, file, args.upload_speed)
 
         if CORE.target_platform in (PLATFORM_RP2040):
-            return upload_using_platformio(config, args.device)
+            return upload_using_platformio(config, host)
 
         if CORE.is_libretiny:
             return upload_using_platformio(config, host)
@@ -379,9 +379,12 @@ def upload_program(config, args, host):
     remote_port = int(ota_conf[CONF_PORT])
     password = ota_conf.get(CONF_PASSWORD, "")
 
+    # Check if we should use MQTT for address resolution
+    # This happens when no device was specified, or the current host is "MQTT"/"OTA"
+    devices = args.device or []
     if (
         CONF_MQTT in config  # pylint: disable=too-many-boolean-expressions
-        and (not args.device or args.device in ("MQTT", "OTA"))
+        and (not devices or host in ("MQTT", "OTA"))
         and (
             ((config[CONF_MDNS][CONF_DISABLED]) and not is_ip_address(CORE.address))
             or get_port_type(host) == "MQTT"
@@ -399,23 +402,28 @@ def upload_program(config, args, host):
     return espota2.run_ota(host, remote_port, password, CORE.firmware_bin)
 
 
-def show_logs(config, args, port):
+def show_logs(config, args, devices):
     if "logger" not in config:
         raise EsphomeError("Logger is not configured!")
+
+    port = devices[0]
+
     if get_port_type(port) == "SERIAL":
         check_permissions(port)
         return run_miniterm(config, port, args)
     if get_port_type(port) == "NETWORK" and "api" in config:
+        addresses_to_use = devices
         if config[CONF_MDNS][CONF_DISABLED] and CONF_MQTT in config:
             from esphome import mqtt
 
-            port = mqtt.get_esphome_device_ip(
+            mqtt_address = mqtt.get_esphome_device_ip(
                 config, args.username, args.password, args.client_id
             )[0]
+            addresses_to_use = [mqtt_address]
 
         from esphome.components.api.client import run_logs
 
-        return run_logs(config, port)
+        return run_logs(config, addresses_to_use)
     if get_port_type(port) == "MQTT" and "mqtt" in config:
         from esphome import mqtt
 
@@ -478,19 +486,31 @@ def command_compile(args, config):
 
 
 def command_upload(args, config):
-    port = choose_upload_log_host(
+    devices = args.device or []
-        default=args.device,
+    if not devices:
+        # No devices specified, use the interactive chooser
+        devices = [
+            choose_upload_log_host(
+                default=None,
                 check_default=None,
                 show_ota=True,
                 show_mqtt=False,
                 show_api=False,
                 purpose="uploading",
             )
-    exit_code = upload_program(config, args, port)
+        ]
-    if exit_code != 0:
+
-        return exit_code
+    # Try each device until one succeeds
+    for device in devices:
+        _LOGGER.info("Uploading to %s", device)
+        exit_code = upload_program(config, args, device)
+        if exit_code == 0:
             _LOGGER.info("Successfully uploaded program.")
             return 0
+        if len(devices) > 1:
+            _LOGGER.warning("Failed to upload to %s", device)
+
+    return exit_code
 
 
 def command_discover(args, config):
@@ -503,15 +523,21 @@ def command_discover(args, config):
 
 
 def command_logs(args, config):
-    port = choose_upload_log_host(
+    devices = args.device or []
-        default=args.device,
+    if not devices:
+        # No devices specified, use the interactive chooser
+        devices = [
+            choose_upload_log_host(
+                default=None,
                 check_default=None,
                 show_ota=False,
                 show_mqtt=True,
                 show_api=True,
                 purpose="logging",
             )
-    return show_logs(config, args, port)
+        ]
+
+    return show_logs(config, args, devices)
 
 
 def command_run(args, config):
@@ -531,29 +557,48 @@ def command_run(args, config):
         program_path = idedata.raw["prog_path"]
         return run_external_process(program_path)
 
-    port = choose_upload_log_host(
+    devices = args.device or []
-        default=args.device,
+    if not devices:
+        # No devices specified, use the interactive chooser
+        devices = [
+            choose_upload_log_host(
+                default=None,
                 check_default=None,
                 show_ota=True,
                 show_mqtt=False,
                 show_api=True,
                 purpose="uploading",
             )
-    exit_code = upload_program(config, args, port)
+        ]
-    if exit_code != 0:
+
-        return exit_code
+    # Try each device for upload until one succeeds
+    successful_device = None
+    for device in devices:
+        _LOGGER.info("Uploading to %s", device)
+        exit_code = upload_program(config, args, device)
+        if exit_code == 0:
             _LOGGER.info("Successfully uploaded program.")
+            successful_device = device
+            break
+        if len(devices) > 1:
+            _LOGGER.warning("Failed to upload to %s", device)
+
+    if successful_device is None:
+        return exit_code
+
     if args.no_logs:
         return 0
+
+    # For logs, prefer the device we successfully uploaded to
     port = choose_upload_log_host(
-        default=args.device,
+        default=successful_device,
-        check_default=port,
+        check_default=successful_device,
         show_ota=False,
         show_mqtt=True,
         show_api=True,
         purpose="logging",
     )
-    return show_logs(config, args, port)
+    return show_logs(config, args, [port])
 
 
 def command_clean_mqtt(args, config):
@@ -854,7 +899,8 @@ def parse_args(argv):
     )
     parser_upload.add_argument(
         "--device",
-        help="Manually specify the serial port/address to use, for example /dev/ttyUSB0.",
+        action="append",
+        help="Manually specify the serial port/address to use, for example /dev/ttyUSB0. Can be specified multiple times for fallback addresses.",
     )
     parser_upload.add_argument(
         "--upload_speed",
@@ -876,7 +922,8 @@ def parse_args(argv):
     )
     parser_logs.add_argument(
         "--device",
-        help="Manually specify the serial port/address to use, for example /dev/ttyUSB0.",
+        action="append",
+        help="Manually specify the serial port/address to use, for example /dev/ttyUSB0. Can be specified multiple times for fallback addresses.",
     )
     parser_logs.add_argument(
         "--reset",
@@ -905,7 +952,8 @@ def parse_args(argv):
     )
     parser_run.add_argument(
         "--device",
-        help="Manually specify the serial port/address to use, for example /dev/ttyUSB0.",
+        action="append",
+        help="Manually specify the serial port/address to use, for example /dev/ttyUSB0. Can be specified multiple times for fallback addresses.",
     )
     parser_run.add_argument(
         "--upload_speed",

esphome/components/api/client.py

@@ -30,7 +30,7 @@ if TYPE_CHECKING:
 _LOGGER = logging.getLogger(__name__)
 
 
-async def async_run_logs(config: dict[str, Any], address: str) -> None:
+async def async_run_logs(config: dict[str, Any], addresses: list[str]) -> None:
     """Run the logs command in the event loop."""
     conf = config["api"]
     name = config["esphome"]["name"]
@@ -39,13 +39,21 @@ async def async_run_logs(config: dict[str, Any], address: str) -> None:
     noise_psk: str | None = None
     if (encryption := conf.get(CONF_ENCRYPTION)) and (key := encryption.get(CONF_KEY)):
         noise_psk = key
-    _LOGGER.info("Starting log output from %s using esphome API", address)
+
+    if len(addresses) == 1:
+        _LOGGER.info("Starting log output from %s using esphome API", addresses[0])
+    else:
+        _LOGGER.info(
+            "Starting log output from %s using esphome API", " or ".join(addresses)
+        )
+
     cli = APIClient(
-        address,
+        addresses[0],  # Primary address for compatibility
         port,
         password,
         client_info=f"ESPHome Logs {__version__}",
         noise_psk=noise_psk,
+        addresses=addresses,  # Pass all addresses for automatic retry
     )
     dashboard = CORE.dashboard
 
@@ -66,7 +74,7 @@ async def async_run_logs(config: dict[str, Any], address: str) -> None:
         await stop()
 
 
-def run_logs(config: dict[str, Any], address: str) -> None:
+def run_logs(config: dict[str, Any], addresses: list[str]) -> None:
     """Run the logs command."""
     with contextlib.suppress(KeyboardInterrupt):
-        asyncio.run(async_run_logs(config, address))
+        asyncio.run(async_run_logs(config, addresses))

esphome/dashboard/web_server.py

@@ -324,6 +324,8 @@ class EsphomePortCommandWebSocket(EsphomeCommandWebSocket):
         configuration = json_message["configuration"]
         config_file = settings.rel_path(configuration)
         port = json_message["port"]
+        addresses: list[str] = []
+
         if (
             port == "OTA"  # pylint: disable=too-many-boolean-expressions
             and (entry := entries.get(config_file))
@@ -333,10 +335,10 @@ class EsphomePortCommandWebSocket(EsphomeCommandWebSocket):
             if (mdns := dashboard.mdns_status) and (
                 address_list := await mdns.async_resolve_host(entry.name)
             ):
-                # Use the IP address if available but only
+                # Use all IP addresses if available but only
                 # if the API is loaded and the device is online
                 # since MQTT logging will not work otherwise
-                port = sort_ip_addresses(address_list)[0]
+                addresses = sort_ip_addresses(address_list)
             elif (
                 entry.address
                 and (
@@ -347,16 +349,25 @@ class EsphomePortCommandWebSocket(EsphomeCommandWebSocket):
                 and not isinstance(address_list, Exception)
             ):
                 # If mdns is not available, try to use the DNS cache
-                port = sort_ip_addresses(address_list)[0]
+                addresses = sort_ip_addresses(address_list)
 
-        return [
+        # Build command with multiple --device arguments for each address
+        command = [
             *DASHBOARD_COMMAND,
             *args,
             config_file,
-            "--device",
-            port,
         ]
 
+        if addresses:
+            # Add multiple --device arguments for each resolved address
+            for address in addresses:
+                command.extend(["--device", address])
+        else:
+            # Fallback to original port if no addresses were resolved
+            command.extend(["--device", port])
+
+        return command
+
 
 class EsphomeLogsHandler(EsphomePortCommandWebSocket):
     async def build_command(self, json_message: dict[str, Any]) -> list[str]: