mirror of
https://github.com/esphome/esphome.git
synced 2026-10-20 04:56:48 +00:00
[core] Use get_declaring_config for parent config lookups
This commit is contained in:
@@ -108,8 +108,7 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
|
||||
|
||||
def _final_validate(config: ConfigType) -> ConfigType:
|
||||
full = fv.full_config.get()
|
||||
hub = full.get_config_for_path(full.get_path_for_id(config[CONF_AS734X_ID])[:-1])
|
||||
hub = fv.get_declaring_config(config[CONF_AS734X_ID])
|
||||
if hub[CONF_TYPE] != "AS7341":
|
||||
return config
|
||||
for key in _BAND_KEYS:
|
||||
|
||||
@@ -66,9 +66,7 @@ CONFIG_SCHEMA = cv.typed_schema(
|
||||
def _final_validate(config: ConfigType) -> None:
|
||||
if config[CONF_TYPE] not in _MAGNETIC_FIELDS:
|
||||
return
|
||||
full_config = fv.full_config.get()
|
||||
bmi270_path = full_config.get_path_for_id(config[CONF_BMI270_ID])[:-1]
|
||||
bmi270_config = full_config.get_config_for_path(bmi270_path)
|
||||
bmi270_config = fv.get_declaring_config(config[CONF_BMI270_ID])
|
||||
if bmi270_config.get(CONF_AUX_DEVICE) != "BMM150":
|
||||
raise cv.Invalid(
|
||||
"This sensor requires the parent bmi270's 'aux_device' to be set to 'BMM150'",
|
||||
|
||||
@@ -7,6 +7,7 @@ from esphome.components import i2c
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_ID, CONF_UPDATE_INTERVAL, CONF_WATCHDOG
|
||||
from esphome.core import CORE, ID, EsphomeError, Lambda, TimePeriod
|
||||
import esphome.final_validate as fv
|
||||
from esphome.types import ConfigType, TemplateArgsType
|
||||
|
||||
CODEOWNERS = ["@Rapsssito"]
|
||||
@@ -361,8 +362,7 @@ async def bq25186_configure_to_code(
|
||||
parent = await cg.get_variable(config[CONF_ID])
|
||||
|
||||
# Options the action leaves out fall back to the component's configuration
|
||||
path = CORE.config.get_path_for_id(config[CONF_ID])[:-1]
|
||||
parent_config = CORE.config.get_config_for_path(path)
|
||||
parent_config = fv.get_declaring_config(config[CONF_ID], CORE.config)
|
||||
merged = {**parent_config, **config}
|
||||
try:
|
||||
_check_watchdog(merged)
|
||||
|
||||
@@ -4,7 +4,6 @@ import esphome.codegen as cg
|
||||
from esphome.components import uart
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_ID, CONF_MODE, CONF_MODEL
|
||||
from esphome.core import ID
|
||||
import esphome.final_validate as fv
|
||||
from esphome.types import ConfigType
|
||||
|
||||
@@ -59,16 +58,11 @@ FINAL_VALIDATE_SCHEMA = uart.final_validate_device_schema(
|
||||
)
|
||||
|
||||
|
||||
def get_hub_config(root: ConfigType, hub_id: ID) -> ConfigType:
|
||||
"""Return the configuration of the hub with the given id."""
|
||||
return next(conf for conf in root[DOMAIN] if conf[CONF_ID] == hub_id)
|
||||
|
||||
|
||||
def final_validate_mode(keys: dict[str, str]) -> Callable[[ConfigType], None]:
|
||||
"""Reject entity keys that need a different hub mode; `keys` maps each key to the mode it needs."""
|
||||
|
||||
def validator(config: ConfigType) -> None:
|
||||
hub = get_hub_config(fv.full_config.get(), config[CONF_DFROBOT_SEN0609_ID])
|
||||
hub = fv.get_declaring_config(config[CONF_DFROBOT_SEN0609_ID])
|
||||
for key, mode in keys.items():
|
||||
if key in config and hub[CONF_MODE] != mode:
|
||||
raise cv.Invalid(
|
||||
|
||||
@@ -18,6 +18,7 @@ from esphome.const import (
|
||||
UNIT_SECOND,
|
||||
)
|
||||
from esphome.core import CORE
|
||||
import esphome.final_validate as fv
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from .. import (
|
||||
@@ -28,7 +29,6 @@ from .. import (
|
||||
DFRobotSEN0609Component,
|
||||
dfrobot_sen0609_ns,
|
||||
final_validate_mode,
|
||||
get_hub_config,
|
||||
)
|
||||
|
||||
DFRobotSEN0609Number = dfrobot_sen0609_ns.class_("DFRobotSEN0609Number", number.Number)
|
||||
@@ -181,7 +181,7 @@ FINAL_VALIDATE_SCHEMA = final_validate_mode(
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
hub_id = config[CONF_DFROBOT_SEN0609_ID]
|
||||
hub = await cg.get_variable(hub_id)
|
||||
max_range = MAX_RANGES[get_hub_config(CORE.config, hub_id)[CONF_MODEL]]
|
||||
max_range = MAX_RANGES[fv.get_declaring_config(hub_id, CORE.config)[CONF_MODEL]]
|
||||
numbers = number.sub_numbers(config, parent=hub)
|
||||
for s in SETTINGS:
|
||||
await numbers(
|
||||
|
||||
@@ -26,7 +26,7 @@ from esphome.const import (
|
||||
SCHEDULER_DONT_RUN,
|
||||
)
|
||||
from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
|
||||
from esphome.final_validate import full_config
|
||||
from esphome.final_validate import full_config, get_declaring_config
|
||||
|
||||
DOMAIN = "display"
|
||||
IS_PLATFORM_COMPONENT = True
|
||||
@@ -195,9 +195,7 @@ def get_display_metadata(display_id: ID) -> DisplayMetaData:
|
||||
return meta
|
||||
# No metadata found, display driver may not yet support it.
|
||||
# Read the raw config to populate the returned data
|
||||
global_config = full_config.get()
|
||||
path = global_config.get_path_for_id(display_id)[:-1]
|
||||
disp_config = global_config.get_config_for_path(path)
|
||||
disp_config = get_declaring_config(display_id)
|
||||
dimensions = disp_config.get(CONF_DIMENSIONS, (0, 0))
|
||||
if isinstance(dimensions, dict):
|
||||
dimensions = (dimensions.get(CONF_WIDTH, 0), dimensions.get(CONF_HEIGHT, 0))
|
||||
|
||||
@@ -32,9 +32,7 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
|
||||
def _final_validate(config: ConfigType) -> None:
|
||||
"""Validate that the channel is within the parent's channel count."""
|
||||
fconf = fv.full_config.get()
|
||||
path = fconf.get_path_for_id(config[CONF_DS248X_ID])[:-1]
|
||||
parent_config = fconf.get_config_for_path(path)
|
||||
parent_config = fv.get_declaring_config(config[CONF_DS248X_ID])
|
||||
channel_count = get_channel_count(parent_config)
|
||||
channel = config[CONF_CHANNEL]
|
||||
|
||||
|
||||
@@ -63,9 +63,7 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
|
||||
|
||||
def _validate_model(config: ConfigType) -> None:
|
||||
full_config = fv.full_config.get()
|
||||
climate_path = full_config.get_path_for_id(config[CONF_GREE_ID])[:-1]
|
||||
climate_conf = full_config.get_config_for_path(climate_path)
|
||||
climate_conf = fv.get_declaring_config(config[CONF_GREE_ID])
|
||||
if climate_conf[CONF_MODEL] not in SUPPORTED_MODELS:
|
||||
raise cv.Invalid(
|
||||
"Gree switches are only supported for the "
|
||||
|
||||
@@ -61,12 +61,10 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
|
||||
|
||||
def _final_validate(config) -> None:
|
||||
full_config = fv.full_config.get()
|
||||
for switch_type in [CONF_BEEPER, CONF_QUIET_MODE]:
|
||||
# Check switches that are only supported for HonClimate
|
||||
if config.get(switch_type):
|
||||
climate_path = full_config.get_path_for_id(config[CONF_HAIER_ID])[:-1]
|
||||
climate_conf = full_config.get_config_for_path(climate_path)
|
||||
climate_conf = fv.get_declaring_config(config[CONF_HAIER_ID])
|
||||
protocol_type = climate_conf.get(CONF_PROTOCOL)
|
||||
if protocol_type.casefold() != PROTOCOL_HON.casefold():
|
||||
raise cv.Invalid(
|
||||
|
||||
@@ -48,9 +48,7 @@ def _final_validate(config: dict[str, Any]) -> None:
|
||||
|
||||
# Get the transmitter configuration
|
||||
transmitter_id = config[CONF_REMOTE_TRANSMITTER_ID]
|
||||
full_config = fv.full_config.get()
|
||||
transmitter_path = full_config.get_path_for_id(transmitter_id)[:-1]
|
||||
transmitter_config = full_config.get_config_for_path(transmitter_path)
|
||||
transmitter_config = fv.get_declaring_config(transmitter_id)
|
||||
|
||||
# Check if carrier_duty_percent set to 0 or 100
|
||||
# Note: remote_transmitter schema requires this field and validates 1-100%,
|
||||
|
||||
@@ -35,11 +35,7 @@ def _final_validate(config: ConfigType) -> None:
|
||||
if CONF_REMOTE_TRANSMITTER_ID not in config:
|
||||
return
|
||||
|
||||
full_config = fv.full_config.get()
|
||||
transmitter_path = full_config.get_path_for_id(config[CONF_REMOTE_TRANSMITTER_ID])[
|
||||
:-1
|
||||
]
|
||||
transmitter_config = full_config.get_config_for_path(transmitter_path)
|
||||
transmitter_config = fv.get_declaring_config(config[CONF_REMOTE_TRANSMITTER_ID])
|
||||
|
||||
duty_percent = transmitter_config.get(CONF_CARRIER_DUTY_PERCENT)
|
||||
if duty_percent is not None and duty_percent != 100:
|
||||
|
||||
@@ -101,8 +101,7 @@ def final_validate(config: ConfigType) -> None:
|
||||
)
|
||||
|
||||
if config.get(CONF_WAKE):
|
||||
hub_path = full_config.get_path_for_id(hub_id)
|
||||
hub_config = full_config.get_config_for_path(hub_path[:-1])
|
||||
hub_config = fv.get_declaring_config(hub_id, full_config)
|
||||
if hub_config.get(CONF_WAKEUP_PIN) is None:
|
||||
raise cv.Invalid(
|
||||
f"{CONF_WAKE} requires {CONF_WAKEUP_PIN} on the parent ld6002b component",
|
||||
|
||||
@@ -326,8 +326,7 @@ def _final_validate(config: ConfigType) -> None:
|
||||
|
||||
for ref in refs:
|
||||
try:
|
||||
light_path = fconf.get_path_for_id(ref.light_id)[:-1]
|
||||
light_config = fconf.get_config_for_path(light_path)
|
||||
light_config = fv.get_declaring_config(ref.light_id, fconf)
|
||||
except KeyError:
|
||||
# Light ID not found — ID validation will have already reported this
|
||||
continue
|
||||
@@ -344,8 +343,7 @@ def _final_validate(config: ConfigType) -> None:
|
||||
|
||||
for ref in cycle_refs:
|
||||
try:
|
||||
light_path = fconf.get_path_for_id(ref.light_id)[:-1]
|
||||
light_config = fconf.get_config_for_path(light_path)
|
||||
light_config = fv.get_declaring_config(ref.light_id, fconf)
|
||||
except KeyError:
|
||||
continue
|
||||
|
||||
|
||||
@@ -34,6 +34,7 @@ from esphome.const import (
|
||||
)
|
||||
from esphome.core import CORE, ID, EsphomeError, Lambda
|
||||
from esphome.cpp_generator import MockObj, TemplateArgsType
|
||||
import esphome.final_validate as fv
|
||||
from esphome.schema_extractors import SCHEMA_EXTRACT, schema_extractor
|
||||
from esphome.types import ConfigType
|
||||
|
||||
@@ -267,8 +268,7 @@ def _resolve_effect_index(config: ConfigType, original_name: str) -> int:
|
||||
if original_name.lower() == "none":
|
||||
return 0
|
||||
light_id = config[CONF_ID]
|
||||
light_path = CORE.config.get_path_for_id(light_id)[:-1]
|
||||
light_config = CORE.config.get_config_for_path(light_path)
|
||||
light_config = fv.get_declaring_config(light_id, CORE.config)
|
||||
effects = light_config.get(CONF_EFFECTS, [])
|
||||
index = find_effect_index(effects, original_name)
|
||||
if index is not None:
|
||||
|
||||
@@ -41,9 +41,7 @@ CONFIG_SCHEMA = cv.All(
|
||||
|
||||
|
||||
def _final_validate(config: ConfigType) -> None:
|
||||
full_config = fv.full_config.get()
|
||||
display_path = full_config.get_path_for_id(config[CONF_DISPLAY_ID])[:-1]
|
||||
display_config = full_config.get_config_for_path(display_path)
|
||||
display_config = fv.get_declaring_config(config[CONF_DISPLAY_ID])
|
||||
if CONF_BRIGHTNESS not in display_config:
|
||||
model = display_config[CONF_MODEL]
|
||||
if model == "CUSTOM":
|
||||
|
||||
@@ -310,9 +310,7 @@ def migrate_custom_command(config: ConfigType) -> None:
|
||||
frame = config.get(CONF_CUSTOM_COMMAND)
|
||||
if frame is None:
|
||||
return
|
||||
fconf = fv.full_config.get()
|
||||
path = fconf.get_path_for_id(config[CONF_MODBUS_CONTROLLER_ID])[:-1]
|
||||
controller = fconf.get_config_for_path(path)
|
||||
controller = fv.get_declaring_config(config[CONF_MODBUS_CONTROLLER_ID])
|
||||
# the controller's DEVICE address (from modbus_device_schema)
|
||||
address = controller[CONF_ADDRESS]
|
||||
if frame[0] != address:
|
||||
@@ -342,9 +340,7 @@ def _reject_continuous_write_custom_pdu(config: ConfigType) -> None:
|
||||
pdu = config.get(CONF_CUSTOM_PDU)
|
||||
if pdu is None or not modbus.is_function_code_write(pdu[0]):
|
||||
return
|
||||
fconf = fv.full_config.get()
|
||||
path = fconf.get_path_for_id(config[CONF_MODBUS_CONTROLLER_ID])[:-1]
|
||||
controller = fconf.get_config_for_path(path)
|
||||
controller = fv.get_declaring_config(config[CONF_MODBUS_CONTROLLER_ID])
|
||||
if controller.get(CONF_CONTINUOUS) is True:
|
||||
raise cv.Invalid(
|
||||
f"a '{CONF_CUSTOM_PDU}' with a write function code (0x{pdu[0] & 0x7F:02X}) can't be polled "
|
||||
@@ -359,9 +355,7 @@ def _reject_broadcastable_custom_pdu(config: ConfigType) -> None:
|
||||
pdu = config.get(CONF_CUSTOM_PDU)
|
||||
if pdu is None or not modbus.is_function_code_broadcastable(pdu[0]):
|
||||
return
|
||||
fconf = fv.full_config.get()
|
||||
path = fconf.get_path_for_id(config[CONF_MODBUS_CONTROLLER_ID])[:-1]
|
||||
controller = fconf.get_config_for_path(path)
|
||||
controller = fv.get_declaring_config(config[CONF_MODBUS_CONTROLLER_ID])
|
||||
if (
|
||||
controller.get(CONF_ADDRESS) == modbus.BROADCAST_ADDRESS
|
||||
and controller.get(modbus.CONF_ALLOW_BROADCAST_READ) is True
|
||||
@@ -384,9 +378,7 @@ def validate_custom_pdu_item(config: ConfigType) -> None:
|
||||
def _reject_write_option_off_broadcast(config: ConfigType) -> None:
|
||||
if not any(config.get(key) is True for key in modbus.broadcast_only_option_keys()):
|
||||
return
|
||||
fconf = fv.full_config.get()
|
||||
path = fconf.get_path_for_id(config[CONF_MODBUS_CONTROLLER_ID])[:-1]
|
||||
controller = fconf.get_config_for_path(path)
|
||||
controller = fv.get_declaring_config(config[CONF_MODBUS_CONTROLLER_ID])
|
||||
if controller.get(CONF_ADDRESS) != modbus.BROADCAST_ADDRESS:
|
||||
raise cv.Invalid(
|
||||
f"'{modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE}' only applies when the "
|
||||
|
||||
@@ -50,9 +50,7 @@ MAX_RECOMMENDED_UPDATE_INTERVAL_MS = 100
|
||||
|
||||
def get_motion_config(motion_id: ID) -> dict:
|
||||
"""Look up the (already-validated) config of the motion hub referenced by motion_id."""
|
||||
full_config = fv.full_config.get()
|
||||
motion_path = full_config.get_path_for_id(motion_id)[:-1]
|
||||
return full_config.get_config_for_path(motion_path)
|
||||
return fv.get_declaring_config(motion_id)
|
||||
|
||||
|
||||
def check_update_interval(motion_id: ID, feature_name: str) -> None:
|
||||
|
||||
@@ -66,8 +66,7 @@ def _final_validate(config: ConfigType) -> None:
|
||||
"Max touch channel must equal the highest binary sensor channel or be removed for auto calculation",
|
||||
path=[CONF_MAX_TOUCH_CHANNEL],
|
||||
)
|
||||
path = fconf.get_path_for_id(config[CONF_ID])[:-1]
|
||||
this_config = fconf.get_config_for_path(path)
|
||||
this_config = fv.get_declaring_config(config[CONF_ID], fconf)
|
||||
this_config[CONF_MAX_TOUCH_CHANNEL] = max_touch_channel
|
||||
|
||||
|
||||
|
||||
@@ -49,9 +49,7 @@ def _final_validate(config: ConfigType) -> None:
|
||||
if config[CONF_TYPE] != CONF_STATUS:
|
||||
# Only run this validation if a status sensor is being configured
|
||||
return
|
||||
full_config = fv.full_config.get()
|
||||
transport_path = full_config.get_path_for_id(config[CONF_TRANSPORT_ID])[:-1]
|
||||
transport_config = full_config.get_config_for_path(transport_path)
|
||||
transport_config = fv.get_declaring_config(config[CONF_TRANSPORT_ID])
|
||||
if transport_config[CONF_PING_PONG_ENABLE] and any(
|
||||
CONF_ENCRYPTION in p
|
||||
for p in transport_config[CONF_PROVIDERS]
|
||||
|
||||
@@ -35,11 +35,8 @@ def validate_from_to(value: ConfigType) -> ConfigType:
|
||||
|
||||
|
||||
def validate_segment(config: ConfigType) -> None:
|
||||
fconf = fv.full_config.get()
|
||||
|
||||
if CONF_ID in config: # only validate addressable segments
|
||||
path = fconf.get_path_for_id(config[CONF_ID])[:-1]
|
||||
segment_light_config = fconf.get_config_for_path(path)
|
||||
segment_light_config = fv.get_declaring_config(config[CONF_ID])
|
||||
|
||||
if CONF_NUM_LEDS in segment_light_config:
|
||||
segment_len = segment_light_config[CONF_NUM_LEDS]
|
||||
|
||||
@@ -56,9 +56,7 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
|
||||
|
||||
def _final_validate(config: ConfigType) -> ConfigType:
|
||||
fconf = fv.full_config.get()
|
||||
hub_path = fconf.get_path_for_id(config[CONF_TAS58XX_ID])
|
||||
hub_conf = fconf.get_config_for_path(hub_path[:-1])
|
||||
hub_conf = fv.get_declaring_config(config[CONF_TAS58XX_ID])
|
||||
|
||||
if hub_conf[CONF_MODEL] == DAC_TAS5825M:
|
||||
return config
|
||||
|
||||
@@ -65,9 +65,7 @@ CONFIG_SCHEMA = cv.All(
|
||||
|
||||
def _final_validate(config: ConfigType) -> None:
|
||||
"""Resolve the key to the number the device reports, checked against the keypad's matrix."""
|
||||
full_config = fv.full_config.get()
|
||||
keypad_path = full_config.get_path_for_id(config[CONF_KEYPAD_ID])[:-1]
|
||||
keypad = full_config.get_config_for_path(keypad_path)
|
||||
keypad = fv.get_declaring_config(config[CONF_KEYPAD_ID])
|
||||
rows = keypad.get(CONF_ROWS, 0)
|
||||
columns = keypad.get(CONF_COLUMNS, 0)
|
||||
|
||||
|
||||
@@ -59,9 +59,7 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
def _final_validate(config: ConfigType) -> ConfigType:
|
||||
if CONF_ACTUAL_CURRENT not in config:
|
||||
return config
|
||||
full_config = fv.full_config.get()
|
||||
path = full_config.get_path_for_id(config[CONF_TMC2209_ID])[:-1]
|
||||
if full_config.get_config_for_path(path).get(CONF_ANALOG_CURRENT_SCALE):
|
||||
if fv.get_declaring_config(config[CONF_TMC2209_ID]).get(CONF_ANALOG_CURRENT_SCALE):
|
||||
raise cv.Invalid(
|
||||
f"'{CONF_ACTUAL_CURRENT}' can't be used with '{CONF_ANALOG_CURRENT_SCALE}', "
|
||||
"the current then depends on VREF",
|
||||
|
||||
@@ -53,8 +53,7 @@ CONFIG_SCHEMA = cv.All(
|
||||
|
||||
def _final_validate(config: ConfigType) -> ConfigType:
|
||||
validate_unique_child(config, "number")
|
||||
full = fv.full_config.get()
|
||||
parent = full.get_config_for_path(full.get_path_for_id(config[CONF_TMP102_ID])[:-1])
|
||||
parent = fv.get_declaring_config(config[CONF_TMP102_ID])
|
||||
add_number_parent(config[CONF_TMP102_ID], parent.get(CONF_EXTENDED_MODE, False))
|
||||
return config
|
||||
|
||||
|
||||
@@ -531,14 +531,12 @@ def inherit_property_from(property_to_inherit, parent_id_property, transform=Non
|
||||
|
||||
# Get config for the parent entity
|
||||
parent_id = _walk_config(config, parent_id_property)
|
||||
parent_path = fconf.get_path_for_id(parent_id)[:-1]
|
||||
parent_config = fconf.get_config_for_path(parent_path)
|
||||
parent_config = fv.get_declaring_config(parent_id, fconf)
|
||||
|
||||
# If parent sensor has the property set, inherit it
|
||||
if property in parent_config:
|
||||
path = fconf.get_path_for_id(config[CONF_ID])[:-1]
|
||||
this_config = _walk_config(
|
||||
fconf.get_config_for_path(path), property_path
|
||||
fv.get_declaring_config(config[CONF_ID], fconf), property_path
|
||||
)
|
||||
value = parent_config[property]
|
||||
if transform:
|
||||
|
||||
+2
-2
@@ -20,6 +20,7 @@ from esphome.const import (
|
||||
)
|
||||
from esphome.core import CORE, ID
|
||||
from esphome.cpp_generator import MockObjClass
|
||||
import esphome.final_validate as fv
|
||||
from esphome.schema_extractors import schema_extractor
|
||||
|
||||
|
||||
@@ -131,8 +132,7 @@ class PinRegistry(dict):
|
||||
with fconf.catch_error([cv.ROOT_CONFIG_PATH] + pin_path):
|
||||
if final_val_fun is not None:
|
||||
# Get the containing path of the config providing this pin.
|
||||
parent_path = fconf.get_path_for_id(client_id)[:-1]
|
||||
parent_config = fconf.get_config_for_path(parent_path)
|
||||
parent_config = fv.get_declaring_config(client_id, fconf)
|
||||
final_val_fun(pin_config, parent_config)
|
||||
allow_others = pin_config.get(CONF_ALLOW_OTHER_USES, False)
|
||||
if count != 1 and not allow_others and not CORE.testing_mode:
|
||||
|
||||
Reference in New Issue
Block a user