[core] Use component DOMAIN constants instead of name literals (#20419)

This commit is contained in:
Jesse Hills
2026-10-09 13:50:29 +13:00
committed by GitHub
parent 0d596a20f8
commit d9795399ad
52 changed files with 198 additions and 165 deletions
+2 -1
View File
@@ -8,6 +8,7 @@ from esphome.automation import Condition
import esphome.codegen as cg
from esphome.components.const import CONF_DESCRIPTION, CONF_HOST
from esphome.components.logger import request_log_listener
from esphome.components.network import DOMAIN as NETWORK_DOMAIN
# ENCRYPTION_SCHEMA and validate_encryption_key are re-exported for external
# components and downstream consumers that import them from api
@@ -514,7 +515,7 @@ def _validate_outgoing_host_ipv6(config: ConfigType) -> ConfigType:
or host.version != 6
):
return config
network_conf = fv.full_config.get().get("network") or {}
network_conf = fv.full_config.get().get(NETWORK_DOMAIN) or {}
if not network_conf.get(CONF_ENABLE_IPV6):
raise cv.Invalid(
"outgoing_connection host is an IPv6 address but IPv6 is not "
+2 -1
View File
@@ -11,6 +11,7 @@ from typing import Any
from esphome import automation
import esphome.codegen as cg
from esphome.components.const import CONF_DESCRIPTION
from esphome.components.homeassistant import DOMAIN as HOMEASSISTANT_DOMAIN
import esphome.config_validation as cv
from esphome.const import (
CONF_DEVICE_CLASS,
@@ -265,7 +266,7 @@ def _validate_wizard_input(conf: ConfigType) -> ConfigType:
fv.full_config.get(), conf[CONF_ENTITY]
)
if (
declaration.get(CONF_PLATFORM) != "homeassistant"
declaration.get(CONF_PLATFORM) != HOMEASSISTANT_DOMAIN
or domain not in WIZARD_INPUT_DOMAINS
):
raise cv.Invalid(
@@ -3,6 +3,7 @@ import logging
import esphome.codegen as cg
from esphome.components import web_server_base, wifi
from esphome.components.web_server_base import CONF_WEB_SERVER_BASE_ID
from esphome.components.wifi import DOMAIN as WIFI_DOMAIN
from esphome.config_helpers import filter_source_files_from_platform
import esphome.config_validation as cv
from esphome.const import (
@@ -64,7 +65,7 @@ CONFIG_SCHEMA = cv.All(
def _final_validate(config: ConfigType) -> None:
full_config = fv.full_config.get()
wifi_conf = full_config.get("wifi")
wifi_conf = full_config.get(WIFI_DOMAIN)
if wifi_conf is None:
# This shouldn't happen due to DEPENDENCIES = ["wifi"], but check anyway
+1 -1
View File
@@ -60,7 +60,7 @@ def pin_mode_check(pin_config: ConfigType, _: ConfigType) -> None:
is_open_drain = pin_config[CONF_MODE][CONF_OPEN_DRAIN]
# Track pin modes per CH423 instance in CORE.data
ch423_modes = CORE.data.setdefault(CONF_CH423, {})
ch423_modes = CORE.data.setdefault(DOMAIN, {})
if ch423_id not in ch423_modes:
ch423_modes[ch423_id] = {"gpio_output": None, "gpo_open_drain": None}
+9 -6
View File
@@ -4,6 +4,9 @@ from typing import Any
import esphome.codegen as cg
from esphome.components import esp32, uart
from esphome.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from esphome.components.sensor import DOMAIN as SENSOR_DOMAIN
from esphome.components.text_sensor import DOMAIN as TEXT_SENSOR_DOMAIN
import esphome.config_validation as cv
from esphome.const import (
CONF_ID,
@@ -211,9 +214,9 @@ async def to_code(config: ConfigType) -> None:
hub_id = config[CONF_ID].id
sensor_count = 0
for sens_conf in CORE.config.get("sensor", []):
for sens_conf in CORE.config.get(SENSOR_DOMAIN, []):
if (
sens_conf.get("platform") == "dlms_meter"
sens_conf.get("platform") == DOMAIN
and sens_conf.get(CONF_DLMS_METER_ID).id == hub_id
):
if CONF_OBIS_CODE in sens_conf:
@@ -224,9 +227,9 @@ async def to_code(config: ConfigType) -> None:
sensor_count += sum(1 for key in NUMERIC_KEYS if key in sens_conf)
text_sensor_count = 0
for sens_conf in CORE.config.get("text_sensor", []):
for sens_conf in CORE.config.get(TEXT_SENSOR_DOMAIN, []):
if (
sens_conf.get("platform") == "dlms_meter"
sens_conf.get("platform") == DOMAIN
and sens_conf.get(CONF_DLMS_METER_ID).id == hub_id
):
if CONF_OBIS_CODE in sens_conf:
@@ -237,9 +240,9 @@ async def to_code(config: ConfigType) -> None:
text_sensor_count += sum(1 for key in TEXT_KEYS if key in sens_conf)
binary_sensor_count = 0
for sens_conf in CORE.config.get("binary_sensor", []):
for sens_conf in CORE.config.get(BINARY_SENSOR_DOMAIN, []):
if (
sens_conf.get("platform") == "dlms_meter"
sens_conf.get("platform") == DOMAIN
and sens_conf.get(CONF_DLMS_METER_ID).id == hub_id
):
binary_sensor_count += 1
+2 -1
View File
@@ -3,6 +3,7 @@ import logging
from esphome import pins
import esphome.codegen as cg
from esphome.components import uart
from esphome.components.uart import DOMAIN as UART_DOMAIN
import esphome.config_validation as cv
from esphome.const import (
CONF_ID,
@@ -92,7 +93,7 @@ async def to_code(config: ConfigType) -> None:
def final_validate(config: ConfigType) -> None:
full_config = fv.full_config.get()
for uart_conf in full_config["uart"]:
for uart_conf in full_config[UART_DOMAIN]:
if uart_conf[CONF_ID] == config[CONF_UART_ID]:
rx_buffer_size = uart_conf[CONF_RX_BUFFER_SIZE]
if rx_buffer_size < 1500:
+5 -3
View File
@@ -3,6 +3,8 @@ from dataclasses import dataclass, field
from esphome import automation
import esphome.codegen as cg
from esphome.components import uart
from esphome.components.sensor import DOMAIN as SENSOR_DOMAIN
from esphome.components.uart import DOMAIN as UART_DOMAIN
import esphome.config_validation as cv
from esphome.const import (
CONF_COMMAND,
@@ -63,13 +65,13 @@ def final_validate(config: ConfigType) -> None:
hub_id = str(config[CONF_ID])
sensor_count = sum(
1
for s in full_config.get("sensor", [])
if s.get("platform") == "emontx" and str(s.get(CONF_EMONTX_ID)) == hub_id
for s in full_config.get(SENSOR_DOMAIN, [])
if s.get("platform") == DOMAIN and str(s.get(CONF_EMONTX_ID)) == hub_id
)
_get_data().sensor_counts[hub_id] = sensor_count
# Ensure UART RX buffer size is large enough to handle data bursts from firmware
for uart_conf in full_config["uart"]:
for uart_conf in full_config[UART_DOMAIN]:
if uart_conf[CONF_ID] == config[CONF_UART_ID]:
current_buffer_size = uart_conf[CONF_RX_BUFFER_SIZE]
if current_buffer_size < MINIMUM_RX_BUFFER_SIZE:
+2 -1
View File
@@ -1,6 +1,7 @@
import esphome.codegen as cg
from esphome.components import i2c
from esphome.components.audio_dac import AudioDac
from esphome.components.speaker import DOMAIN as SPEAKER_DOMAIN
import esphome.config_validation as cv
from esphome.const import CONF_AUDIO_DAC, CONF_BITS_PER_SAMPLE, CONF_ID
import esphome.final_validate as fv
@@ -27,7 +28,7 @@ def _final_validate(config: ConfigType) -> None:
full_config = fv.full_config.get()
# Check all speaker configurations for ones that reference this es8156
speaker_configs = full_config.get("speaker", [])
speaker_configs = full_config.get(SPEAKER_DOMAIN, [])
for speaker_config in speaker_configs:
audio_dac_id = speaker_config.get(CONF_AUDIO_DAC)
if (
+7 -5
View File
@@ -22,6 +22,8 @@ from esphome.components.esp32 import (
request_bluetooth,
)
from esphome.components.esp32.const import VARIANT_ESP32C2
from esphome.components.esp32_hosted import DOMAIN as ESP32_HOSTED_DOMAIN
from esphome.components.psram import DOMAIN as PSRAM_DOMAIN
from esphome.config_helpers import filter_source_files_from_defines
import esphome.config_validation as cv
from esphome.const import (
@@ -277,7 +279,7 @@ DEFAULT_MAX_CONNECTIONS = 3
IDF_MAX_CONNECTIONS = 9
# Connection slot tracking keys
KEY_ESP32_BLE = "esp32_ble"
KEY_ESP32_BLE = DOMAIN
KEY_USED_CONNECTION_SLOTS = "used_connection_slots"
# Export for use by other components (bluetooth_proxy, etc.)
@@ -361,7 +363,7 @@ CONFIG_SCHEMA = cv.Schema(
cv.positive_int, cv.Range(min=1, max=IDF_MAX_CONNECTIONS)
),
cv.Optional(CONF_USE_PSRAM): cv.All(
cv.only_on_esp32, cv.requires_component("psram"), cv.boolean
cv.only_on_esp32, cv.requires_component(PSRAM_DOMAIN), cv.boolean
),
}
).extend(cv.COMPONENT_SCHEMA)
@@ -404,7 +406,7 @@ def consume_connection_slots(
"""
def _consume_connection_slots(config: MutableMapping) -> MutableMapping:
data: dict[str, Any] = CORE.data.setdefault(KEY_ESP32_BLE, {})
data: dict[str, Any] = CORE.data.setdefault(DOMAIN, {})
slots: list[str] = data.setdefault(KEY_USED_CONNECTION_SLOTS, [])
slots.extend([consumer] * value)
return config
@@ -418,7 +420,7 @@ def validate_connection_slots(max_connections: int) -> None:
if CORE.testing_mode:
return
ble_data = CORE.data.get(KEY_ESP32_BLE, {})
ble_data = CORE.data.get(DOMAIN, {})
used_slots = ble_data.get(KEY_USED_CONNECTION_SLOTS, [])
num_used = len(used_slots)
@@ -473,7 +475,7 @@ def final_validation(config: ConfigType) -> None:
validate_connection_slots(max_connections)
# Check if hosted bluetooth is being used
if "esp32_hosted" in full_config:
if ESP32_HOSTED_DOMAIN in full_config:
from esphome.components.esp32_hosted import uses_esp_hosted_3x
add_idf_sdkconfig_option("CONFIG_BT_CLASSIC_ENABLED", False)
@@ -5,7 +5,7 @@ import esphome.codegen as cg
from esphome.components import esp32_ble
from esphome.components.const import CONF_DESCRIPTION, CONF_MANUFACTURER
from esphome.components.esp32 import request_bluetooth
from esphome.components.esp32_ble import BTLoggers, bt_uuid
from esphome.components.esp32_ble import DOMAIN as ESP32_BLE_DOMAIN, BTLoggers, bt_uuid
import esphome.config_validation as cv
from esphome.config_validation import UNDEFINED
from esphome.const import (
@@ -311,7 +311,7 @@ def final_validate_config(config) -> None:
max_clients = config[CONF_MAX_CLIENTS]
if max_clients > 1:
full_config = fv.full_config.get()
ble_config = full_config.get("esp32_ble", {})
ble_config = full_config.get(ESP32_BLE_DOMAIN, {})
max_connections = ble_config.get(
"max_connections", esp32_ble.DEFAULT_MAX_CONNECTIONS
)
+3 -2
View File
@@ -10,8 +10,9 @@ from esphome.components.esp32.const import (
VARIANT_ESP32S2,
VARIANT_ESP32S3,
)
from esphome.components.wifi import DOMAIN as WIFI_DOMAIN
import esphome.config_validation as cv
from esphome.const import CONF_ID, CONF_WIFI
from esphome.const import CONF_ID
from esphome.core import CORE
from esphome.types import ConfigType
@@ -162,7 +163,7 @@ async def to_code(config: ConfigType) -> None:
if (off_hits := config.get(CONF_MOTION_OFF_HITS)) is not None:
cg.add(var.set_motion_off_hits(off_hits))
if esp32.get_esp32_variant() == VARIANT_ESP32C5:
band = CORE.config[CONF_WIFI].get(wifi.CONF_BAND_MODE, "AUTO")
band = CORE.config[WIFI_DOMAIN].get(wifi.CONF_BAND_MODE, "AUTO")
cg.add(
var.set_wifi_band_policy(
{
+11 -4
View File
@@ -6,7 +6,12 @@ from esphome.components.noise import (
new_psk_progmem,
static_encryption_key,
)
from esphome.components.ota import BASE_OTA_SCHEMA, OTAComponent, ota_to_code
from esphome.components.ota import (
BASE_OTA_SCHEMA,
DOMAIN as OTA_DOMAIN,
OTAComponent,
ota_to_code,
)
from esphome.config_helpers import filter_source_files_from_defines, merge_config
import esphome.config_validation as cv
from esphome.const import (
@@ -31,6 +36,8 @@ from esphome.espota2 import CONF_ALLOW_PLAINTEXT_UPLOAD
import esphome.final_validate as fv
from esphome.types import ConfigType
from .. import DOMAIN
CONF_ALLOW_PARTITION_ACCESS = "allow_partition_access"
_LOGGER = logging.getLogger(__name__)
@@ -56,12 +63,12 @@ ESPHomeOTAComponent = esphome.class_("ESPHomeOTAComponent", OTAComponent)
def ota_esphome_final_validate(config: ConfigType) -> None:
full_conf = fv.full_config.get()
full_ota_conf = full_conf[CONF_OTA]
full_ota_conf = full_conf[OTA_DOMAIN]
new_ota_conf = []
merged_ota_esphome_configs_by_port = {}
ports_with_merged_configs = []
for ota_conf in full_ota_conf:
if ota_conf.get(CONF_PLATFORM) == CONF_ESPHOME:
if ota_conf.get(CONF_PLATFORM) == DOMAIN:
if (
conf_port := ota_conf.get(CONF_PORT)
) not in merged_ota_esphome_configs_by_port:
@@ -182,7 +189,7 @@ def ota_esphome_final_validate(config: ConfigType) -> None:
CONF_WEB_SERVER,
)
full_conf[CONF_OTA] = new_ota_conf
full_conf[OTA_DOMAIN] = new_ota_conf
fv.full_config.set(full_conf)
if len(ports_with_merged_configs) > 0:
+4 -2
View File
@@ -4,7 +4,9 @@ from esphome import automation, core
import esphome.codegen as cg
from esphome.components import wifi
from esphome.components.esp32 import VARIANT_ESP32P4, get_esp32_variant
from esphome.components.esp32_hosted import DOMAIN as ESP32_HOSTED_DOMAIN
from esphome.components.udp import CONF_ON_RECEIVE
from esphome.components.wifi import DOMAIN as WIFI_DOMAIN
import esphome.config_validation as cv
from esphome.const import (
CONF_ADDRESS,
@@ -141,7 +143,7 @@ def _validate_variant(config: ConfigType) -> ConfigType:
return config
if variant != VARIANT_ESP32P4:
raise cv.Invalid(f"ESP-NOW is not supported on {variant} (no Wi-Fi radio)")
if "esp32_hosted" not in fv.full_config.get():
if ESP32_HOSTED_DOMAIN not in fv.full_config.get():
raise cv.Invalid(f"ESP-NOW on {variant} requires the esp32_hosted component")
return config
@@ -177,7 +179,7 @@ async def to_code(config: ConfigType) -> None:
include_builtin_idf_component("esp_wifi")
if CONF_WIFI in CORE.config:
if WIFI_DOMAIN in CORE.config:
# Track the Wi-Fi channel via connect events instead of polling every loop
wifi.request_wifi_connect_state_listener()
if wifi_channel := config.get(CONF_CHANNEL):
+2 -3
View File
@@ -68,7 +68,6 @@ LOGGER = logging.getLogger(__name__)
ETHERNET_IP_STATE_LISTENERS_KEY = "ethernet_ip_state_listeners"
# Key for tracking configured ethernet type
ETHERNET_TYPE_KEY = "ethernet_type"
KEY_ETHERNET = "ethernet"
def request_ethernet_ip_state_listener() -> None:
@@ -673,7 +672,7 @@ async def to_code(config: ConfigType) -> None:
# enable_on_boot defaults to true in C++ - only set if false
if not config[CONF_ENABLE_ON_BOOT]:
cg.add(var.set_enable_on_boot(False))
CORE.data.setdefault(KEY_ETHERNET, {})[ETHERNET_TYPE_KEY] = config[CONF_TYPE]
CORE.data.setdefault(DOMAIN, {})[ETHERNET_TYPE_KEY] = config[CONF_TYPE]
if CONF_MANUAL_IP in config:
cg.add_define("USE_ETHERNET_MANUAL_IP")
@@ -914,7 +913,7 @@ _define_filter = filter_source_files_from_defines(
def _filter_source_files() -> list[str]:
excluded = _platform_filter() + _define_filter()
eth_data = CORE.data.get(KEY_ETHERNET, {})
eth_data = CORE.data.get(DOMAIN, {})
eth_type = eth_data.get(ETHERNET_TYPE_KEY)
# Only compile the custom JL1101 driver when JL1101 is configured
# and pioarduino doesn't have it builtin (IDF 5.4.2 to 5.x)
+5 -5
View File
@@ -4,13 +4,14 @@ from esphome import automation
import esphome.codegen as cg
from esphome.components import climate, logger, uart
from esphome.components.climate import ClimateMode, ClimatePreset, ClimateSwingMode
from esphome.components.logger import DOMAIN as LOGGER_DOMAIN
from esphome.components.wifi import DOMAIN as WIFI_DOMAIN
import esphome.config_validation as cv
from esphome.const import (
CONF_BEEPER,
CONF_CURRENT_TEMPERATURE,
CONF_DISPLAY,
CONF_LEVEL,
CONF_LOGGER,
CONF_LOGS,
CONF_MAX_TEMPERATURE,
CONF_MIN_TEMPERATURE,
@@ -22,7 +23,6 @@ from esphome.const import (
CONF_TARGET_TEMPERATURE,
CONF_TEMPERATURE_STEP,
CONF_VISUAL,
CONF_WIFI,
)
from esphome.cpp_generator import MockObjClass
import esphome.final_validate as fv
@@ -326,9 +326,9 @@ automation.register_apply_action(
def _final_validate(config) -> None:
full_config = fv.full_config.get()
if CONF_LOGGER in full_config:
if LOGGER_DOMAIN in full_config:
_level = "NONE"
logger_config = full_config[CONF_LOGGER]
logger_config = full_config[LOGGER_DOMAIN]
if CONF_LOGS in logger_config:
if "haier.protocol" in logger_config[CONF_LOGS]:
_level = logger_config[CONF_LOGS]["haier.protocol"]
@@ -344,7 +344,7 @@ def _final_validate(config) -> None:
"No logger component found, logging for Haier protocol is disabled"
)
cg.add_build_flag("-DHAIER_LOG_LEVEL=0")
if config.get(CONF_WIFI_SIGNAL) and CONF_WIFI not in full_config:
if config.get(CONF_WIFI_SIGNAL) and WIFI_DOMAIN not in full_config:
raise cv.Invalid(
f"No WiFi configured, if you want to use haier climate without WiFi add {CONF_WIFI_SIGNAL}: false to climate configuration"
)
+1 -2
View File
@@ -38,7 +38,6 @@ from esphome.const import (
CONF_ADDRESS,
CONF_DEVICE,
CONF_FREQUENCY,
CONF_I2C,
CONF_I2C_ID,
CONF_ID,
CONF_LOW_POWER_MODE,
@@ -233,7 +232,7 @@ CONFIG_SCHEMA = cv.All(
def _final_validate(config: ConfigType) -> None:
full_config = fv.full_config.get()[CONF_I2C]
full_config = fv.full_config.get()[DOMAIN]
if CORE.using_zephyr and len(full_config) > 1:
raise cv.Invalid("Second i2c is not implemented on Zephyr yet")
if CORE.is_rp2:
+7 -7
View File
@@ -18,6 +18,7 @@ from esphome.components.esp32.const import (
VARIANT_ESP32S2,
VARIANT_ESP32S3,
)
from esphome.components.microphone import DOMAIN as MICROPHONE_DOMAIN
import esphome.config_validation as cv
from esphome.const import CONF_BITS_PER_SAMPLE, CONF_CHANNEL, CONF_ID, CONF_SAMPLE_RATE
from esphome.core import CORE
@@ -40,7 +41,6 @@ CONF_I2S_MCLK_PIN = "i2s_mclk_pin"
CONF_I2S_BCLK_PIN = "i2s_bclk_pin"
CONF_I2S_LRCLK_PIN = "i2s_lrclk_pin"
CONF_I2S_AUDIO = "i2s_audio"
CONF_I2S_AUDIO_ID = "i2s_audio_id"
CONF_I2S_MODE = "i2s_mode"
@@ -219,9 +219,9 @@ class I2SAudioData:
def _get_data() -> I2SAudioData:
if CONF_I2S_AUDIO not in CORE.data:
CORE.data[CONF_I2S_AUDIO] = I2SAudioData()
return CORE.data[CONF_I2S_AUDIO]
if DOMAIN not in CORE.data:
CORE.data[DOMAIN] = I2SAudioData()
return CORE.data[DOMAIN]
def _assign_ports() -> None:
@@ -235,12 +235,12 @@ def _assign_ports() -> None:
return
full_config = fv.full_config.get()
i2s_configs = full_config[CONF_I2S_AUDIO]
i2s_configs = full_config[DOMAIN]
# Find i2s_audio instances with microphones that require port 0
# (PDM and internal ADC only work on I2S port 0)
port0_parent_id = None
for mic_config in full_config.get("microphone", []):
for mic_config in full_config.get(MICROPHONE_DOMAIN, []):
if CONF_I2S_AUDIO_ID not in mic_config:
continue
if mic_config.get(CONF_PDM) or mic_config.get(CONF_ADC_TYPE) == "internal":
@@ -263,7 +263,7 @@ def _assign_ports() -> None:
def _final_validate(_: ConfigType) -> None:
i2s_audio_configs = fv.full_config.get()[CONF_I2S_AUDIO]
i2s_audio_configs = fv.full_config.get()[DOMAIN]
variant = get_esp32_variant()
if variant not in I2S_PORTS:
raise cv.Invalid(f"Unsupported variant {variant}")
+3 -9
View File
@@ -1,15 +1,9 @@
import esphome.codegen as cg
from esphome.components import improv_base, uart
from esphome.components.esp32 import VARIANT_ESP32S3, get_esp32_variant
from esphome.components.logger import USB_CDC
from esphome.components.logger import DOMAIN as LOGGER_DOMAIN, USB_CDC
import esphome.config_validation as cv
from esphome.const import (
CONF_BAUD_RATE,
CONF_HARDWARE_UART,
CONF_ID,
CONF_LOGGER,
CONF_UART_ID,
)
from esphome.const import CONF_BAUD_RATE, CONF_HARDWARE_UART, CONF_ID, CONF_UART_ID
from esphome.core import CORE
import esphome.final_validate as fv
from esphome.types import ConfigType
@@ -53,7 +47,7 @@ def validate_transport(config: ConfigType) -> None:
# The host logger has no serial port for Improv to share
if CORE.is_host:
raise cv.Invalid("improv_serial on the host platform requires uart_id")
logger_conf = fv.full_config.get()[CONF_LOGGER]
logger_conf = fv.full_config.get()[LOGGER_DOMAIN]
if logger_conf[CONF_BAUD_RATE] == 0:
raise cv.Invalid("improv_serial requires the logger baud_rate to be not 0")
if CORE.is_esp32 and (
+2 -2
View File
@@ -2,7 +2,7 @@
import esphome.codegen as cg
from esphome.components import remote_base
from esphome.const import CONF_API
from esphome.components.api import DOMAIN as API_DOMAIN
from esphome.core import CORE
from esphome.core.entity_helpers import queue_entity_register
from esphome.types import ConfigType
@@ -25,7 +25,7 @@ async def attach_transmitter(var: cg.MockObj, config: ConfigType, key: str) -> N
soon as the frame is handed over.
"""
await remote_base.register_transmittable(var, config, key)
if CONF_API in CORE.config:
if API_DOMAIN in CORE.config:
cg.add_define("USE_IR_RF_TRANSMIT_COMPLETE")
@@ -1,5 +1,6 @@
import esphome.codegen as cg
from esphome.components import button
from esphome.components.select import DOMAIN as SELECT_DOMAIN
import esphome.config_validation as cv
from esphome.const import (
CONF_AREA_ID,
@@ -91,7 +92,7 @@ def final_validate(config: ConfigType) -> None:
if config.get(CONF_APPLY_AREA):
has_area_id_select = any(
entry.get(CONF_LD6002B_ID) == hub_id and entry.get(CONF_AREA_ID) is not None
for entry in full_config.get("select", [])
for entry in full_config.get(SELECT_DOMAIN, [])
)
if not has_area_id_select:
raise cv.Invalid(
@@ -1,9 +1,10 @@
import esphome.codegen as cg
from esphome.components import number
from esphome.components.button import DOMAIN as BUTTON_DOMAIN
from esphome.components.select import DOMAIN as SELECT_DOMAIN
import esphome.config_validation as cv
from esphome.const import (
CONF_AREA_ID,
CONF_BUTTON,
DEVICE_CLASS_DISTANCE,
DEVICE_CLASS_DURATION,
ENTITY_CATEGORY_CONFIG,
@@ -114,7 +115,7 @@ def final_validate(config: ConfigType) -> None:
has_apply_area = any(
entry.get(CONF_LD6002B_ID) == hub_id and entry.get(CONF_APPLY_AREA) is not None
for entry in full_config.get(CONF_BUTTON, [])
for entry in full_config.get(BUTTON_DOMAIN, [])
)
if not has_apply_area:
raise cv.Invalid(
@@ -124,7 +125,7 @@ def final_validate(config: ConfigType) -> None:
has_area_id_select = any(
entry.get(CONF_LD6002B_ID) == hub_id and entry.get(CONF_AREA_ID) is not None
for entry in full_config.get("select", [])
for entry in full_config.get(SELECT_DOMAIN, [])
)
if not has_area_id_select:
raise cv.Invalid(
+1 -2
View File
@@ -45,7 +45,6 @@ from esphome.const import (
CONF_HARDWARE_UART,
CONF_ID,
CONF_LEVEL,
CONF_LOGGER,
CONF_LOGS,
CONF_ON_MESSAGE,
CONF_TAG,
@@ -365,7 +364,7 @@ CONFIG_SCHEMA = cv.All(
async def to_code(config: ConfigType) -> None:
baud_rate: int = config[CONF_BAUD_RATE]
level = config[CONF_LEVEL]
CORE.data.setdefault(CONF_LOGGER, {})[CONF_LEVEL] = level
CORE.data.setdefault(DOMAIN, {})[CONF_LEVEL] = level
tx_buffer_size = config[CONF_TX_BUFFER_SIZE]
cg.add_define("ESPHOME_LOGGER_TX_BUFFER_SIZE", tx_buffer_size)
# Determine task log buffer size. The buffer is a direct member of Logger
+3 -2
View File
@@ -1,13 +1,14 @@
import esphome.codegen as cg
from esphome.components import select
import esphome.config_validation as cv
from esphome.const import CONF_LEVEL, CONF_LOGGER, ENTITY_CATEGORY_CONFIG, ICON_BUG
from esphome.const import CONF_LEVEL, ENTITY_CATEGORY_CONFIG, ICON_BUG
from esphome.core import CORE
from esphome.cpp_helpers import register_component, register_parented
from esphome.types import ConfigType
from .. import (
CONF_LOGGER_ID,
DOMAIN,
LOG_LEVELS,
Logger,
logger_ns,
@@ -31,7 +32,7 @@ async def to_code(config: ConfigType) -> None:
request_logger_level_listeners()
parent = await cg.get_variable(config[CONF_LOGGER_ID])
levels = list(LOG_LEVELS)
index = levels.index(CORE.data[CONF_LOGGER][CONF_LEVEL])
index = levels.index(CORE.data[DOMAIN][CONF_LEVEL])
levels = levels[: index + 1]
var = await select.new_select(config, options=levels)
await register_parented(var, parent)
+2 -1
View File
@@ -1,5 +1,6 @@
import esphome.codegen as cg
from esphome.components.binary_sensor import BinarySensor
from esphome.components.rotary_encoder import DOMAIN as ROTARY_ENCODER_DOMAIN
from esphome.components.rotary_encoder.sensor import RotaryEncoderSensor
import esphome.config_validation as cv
from esphome.const import CONF_GROUP, CONF_ID, CONF_SENSOR
@@ -27,7 +28,7 @@ ENCODERS_CONFIG = cv.ensure_list(
cv.Required(CONF_SENSOR): cv.Any(
cv.All(
cv.use_id(RotaryEncoderSensor),
cv.requires_component("rotary_encoder"),
cv.requires_component(ROTARY_ENCODER_DOMAIN),
),
cv.Schema(
{
+10 -12
View File
@@ -3,19 +3,17 @@ from typing import Any
import esphome.codegen as cg
from esphome.components import image
from esphome.components.color import CONF_HEX, ColorStruct, from_rgbw
from esphome.components.color import (
CONF_HEX,
DOMAIN as COLOR_DOMAIN,
ColorStruct,
from_rgbw,
)
from esphome.components.const.css_colors import CSS_COLORS
from esphome.components.font import Font
from esphome.components.font import DOMAIN as FONT_DOMAIN, Font
from esphome.components.image import Image_
import esphome.config_validation as cv
from esphome.const import (
CONF_ARGS,
CONF_COLOR,
CONF_FORMAT,
CONF_ID,
CONF_TIME,
CONF_VALUE,
)
from esphome.const import CONF_ARGS, CONF_FORMAT, CONF_ID, CONF_TIME, CONF_VALUE
from esphome.core import CORE, ID, Lambda
from esphome.cpp_generator import MockObj, StaticCastExpression, call_lambda
from esphome.cpp_types import ESPTime, int32, uint32
@@ -81,7 +79,7 @@ def get_component_colors(value):
if isinstance(value, int):
return value >> 16, value >> 8 & 0xFF, value & 0xFF
if isinstance(value, ID):
cval = [x for x in CORE.config[CONF_COLOR] if x[CONF_ID] == value][0]
cval = [x for x in CORE.config[COLOR_DOMAIN] if x[CONF_ID] == value][0]
if CONF_HEX in cval:
r, g, b = cval[CONF_HEX]
else:
@@ -437,7 +435,7 @@ class LvFont(LValidator):
add_lv_use("font")
fontval = cv.use_id(Font)(value)
get_esphome_fonts_used().add(fontval)
return cv.requires_component("font")(fontval)
return cv.requires_component(FONT_DOMAIN)(fontval)
# Use font::Font* as return type for lambdas returning ESPHome fonts
# The inline overloads in lvgl_esphome.h handle conversion to lv_font_t*
+3 -1
View File
@@ -4,6 +4,7 @@ from typing import Any
from esphome import config_validation as cv
from esphome.automation import Trigger, validate_automation
from esphome.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from esphome.components.mapping import mapping_class
from esphome.components.time import RealTimeClock
from esphome.config_validation import prepend_path
@@ -148,7 +149,8 @@ PRESS_TIME = cv.All(
ENCODER_SCHEMA = cv.Schema(
{
cv.GenerateID(): cv.All(
cv.declare_id(LVEncoderListener), cv.requires_component("binary_sensor")
cv.declare_id(LVEncoderListener),
cv.requires_component(BINARY_SENSOR_DOMAIN),
),
cv.Optional(CONF_GROUP): cv.declare_id(lv_group_t),
cv.Optional(df.CONF_INITIAL_FOCUS): cv.All(
+7 -3
View File
@@ -1,5 +1,9 @@
import esphome.codegen as cg
from esphome.components.touchscreen import CONF_TOUCHSCREEN_ID, Touchscreen
from esphome.components.touchscreen import (
CONF_TOUCHSCREEN_ID,
DOMAIN as TOUCHSCREEN_DOMAIN,
Touchscreen,
)
import esphome.config_validation as cv
from esphome.const import CONF_ID
from esphome.core import CORE
@@ -17,7 +21,7 @@ CONF_TOUCHSCREEN = "touchscreen"
TOUCHSCREENS_CONFIG = cv.maybe_simple_value(
{
cv.Required(CONF_TOUCHSCREEN_ID): cv.All(
cv.use_id(Touchscreen), cv.requires_component(CONF_TOUCHSCREEN)
cv.use_id(Touchscreen), cv.requires_component(TOUCHSCREEN_DOMAIN)
),
cv.Optional(CONF_LONG_PRESS_TIME, default="400ms"): PRESS_TIME,
cv.Optional(CONF_LONG_PRESS_REPEAT_TIME, default="100ms"): PRESS_TIME,
@@ -29,7 +33,7 @@ TOUCHSCREENS_CONFIG = cv.maybe_simple_value(
def touchscreen_schema(config):
value = cv.ensure_list(TOUCHSCREENS_CONFIG)(config)
if value or CONF_TOUCHSCREEN not in CORE.loaded_integrations:
if value or TOUCHSCREEN_DOMAIN not in CORE.loaded_integrations:
return value
return [TOUCHSCREENS_CONFIG(config)]
+2 -2
View File
@@ -1,6 +1,6 @@
from esphome import automation
import esphome.codegen as cg
from esphome.components.image import get_image_metadata
from esphome.components.image import DOMAIN as IMAGE_DOMAIN, get_image_metadata
import esphome.config_validation as cv
from esphome.const import (
CONF_COLOR,
@@ -214,7 +214,7 @@ INDICATOR_SCHEMA = cv.Schema(
cv.GenerateID(CONF_IMAGE_ID): cv.declare_id(lv_image_t),
}
),
cv.requires_component("image"),
cv.requires_component(IMAGE_DOMAIN),
),
cv.Exclusive(CONF_ARC, CONF_INDICATORS): INDICATOR_ARC_SCHEMA.extend(
{
+1 -2
View File
@@ -5,7 +5,6 @@ import esphome.config_validation as cv
from esphome.const import (
CONF_DISABLED,
CONF_ID,
CONF_MDNS,
CONF_OPENTHREAD,
CONF_PORT,
CONF_PROTOCOL,
@@ -195,7 +194,7 @@ def request_service_enable_disable() -> bool:
Public API for external components. Do not remove.
"""
mdns_config = CORE.config.get(CONF_MDNS)
mdns_config = CORE.config.get(DOMAIN)
if (
mdns_config is None
or mdns_config[CONF_DISABLED]
+1 -1
View File
@@ -278,7 +278,7 @@ def get_priority_interfaces_from_full_config(full_config: ConfigType) -> set[str
"""
return {
entry["interface"]
for entry in full_config.get("network", {}).get(CONF_PRIORITY, [])
for entry in full_config.get(DOMAIN, {}).get(CONF_PRIORITY, [])
}
+2 -2
View File
@@ -19,7 +19,7 @@ from esphome.components.esp32 import (
require_vfs_select,
)
from esphome.components.mdns import MDNSComponent, enable_mdns_storage
from esphome.components.network import add_use_address
from esphome.components.network import DOMAIN as NETWORK_DOMAIN, add_use_address
from esphome.components.nrf52.framework import include_west_project
from esphome.components.zephyr import zephyr_add_prj_conf
from esphome.config_helpers import filter_source_files_from_platform
@@ -261,7 +261,7 @@ CONFIG_SCHEMA = cv.All(
def _final_validate(_: ConfigType) -> None:
full_config = fv.full_config.get()
network_config = full_config.get("network", {})
network_config = full_config.get(NETWORK_DOMAIN, {})
if not network_config.get(CONF_ENABLE_IPV6, False):
raise cv.Invalid(
"OpenThread requires IPv6 to be enabled in the network component. "
+3 -2
View File
@@ -7,6 +7,7 @@ from typing import Any
from esphome import git, yaml_util
from esphome.components.substitutions import (
DOMAIN as SUBSTITUTIONS_DOMAIN,
ContextVars,
ErrList,
push_context,
@@ -353,7 +354,7 @@ def _walk_packages(
packages = config[CONF_PACKAGES]
packages_path = (path or []) + [CONF_PACKAGES]
with cv.prepend_path(CONF_PACKAGES):
with cv.prepend_path(DOMAIN):
if isinstance(packages, yaml_util.IncludeFile):
# If the packages key is an IncludeFile, resolve it first before processing.
packages = resolve_include(
@@ -599,7 +600,7 @@ def do_packages_pass(
if CONF_PACKAGES not in config:
return config
with cv.prepend_path(CONF_SUBSTITUTIONS):
with cv.prepend_path(SUBSTITUTIONS_DOMAIN):
substitutions = UserDict(
resolve_substitutions_block(
config.pop(CONF_SUBSTITUTIONS, {}), command_line_substitutions
+2 -3
View File
@@ -12,7 +12,6 @@ DOMAIN = "rp2040_ble"
CONF_RP2040_BLE_ID = "rp2040_ble_id"
KEY_RP2040_BLE = "rp2040_ble"
KEY_USED_CONNECTION_SLOTS = "used_connection_slots"
# Hard platform cap on concurrent GATT connections: the BTstack pool overrides
@@ -50,7 +49,7 @@ def consume_connection_slots(
the total is checked against MAX_CONNECTIONS in final validation."""
def _consume_connection_slots(config: MutableMapping) -> MutableMapping:
data: dict = CORE.data.setdefault(KEY_RP2040_BLE, {})
data: dict = CORE.data.setdefault(DOMAIN, {})
slots: list[str] = data.setdefault(KEY_USED_CONNECTION_SLOTS, [])
slots.extend([consumer] * value)
return config
@@ -63,7 +62,7 @@ def validate_connection_slots() -> None:
# Skip in testing mode to allow component grouping (esp32_ble parity).
if CORE.testing_mode:
return
used = CORE.data.get(KEY_RP2040_BLE, {}).get(KEY_USED_CONNECTION_SLOTS, [])
used = CORE.data.get(DOMAIN, {}).get(KEY_USED_CONNECTION_SLOTS, [])
if len(used) > MAX_CONNECTIONS:
raise cv.Invalid(
f"BLE components require {len(used)} connection slots but the "
+2 -3
View File
@@ -6,7 +6,6 @@ from esphome.const import (
CONF_ID,
CONF_NUM_ATTEMPTS,
CONF_REBOOT_TIMEOUT,
CONF_SAFE_MODE,
CONF_STORAGE,
KEY_PAST_SAFE_MODE,
)
@@ -93,5 +92,5 @@ async def to_code(config: ConfigType) -> None:
)
cg.add(RawExpression(f"if ({condition}) return"))
CORE.data[CONF_SAFE_MODE] = {}
CORE.data[CONF_SAFE_MODE][KEY_PAST_SAFE_MODE] = True
CORE.data[DOMAIN] = {}
CORE.data[DOMAIN][KEY_PAST_SAFE_MODE] = True
+1 -2
View File
@@ -16,7 +16,6 @@ RestartScript = script_ns.class_("RestartScript", Script)
QueueingScript = script_ns.class_("QueueingScript", Script, cg.Component)
ParallelScript = script_ns.class_("ParallelScript", Script)
CONF_SCRIPT = "script"
CONF_SINGLE = "single"
CONF_QUEUED = "queued"
CONF_PARALLEL = "parallel"
@@ -36,7 +35,7 @@ PARAMETER_TYPE_TRANSLATIONS = {
def get_script(script_id):
scripts = CORE.config.get(CONF_SCRIPT, {})
scripts = CORE.config.get(DOMAIN, {})
for script in scripts:
if script.get(CONF_ID, None) == script_id:
return script
+2 -2
View File
@@ -6,6 +6,7 @@ from esphome import automation
import esphome.codegen as cg
from esphome.components import esp32, mdns, network, psram, socket, wifi
from esphome.components.const import CONF_ENABLED, CONF_MANUFACTURER
from esphome.components.mdns import DOMAIN as MDNS_DOMAIN
import esphome.config_validation as cv
from esphome.const import (
CONF_BUFFER_SIZE,
@@ -13,7 +14,6 @@ from esphome.const import (
CONF_FORMAT,
CONF_HEIGHT,
CONF_ID,
CONF_MDNS,
CONF_MODEL,
CONF_NAME,
CONF_PROJECT,
@@ -428,7 +428,7 @@ async def to_code(config: ConfigType) -> None:
# Service starts disabled and the hub enables it; always advertised where unsupported
if mdns.request_service_enable_disable():
mdns_var = await cg.get_variable(CORE.config[CONF_MDNS][CONF_ID])
mdns_var = await cg.get_variable(CORE.config[MDNS_DOMAIN][CONF_ID])
cg.add(var.set_mdns(mdns_var))
data = _get_data()
@@ -3,8 +3,9 @@ import logging
import esphome.codegen as cg
from esphome.components import switch
from esphome.components.const import CONF_ENABLED
from esphome.components.switch import DOMAIN as SWITCH_DOMAIN
import esphome.config_validation as cv
from esphome.const import CONF_PLATFORM, CONF_SWITCH, CONF_TYPE, ENTITY_CATEGORY_CONFIG
from esphome.const import CONF_PLATFORM, CONF_TYPE, ENTITY_CATEGORY_CONFIG
import esphome.final_validate as fv
from esphome.types import ConfigType
@@ -69,7 +70,7 @@ def _final_validate(config: ConfigType) -> ConfigType:
switch_type = config[CONF_TYPE]
same_type = [
conf
for conf in full_config.get(CONF_SWITCH, [])
for conf in full_config.get(SWITCH_DOMAIN, [])
if conf.get(CONF_PLATFORM) == DOMAIN and conf.get(CONF_TYPE) == switch_type
]
# Two switches of one type would each drive the same hub setting.
+6 -5
View File
@@ -2,13 +2,13 @@ import logging
import esphome.codegen as cg
from esphome.components import time as time_
from esphome.components.time import DOMAIN as TIME_DOMAIN
from esphome.config_helpers import merge_config
import esphome.config_validation as cv
from esphome.const import (
CONF_ID,
CONF_PLATFORM,
CONF_SERVERS,
CONF_TIME,
PLATFORM_BK72XX,
PLATFORM_ESP32,
PLATFORM_ESP8266,
@@ -20,11 +20,12 @@ from esphome.core import CORE
import esphome.final_validate as fv
from esphome.types import ConfigType
from . import DOMAIN
_LOGGER = logging.getLogger(__name__)
DEPENDENCIES = ["network"]
CONF_SNTP = "sntp"
sntp_ns = cg.esphome_ns.namespace("sntp")
SNTPComponent = sntp_ns.class_("SNTPComponent", time_.RealTimeClock)
@@ -35,13 +36,13 @@ DEFAULT_SERVERS = ["0.pool.ntp.org", "1.pool.ntp.org", "2.pool.ntp.org"]
def _sntp_final_validate(config: ConfigType) -> None:
"""Merge multiple SNTP instances into one, similar to OTA merging behavior."""
full_conf = fv.full_config.get()
time_confs = full_conf.get(CONF_TIME, [])
time_confs = full_conf.get(TIME_DOMAIN, [])
sntp_configs: list[ConfigType] = []
other_time_configs: list[ConfigType] = []
for time_conf in time_confs:
if time_conf.get(CONF_PLATFORM) == CONF_SNTP:
if time_conf.get(CONF_PLATFORM) == DOMAIN:
sntp_configs.append(time_conf)
else:
other_time_configs.append(time_conf)
@@ -81,7 +82,7 @@ def _sntp_final_validate(config: ConfigType) -> None:
# Replace time configs with merged SNTP + other time platforms
other_time_configs.append(merged)
full_conf[CONF_TIME] = other_time_configs
full_conf[TIME_DOMAIN] = other_time_configs
fv.full_config.set(full_conf)
+2 -1
View File
@@ -18,6 +18,7 @@ from esphome.components.esp32 import (
VARIANT_ESP32S31,
only_on_variant,
)
from esphome.components.psram import DOMAIN as PSRAM_DOMAIN
from esphome.config_helpers import filter_source_files_from_platform
import esphome.config_validation as cv
from esphome.const import (
@@ -158,7 +159,7 @@ def _validate_psram_dma(value: Any) -> bool:
msg_prefix="PSRAM DMA",
),
cv.require_framework_version(esp_idf=cv.Version(5, 5, 3)),
cv.requires_component("psram"),
cv.requires_component(PSRAM_DOMAIN),
)(value)
+2 -2
View File
@@ -518,7 +518,7 @@ def resolve_substitutions_block(
"""Resolve a deferred ``substitutions: !include file.yaml`` and validate the shape.
The caller is responsible for wrapping the call in
``cv.prepend_path(CONF_SUBSTITUTIONS)`` for error reporting.
``cv.prepend_path(DOMAIN)`` for error reporting.
``command_line_substitutions`` seeds the filename context so
``substitutions: !include ${var}.yaml`` can reference CLI-provided vars.
"""
@@ -553,7 +553,7 @@ def do_substitution_pass(
# Extract substitutions from config, overriding with substitutions coming from command line:
# Use merge_dicts_ordered to preserve OrderedDict type for move_to_end()
substitutions = config.pop(CONF_SUBSTITUTIONS, {})
with cv.prepend_path(CONF_SUBSTITUTIONS):
with cv.prepend_path(DOMAIN):
substitutions = resolve_substitutions_block(
substitutions, command_line_substitutions
)
+2 -1
View File
@@ -10,6 +10,7 @@ from esphome.components.esp32 import (
add_idf_component,
add_idf_sdkconfig_option,
)
from esphome.components.logger import DOMAIN as LOGGER_DOMAIN
import esphome.config_validation as cv
from esphome.const import CONF_HARDWARE_UART, CONF_ID
from esphome.types import ConfigType
@@ -96,7 +97,7 @@ def _final_validate(config: ConfigType) -> None:
# tinyusb owns the USB OTG peripheral. The logger's USB_CDC backend routes
# the ROM console through that same peripheral, so the two cannot coexist.
# (USB_SERIAL_JTAG is a separate peripheral and is fine alongside tinyusb.)
logger_config = full_config.get("logger")
logger_config = full_config.get(LOGGER_DOMAIN)
if logger_config and logger_config.get(CONF_HARDWARE_UART) == "USB_CDC":
raise cv.Invalid(
"'tinyusb' cannot be used with 'logger.hardware_uart: USB_CDC' "
+2 -1
View File
@@ -7,6 +7,7 @@ from esphome.components.const import (
CONF_RECONNECT_INTERVAL,
CONF_ROLE,
)
from esphome.components.tcp_uart import DOMAIN as TCP_UART_DOMAIN
import esphome.config_validation as cv
from esphome.const import (
CONF_DEBUG,
@@ -125,7 +126,7 @@ def _final_validate(config: ConfigType) -> ConfigType:
ports = data.setdefault(CONF_PORT, set())
if port in ports or any(
conf[CONF_ROLE] == "server" and conf[CONF_PORT] == port
for conf in full_config.get("tcp_uart", [])
for conf in full_config.get(TCP_UART_DOMAIN, [])
):
raise cv.Invalid(
f"Port {port} is already the listen port of another uart_tcp "
+3 -1
View File
@@ -1,5 +1,6 @@
import esphome.codegen as cg
from esphome.components import sensor, time
from esphome.components.time import DOMAIN as TIME_DOMAIN
from esphome.config_helpers import filter_source_files_from_defines
import esphome.config_validation as cv
from esphome.const import (
@@ -44,7 +45,8 @@ CONFIG_SCHEMA = cv.typed_schema(
cv.Schema(
{
cv.GenerateID(CONF_TIME_ID): cv.All(
cv.requires_component("time"), cv.use_id(time.RealTimeClock)
cv.requires_component(TIME_DOMAIN),
cv.use_id(time.RealTimeClock),
),
}
)
+1 -1
View File
@@ -34,7 +34,7 @@ USBUartChannel = usb_uart_ns.class_("USBUartChannel", UARTComponent)
def is_usb_uart_channel(uart_id: ID, full_config: ConfigType) -> bool:
return any(
channel[CONF_ID] == uart_id
for device in full_config.get("usb_uart") or []
for device in full_config.get(DOMAIN) or []
for channel in device[CONF_CHANNELS]
)
+3 -3
View File
@@ -240,11 +240,11 @@ WEBSERVER_SORTING_SCHEMA = cv.Schema(
{
cv.OnlyWith(CONF_WEB_SERVER_ID, "web_server"): cv.use_id(WebServer),
cv.Optional(CONF_SORTING_WEIGHT): cv.All(
cv.requires_component("web_server"),
cv.requires_component(DOMAIN),
cv.float_,
),
cv.Optional(CONF_SORTING_GROUP_ID): cv.All(
cv.requires_component("web_server"),
cv.requires_component(DOMAIN),
cv.use_id(cg.int_),
),
}
@@ -474,7 +474,7 @@ def FILTER_SOURCE_FILES() -> list[str]:
files_to_filter: list[str] = []
# web_server_v1.cpp is only needed when version is 1
config = CORE.config.get("web_server", {})
config = CORE.config.get(DOMAIN, {})
if config.get(CONF_VERSION, 2) != 1:
files_to_filter.append("web_server_v1.cpp")
+12 -5
View File
@@ -2,15 +2,22 @@ import logging
import esphome.codegen as cg
from esphome.components.esp32 import add_idf_component
from esphome.components.ota import BASE_OTA_SCHEMA, OTAComponent, ota_to_code
from esphome.components.ota import (
BASE_OTA_SCHEMA,
DOMAIN as OTA_DOMAIN,
OTAComponent,
ota_to_code,
)
from esphome.config_helpers import merge_config
import esphome.config_validation as cv
from esphome.const import CONF_ID, CONF_OTA, CONF_PLATFORM, CONF_WEB_SERVER
from esphome.const import CONF_ID, CONF_PLATFORM
from esphome.core import CORE, coroutine_with_priority
from esphome.coroutine import CoroPriority
import esphome.final_validate as fv
from esphome.types import ConfigType
from .. import DOMAIN
_LOGGER = logging.getLogger(__name__)
CODEOWNERS = ["@esphome/core"]
@@ -27,13 +34,13 @@ def _web_server_ota_final_validate(config: ConfigType) -> None:
causing undefined behavior. Merge them into a single instance.
"""
full_conf = fv.full_config.get()
ota_confs = full_conf.get(CONF_OTA, [])
ota_confs = full_conf.get(OTA_DOMAIN, [])
web_server_ota_configs: list[ConfigType] = []
other_ota_configs: list[ConfigType] = []
for ota_conf in ota_confs:
if ota_conf.get(CONF_PLATFORM) == CONF_WEB_SERVER:
if ota_conf.get(CONF_PLATFORM) == DOMAIN:
web_server_ota_configs.append(ota_conf)
else:
other_ota_configs.append(ota_conf)
@@ -62,7 +69,7 @@ def _web_server_ota_final_validate(config: ConfigType) -> None:
# Replace OTA configs with merged web_server + other OTA platforms
other_ota_configs.append(merged)
full_conf[CONF_OTA] = other_ota_configs
full_conf[OTA_DOMAIN] = other_ota_configs
fv.full_config.set(full_conf)
+1 -2
View File
@@ -61,7 +61,6 @@ from esphome.const import (
CONF_TTLS_PHASE_2,
CONF_USE_ADDRESS,
CONF_USERNAME,
CONF_WIFI,
PLACEHOLDER_WIFI_SSID,
Platform,
PlatformFramework,
@@ -1089,7 +1088,7 @@ def _placeholder_wifi_credentials(config: ConfigType) -> list[str]:
values still appear. Empty list means no placeholders were found.
"""
placeholders: list[str] = []
wifi_conf = config.get(CONF_WIFI)
wifi_conf = config.get(DOMAIN)
if not wifi_conf:
return placeholders
+2 -2
View File
@@ -4,6 +4,7 @@ from typing import Any
from esphome import pins
import esphome.codegen as cg
from esphome.components import output
from esphome.components.output import DOMAIN as OUTPUT_DOMAIN
from esphome.components.zephyr import zephyr_add_overlay_builder, zephyr_add_prj_conf
import esphome.config_validation as cv
from esphome.const import (
@@ -12,7 +13,6 @@ from esphome.const import (
CONF_ID,
CONF_INVERTED,
CONF_NUMBER,
CONF_OUTPUT,
CONF_PIN,
CONF_PLATFORM,
)
@@ -74,7 +74,7 @@ def _allocate_blocks() -> None:
full_config = fv.full_config.get()
zephyr_pwm_conf = [
cfg
for cfg in full_config.get(CONF_OUTPUT, [])
for cfg in full_config.get(OUTPUT_DOMAIN, [])
if cfg.get(CONF_PLATFORM) == DOMAIN
]
+17 -17
View File
@@ -3,7 +3,7 @@ from typing import Any
from esphome import automation
import esphome.codegen as cg
from esphome.components.esp32 import only_on_variant
from esphome.components.esp32 import DOMAIN as ESP32_DOMAIN, only_on_variant
from esphome.components.esp32.const import (
VARIANT_ESP32C5,
VARIANT_ESP32C6,
@@ -74,19 +74,19 @@ def _get_base_schema(cluster_options: list[str] | None = None) -> cv.Schema:
schema = cv.Schema(
{
cv.Optional(CONF_REPORT): cv.All(
cv.requires_component("zigbee"),
cv.requires_component("esp32"),
cv.requires_component(DOMAIN),
cv.requires_component(ESP32_DOMAIN),
_check_report_deprecation,
cv.enum(REPORT, lower=True),
),
cv.Optional(CONF_ENDPOINT): cv.All(
cv.requires_component("zigbee"),
cv.requires_component("esp32"),
cv.requires_component(DOMAIN),
cv.requires_component(ESP32_DOMAIN),
cv.int_range(1, CONF_MAX_EP_NUMBER),
),
cv.Optional(CONF_USE_DEVICE_TYPE): cv.All(
cv.requires_component("zigbee"),
cv.requires_component("esp32"),
cv.requires_component(DOMAIN),
cv.requires_component(ESP32_DOMAIN),
cv.boolean,
),
}
@@ -97,8 +97,8 @@ def _get_base_schema(cluster_options: list[str] | None = None) -> cv.Schema:
cv.OnlyWith(
CONF_CLUSTER, ["esp32", "zigbee"], default=cluster_options[0]
): cv.All(
cv.requires_component("zigbee"),
cv.requires_component("esp32"),
cv.requires_component(DOMAIN),
cv.requires_component(ESP32_DOMAIN),
cv.one_of(*cluster_options, lower=True),
),
}
@@ -231,7 +231,7 @@ async def to_code(config: ConfigType) -> None:
async def setup_binary_sensor(entity: cg.MockObj, config: ConfigType) -> None:
if "zigbee" not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
if DOMAIN not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
return
if CORE.using_zephyr:
if not config.get(CONF_ZIGBEE_ID):
@@ -243,7 +243,7 @@ async def setup_binary_sensor(entity: cg.MockObj, config: ConfigType) -> None:
async def setup_sensor(entity: cg.MockObj, config: ConfigType) -> None:
if "zigbee" not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
if DOMAIN not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
return
if CORE.using_zephyr:
if not config.get(CONF_ZIGBEE_ID):
@@ -255,7 +255,7 @@ async def setup_sensor(entity: cg.MockObj, config: ConfigType) -> None:
async def setup_switch(entity: cg.MockObj, config: ConfigType) -> None:
if "zigbee" not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
if DOMAIN not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
return
if CORE.using_zephyr:
if not config.get(CONF_ZIGBEE_ID):
@@ -273,7 +273,7 @@ async def setup_number(
max_value: float,
step: float,
) -> None:
if "zigbee" not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
if DOMAIN not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
return
if CORE.using_zephyr:
if not config.get(CONF_ZIGBEE_ID):
@@ -303,7 +303,7 @@ def consume_endpoint(config: ConfigType) -> ConfigType:
def validate_binary_sensor(config: ConfigType) -> ConfigType:
if "zigbee" not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
if DOMAIN not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
return config
if CORE.is_esp32:
return validate_binary_sensor_esp32(config)
@@ -311,7 +311,7 @@ def validate_binary_sensor(config: ConfigType) -> ConfigType:
def validate_sensor(config: ConfigType) -> ConfigType:
if "zigbee" not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
if DOMAIN not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
return config
if CORE.is_esp32:
return validate_sensor_esp32(config)
@@ -319,7 +319,7 @@ def validate_sensor(config: ConfigType) -> ConfigType:
def validate_switch(config: ConfigType) -> ConfigType:
if "zigbee" not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
if DOMAIN not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
return config
if CORE.is_esp32:
return validate_switch_esp32(config)
@@ -327,7 +327,7 @@ def validate_switch(config: ConfigType) -> ConfigType:
def validate_number(config: ConfigType) -> ConfigType:
if "zigbee" not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
if DOMAIN not in CORE.loaded_integrations or config.get(CONF_INTERNAL):
return config
if CORE.is_esp32:
return validate_number_esp32(config)
+2 -1
View File
@@ -1,5 +1,6 @@
import esphome.codegen as cg
from esphome.components import uart
from esphome.components.wifi import DOMAIN as WIFI_DOMAIN
import esphome.config_validation as cv
from esphome.const import CONF_ID, CONF_POWER_SAVE_MODE, CONF_WIFI
import esphome.final_validate as fv
@@ -15,7 +16,7 @@ ZWaveProxy = zwave_proxy_ns.class_("ZWaveProxy", cg.Component, uart.UARTDevice)
def final_validate(config: ConfigType) -> None:
full_config = fv.full_config.get()
if (wifi_conf := full_config.get(CONF_WIFI)) and (
if (wifi_conf := full_config.get(WIFI_DOMAIN)) and (
wifi_conf.get(CONF_POWER_SAVE_MODE).lower() != "none"
):
raise cv.Invalid(
@@ -20,7 +20,7 @@ def test_proxy_claims_its_slots_through_the_shared_accounting(
# A default (3-slot) proxy build records one claim per slot, attributed
# to the consumer, and passes final validation.
generate_main(component_config_path("rp2_proxy_default.yaml"))
used = CORE.data[rp2040_ble.KEY_RP2040_BLE][rp2040_ble.KEY_USED_CONNECTION_SLOTS]
used = CORE.data[rp2040_ble.DOMAIN][rp2040_ble.KEY_USED_CONNECTION_SLOTS]
assert used == ["bluetooth_proxy"] * 3
+19 -18
View File
@@ -8,7 +8,8 @@ from typing import Any
import pytest
from esphome import config_validation as cv
from esphome.components.sntp.time import CONF_SNTP, _sntp_final_validate
from esphome.components.sntp import DOMAIN as SNTP_DOMAIN
from esphome.components.sntp.time import _sntp_final_validate
from esphome.const import CONF_ID, CONF_PLATFORM, CONF_SERVERS, CONF_TIME
from esphome.core import ID
import esphome.final_validate as fv
@@ -20,7 +21,7 @@ import esphome.final_validate as fv
pytest.param(
[
{
CONF_PLATFORM: CONF_SNTP,
CONF_PLATFORM: SNTP_DOMAIN,
CONF_ID: ID("sntp_time", is_manual=False),
CONF_SERVERS: ["192.168.1.1", "pool.ntp.org"],
}
@@ -33,12 +34,12 @@ import esphome.final_validate as fv
pytest.param(
[
{
CONF_PLATFORM: CONF_SNTP,
CONF_PLATFORM: SNTP_DOMAIN,
CONF_ID: ID("sntp_time_1", is_manual=False),
CONF_SERVERS: ["192.168.1.1", "pool.ntp.org"],
},
{
CONF_PLATFORM: CONF_SNTP,
CONF_PLATFORM: SNTP_DOMAIN,
CONF_ID: ID("sntp_time_2", is_manual=False),
CONF_SERVERS: ["192.168.1.2"],
},
@@ -51,12 +52,12 @@ import esphome.final_validate as fv
pytest.param(
[
{
CONF_PLATFORM: CONF_SNTP,
CONF_PLATFORM: SNTP_DOMAIN,
CONF_ID: ID("sntp_time_1", is_manual=False),
CONF_SERVERS: ["192.168.1.1", "pool.ntp.org"],
},
{
CONF_PLATFORM: CONF_SNTP,
CONF_PLATFORM: SNTP_DOMAIN,
CONF_ID: ID("sntp_time_2", is_manual=False),
CONF_SERVERS: ["pool.ntp.org", "192.168.1.2"],
},
@@ -69,17 +70,17 @@ import esphome.final_validate as fv
pytest.param(
[
{
CONF_PLATFORM: CONF_SNTP,
CONF_PLATFORM: SNTP_DOMAIN,
CONF_ID: ID("sntp_time_1", is_manual=False),
CONF_SERVERS: ["192.168.1.1", "pool.ntp.org"],
},
{
CONF_PLATFORM: CONF_SNTP,
CONF_PLATFORM: SNTP_DOMAIN,
CONF_ID: ID("sntp_time_2", is_manual=False),
CONF_SERVERS: ["192.168.1.2", "pool2.ntp.org"],
},
{
CONF_PLATFORM: CONF_SNTP,
CONF_PLATFORM: SNTP_DOMAIN,
CONF_ID: ID("sntp_time_3", is_manual=False),
CONF_SERVERS: ["pool3.ntp.org"],
},
@@ -95,7 +96,7 @@ import esphome.final_validate as fv
pytest.param(
[
{
CONF_PLATFORM: CONF_SNTP,
CONF_PLATFORM: SNTP_DOMAIN,
CONF_ID: ID("sntp_time_1", is_manual=False),
CONF_SERVERS: [
"192.168.1.1",
@@ -105,7 +106,7 @@ import esphome.final_validate as fv
],
},
{
CONF_PLATFORM: CONF_SNTP,
CONF_PLATFORM: SNTP_DOMAIN,
CONF_ID: ID("sntp_time_2", is_manual=False),
CONF_SERVERS: ["pool.ntp.org", "192.168.1.2"],
},
@@ -143,7 +144,7 @@ def test_sntp_instance_merging(
sntp_instances = [
tc
for tc in updated_conf[CONF_TIME]
if tc.get(CONF_PLATFORM) == CONF_SNTP
if tc.get(CONF_PLATFORM) == SNTP_DOMAIN
]
assert len(sntp_instances) == expected_count
@@ -169,12 +170,12 @@ def test_sntp_inconsistent_manual_ids() -> None:
# Create configs with manual IDs that are inconsistent
time_configs = [
{
CONF_PLATFORM: CONF_SNTP,
CONF_PLATFORM: SNTP_DOMAIN,
CONF_ID: ID("sntp_time_1", is_manual=True),
CONF_SERVERS: ["192.168.1.1"],
},
{
CONF_PLATFORM: CONF_SNTP,
CONF_PLATFORM: SNTP_DOMAIN,
CONF_ID: ID("sntp_time_2", is_manual=True),
CONF_SERVERS: ["192.168.1.2"],
},
@@ -197,7 +198,7 @@ def test_sntp_with_other_time_platforms(caplog: pytest.LogCaptureFixture) -> Non
"""Test that SNTP merging doesn't affect other time platforms."""
time_configs = [
{
CONF_PLATFORM: CONF_SNTP,
CONF_PLATFORM: SNTP_DOMAIN,
CONF_ID: ID("sntp_time_1", is_manual=False),
CONF_SERVERS: ["192.168.1.1"],
},
@@ -206,7 +207,7 @@ def test_sntp_with_other_time_platforms(caplog: pytest.LogCaptureFixture) -> Non
CONF_ID: ID("homeassistant_time", is_manual=False),
},
{
CONF_PLATFORM: CONF_SNTP,
CONF_PLATFORM: SNTP_DOMAIN,
CONF_ID: ID("sntp_time_2", is_manual=False),
CONF_SERVERS: ["192.168.1.2"],
},
@@ -226,11 +227,11 @@ def test_sntp_with_other_time_platforms(caplog: pytest.LogCaptureFixture) -> Non
# Find the platforms
platforms = {tc[CONF_PLATFORM] for tc in updated_conf[CONF_TIME]}
assert platforms == {CONF_SNTP, "homeassistant"}
assert platforms == {SNTP_DOMAIN, "homeassistant"}
# Verify SNTP was merged
sntp_instances = [
tc for tc in updated_conf[CONF_TIME] if tc[CONF_PLATFORM] == CONF_SNTP
tc for tc in updated_conf[CONF_TIME] if tc[CONF_PLATFORM] == SNTP_DOMAIN
]
assert len(sntp_instances) == 1
assert sntp_instances[0][CONF_SERVERS] == ["192.168.1.1", "192.168.1.2"]