mirror of
https://github.com/esphome/esphome.git
synced 2026-10-24 06:55:37 +00:00
Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e263d52405 | ||
|
|
f77796877d | ||
|
|
2931a58a9b | ||
|
|
9d30bcc721 |
No files matched your search
@@ -108,7 +108,8 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
|
||||
|
||||
def _final_validate(config: ConfigType) -> ConfigType:
|
||||
hub = fv.get_declaring_config(config[CONF_AS734X_ID])
|
||||
full = fv.full_config.get()
|
||||
hub = full.get_config_for_path(full.get_path_for_id(config[CONF_AS734X_ID])[:-1])
|
||||
if hub[CONF_TYPE] != "AS7341":
|
||||
return config
|
||||
for key in _BAND_KEYS:
|
||||
|
||||
@@ -66,7 +66,9 @@ CONFIG_SCHEMA = cv.typed_schema(
|
||||
def _final_validate(config: ConfigType) -> None:
|
||||
if config[CONF_TYPE] not in _MAGNETIC_FIELDS:
|
||||
return
|
||||
bmi270_config = fv.get_declaring_config(config[CONF_BMI270_ID])
|
||||
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)
|
||||
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,7 +7,6 @@ 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"]
|
||||
@@ -360,7 +359,8 @@ 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
|
||||
parent_config = fv.get_declaring_config(config[CONF_ID], CORE.config)
|
||||
path = CORE.config.get_path_for_id(config[CONF_ID])[:-1]
|
||||
parent_config = CORE.config.get_config_for_path(path)
|
||||
merged = {**parent_config, **config}
|
||||
try:
|
||||
_check_watchdog(merged)
|
||||
|
||||
@@ -3,7 +3,8 @@ from collections.abc import Callable
|
||||
import esphome.codegen as cg
|
||||
from esphome.components import uart
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_MODE, CONF_MODEL
|
||||
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
|
||||
|
||||
@@ -58,11 +59,16 @@ 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 = fv.get_declaring_config(config[CONF_DFROBOT_SEN0609_ID])
|
||||
hub = get_hub_config(fv.full_config.get(), config[CONF_DFROBOT_SEN0609_ID])
|
||||
for key, mode in keys.items():
|
||||
if key in config and hub[CONF_MODE] != mode:
|
||||
raise cv.Invalid(
|
||||
|
||||
@@ -18,7 +18,6 @@ from esphome.const import (
|
||||
UNIT_SECOND,
|
||||
)
|
||||
from esphome.core import CORE
|
||||
import esphome.final_validate as fv
|
||||
from esphome.types import ConfigType
|
||||
|
||||
from .. import (
|
||||
@@ -29,6 +28,7 @@ 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[fv.get_declaring_config(hub_id, CORE.config)[CONF_MODEL]]
|
||||
max_range = MAX_RANGES[get_hub_config(CORE.config, hub_id)[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, get_declaring_config
|
||||
from esphome.final_validate import full_config
|
||||
from esphome.types import ConfigType
|
||||
|
||||
DOMAIN = "display"
|
||||
@@ -196,7 +196,9 @@ 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
|
||||
disp_config = get_declaring_config(display_id)
|
||||
global_config = full_config.get()
|
||||
path = global_config.get_path_for_id(display_id)[:-1]
|
||||
disp_config = global_config.get_config_for_path(path)
|
||||
dimensions = disp_config.get(CONF_DIMENSIONS, (0, 0))
|
||||
if isinstance(dimensions, dict):
|
||||
dimensions = (dimensions.get(CONF_WIDTH, 0), dimensions.get(CONF_HEIGHT, 0))
|
||||
|
||||
@@ -32,7 +32,9 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
|
||||
def _final_validate(config: ConfigType) -> None:
|
||||
"""Validate that the channel is within the parent's channel count."""
|
||||
parent_config = fv.get_declaring_config(config[CONF_DS248X_ID])
|
||||
fconf = fv.full_config.get()
|
||||
path = fconf.get_path_for_id(config[CONF_DS248X_ID])[:-1]
|
||||
parent_config = fconf.get_config_for_path(path)
|
||||
channel_count = get_channel_count(parent_config)
|
||||
channel = config[CONF_CHANNEL]
|
||||
|
||||
|
||||
@@ -63,7 +63,9 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
|
||||
|
||||
def _validate_model(config: ConfigType) -> None:
|
||||
climate_conf = fv.get_declaring_config(config[CONF_GREE_ID])
|
||||
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)
|
||||
if climate_conf[CONF_MODEL] not in SUPPORTED_MODELS:
|
||||
raise cv.Invalid(
|
||||
"Gree switches are only supported for the "
|
||||
|
||||
@@ -61,10 +61,12 @@ 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_conf = fv.get_declaring_config(config[CONF_HAIER_ID])
|
||||
climate_path = full_config.get_path_for_id(config[CONF_HAIER_ID])[:-1]
|
||||
climate_conf = full_config.get_config_for_path(climate_path)
|
||||
protocol_type = climate_conf.get(CONF_PROTOCOL)
|
||||
if protocol_type.casefold() != PROTOCOL_HON.casefold():
|
||||
raise cv.Invalid(
|
||||
|
||||
@@ -42,7 +42,9 @@ def _final_validate(config: dict[str, Any]) -> None:
|
||||
|
||||
# Get the transmitter configuration
|
||||
transmitter_id = config[CONF_REMOTE_TRANSMITTER_ID]
|
||||
transmitter_config = fv.get_declaring_config(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)
|
||||
|
||||
# Check if carrier_duty_percent set to 0 or 100
|
||||
# Note: remote_transmitter schema requires this field and validates 1-100%,
|
||||
|
||||
@@ -35,7 +35,11 @@ def _final_validate(config: ConfigType) -> None:
|
||||
if CONF_REMOTE_TRANSMITTER_ID not in config:
|
||||
return
|
||||
|
||||
transmitter_config = fv.get_declaring_config(config[CONF_REMOTE_TRANSMITTER_ID])
|
||||
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)
|
||||
|
||||
duty_percent = transmitter_config.get(CONF_CARRIER_DUTY_PERCENT)
|
||||
if duty_percent is not None and duty_percent != 100:
|
||||
|
||||
@@ -101,7 +101,8 @@ def final_validate(config: ConfigType) -> None:
|
||||
)
|
||||
|
||||
if config.get(CONF_WAKE):
|
||||
hub_config = fv.get_declaring_config(hub_id, full_config)
|
||||
hub_path = full_config.get_path_for_id(hub_id)
|
||||
hub_config = full_config.get_config_for_path(hub_path[:-1])
|
||||
if hub_config.get(CONF_WAKEUP_PIN) is None:
|
||||
raise cv.Invalid(
|
||||
f"{CONF_WAKE} requires {CONF_WAKEUP_PIN} on the parent ld6002b component",
|
||||
|
||||
@@ -380,7 +380,8 @@ def _final_validate(config: ConfigType) -> None:
|
||||
|
||||
for ref in refs:
|
||||
try:
|
||||
light_config = fv.get_declaring_config(ref.light_id, fconf)
|
||||
light_path = fconf.get_path_for_id(ref.light_id)[:-1]
|
||||
light_config = fconf.get_config_for_path(light_path)
|
||||
except KeyError:
|
||||
# Light ID not found — ID validation will have already reported this
|
||||
continue
|
||||
@@ -397,7 +398,8 @@ def _final_validate(config: ConfigType) -> None:
|
||||
|
||||
for ref in cycle_refs:
|
||||
try:
|
||||
light_config = fv.get_declaring_config(ref.light_id, fconf)
|
||||
light_path = fconf.get_path_for_id(ref.light_id)[:-1]
|
||||
light_config = fconf.get_config_for_path(light_path)
|
||||
except KeyError:
|
||||
continue
|
||||
|
||||
|
||||
@@ -34,7 +34,6 @@ 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
|
||||
|
||||
@@ -254,7 +253,8 @@ def _resolve_effect_index(config: ConfigType, original_name: str) -> int:
|
||||
from . import effect_not_found_message, find_effect_index
|
||||
|
||||
light_id = config[CONF_ID]
|
||||
light_config = fv.get_declaring_config(light_id, CORE.config)
|
||||
light_path = CORE.config.get_path_for_id(light_id)[:-1]
|
||||
light_config = CORE.config.get_config_for_path(light_path)
|
||||
effects = light_config.get(CONF_EFFECTS, [])
|
||||
index = find_effect_index(effects, original_name)
|
||||
if index is not None:
|
||||
|
||||
@@ -41,7 +41,9 @@ CONFIG_SCHEMA = cv.All(
|
||||
|
||||
|
||||
def _final_validate(config: ConfigType) -> None:
|
||||
display_config = fv.get_declaring_config(config[CONF_DISPLAY_ID])
|
||||
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)
|
||||
if CONF_BRIGHTNESS not in display_config:
|
||||
model = display_config[CONF_MODEL]
|
||||
if model == "CUSTOM":
|
||||
|
||||
@@ -310,7 +310,9 @@ def migrate_custom_command(config: ConfigType) -> None:
|
||||
frame = config.get(CONF_CUSTOM_COMMAND)
|
||||
if frame is None:
|
||||
return
|
||||
controller = fv.get_declaring_config(config[CONF_MODBUS_CONTROLLER_ID])
|
||||
fconf = fv.full_config.get()
|
||||
path = fconf.get_path_for_id(config[CONF_MODBUS_CONTROLLER_ID])[:-1]
|
||||
controller = fconf.get_config_for_path(path)
|
||||
# the controller's DEVICE address (from modbus_device_schema)
|
||||
address = controller[CONF_ADDRESS]
|
||||
if frame[0] != address:
|
||||
@@ -340,7 +342,9 @@ 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
|
||||
controller = fv.get_declaring_config(config[CONF_MODBUS_CONTROLLER_ID])
|
||||
fconf = fv.full_config.get()
|
||||
path = fconf.get_path_for_id(config[CONF_MODBUS_CONTROLLER_ID])[:-1]
|
||||
controller = fconf.get_config_for_path(path)
|
||||
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 "
|
||||
@@ -355,7 +359,9 @@ 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
|
||||
controller = fv.get_declaring_config(config[CONF_MODBUS_CONTROLLER_ID])
|
||||
fconf = fv.full_config.get()
|
||||
path = fconf.get_path_for_id(config[CONF_MODBUS_CONTROLLER_ID])[:-1]
|
||||
controller = fconf.get_config_for_path(path)
|
||||
if (
|
||||
controller.get(CONF_ADDRESS) == modbus.BROADCAST_ADDRESS
|
||||
and controller.get(modbus.CONF_ALLOW_BROADCAST_READ) is True
|
||||
@@ -378,7 +384,9 @@ 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
|
||||
controller = fv.get_declaring_config(config[CONF_MODBUS_CONTROLLER_ID])
|
||||
fconf = fv.full_config.get()
|
||||
path = fconf.get_path_for_id(config[CONF_MODBUS_CONTROLLER_ID])[:-1]
|
||||
controller = fconf.get_config_for_path(path)
|
||||
if controller.get(CONF_ADDRESS) != modbus.BROADCAST_ADDRESS:
|
||||
raise cv.Invalid(
|
||||
f"'{modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE}' only applies when the "
|
||||
|
||||
@@ -50,7 +50,9 @@ 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."""
|
||||
return fv.get_declaring_config(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)
|
||||
|
||||
|
||||
def check_update_interval(motion_id: ID, feature_name: str) -> None:
|
||||
|
||||
@@ -67,7 +67,8 @@ 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],
|
||||
)
|
||||
this_config = fv.get_declaring_config(config[CONF_ID], fconf)
|
||||
path = fconf.get_path_for_id(config[CONF_ID])[:-1]
|
||||
this_config = fconf.get_config_for_path(path)
|
||||
this_config[CONF_MAX_TOUCH_CHANNEL] = max_touch_channel
|
||||
|
||||
|
||||
|
||||
@@ -48,7 +48,9 @@ def _final_validate(config: ConfigType) -> None:
|
||||
if config[CONF_TYPE] != CONF_STATUS:
|
||||
# Only run this validation if a status sensor is being configured
|
||||
return
|
||||
transport_config = fv.get_declaring_config(config[CONF_TRANSPORT_ID])
|
||||
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)
|
||||
if transport_config[CONF_PING_PONG_ENABLE] and any(
|
||||
CONF_ENCRYPTION in p
|
||||
for p in transport_config[CONF_PROVIDERS]
|
||||
|
||||
@@ -35,8 +35,11 @@ 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
|
||||
segment_light_config = fv.get_declaring_config(config[CONF_ID])
|
||||
path = fconf.get_path_for_id(config[CONF_ID])[:-1]
|
||||
segment_light_config = fconf.get_config_for_path(path)
|
||||
|
||||
if CONF_NUM_LEDS in segment_light_config:
|
||||
segment_len = segment_light_config[CONF_NUM_LEDS]
|
||||
|
||||
@@ -56,7 +56,9 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
|
||||
|
||||
def _final_validate(config: ConfigType) -> ConfigType:
|
||||
hub_conf = fv.get_declaring_config(config[CONF_TAS58XX_ID])
|
||||
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])
|
||||
|
||||
if hub_conf[CONF_MODEL] == DAC_TAS5825M:
|
||||
return config
|
||||
|
||||
@@ -65,7 +65,9 @@ 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."""
|
||||
keypad = fv.get_declaring_config(config[CONF_KEYPAD_ID])
|
||||
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)
|
||||
rows = keypad.get(CONF_ROWS, 0)
|
||||
columns = keypad.get(CONF_COLUMNS, 0)
|
||||
|
||||
|
||||
@@ -59,7 +59,9 @@ CONFIG_SCHEMA = cv.Schema(
|
||||
def _final_validate(config: ConfigType) -> ConfigType:
|
||||
if CONF_ACTUAL_CURRENT not in config:
|
||||
return config
|
||||
if fv.get_declaring_config(config[CONF_TMC2209_ID]).get(CONF_ANALOG_CURRENT_SCALE):
|
||||
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):
|
||||
raise cv.Invalid(
|
||||
f"'{CONF_ACTUAL_CURRENT}' can't be used with '{CONF_ANALOG_CURRENT_SCALE}', "
|
||||
"the current then depends on VREF",
|
||||
|
||||
@@ -56,7 +56,8 @@ _validate_unique = unique_child("number")
|
||||
|
||||
def _final_validate(config: ConfigType) -> ConfigType:
|
||||
_validate_unique(config)
|
||||
parent = fv.get_declaring_config(config[CONF_TMP102_ID])
|
||||
full = fv.full_config.get()
|
||||
parent = full.get_config_for_path(full.get_path_for_id(config[CONF_TMP102_ID])[:-1])
|
||||
add_number_parent(config[CONF_TMP102_ID], parent.get(CONF_EXTENDED_MODE, False))
|
||||
return config
|
||||
|
||||
|
||||
+3
-1
@@ -1136,7 +1136,9 @@ class IDPassValidationStep(ConfigValidationStep):
|
||||
m
|
||||
for m in matches
|
||||
if isinstance(
|
||||
candidate_conf := fv.get_declaring_config(m, result),
|
||||
candidate_conf := result.get_config_for_path(
|
||||
result.get_path_for_id(m)[:-1]
|
||||
),
|
||||
dict,
|
||||
)
|
||||
and all(
|
||||
|
||||
@@ -646,12 +646,14 @@ 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_config = fv.get_declaring_config(parent_id, fconf)
|
||||
parent_path = fconf.get_path_for_id(parent_id)[:-1]
|
||||
parent_config = fconf.get_config_for_path(parent_path)
|
||||
|
||||
# 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(
|
||||
fv.get_declaring_config(config[CONF_ID], fconf), property_path
|
||||
fconf.get_config_for_path(path), property_path
|
||||
)
|
||||
value = parent_config[property]
|
||||
if transform:
|
||||
|
||||
+23
-7
@@ -177,14 +177,13 @@ template<typename T> class ConstVector<T, true> {
|
||||
this->data_ = data;
|
||||
this->size_ = size;
|
||||
}
|
||||
/// Appends count elements into one new heap array this owns; the current list and extra may be codegen tables in
|
||||
/// flash.
|
||||
void append_copy(const T *extra, size_t count) { this->merge_copy_(this->data_, this->size(), extra, count); }
|
||||
/// Like append_copy(), with extra placed in front of the current list.
|
||||
void prepend_copy(const T *extra, size_t count) { this->merge_copy_(extra, count, this->data_, this->size()); }
|
||||
/// Copies the list into a heap array this owns, freeing a previous owned copy.
|
||||
void assign_copy(const T *data, size_t size) {
|
||||
auto *table = new T[size]; // NOLINT(cppcoreguidelines-owning-memory)
|
||||
std::copy(data, data + size, table);
|
||||
this->release_();
|
||||
this->data_ = table;
|
||||
this->size_ = size | OWNED_BIT;
|
||||
}
|
||||
void assign_copy(const T *data, size_t size) { this->merge_copy_(data, size, nullptr, 0); }
|
||||
|
||||
protected:
|
||||
static constexpr size_t OWNED_BIT = size_t{1} << (sizeof(size_t) * 8 - 1);
|
||||
@@ -193,6 +192,23 @@ template<typename T> class ConstVector<T, true> {
|
||||
if (this->size_ & OWNED_BIT)
|
||||
delete[] this->data_; // NOLINT(cppcoreguidelines-owning-memory)
|
||||
}
|
||||
void merge_copy_(const T *first, size_t first_count, const T *second, size_t second_count) {
|
||||
auto *table = new T[first_count + second_count]; // NOLINT(cppcoreguidelines-owning-memory)
|
||||
copy_words_(first, first_count, table);
|
||||
copy_words_(second, second_count, table + first_count);
|
||||
this->release_(); // after the copies: first or second may be the current list
|
||||
this->data_ = table;
|
||||
this->size_ = (first_count + second_count) | OWNED_BIT;
|
||||
}
|
||||
// Word loads through volatile, so the loop never becomes a memcpy that reads ESP8266 flash byte by byte
|
||||
static void copy_words_(const T *src, size_t count, T *dst) {
|
||||
auto *from = reinterpret_cast<const volatile uint32_t *>(src);
|
||||
auto *to = reinterpret_cast<uint8_t *>(dst);
|
||||
for (size_t i = 0; i < count * sizeof(T) / sizeof(uint32_t); i++) {
|
||||
const uint32_t word = from[i];
|
||||
std::memcpy(to + i * sizeof(uint32_t), &word, sizeof(word)); // into RAM: no aliasing through uint32_t
|
||||
}
|
||||
}
|
||||
|
||||
const T *data_{nullptr};
|
||||
size_t size_{0}; // top bit set when data_ is an owned heap copy
|
||||
|
||||
@@ -296,7 +296,9 @@ def _external_manifests() -> list[dict[str, Any]]:
|
||||
found[component_dir] = source
|
||||
break
|
||||
else:
|
||||
if not component_dir.is_relative_to(loader.CORE_COMPONENTS_PATH):
|
||||
# Built in means anywhere in the package: the esphome domain itself
|
||||
# is esphome/core, not under esphome/components
|
||||
if not component_dir.is_relative_to(loader.CORE_COMPONENTS_PATH.parent):
|
||||
_LOGGER.warning(
|
||||
"SBOM: component '%s' in %s matches no external component "
|
||||
"source and is listed as part of ESPHome",
|
||||
|
||||
+2
-2
@@ -21,7 +21,6 @@ from esphome.const import (
|
||||
)
|
||||
from esphome.core import CORE, ID
|
||||
from esphome.cpp_generator import MockObj, MockObjClass
|
||||
import esphome.final_validate as fv
|
||||
from esphome.schema_extractors import schema_extractor
|
||||
from esphome.types import ConfigType, SafeExpType
|
||||
|
||||
@@ -134,7 +133,8 @@ 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_config = fv.get_declaring_config(client_id, fconf)
|
||||
parent_path = fconf.get_path_for_id(client_id)[:-1]
|
||||
parent_config = fconf.get_config_for_path(parent_path)
|
||||
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:
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
esphome:
|
||||
name: test
|
||||
|
||||
esp32:
|
||||
board: esp32dev
|
||||
|
||||
logger:
|
||||
|
||||
remote_receiver:
|
||||
- id: ir_rcvr
|
||||
pin: GPIO4
|
||||
dump:
|
||||
- nec
|
||||
- rc_switch
|
||||
- raw
|
||||
- id: rf_rcvr
|
||||
pin: GPIO5
|
||||
dump:
|
||||
- rc_switch
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
from collections.abc import Callable, Generator
|
||||
from pathlib import Path
|
||||
import re
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
@@ -135,3 +136,24 @@ def test_dump_list_is_deduplicated_across_forms() -> None:
|
||||
def test_dump_list_rejects_invalid_entries_with_a_validation_error(bad: list) -> None:
|
||||
with pytest.raises(cv.Invalid):
|
||||
remote_base.validate_dumpers(bad)
|
||||
|
||||
|
||||
def test_dumper_count_is_the_busiest_receivers(
|
||||
generate_main: Callable[[str | Path], str],
|
||||
component_config_path: Callable[[str], Path],
|
||||
) -> None:
|
||||
generate_main(component_config_path("receiver_two_dumper_lists.yaml"))
|
||||
# nec, rc_switch and raw on one receiver, rc_switch on the other
|
||||
assert get_define_value("REMOTE_BASE_DUMPER_COUNT") == "3"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("restore_protocol_registries")
|
||||
def test_external_dumpers_are_constructed_in_ram(
|
||||
generate_main: Callable[[str | Path], str],
|
||||
component_config_path: Callable[[str], Path],
|
||||
) -> None:
|
||||
"""A dumper registered outside remote_base may hold state."""
|
||||
main_cpp = generate_main(
|
||||
component_config_path("receiver_with_external_protocol.yaml")
|
||||
)
|
||||
assert re.search(r"new\(\w+\) fake_protocol::FakeDumper\(\);", main_cpp)
|
||||
@@ -78,6 +78,40 @@ TEST(ConstVector, CopyFromItsOwnStorage) {
|
||||
EXPECT_STREQ(list[2], "c");
|
||||
}
|
||||
|
||||
TEST(ConstVector, AppendCopiesTheTableAndTheExtrasIntoOneOwnedArray) {
|
||||
static constexpr const char *const EXTRA[] = {"d", "e"};
|
||||
ProbeVector empty;
|
||||
empty.append_copy(EXTRA, 1);
|
||||
EXPECT_TRUE(empty.owned());
|
||||
ASSERT_EQ(empty.size(), 1U);
|
||||
EXPECT_STREQ(empty[0], "d");
|
||||
|
||||
ProbeVector list;
|
||||
list.assign_static(TABLE, 3);
|
||||
list.append_copy(EXTRA, 2);
|
||||
EXPECT_TRUE(list.owned());
|
||||
EXPECT_NE(list.data(), TABLE);
|
||||
ASSERT_EQ(list.size(), 5U);
|
||||
EXPECT_TRUE(std::equal(TABLE, TABLE + 3, list.begin()));
|
||||
EXPECT_STREQ(list[3], "d");
|
||||
EXPECT_STREQ(list[4], "e");
|
||||
list.append_copy(EXTRA, 1); // frees the previous owned array
|
||||
EXPECT_EQ(list.size(), 6U);
|
||||
EXPECT_STREQ(list[5], "d");
|
||||
}
|
||||
|
||||
TEST(ConstVector, PrependPutsTheExtrasInFront) {
|
||||
static constexpr const char *const EXTRA[] = {"d", "e"};
|
||||
ProbeVector list;
|
||||
list.assign_static(TABLE, 3);
|
||||
list.append_copy(EXTRA, 1);
|
||||
list.prepend_copy(EXTRA + 1, 1); // frees the previous owned array
|
||||
ASSERT_EQ(list.size(), 5U);
|
||||
EXPECT_STREQ(list[0], "e");
|
||||
EXPECT_TRUE(std::equal(TABLE, TABLE + 3, list.begin() + 1));
|
||||
EXPECT_STREQ(list[4], "d");
|
||||
}
|
||||
|
||||
TEST(ConstVector, PlainViewHasNoOwnershipCost) {
|
||||
static_assert(std::is_trivially_copyable_v<ConstVector<const char *>>);
|
||||
static_assert(std::is_trivially_destructible_v<ConstVector<const char *>>);
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import esphome.codegen as cg
|
||||
from esphome.types import ConfigType
|
||||
from tests.testing_helpers import ComponentManifestOverride
|
||||
|
||||
@@ -24,6 +25,8 @@ def override_manifest(manifest: ComponentManifestOverride) -> None:
|
||||
"panasonic",
|
||||
):
|
||||
request_protocol(protocol)
|
||||
# receiver_dumpers_test.cpp registers dumpers by hand
|
||||
cg.add_define("REMOTE_BASE_DUMPER_COUNT", 4)
|
||||
|
||||
manifest.to_code = to_code_testing
|
||||
# AUTO_LOAD is not resolved by the unit test build.
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
#include <gtest/gtest.h>
|
||||
|
||||
#include <vector>
|
||||
|
||||
#include "esphome/components/remote_base/remote_base.h"
|
||||
|
||||
namespace esphome::remote_base::testing {
|
||||
|
||||
class RecordingDumper : public RemoteReceiverDumperBase {
|
||||
public:
|
||||
RecordingDumper(std::vector<int> *calls, int id, bool matches, bool secondary = false)
|
||||
: calls_(calls), id_(id), matches_(matches), secondary_(secondary) {}
|
||||
bool dump(RemoteReceiveData /*src*/) override {
|
||||
this->calls_->push_back(this->id_);
|
||||
return this->matches_;
|
||||
}
|
||||
bool is_secondary() override { return this->secondary_; }
|
||||
|
||||
protected:
|
||||
std::vector<int> *calls_;
|
||||
int id_;
|
||||
bool matches_;
|
||||
bool secondary_;
|
||||
};
|
||||
|
||||
class TestReceiver : public RemoteReceiverBase {
|
||||
public:
|
||||
TestReceiver() : RemoteReceiverBase(nullptr) {}
|
||||
void receive() { this->call_listeners_dumpers_(); }
|
||||
};
|
||||
|
||||
TEST(RemoteReceiverDumpers, PrimariesRunInOrderAndAMatchSkipsTheSecondary) {
|
||||
std::vector<int> calls;
|
||||
RecordingDumper raw(&calls, 0, true, true), first(&calls, 1, false), second(&calls, 2, true), third(&calls, 3, false);
|
||||
TestReceiver receiver;
|
||||
receiver.register_dumper(&raw);
|
||||
receiver.register_dumper(&first);
|
||||
receiver.register_dumper(&second);
|
||||
receiver.register_dumper(&third);
|
||||
receiver.receive();
|
||||
EXPECT_EQ(calls, (std::vector<int>{1, 2, 3}));
|
||||
}
|
||||
|
||||
TEST(RemoteReceiverDumpers, SecondaryRunsWhenNoPrimaryMatched) {
|
||||
std::vector<int> calls;
|
||||
RecordingDumper raw(&calls, 0, true, true), miss(&calls, 1, false);
|
||||
TestReceiver receiver;
|
||||
receiver.register_dumper(&miss);
|
||||
receiver.register_dumper(&raw);
|
||||
receiver.receive();
|
||||
EXPECT_EQ(calls, (std::vector<int>{1, 0}));
|
||||
}
|
||||
|
||||
TEST(RemoteReceiverDumpers, OnlyTheFirstSecondaryRuns) {
|
||||
std::vector<int> calls;
|
||||
RecordingDumper raw(&calls, 0, true, true), extra_raw(&calls, 9, true, true);
|
||||
TestReceiver receiver;
|
||||
receiver.register_dumper(&raw);
|
||||
receiver.register_dumper(&extra_raw);
|
||||
receiver.receive();
|
||||
EXPECT_EQ(calls, (std::vector<int>{0}));
|
||||
}
|
||||
|
||||
// External components (vevor decoders) register themselves from their own setup()
|
||||
TEST(RemoteReceiverDumpers, RegisteringAfterConstructionAddsTheDumper) {
|
||||
std::vector<int> calls;
|
||||
RecordingDumper early(&calls, 1, false), late(&calls, 2, false);
|
||||
TestReceiver receiver;
|
||||
receiver.register_dumper(&early);
|
||||
receiver.receive();
|
||||
receiver.register_dumper(&late);
|
||||
receiver.receive();
|
||||
EXPECT_EQ(calls, (std::vector<int>{1, 1, 2}));
|
||||
}
|
||||
|
||||
} // namespace esphome::remote_base::testing
|
||||
@@ -386,7 +386,7 @@ def test_external_manifests_skip_other_components(
|
||||
components_dir = tmp_path / "components"
|
||||
get_sources().append(ExternalSource(components_dir=components_dir))
|
||||
_add_external_component(components_dir, "widget")
|
||||
_add_external_component(tmp_path / "builtin", "wifi")
|
||||
_add_external_component(tmp_path / "pkg" / "components", "wifi")
|
||||
CORE.loaded_integrations.update({"missing", "namespace"})
|
||||
sbom.loader._COMPONENT_CACHE["namespace"] = SimpleNamespace(
|
||||
module=SimpleNamespace()
|
||||
@@ -399,7 +399,9 @@ def test_external_manifests_skip_other_components(
|
||||
side_effect=sbom.loader._COMPONENT_CACHE.get,
|
||||
),
|
||||
patch.object(
|
||||
sbom.loader, "CORE_COMPONENTS_PATH", (tmp_path / "builtin").resolve()
|
||||
sbom.loader,
|
||||
"CORE_COMPONENTS_PATH",
|
||||
(tmp_path / "pkg" / "components").resolve(),
|
||||
),
|
||||
):
|
||||
manifests = sbom._external_manifests()
|
||||
@@ -703,3 +705,12 @@ def test_checkout_remote_detached(tmp_path: Path) -> None:
|
||||
"https://user:token@github.com/someone/esphome.git",
|
||||
)
|
||||
assert not any(call.args[0][1] == "config" for call in run.call_args_list)
|
||||
|
||||
|
||||
def test_external_manifests_esphome_core_is_built_in(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""The esphome domain resolves to esphome/core, which is still part of ESPHome."""
|
||||
CORE.loaded_integrations.add("esphome")
|
||||
assert sbom._external_manifests() == []
|
||||
assert "SBOM" not in caplog.text
|
||||
Reference in new issue
Block a user