sascha/esphome
1
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2025-09-27 23:52:28 +01:00
Code Issues Releases Wiki Activity

dry

Browse Source
This commit is contained in:
J. Nick Koston
2025-09-12 16:36:57 -05:00
parent 3909d271f7
commit 4b15421d42
1 changed files with 11 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
esphome/__main__.py
View File

@@ -114,6 +114,14 @@ class Purpose(StrEnum):
LOGGING = "logging" LOGGING = "logging"
def _resolve_with_cache(address: str, purpose: Purpose) -> list[str]:
"""Resolve an address using cache if available, otherwise return the address itself."""
if CORE.address_cache and (cached := CORE.address_cache.get_addresses(address)):
_LOGGER.debug("Using cached addresses for %s: %s", purpose.value, cached)
return cached
return [address]
def choose_upload_log_host( def choose_upload_log_host(
default: list[str] | str | None, default: list[str] | str | None,
check_default: str | None, check_default: str | None,
@@ -142,14 +150,7 @@ def choose_upload_log_host(
(purpose == Purpose.LOGGING and has_api()) (purpose == Purpose.LOGGING and has_api())
or (purpose == Purpose.UPLOADING and has_ota()) or (purpose == Purpose.UPLOADING and has_ota())
): ):
# Check if we have cached addresses for CORE.address resolved.extend(_resolve_with_cache(CORE.address, purpose))
if CORE.address_cache and (
cached := CORE.address_cache.get_addresses(CORE.address)
):
_LOGGER.debug("Using cached addresses for OTA: %s", cached)
resolved.extend(cached)
else:
resolved.append(CORE.address)
if purpose == Purpose.LOGGING: if purpose == Purpose.LOGGING:
if has_api() and has_mqtt_ip_lookup(): if has_api() and has_mqtt_ip_lookup():
@@ -159,32 +160,14 @@ def choose_upload_log_host(
resolved.append("MQTT") resolved.append("MQTT")
if has_api() and has_non_ip_address(): if has_api() and has_non_ip_address():
# Check if we have cached addresses for CORE.address resolved.extend(_resolve_with_cache(CORE.address, purpose))
if CORE.address_cache and (
cached := CORE.address_cache.get_addresses(CORE.address)
):
_LOGGER.debug(
"Using cached addresses for logging: %s", cached
)
resolved.extend(cached)
else:
resolved.append(CORE.address)
elif purpose == Purpose.UPLOADING: elif purpose == Purpose.UPLOADING:
if has_ota() and has_mqtt_ip_lookup(): if has_ota() and has_mqtt_ip_lookup():
resolved.append("MQTTIP") resolved.append("MQTTIP")
if has_ota() and has_non_ip_address(): if has_ota() and has_non_ip_address():
# Check if we have cached addresses for CORE.address resolved.extend(_resolve_with_cache(CORE.address, purpose))
if CORE.address_cache and (
cached := CORE.address_cache.get_addresses(CORE.address)
):
_LOGGER.debug(
"Using cached addresses for uploading: %s", cached
)
resolved.extend(cached)
else:
resolved.append(CORE.address)
else: else:
resolved.append(device) resolved.append(device)
if not resolved: if not resolved: