From bc03538e25a954cd78bac6b35de08a149988b026 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Mon, 4 Aug 2025 18:40:46 -1000 Subject: [PATCH] Support multiple --device arguments for address fallback (#10003) Co-authored-by: pre-commit-ci-lite[bot] <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com> --- esphome/__main__.py | 224 +++++++++++++++++++++---------- esphome/components/api/client.py | 18 ++- esphome/dashboard/web_server.py | 44 +++--- esphome/util.py | 11 +- 4 files changed, 201 insertions(+), 96 deletions(-) diff --git a/esphome/__main__.py b/esphome/__main__.py index 47e1c774ac..7cc8296e7e 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -9,6 +9,7 @@ import os import re import sys import time +from typing import Protocol import argcomplete @@ -44,6 +45,7 @@ from esphome.const import ( from esphome.core import CORE, EsphomeError, coroutine 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_serial_ports, list_yaml_files, @@ -55,6 +57,23 @@ from esphome.util import ( _LOGGER = logging.getLogger(__name__) +class ArgsProtocol(Protocol): + device: list[str] | None + reset: bool + username: str | None + password: str | None + client_id: str | None + topic: str | None + file: str | None + no_logs: bool + only_generate: bool + show_secrets: bool + dashboard: bool + configuration: str + name: str + upload_speed: str | None + + def choose_prompt(options, purpose: str = None): if not options: raise EsphomeError( @@ -88,30 +107,54 @@ def choose_prompt(options, purpose: str = None): def choose_upload_log_host( - default, check_default, show_ota, show_mqtt, show_api, purpose: str = None -): + default: list[str] | str | None, + check_default: str | None, + show_ota: bool, + show_mqtt: bool, + show_api: bool, + purpose: str | None = None, +) -> list[str]: + # Convert to list for uniform handling + defaults = [default] if isinstance(default, str) else default or [] + + # If devices specified, resolve them + if defaults: + resolved: list[str] = [] + for device in defaults: + if device == "SERIAL": + serial_ports = get_serial_ports() + if not serial_ports: + _LOGGER.warning("No serial ports found, skipping SERIAL device") + continue + options = [ + (f"{port.path} ({port.description})", port.path) + for port in serial_ports + ] + resolved.append(choose_prompt(options, purpose=purpose)) + elif device == "OTA": + if (show_ota and "ota" in CORE.config) or ( + show_api and "api" in CORE.config + ): + resolved.append(CORE.address) + elif show_mqtt and has_mqtt_logging(): + resolved.append("MQTT") + else: + resolved.append(device) + return resolved + + # No devices specified, show interactive chooser options = [ (f"{port.path} ({port.description})", port.path) for port in get_serial_ports() ] - if default == "SERIAL": - return choose_prompt(options, purpose=purpose) if (show_ota and "ota" in CORE.config) or (show_api and "api" in CORE.config): options.append((f"Over The Air ({CORE.address})", CORE.address)) - if default == "OTA": - return CORE.address - if ( - show_mqtt - and (mqtt_config := CORE.config.get(CONF_MQTT)) - and mqtt_logging_enabled(mqtt_config) - ): + if show_mqtt and has_mqtt_logging(): + mqtt_config = CORE.config[CONF_MQTT] options.append((f"MQTT ({mqtt_config[CONF_BROKER]})", "MQTT")) - if default == "OTA": - return "MQTT" - if default is not None: - return default + if check_default is not None and check_default in [opt[1] for opt in options]: - return check_default - return choose_prompt(options, purpose=purpose) + return [check_default] + return [choose_prompt(options, purpose=purpose)] def mqtt_logging_enabled(mqtt_config): @@ -123,7 +166,14 @@ def mqtt_logging_enabled(mqtt_config): return log_topic.get(CONF_LEVEL, None) != "NONE" -def get_port_type(port): +def has_mqtt_logging() -> bool: + """Check if MQTT logging is available.""" + return (mqtt_config := CORE.config.get(CONF_MQTT)) and mqtt_logging_enabled( + mqtt_config + ) + + +def get_port_type(port: str) -> str: if port.startswith("/") or port.startswith("COM"): return "SERIAL" if port == "MQTT": @@ -131,7 +181,7 @@ def get_port_type(port): return "NETWORK" -def run_miniterm(config, port, args): +def run_miniterm(config: ConfigType, port: str, args) -> int: from aioesphomeapi import LogParser import serial @@ -208,7 +258,7 @@ def wrap_to_code(name, comp): return wrapped -def write_cpp(config): +def write_cpp(config: ConfigType) -> int: if not get_bool_env(ENV_NOGITIGNORE): writer.write_gitignore() @@ -216,7 +266,7 @@ def write_cpp(config): return write_cpp_file() -def generate_cpp_contents(config): +def generate_cpp_contents(config: ConfigType) -> None: _LOGGER.info("Generating C++ source...") for name, component, conf in iter_component_configs(CORE.config): @@ -227,7 +277,7 @@ def generate_cpp_contents(config): CORE.flush_tasks() -def write_cpp_file(): +def write_cpp_file() -> int: code_s = indent(CORE.cpp_main_section) writer.write_cpp(code_s) @@ -238,7 +288,7 @@ def write_cpp_file(): return 0 -def compile_program(args, config): +def compile_program(args: ArgsProtocol, config: ConfigType) -> int: from esphome import platformio_api _LOGGER.info("Compiling app...") @@ -249,7 +299,9 @@ def compile_program(args, config): return 0 if idedata is not None else 1 -def upload_using_esptool(config, port, file, speed): +def upload_using_esptool( + config: ConfigType, port: str, file: str, speed: int +) -> str | int: from esphome import platformio_api first_baudrate = speed or config[CONF_ESPHOME][CONF_PLATFORMIO_OPTIONS].get( @@ -314,7 +366,7 @@ def upload_using_esptool(config, port, file, speed): return run_esptool(115200) -def upload_using_platformio(config, port): +def upload_using_platformio(config: ConfigType, port: str): from esphome import platformio_api upload_args = ["-t", "upload", "-t", "nobuild"] @@ -323,7 +375,7 @@ def upload_using_platformio(config, port): return platformio_api.run_platformio_cli_run(config, CORE.verbose, *upload_args) -def check_permissions(port): +def check_permissions(port: str): if os.name == "posix" and get_port_type(port) == "SERIAL": # Check if we can open selected serial port if not os.access(port, os.F_OK): @@ -341,7 +393,7 @@ def check_permissions(port): ) -def upload_program(config, args, host): +def upload_program(config: ConfigType, args: ArgsProtocol, host: str) -> int | str: try: module = importlib.import_module("esphome.components." + CORE.target_platform) if getattr(module, "upload_program")(config, args, host): @@ -356,7 +408,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 +431,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: list[str] = 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 +454,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: ConfigType, args: ArgsProtocol, devices: list[str]) -> int | None: 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 @@ -426,7 +486,7 @@ def show_logs(config, args, port): raise EsphomeError("No remote or local logging method configured (api/mqtt/logger)") -def clean_mqtt(config, args): +def clean_mqtt(config: ConfigType, args: ArgsProtocol) -> int | None: from esphome import mqtt return mqtt.clear_topic( @@ -434,13 +494,13 @@ def clean_mqtt(config, args): ) -def command_wizard(args): +def command_wizard(args: ArgsProtocol) -> int | None: from esphome import wizard return wizard.wizard(args.configuration) -def command_config(args, config): +def command_config(args: ArgsProtocol, config: ConfigType) -> int | None: if not CORE.verbose: config = strip_default_ids(config) output = yaml_util.dump(config, args.show_secrets) @@ -455,7 +515,7 @@ def command_config(args, config): return 0 -def command_vscode(args): +def command_vscode(args: ArgsProtocol) -> int | None: from esphome import vscode logging.disable(logging.INFO) @@ -463,7 +523,7 @@ def command_vscode(args): vscode.read_config(args) -def command_compile(args, config): +def command_compile(args: ArgsProtocol, config: ConfigType) -> int | None: exit_code = write_cpp(config) if exit_code != 0: return exit_code @@ -477,8 +537,9 @@ def command_compile(args, config): return 0 -def command_upload(args, config): - port = choose_upload_log_host( +def command_upload(args: ArgsProtocol, config: ConfigType) -> int | None: + # Get devices, resolving special identifiers like OTA + devices = choose_upload_log_host( default=args.device, check_default=None, show_ota=True, @@ -486,14 +547,22 @@ def command_upload(args, config): show_api=False, purpose="uploading", ) - exit_code = upload_program(config, args, port) - if exit_code != 0: - return exit_code - _LOGGER.info("Successfully uploaded program.") - return 0 + + # Try each device until one succeeds + exit_code = 1 + 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): +def command_discover(args: ArgsProtocol, config: ConfigType) -> int | None: if "mqtt" in config: from esphome import mqtt @@ -502,8 +571,9 @@ def command_discover(args, config): raise EsphomeError("No discover method configured (mqtt)") -def command_logs(args, config): - port = choose_upload_log_host( +def command_logs(args: ArgsProtocol, config: ConfigType) -> int | None: + # Get devices, resolving special identifiers like OTA + devices = choose_upload_log_host( default=args.device, check_default=None, show_ota=False, @@ -511,10 +581,10 @@ def command_logs(args, config): show_api=True, purpose="logging", ) - return show_logs(config, args, port) + return show_logs(config, args, devices) -def command_run(args, config): +def command_run(args: ArgsProtocol, config: ConfigType) -> int | None: exit_code = write_cpp(config) if exit_code != 0: return exit_code @@ -531,7 +601,8 @@ def command_run(args, config): program_path = idedata.raw["prog_path"] return run_external_process(program_path) - port = choose_upload_log_host( + # Get devices, resolving special identifiers like OTA + devices = choose_upload_log_host( default=args.device, check_default=None, show_ota=True, @@ -539,39 +610,53 @@ def command_run(args, config): show_api=True, purpose="uploading", ) - exit_code = upload_program(config, args, port) - if exit_code != 0: + + # Try each device for upload until one succeeds + successful_device: str | None = 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 - _LOGGER.info("Successfully uploaded program.") + if args.no_logs: return 0 - port = choose_upload_log_host( - default=args.device, - check_default=port, + + # For logs, prefer the device we successfully uploaded to + devices = choose_upload_log_host( + default=successful_device, + 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, devices) -def command_clean_mqtt(args, config): +def command_clean_mqtt(args: ArgsProtocol, config: ConfigType) -> int | None: return clean_mqtt(config, args) -def command_mqtt_fingerprint(args, config): +def command_mqtt_fingerprint(args: ArgsProtocol, config: ConfigType) -> int | None: from esphome import mqtt return mqtt.get_fingerprint(config) -def command_version(args): +def command_version(args: ArgsProtocol) -> int | None: safe_print(f"Version: {const.__version__}") return 0 -def command_clean(args, config): +def command_clean(args: ArgsProtocol, config: ConfigType) -> int | None: try: writer.clean_build() except OSError as err: @@ -581,13 +666,13 @@ def command_clean(args, config): return 0 -def command_dashboard(args): +def command_dashboard(args: ArgsProtocol) -> int | None: from esphome.dashboard import dashboard return dashboard.start_dashboard(args) -def command_update_all(args): +def command_update_all(args: ArgsProtocol) -> int | None: import click success = {} @@ -634,7 +719,7 @@ def command_update_all(args): return failed -def command_idedata(args, config): +def command_idedata(args: ArgsProtocol, config: ConfigType) -> int: import json from esphome import platformio_api @@ -650,7 +735,7 @@ def command_idedata(args, config): return 0 -def command_rename(args, config): +def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None: for c in args.name: if c not in ALLOWED_NAME_CHARS: print( @@ -860,7 +945,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", @@ -882,7 +968,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", @@ -911,7 +998,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", diff --git a/esphome/components/api/client.py b/esphome/components/api/client.py index 5239e07435..ce018b3b98 100644 --- a/esphome/components/api/client.py +++ b/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)) diff --git a/esphome/dashboard/web_server.py b/esphome/dashboard/web_server.py index 286dc9e1d7..46f09336bb 100644 --- a/esphome/dashboard/web_server.py +++ b/esphome/dashboard/web_server.py @@ -324,39 +324,47 @@ class EsphomePortCommandWebSocket(EsphomeCommandWebSocket): configuration = json_message["configuration"] config_file = settings.rel_path(configuration) port = json_message["port"] + addresses: list[str] = [port] if ( port == "OTA" # pylint: disable=too-many-boolean-expressions and (entry := entries.get(config_file)) and entry.loaded_integrations and "api" in entry.loaded_integrations ): - if (mdns := dashboard.mdns_status) and ( - address_list := await mdns.async_resolve_host(entry.name) - ): - # Use the IP address 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] - elif ( - entry.address + addresses = [] + # First priority: entry.address AKA use_address + if ( + (use_address := entry.address) and ( address_list := await dashboard.dns_cache.async_resolve( - entry.address, time.monotonic() + use_address, time.monotonic() ) ) 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.extend(sort_ip_addresses(address_list)) - return [ - *DASHBOARD_COMMAND, - *args, - config_file, - "--device", - port, + # Second priority: mDNS + if ( + (mdns := dashboard.mdns_status) + and (address_list := await mdns.async_resolve_host(entry.name)) + and ( + new_addresses := [ + addr for addr in address_list if addr not in addresses + ] + ) + ): + # Use the IP address if available but only + # if the API is loaded and the device is online + # since MQTT logging will not work otherwise + addresses.extend(sort_ip_addresses(new_addresses)) + + device_args: list[str] = [ + arg for address in addresses for arg in ("--device", address) ] + return [*DASHBOARD_COMMAND, *args, config_file, *device_args] + class EsphomeLogsHandler(EsphomePortCommandWebSocket): async def build_command(self, json_message: dict[str, Any]) -> list[str]: diff --git a/esphome/util.py b/esphome/util.py index 9aa0f6b9d8..ed9ab4a446 100644 --- a/esphome/util.py +++ b/esphome/util.py @@ -6,6 +6,7 @@ from pathlib import Path import re import subprocess import sys +from typing import Any from esphome import const @@ -110,7 +111,7 @@ class RedirectText: def __getattr__(self, item): return getattr(self._out, item) - def _write_color_replace(self, s): + def _write_color_replace(self, s: str | bytes) -> None: from esphome.core import CORE if CORE.dashboard: @@ -121,7 +122,7 @@ class RedirectText: s = s.replace("\033", "\\033") self._out.write(s) - def write(self, s): + def write(self, s: str | bytes) -> int: # s is usually a str already (self._out is of type TextIOWrapper) # However, s is sometimes also a bytes object in python3. Let's make sure it's a # str @@ -223,7 +224,7 @@ def run_external_command( return retval -def run_external_process(*cmd, **kwargs): +def run_external_process(*cmd: str, **kwargs: Any) -> int | str: full_cmd = " ".join(shlex_quote(x) for x in cmd) _LOGGER.debug("Running: %s", full_cmd) filter_lines = kwargs.get("filter_lines") @@ -266,7 +267,7 @@ class OrderedDict(collections.OrderedDict): return dict(self).__repr__() -def list_yaml_files(folders): +def list_yaml_files(folders: list[str]) -> list[str]: files = filter_yaml_files( [os.path.join(folder, p) for folder in folders for p in os.listdir(folder)] ) @@ -274,7 +275,7 @@ def list_yaml_files(folders): return files -def filter_yaml_files(files): +def filter_yaml_files(files: list[str]) -> list[str]: return [ f for f in files