Fix timeout to use ESPhome built-in function.

This commit is contained in:
Roy Walker
2026-02-28 16:22:28 -06:00
parent 1a61cd622e
commit d9b712ee5f
+136 -17
View File
@@ -5,7 +5,7 @@ import esphome.codegen as cg
from esphome.components.esp32 import add_idf_sdkconfig_option
from esphome.components.psram import is_guaranteed as psram_is_guaranteed
import esphome.config_validation as cv
from esphome.const import CONF_ENABLE_IPV6, CONF_MIN_IPV6_ADDR_COUNT, CONF_PRIORITY
from esphome.const import CONF_ENABLE_IPV6, CONF_MIN_IPV6_ADDR_COUNT, CONF_PRIORITY, CONF_TIMEOUT
from esphome.core import CORE, CoroPriority, coroutine_with_priority
import esphome.final_validate as fv
@@ -19,12 +19,19 @@ _LOGGER = logging.getLogger(__name__)
KEY_HIGH_PERFORMANCE_NETWORKING = "high_performance_networking"
CONF_ENABLE_HIGH_PERFORMANCE = "enable_high_performance"
# Network priority tracking
# Network priority tracking infrastructure
# Components can query this to determine their relative setup priority and fallback timeout.
# CORE.data[KEY_NETWORK_PRIORITY] is a list of dicts:
# [{"interface": "ethernet", "timeout": 30000}, {"interface": "wifi", "timeout": None}, ...]
# where timeout is in milliseconds, or None meaning "start the next interface immediately".
KEY_NETWORK_PRIORITY = "network_priority"
VALID_NETWORK_TYPES = ["ethernet", "openthread", "wifi", "modem"]
# Setup priority base values — first in list gets the highest priority
NETWORK_PRIORITY_BASE = 300.0
NETWORK_PRIORITY_STEP = 100.0
network_ns = cg.esphome_ns.namespace("network")
IPAddress = network_ns.class_("IPAddress")
@@ -112,6 +119,18 @@ def has_high_performance_networking() -> bool:
return CORE.data.get(KEY_HIGH_PERFORMANCE_NETWORKING, False)
def _get_priority_entry(iface: str) -> dict | None:
"""Return the priority entry dict for the given interface, or None if not configured."""
priority_list = CORE.data.get(KEY_NETWORK_PRIORITY)
if priority_list is None:
return None
iface_lower = iface.lower()
for entry in priority_list:
if entry["interface"] == iface_lower:
return entry
return None
def get_network_priority(iface: str) -> float | None:
"""Get the setup priority for the given network interface type.
@@ -123,8 +142,8 @@ def get_network_priority(iface: str) -> float | None:
the calling component should fall back to its own default setup priority.
Args:
iface: Interface type string — one of ``"ethernet"`` or ``"wifi"``
(case-insensitive).
iface: Interface type string — one of ``"ethernet"``, ``"wifi"``,
``"openthread"`` or ``"modem"`` (case-insensitive).
Returns:
float setup priority, or None if no priority list was configured.
@@ -146,19 +165,109 @@ def get_network_priority(iface: str) -> float | None:
if priority_list is None:
return None
iface_lower = iface.lower()
try:
idx = priority_list.index(iface_lower)
except ValueError:
for idx, entry in enumerate(priority_list):
if entry["interface"] == iface_lower:
return NETWORK_PRIORITY_BASE - (idx * NETWORK_PRIORITY_STEP)
return None
def get_network_timeout(iface: str) -> int | None:
"""Get the fallback timeout in milliseconds for the given network interface.
Returns the timeout (in ms) that the runtime should wait for ``iface`` to
connect before attempting to bring up the next interface in the priority
list. Returns ``None`` if no timeout was configured for this interface,
meaning the next interface should start immediately.
Args:
iface: Interface type string — one of ``"ethernet"``, ``"wifi"``,
``"openthread"`` or ``"modem"`` (case-insensitive).
Returns:
int timeout in milliseconds, or None if no timeout is configured.
Example usage inside a component's ``to_code``::
from esphome.components import network
async def to_code(config):
...
timeout_ms = network.get_network_timeout("ethernet")
if timeout_ms is not None:
cg.add(var.set_fallback_timeout(timeout_ms))
...
"""
entry = _get_priority_entry(iface)
if entry is None:
return None
return NETWORK_PRIORITY_BASE - (idx * NETWORK_PRIORITY_STEP)
return entry.get("timeout")
def _validate_timeout(value):
"""Accept any common ESPHome/HA time period format, or a plain integer as seconds.
Accepted formats: 30s, 10sec, 1min, 500ms, 1h, 1.5h, 30 (plain int → 30s).
"""
if isinstance(value, int):
# Plain integer — treat as seconds, e.g. timeout: 30 means 30s
return cv.positive_time_period_milliseconds(f"{value}s")
return cv.positive_time_period_milliseconds(value)
def _priority_entry_schema(value):
"""Validate a single priority list entry in either plain string or mapping form.
Plain string form (no timeout):
- ethernet
Mapping form with optional timeout:
- ethernet:
timeout: 30s
"""
if isinstance(value, str):
return cv.one_of(*VALID_NETWORK_TYPES, lower=True)(value)
if isinstance(value, dict):
if len(value) != 1:
raise cv.Invalid(
"Each priority entry must have exactly one interface name as its key"
)
iface = next(iter(value))
cv.one_of(*VALID_NETWORK_TYPES, lower=True)(iface)
opts = cv.Schema(
{
cv.Optional(CONF_TIMEOUT): _validate_timeout,
}
)(value[iface] or {})
return {iface: opts}
raise cv.Invalid(
f"Expected an interface name string or a mapping, got {type(value).__name__}"
)
def _normalize_priority_entry(value) -> dict:
"""Normalize a validated priority entry to a canonical dict.
Returns a dict with keys:
- ``interface``: str, lowercase interface name
- ``timeout``: int milliseconds, or None
"""
if isinstance(value, str):
return {"interface": value, "timeout": None}
# Mapping form — exactly one key (the interface name)
iface, opts = next(iter(value.items()))
timeout = opts.get(CONF_TIMEOUT)
timeout_ms = int(timeout.total_milliseconds) if timeout is not None else None
return {"interface": iface, "timeout": timeout_ms}
def _validate_priority_list(value):
"""Ensure the priority list has no duplicates and only valid interface names."""
value = cv.ensure_list(cv.one_of(*VALID_NETWORK_TYPES, lower=True))(value)
if len(value) != len(set(value)):
"""Validate and normalize the full priority list, rejecting duplicates."""
raw = cv.ensure_list(_priority_entry_schema)(value)
entries = [_normalize_priority_entry(e) for e in raw]
interfaces = [e["interface"] for e in entries]
if len(interfaces) != len(set(interfaces)):
raise cv.Invalid("Duplicate entries are not allowed in 'priority'")
return value
return entries
CONFIG_SCHEMA = cv.Schema(
@@ -194,7 +303,8 @@ CONFIG_SCHEMA = cv.Schema(
def _final_validate(config):
"""Check that every interface named in 'priority' has a corresponding component block."""
full = fv.full_config.get()
for iface in config.get(CONF_PRIORITY, []):
for entry in config.get(CONF_PRIORITY, []):
iface = entry["interface"]
if iface not in full:
raise cv.Invalid(
f"'{iface}' is listed in 'network: priority:' but no '{iface}:' "
@@ -211,13 +321,22 @@ async def to_code(config):
cg.add_define("USE_NETWORK")
# ESP32 with Arduino uses ESP-IDF network APIs directly, no Arduino Network library needed
# Store the user-declared network priority list in CORE.data so that the
# ethernet and wifi components can query it via get_network_priority() during
# their own to_code phase.
# Store the user-declared network priority list in CORE.data so that ethernet,
# wifi and other network components can query it via get_network_priority() and
# get_network_timeout() during their own to_code phase.
if CONF_PRIORITY in config:
priority_list = config[CONF_PRIORITY]
CORE.data[KEY_NETWORK_PRIORITY] = priority_list
_LOGGER.info("Network interface priority: %s", " > ".join(priority_list))
def _fmt(entry):
if entry["timeout"] is not None:
return f"{entry['interface']} (timeout: {entry['timeout']}ms)"
return entry["interface"]
_LOGGER.info(
"Network interface priority: %s",
" > ".join(_fmt(e) for e in priority_list),
)
# Apply high performance networking settings
# Config can explicitly enable/disable, or default to component-driven behavior