Merge branch 'dev' into bt_congestion_warn_once

This commit is contained in:
J. Nick Koston
2026-08-21 16:16:06 -05:00
committed by GitHub
59 changed files with 521 additions and 163 deletions
+1 -1
View File
@@ -22,7 +22,7 @@ RUN \
-r /requirements.txt
# Install the ESPHome Device Builder dashboard.
RUN uv pip install --no-cache-dir esphome-device-builder==1.12.2
RUN uv pip install --no-cache-dir esphome-device-builder==1.12.3
RUN \
platformio settings set enable_telemetry No \
+16 -3
View File
@@ -17,6 +17,9 @@ from esphome.const import (
UNIT_OHM,
UNIT_PARTS_PER_BILLION,
)
from esphome.core import ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
CONF_RESISTANCE = "resistance"
@@ -62,7 +65,7 @@ CONFIG_SCHEMA = (
FINAL_VALIDATE_SCHEMA = i2c.final_validate_device_schema("ags10", max_frequency="15khz")
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
await i2c.register_i2c_device(var, config)
@@ -94,7 +97,12 @@ AGS10_NEW_I2C_ADDRESS_SCHEMA = cv.maybe_simple_value(
AGS10_NEW_I2C_ADDRESS_SCHEMA,
synchronous=True,
)
async def ags10newi2caddress_to_code(config, action_id, template_arg, args):
async def ags10newi2caddress_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID])
address = await cg.templatable(config[CONF_ADDRESS], args, cg.uint8)
@@ -126,7 +134,12 @@ AGS10_SET_ZERO_POINT_SCHEMA = cv.Schema(
AGS10_SET_ZERO_POINT_SCHEMA,
synchronous=True,
)
async def ags10setzeropoint_to_code(config, action_id, template_arg, args):
async def ags10setzeropoint_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID])
mode = await cg.templatable(
+16 -3
View File
@@ -4,6 +4,9 @@ import esphome.codegen as cg
from esphome.components import i2c
import esphome.config_validation as cv
from esphome.const import CONF_FREQUENCY, CONF_ID
from esphome.core import ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
CODEOWNERS = ["@X-Ryl669"]
DEPENDENCIES = ["i2c"]
@@ -70,7 +73,7 @@ CONFIG_SCHEMA = cv.All(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
await i2c.register_i2c_device(var, config)
@@ -91,7 +94,12 @@ AT581XSettingsAction = at581x_ns.class_("AT581XSettingsAction", automation.Actio
),
synchronous=True,
)
async def at581x_reset_to_code(config, action_id, template_arg, args):
async def at581x_reset_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID])
@@ -163,7 +171,12 @@ RADAR_SETTINGS_SCHEMA = cv.Schema(
RADAR_SETTINGS_SCHEMA,
synchronous=True,
)
async def at581x_settings_to_code(config, action_id, template_arg, args):
async def at581x_settings_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID])
+2 -1
View File
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import switch
import esphome.config_validation as cv
from esphome.const import DEVICE_CLASS_SWITCH, ICON_WIFI
from esphome.types import ConfigType
from .. import CONF_AT581X_ID, AT581XComponent, at581x_ns
@@ -22,7 +23,7 @@ CONFIG_SCHEMA = switch.switch_schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
at581x_component = await cg.get_variable(config[CONF_AT581X_ID])
s = await switch.new_switch(config)
await cg.register_parented(s, config[CONF_AT581X_ID])
+14 -6
View File
@@ -1,10 +1,13 @@
import re
from typing import Any
from esphome import automation
import esphome.codegen as cg
import esphome.config_validation as cv
from esphome.const import CONF_DATA, CONF_ID, CONF_TRIGGER_ID
from esphome.core import CORE, ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
CODEOWNERS = ["@mvturnho", "@danielschramm"]
IS_PLATFORM_COMPONENT = True
@@ -18,7 +21,7 @@ CONF_BIT_RATE = "bit_rate"
CONF_ON_FRAME = "on_frame"
def validate_id(config):
def validate_id(config: ConfigType) -> ConfigType:
if CONF_CAN_ID in config:
can_id = config[CONF_CAN_ID]
id_ext = config[CONF_USE_EXTENDED_ID]
@@ -27,7 +30,7 @@ def validate_id(config):
return config
def validate_raw_data(value):
def validate_raw_data(value: Any) -> bytes | list:
if isinstance(value, str):
return value.encode("utf-8")
if isinstance(value, list):
@@ -71,7 +74,7 @@ CAN_SPEEDS = {
}
def get_rate(value):
def get_rate(value: str) -> int:
match = re.match(r"(\d+)(?:K(\d+)?)?BPS", value, re.IGNORECASE)
if not match:
raise ValueError(f"Invalid rate format: {value}")
@@ -103,7 +106,7 @@ CANBUS_SCHEMA = cv.Schema(
CANBUS_SCHEMA.add_extra(validate_id)
async def setup_canbus_core_(var, config):
async def setup_canbus_core_(var: MockObj, config: ConfigType) -> None:
await cg.register_component(var, config)
cg.add(var.set_can_id([config[CONF_CAN_ID]]))
cg.add(var.set_use_extended_id([config[CONF_USE_EXTENDED_ID]]))
@@ -134,7 +137,7 @@ async def setup_canbus_core_(var, config):
)
async def register_canbus(var, config):
async def register_canbus(var: MockObj, config: ConfigType) -> None:
if not CORE.has_id(config[CONF_ID]):
var = cg.new_Pvariable(config[CONF_ID], var)
await setup_canbus_core_(var, config)
@@ -157,7 +160,12 @@ async def register_canbus(var, config):
),
synchronous=True,
)
async def canbus_action_to_code(config, action_id, template_arg, args):
async def canbus_action_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_CANBUS_ID])
+2 -1
View File
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import uart
import esphome.config_validation as cv
from esphome.const import CONF_ADDRESS, CONF_ID
from esphome.types import ConfigType
CODEOWNERS = ["@s1lvi0"]
MULTI_CONF = True
@@ -26,7 +27,7 @@ CONFIG_SCHEMA = (
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
await uart.register_uart_device(var, config)
+4 -2
View File
@@ -1,6 +1,8 @@
import esphome.codegen as cg
from esphome.components import binary_sensor
import esphome.config_validation as cv
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
from . import CONF_BMS_DALY_ID, DalyBmsComponent
@@ -27,13 +29,13 @@ CONFIG_SCHEMA = cv.All(
)
async def setup_conf(config, key, hub):
async def setup_conf(config: ConfigType, key: str, hub: MockObj) -> None:
if sensor_config := config.get(key):
var = await binary_sensor.new_binary_sensor(sensor_config)
cg.add(getattr(hub, f"set_{key}_binary_sensor")(var))
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_BMS_DALY_ID])
for key in TYPES:
await setup_conf(config, key, hub)
+4 -2
View File
@@ -23,6 +23,8 @@ from esphome.const import (
UNIT_PERCENT,
UNIT_VOLT,
)
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
from . import CONF_BMS_DALY_ID, DalyBmsComponent
@@ -222,13 +224,13 @@ CONFIG_SCHEMA = cv.All(
)
async def setup_conf(config, key, hub):
async def setup_conf(config: ConfigType, key: str, hub: MockObj) -> None:
if sensor_config := config.get(key):
sens = await sensor.new_sensor(sensor_config)
cg.add(getattr(hub, f"set_{key}_sensor")(sens))
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_BMS_DALY_ID])
for key in TYPES:
await setup_conf(config, key, hub)
+4 -2
View File
@@ -2,6 +2,8 @@ import esphome.codegen as cg
from esphome.components import text_sensor
import esphome.config_validation as cv
from esphome.const import CONF_STATUS
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
from . import CONF_BMS_DALY_ID, DalyBmsComponent
@@ -23,13 +25,13 @@ CONFIG_SCHEMA = cv.All(
)
async def setup_conf(config, key, hub):
async def setup_conf(config: ConfigType, key: str, hub: MockObj) -> None:
if sensor_config := config.get(key):
sens = await text_sensor.new_text_sensor(sensor_config)
cg.add(getattr(hub, f"set_{key}_text_sensor")(sens))
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_BMS_DALY_ID])
for key in TYPES:
await setup_conf(config, key, hub)
+16 -5
View File
@@ -38,7 +38,8 @@ from esphome.const import (
PLATFORM_NRF52,
PlatformFramework,
)
from esphome.core import CORE
from esphome.core import CORE, ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
WAKEUP_PINS = {
@@ -174,7 +175,7 @@ def validate_config(config: ConfigType) -> ConfigType:
return config
def _validate_ex1_wakeup_mode(value):
def _validate_ex1_wakeup_mode(value: str) -> str:
if value == "ALL_LOW":
esp32.only_on_variant(supported=[VARIANT_ESP32], msg_prefix="ALL_LOW")(value)
if value == "ANY_LOW":
@@ -345,7 +346,7 @@ CONFIG_SCHEMA = cv.All(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
@@ -458,7 +459,12 @@ DEEP_SLEEP_ENTER_SCHEMA = cv.All(
DEEP_SLEEP_ENTER_SCHEMA,
synchronous=True,
)
async def deep_sleep_enter_to_code(config, action_id, template_arg, args):
async def deep_sleep_enter_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren)
if CONF_SLEEP_DURATION in config:
@@ -487,7 +493,12 @@ async def deep_sleep_enter_to_code(config, action_id, template_arg, args):
automation.maybe_simple_id(DEEP_SLEEP_ACTION_SCHEMA),
synchronous=True,
)
async def deep_sleep_action_to_code(config, action_id, template_arg, args):
async def deep_sleep_action_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID])
return var
+16 -3
View File
@@ -3,6 +3,9 @@ import esphome.codegen as cg
from esphome.components import i2c, time
import esphome.config_validation as cv
from esphome.const import CONF_ID
from esphome.core import ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
CODEOWNERS = ["@badbadc0ffee"]
DEPENDENCIES = ["i2c"]
@@ -29,7 +32,12 @@ CONFIG_SCHEMA = time.TIME_SCHEMA.extend(
),
synchronous=True,
)
async def ds1307_write_time_to_code(config, action_id, template_arg, args):
async def ds1307_write_time_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID])
return var
@@ -45,13 +53,18 @@ async def ds1307_write_time_to_code(config, action_id, template_arg, args):
),
synchronous=True,
)
async def ds1307_read_time_to_code(config, action_id, template_arg, args):
async def ds1307_read_time_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID])
return var
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
+37 -26
View File
@@ -68,6 +68,7 @@ PATTERN_CONFIGS = {
"PULSE": {
CONF_UNIT_OF_MEASUREMENT: UNIT_PULSES,
CONF_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
CONF_STATE_CLASS: STATE_CLASS_TOTAL_INCREASING,
CONF_ACCURACY_DECIMALS: 0,
},
"PF": {
@@ -78,12 +79,13 @@ PATTERN_CONFIGS = {
},
}
# Create a base schema that's flexible for any tag
BASE_SCHEMA = sensor.sensor_schema(
EmonTxSensor,
state_class=STATE_CLASS_MEASUREMENT,
accuracy_decimals=0,
).extend(
# BASE_SCHEMA intentionally omits state_class and accuracy_decimals defaults.
# Passing them to sensor_schema() would register them via cv.Optional(key, default=...),
# making them always present in the validated config dict and preventing
# apply_tag_defaults from overriding them with the correct per-prefix values.
# They are injected by apply_tag_defaults below, after running through
# sensor.validate_state_class() so the value is code-generation-ready.
BASE_SCHEMA = sensor.sensor_schema(EmonTxSensor).extend(
{
cv.GenerateID(CONF_EMONTX_ID): cv.use_id(EmonTx),
cv.Required(CONF_TAG_NAME): cv.string,
@@ -91,34 +93,43 @@ BASE_SCHEMA = sensor.sensor_schema(
)
def _apply_defaults(config: ConfigType, defaults: dict) -> None:
"""Inject defaults into config, skipping keys already set by the user.
state_class values are run through validate_state_class so they are
code-generation-ready, matching what sensor_schema() would normally do."""
for key, value in defaults.items():
if key not in config:
if key == CONF_STATE_CLASS:
value = sensor.validate_state_class(value)
config[key] = value
def apply_tag_defaults(config: ConfigType) -> ConfigType:
"""Apply defaults based on tag prefix if applicable, but don't restrict any tags."""
tag = config[CONF_TAG_NAME]
# Skip if tag is too short
if len(tag) < 2:
return config
if len(tag) >= 2:
tag_upper = tag.upper()
# Check if this tag starts with a known prefix
tag_upper = tag.upper()
for pattern, pattern_config in PATTERN_CONFIGS.items():
if tag_upper.startswith(pattern):
_apply_defaults(config, pattern_config)
return config
for pattern, pattern_config in PATTERN_CONFIGS.items():
if tag_upper.startswith(pattern):
# Apply pattern defaults if not overridden by user
for key, value in pattern_config.items():
if key not in config:
config[key] = value
# Only apply defaults for known prefixes with numeric indices (e.g. E1, V2, T3)
prefix = tag_upper[0]
if prefix in SENSOR_CONFIGS and tag[1:].isdigit():
_apply_defaults(config, SENSOR_CONFIGS[prefix])
return config
# Only apply defaults for known prefixes with numeric indices
prefix = tag_upper[0]
if prefix in SENSOR_CONFIGS and len(tag) > 1 and tag[1:].isdigit():
# Apply defaults for known tag types, but only if not overridden by user
defaults = SENSOR_CONFIGS[prefix]
for key, value in defaults.items():
if key not in config:
config[key] = value
# Fall back to generic defaults for tags with no known prefix
_apply_defaults(
config,
{
CONF_STATE_CLASS: STATE_CLASS_MEASUREMENT,
CONF_ACCURACY_DECIMALS: 0,
},
)
return config
@@ -38,7 +38,8 @@ from esphome.const import (
CONF_SERVICE_UUID,
CONF_TRIGGER_ID,
)
from esphome.core import CORE, CoroPriority, TimePeriod, coroutine_with_priority
from esphome.core import CORE, ID, CoroPriority, TimePeriod, coroutine_with_priority
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.enum import StrEnum
from esphome.types import ConfigType
@@ -262,7 +263,7 @@ ESP_BLE_DEVICE_SCHEMA = cv.Schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
# Register the loggers this component needs
esp32_ble.register_bt_logger(BTLoggers.BLE_SCAN)
@@ -360,7 +361,7 @@ async def to_code(config):
# chance to call register_ble_tracker and register_client before the list is checked
# and added to the global defines list.
@coroutine_with_priority(CoroPriority.FINAL)
async def _add_ble_features():
async def _add_ble_features() -> None:
# Add feature-specific defines based on what's needed
required_features = _get_required_features()
# Sensors registered through the neutral ble_device_base path (BLEHub) need
@@ -389,8 +390,11 @@ ESP32_BLE_START_SCAN_ACTION_SCHEMA = cv.Schema(
synchronous=True,
)
async def esp32_ble_tracker_start_scan_action_to_code(
config, action_id, template_arg, args
):
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren)
template_ = await cg.templatable(config[CONF_CONTINUOUS], args, cg.bool_)
@@ -414,8 +418,11 @@ ESP32_BLE_STOP_SCAN_ACTION_SCHEMA = automation.maybe_simple_id(
synchronous=True,
)
async def esp32_ble_tracker_stop_scan_action_to_code(
config, action_id, template_arg, args
):
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID])
return var
@@ -135,6 +135,10 @@ void Esp32HostedUpdate::setup() {
// Publish state
this->status_clear_error();
this->publish_state();
// Defer so the automation runs on the main loop after setup, not during App.setup()
if (this->state_ == update::UPDATE_STATE_AVAILABLE && this->update_available_trigger_) {
this->defer([this]() { this->update_available_trigger_->trigger(this->update_info_); });
}
#else
// HTTP mode: check every 10s until network is ready (max 6 attempts)
// Only if update interval is > 1 minute to avoid redundant checks
@@ -185,6 +189,8 @@ void Esp32HostedUpdate::check() {
return;
}
const bool was_available = this->state_ == update::UPDATE_STATE_AVAILABLE;
// Compare versions
if (this->update_info_.latest_version.empty() ||
this->update_info_.latest_version == this->update_info_.current_version) {
@@ -197,6 +203,9 @@ void Esp32HostedUpdate::check() {
this->update_info_.progress = 0.0f;
this->status_clear_error();
this->publish_state();
if (this->state_ == update::UPDATE_STATE_AVAILABLE && !was_available && this->update_available_trigger_) {
this->update_available_trigger_->trigger(this->update_info_);
}
#endif
}
@@ -129,14 +129,17 @@ void on_data_received(const esp_now_recv_info_t *info, const uint8_t *data, int
ESPNowComponent::ESPNowComponent() { global_esp_now = this; }
void ESPNowComponent::dump_config() {
uint32_t version = 0;
esp_now_get_version(&version);
ESP_LOGCONFIG(TAG, "espnow:");
if (this->is_disabled()) {
ESP_LOGCONFIG(TAG, " Disabled");
// Only report driver details once enabled; with enable_on_boot: false the
// Wi-Fi driver is not initialized yet and esp_now_get_version() would crash,
// and after a failed enable_() the values would be meaningless.
if (this->state_ != ESPNOW_STATE_ENABLED) {
// OFF here means enable_() failed; the core logs the FAILED marker separately
ESP_LOGCONFIG(TAG, " %s", this->is_disabled() ? LOG_STR_LITERAL("Disabled") : LOG_STR_LITERAL("Not enabled"));
return;
}
uint32_t version = 0;
esp_now_get_version(&version);
char own_addr_buf[MAC_ADDRESS_PRETTY_BUFFER_SIZE];
format_mac_addr_upper(this->own_address_, own_addr_buf);
ESP_LOGCONFIG(TAG,
+17 -10
View File
@@ -48,10 +48,12 @@ from esphome.const import (
)
from esphome.core import (
CORE,
ID,
CoroPriority,
TimePeriodMilliseconds,
coroutine_with_priority,
)
from esphome.cpp_generator import MockObj, TemplateArgsType
import esphome.final_validate as fv
from esphome.types import ConfigType
@@ -276,7 +278,7 @@ def _validate_spi_interface(config: ConfigType) -> ConfigType:
return config
def _validate(config):
def _validate(config: ConfigType) -> ConfigType:
if CONF_USE_ADDRESS not in config:
if CONF_MANUAL_IP in config:
use_address = str(config[CONF_MANUAL_IP][CONF_STATIC_IP])
@@ -441,7 +443,7 @@ GENERIC_SCHEMA = cv.All(
)
def _spi_schema(default_clock: str = "26.67MHz", max_clock: int = int(80e6)):
def _spi_schema(default_clock: str = "26.67MHz", max_clock: int = int(80e6)) -> cv.All:
return cv.All(
BASE_SCHEMA.extend(
cv.Schema(
@@ -517,7 +519,7 @@ CONFIG_SCHEMA = cv.All(
)
def _final_validate_spi(config):
def _final_validate_spi(config: ConfigType) -> None:
if not CORE.is_esp32:
return # SPI interface validation is ESP32-only
if config[CONF_TYPE] not in SPI_ETHERNET_TYPES:
@@ -537,7 +539,7 @@ def _final_validate_spi(config):
)
def manual_ip(config):
def manual_ip(config: ConfigType) -> cg.StructInitializer:
return cg.StructInitializer(
ManualIP,
("static_ip", ip_address_literal(config[CONF_STATIC_IP])),
@@ -548,7 +550,7 @@ def manual_ip(config):
)
def phy_register(address: int, value: int, page: int):
def phy_register(address: int, value: int, page: int) -> cg.StructInitializer:
return cg.StructInitializer(
PHYRegister,
("address", address),
@@ -558,7 +560,7 @@ def phy_register(address: int, value: int, page: int):
@coroutine_with_priority(CoroPriority.COMMUNICATION)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
# Apply network priority before register_component (which emits the user's
@@ -610,7 +612,7 @@ async def to_code(config):
CORE.add_job(final_step)
async def _to_code_esp32(var: cg.Pvariable, config: ConfigType) -> None:
async def _to_code_esp32(var: cg.MockObj, config: ConfigType) -> None:
from esphome.components.esp32 import (
add_idf_component,
add_idf_sdkconfig_option,
@@ -698,7 +700,7 @@ async def _to_code_esp32(var: cg.Pvariable, config: ConfigType) -> None:
add_idf_component(name=component.name, ref=component.version)
async def _to_code_rp2040(var: cg.Pvariable, config: ConfigType) -> None:
async def _to_code_rp2040(var: cg.MockObj, config: ConfigType) -> None:
cg.add(var.set_clk_pin(config[CONF_CLK_PIN]))
cg.add(var.set_miso_pin(config[CONF_MISO_PIN]))
cg.add(var.set_mosi_pin(config[CONF_MOSI_PIN]))
@@ -793,7 +795,7 @@ FINAL_VALIDATE_SCHEMA = _final_validate
@coroutine_with_priority(CoroPriority.FINAL)
async def final_step():
async def final_step() -> None:
"""Final code generation step to configure optional Ethernet features."""
if ip_state_count := CORE.data.get(ETHERNET_IP_STATE_LISTENERS_KEY, 0):
cg.add_define("USE_ETHERNET_IP_STATE_LISTENERS")
@@ -845,7 +847,12 @@ def _filter_source_files() -> list[str]:
FILTER_SOURCE_FILES = _filter_source_files
async def _new_pvariable_to_code(config, id_, template_arg, args):
async def _new_pvariable_to_code(
config: ConfigType,
id_: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
return cg.new_Pvariable(id_, template_arg)
+19 -4
View File
@@ -1,3 +1,5 @@
from typing import Any
from esphome import automation
from esphome.automation import maybe_simple_id
import esphome.codegen as cg
@@ -16,6 +18,9 @@ from esphome.const import (
UNIT_CELSIUS,
UNIT_PERCENT,
)
from esphome.core import ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
DEPENDENCIES = ["i2c"]
@@ -62,7 +67,7 @@ CONFIG_SCHEMA = (
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
await i2c.register_i2c_device(var, config)
@@ -86,7 +91,7 @@ HDC302X_HEATER_POWER_MAP = {
}
def heater_power_value(value):
def heater_power_value(value: Any) -> cv.Lambda | int:
"""Accept enum names or raw uint16 values"""
if isinstance(value, cv.Lambda):
return value
@@ -119,7 +124,12 @@ HDC302X_HEATER_ON_ACTION_SCHEMA = maybe_simple_id(
HDC302X_HEATER_ON_ACTION_SCHEMA,
synchronous=True,
)
async def hdc302x_heater_on_to_code(config, action_id, template_arg, args):
async def hdc302x_heater_on_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID])
template_ = await cg.templatable(config[CONF_POWER], args, cg.uint16)
@@ -135,7 +145,12 @@ async def hdc302x_heater_on_to_code(config, action_id, template_arg, args):
HDC302X_ACTION_SCHEMA,
synchronous=True,
)
async def hdc302x_heater_off_to_code(config, action_id, template_arg, args):
async def hdc302x_heater_off_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID])
return var
+16 -3
View File
@@ -17,6 +17,9 @@ from esphome.const import (
UNIT_EMPTY,
UNIT_PERCENT,
)
from esphome.core import ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
DEPENDENCIES = ["i2c"]
@@ -63,7 +66,7 @@ CONFIG_SCHEMA = (
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
await i2c.register_i2c_device(var, config)
@@ -95,7 +98,12 @@ async def to_code(config):
),
synchronous=True,
)
async def set_heater_level_to_code(config, action_id, template_arg, args):
async def set_heater_level_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID])
level_ = await cg.templatable(config[CONF_LEVEL], args, cg.uint8)
@@ -115,7 +123,12 @@ async def set_heater_level_to_code(config, action_id, template_arg, args):
),
synchronous=True,
)
async def set_heater_to_code(config, action_id, template_arg, args):
async def set_heater_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID])
status_ = await cg.templatable(config[CONF_STATUS], args, cg.bool_)
+1 -1
View File
@@ -60,7 +60,7 @@ FINAL_VALIDATE_SCHEMA = uart.final_validate_device_schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
await uart.register_uart_device(var, config)
+2 -1
View File
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import binary_sensor
import esphome.config_validation as cv
from esphome.const import CONF_TARGET, DEVICE_CLASS_OCCUPANCY
from esphome.types import ConfigType
from . import LD6002BComponent
from .const import AREA_COUNT, CONF_LD6002B_ID, MAX_TARGETS
@@ -36,7 +37,7 @@ CONFIG_SCHEMA = (
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_LD6002B_ID])
if target_config := config.get(CONF_TARGET):
@@ -129,7 +129,7 @@ BUTTON_MAP = {
}
async def to_code(config):
async def to_code(config: ConfigType) -> None:
for key, button_type in BUTTON_MAP.items():
if button_config := config.get(key):
b = cg.new_Pvariable(button_config[CONF_ID], button_type)
@@ -136,7 +136,7 @@ def final_validate(config: ConfigType) -> None:
FINAL_VALIDATE_SCHEMA = final_validate
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_LD6002B_ID])
for key, number_type, setter, min_value, max_value, step in (
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import select
import esphome.config_validation as cv
from esphome.const import CONF_AREA_ID, CONF_SENSITIVITY, ENTITY_CATEGORY_CONFIG
from esphome.types import ConfigType
from .. import LD6002BComponent, ld6002b_ns
from ..const import CONF_INSTALLATION_MODE, CONF_LD6002B_ID, CONF_TRIGGER_SPEED
@@ -64,7 +65,7 @@ SELECT_MAP = (
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_LD6002B_ID])
for key, select_type, setter, options in SELECT_MAP:
+2 -1
View File
@@ -9,6 +9,7 @@ from esphome.const import (
STATE_CLASS_MEASUREMENT,
UNIT_METER,
)
from esphome.types import ConfigType
from . import LD6002BComponent
from .const import (
@@ -150,7 +151,7 @@ CONFIG_SCHEMA = (
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_LD6002B_ID])
if target_count_config := config.get(CONF_TARGET_COUNT):
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import switch
import esphome.config_validation as cv
from esphome.const import DEVICE_CLASS_SWITCH, ENTITY_CATEGORY_CONFIG
from esphome.types import ConfigType
from .. import LD6002BComponent, ld6002b_ns
from ..const import (
@@ -46,7 +47,7 @@ CONFIG_SCHEMA = cv.Schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_LD6002B_ID])
for key, switch_type, setter in (
+2 -1
View File
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import text_sensor
import esphome.config_validation as cv
from esphome.const import ENTITY_CATEGORY_DIAGNOSTIC
from esphome.types import ConfigType
from . import LD6002BComponent
from .const import CONF_LD6002B_ID, CONF_OTA_VERSION, CONF_WORK_MODE
@@ -21,7 +22,7 @@ CONFIG_SCHEMA = cv.Schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_LD6002B_ID])
if work_mode_config := config.get(CONF_WORK_MODE):
sens = await text_sensor.new_text_sensor(work_mode_config)
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import i2c
import esphome.config_validation as cv
from esphome.const import CONF_ID
from esphome.types import ConfigType
DEPENDENCIES = ["i2c"]
CODEOWNERS = ["@rnauber"]
@@ -26,7 +27,7 @@ CONFIG_SCHEMA = cv.Schema(
).extend(i2c.i2c_device_schema(0x43))
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
await i2c.register_i2c_device(var, config)
@@ -1,6 +1,7 @@
import esphome.codegen as cg
from esphome.components import binary_sensor
import esphome.config_validation as cv
from esphome.types import ConfigType
from .. import CONF_M5STACK_8ANGLE_ID, M5Stack8AngleComponent, m5stack_8angle_ns
@@ -22,7 +23,7 @@ CONFIG_SCHEMA = cv.All(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_M5STACK_8ANGLE_ID])
sens = await binary_sensor.new_binary_sensor(config)
cg.add(sens.set_parent(hub))
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import light
import esphome.config_validation as cv
from esphome.const import CONF_OUTPUT_ID
from esphome.types import ConfigType
from .. import CONF_M5STACK_8ANGLE_ID, M5Stack8AngleComponent, m5stack_8angle_ns
@@ -21,7 +22,7 @@ CONFIG_SCHEMA = cv.All(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
hub = await cg.get_variable(config[CONF_M5STACK_8ANGLE_ID])
lights = cg.new_Pvariable(config[CONF_OUTPUT_ID])
await light.register_light(lights, config)
@@ -8,6 +8,7 @@ from esphome.const import (
ICON_ROTATE_RIGHT,
STATE_CLASS_MEASUREMENT,
)
from esphome.types import ConfigType
from .. import (
CONF_M5STACK_8ANGLE_ID,
@@ -55,7 +56,7 @@ CONFIG_SCHEMA = cv.All(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = await sensor.new_sensor(config)
await cg.register_component(var, config)
await cg.register_parented(var, config[CONF_M5STACK_8ANGLE_ID])
+12 -8
View File
@@ -8,8 +8,10 @@ import esphome.codegen as cg
from esphome.components import uart
import esphome.config_validation as cv
from esphome.const import CONF_ADDRESS, CONF_DISABLE_CRC, CONF_FLOW_CONTROL_PIN, CONF_ID
from esphome.cpp_generator import MockObj
from esphome.cpp_helpers import gpio_pin_expression
import esphome.final_validate as fv
from esphome.types import ConfigType
_LOGGER = logging.getLogger(__name__)
@@ -84,7 +86,7 @@ CONFIG_SCHEMA = cv.typed_schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
cg.add_global(modbus_ns.using)
var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config)
@@ -112,7 +114,9 @@ def _validate_server_address(value: Any) -> int:
return address
def modbus_device_schema(default_address, role: Literal["client", "server"] = "client"):
def modbus_device_schema(
default_address: int | None, role: Literal["client", "server"] = "client"
) -> cv.Schema:
hub_type = ModbusClient if role == "client" else ModbusServer
address_validator = _validate_server_address if role == "server" else cv.hex_uint8_t
schema = {
@@ -127,14 +131,14 @@ def modbus_device_schema(default_address, role: Literal["client", "server"] = "c
def final_validate_modbus_device(
name: str, *, role: Literal["server", "client"] | None = None
):
def validate_role(value):
) -> cv.Schema:
def validate_role(value: str) -> str:
assert role in MODBUS_ROLES
if value != role:
raise cv.Invalid(f"Component {name} requires role to be {role}")
return value
def validate_hub(hub_config):
def validate_hub(hub_config: ConfigType) -> ConfigType:
hub_schema = {}
if role is not None:
hub_schema[cv.Required(CONF_ROLE)] = validate_role
@@ -147,19 +151,19 @@ def final_validate_modbus_device(
)
async def register_modbus_client_device(var, config):
async def register_modbus_client_device(var: MockObj, config: ConfigType) -> None:
parent = await cg.get_variable(config[CONF_MODBUS_ID])
cg.add(var.set_parent(parent))
cg.add(var.set_address(config[CONF_ADDRESS]))
async def register_modbus_server_device(var, config):
async def register_modbus_server_device(var: MockObj, config: ConfigType) -> None:
parent = await cg.get_variable(config[CONF_MODBUS_ID])
cg.add(var.set_address(config[CONF_ADDRESS]))
cg.add(parent.register_device(var))
async def register_modbus_device(var, config):
async def register_modbus_device(var: MockObj, config: ConfigType) -> None:
# Remove before 2026.12.0
_LOGGER.warning(
"'register_modbus_device' is deprecated, use 'register_modbus_client_device' "
+17 -8
View File
@@ -1,3 +1,5 @@
from typing import Any
from esphome import automation
import esphome.codegen as cg
from esphome.components.esp32 import (
@@ -31,10 +33,12 @@ from esphome.const import (
)
from esphome.core import (
CORE,
ID,
CoroPriority,
TimePeriodMilliseconds,
coroutine_with_priority,
)
from esphome.cpp_generator import MockObj, TemplateArgsType
import esphome.final_validate as fv
from esphome.types import ConfigType
@@ -76,7 +80,7 @@ CONF_DEVICE_TYPES = [
]
def _validate_txpower(value):
def _validate_txpower(value: Any) -> int | float:
if CORE.is_esp32:
variant = get_esp32_variant()
@@ -90,7 +94,7 @@ def _validate_txpower(value):
return value # Unsupported, fail later with clear error
def set_sdkconfig_options(config):
def set_sdkconfig_options(config: ConfigType) -> None:
# and expose options for using SPI/UART RCPs
add_idf_sdkconfig_option("CONFIG_IEEE802154_ENABLED", True)
add_idf_sdkconfig_option("CONFIG_OPENTHREAD_RADIO_NATIVE", True)
@@ -180,7 +184,7 @@ def _validate(config: ConfigType) -> ConfigType:
return config
def _require_vfs_select(config):
def _require_vfs_select(config: ConfigType) -> ConfigType:
"""Register VFS select requirement during config validation."""
# OpenThread uses esp_vfs_eventfd which requires VFS select support (ESP32 only)
if CORE.is_esp32:
@@ -188,7 +192,7 @@ def _require_vfs_select(config):
return config
def _validate_platform(config):
def _validate_platform(config: ConfigType) -> ConfigType:
if CORE.using_zephyr:
return config
return only_on_variant(
@@ -203,7 +207,7 @@ def _validate_platform(config):
)(config)
def _validate_tlv_hex(value):
def _validate_tlv_hex(value: Any) -> str:
s = cv.string_strict(value)
if len(s) % 2 != 0:
raise cv.Invalid("TLV must have an even number of hex characters")
@@ -242,7 +246,7 @@ CONFIG_SCHEMA = cv.All(
)
def _final_validate(_):
def _final_validate(_: ConfigType) -> None:
full_config = fv.full_config.get()
network_config = full_config.get("network", {})
if not network_config.get(CONF_ENABLE_IPV6, False):
@@ -274,7 +278,7 @@ FILTER_SOURCE_FILES = filter_source_files_from_platform(
@coroutine_with_priority(CoroPriority.COMMUNICATION)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
# Re-enable openthread IDF component (excluded by default)
if CORE.is_esp32:
include_builtin_idf_component("openthread")
@@ -339,7 +343,12 @@ POLL_PERIOD_ACTION_SCHEMA = automation.maybe_conf(
POLL_PERIOD_ACTION_SCHEMA,
synchronous=True,
)
async def openthread_poll_period_action_to_code(config, action_id, template_arg, args):
async def openthread_poll_period_action_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren)
template_ = await cg.templatable(config[CONF_POLL_PERIOD], args, cg.uint32)
+15 -6
View File
@@ -1,3 +1,5 @@
from typing import Any
from esphome import automation, pins
import esphome.codegen as cg
from esphome.components import sensor
@@ -19,7 +21,9 @@ from esphome.const import (
UNIT_PULSES,
UNIT_PULSES_PER_MINUTE,
)
from esphome.core import CORE
from esphome.core import CORE, ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
CONF_USE_PCNT = "use_pcnt"
@@ -42,7 +46,7 @@ SetTotalPulsesAction = pulse_counter_ns.class_(
)
def validate_internal_filter(value):
def validate_internal_filter(value: ConfigType) -> ConfigType:
use_pcnt = value.get(CONF_USE_PCNT)
if CORE.is_esp8266 and use_pcnt:
raise cv.Invalid(
@@ -63,7 +67,7 @@ def validate_internal_filter(value):
return value
def validate_pulse_counter_pin(value):
def validate_pulse_counter_pin(value: Any) -> ConfigType:
value = pins.internal_gpio_input_pin_schema(value)
if CORE.is_esp8266 and value[CONF_NUMBER] >= 16:
raise cv.Invalid(
@@ -72,7 +76,7 @@ def validate_pulse_counter_pin(value):
return value
def validate_count_mode(value):
def validate_count_mode(value: ConfigType) -> ConfigType:
rising_edge = value[CONF_RISING_EDGE]
falling_edge = value[CONF_FALLING_EDGE]
if rising_edge == "DISABLE" and falling_edge == "DISABLE":
@@ -126,7 +130,7 @@ CONFIG_SCHEMA = cv.All(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
use_pcnt = config.get(CONF_USE_PCNT)
if CORE.is_esp32 and use_pcnt:
include_builtin_idf_component("esp_driver_pcnt")
@@ -157,7 +161,12 @@ async def to_code(config):
),
synchronous=True,
)
async def set_total_action_to_code(config, action_id, template_arg, args):
async def set_total_action_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren)
template_ = await cg.templatable(config[CONF_VALUE], args, cg.uint32)
+15 -6
View File
@@ -1,3 +1,5 @@
from typing import Any
from esphome import automation, pins
import esphome.codegen as cg
from esphome.components import sensor
@@ -17,7 +19,9 @@ from esphome.const import (
UNIT_PULSES,
UNIT_PULSES_PER_MINUTE,
)
from esphome.core import CORE
from esphome.core import CORE, ID, TimePeriodMicroseconds
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
CODEOWNERS = ["@stevebaxter", "@cstaahl", "@TrentHouliston"]
@@ -37,18 +41,18 @@ FILTER_MODES = {
SetTotalPulsesAction = pulse_meter_ns.class_("SetTotalPulsesAction", automation.Action)
def validate_internal_filter(value):
def validate_internal_filter(value: Any) -> TimePeriodMicroseconds:
return cv.positive_time_period_microseconds(value)
def validate_timeout(value):
def validate_timeout(value: Any) -> TimePeriodMicroseconds:
value = cv.positive_time_period_microseconds(value)
if value.total_minutes > 70:
raise cv.Invalid("Maximum timeout is 70 minutes")
return value
def validate_pulse_meter_pin(value):
def validate_pulse_meter_pin(value: Any) -> ConfigType:
value = pins.internal_gpio_input_pin_schema(value)
if CORE.is_esp8266 and value[CONF_NUMBER] >= 16:
raise cv.Invalid(
@@ -81,7 +85,7 @@ CONFIG_SCHEMA = sensor.sensor_schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
var = await sensor.new_sensor(config)
await cg.register_component(var, config)
@@ -107,7 +111,12 @@ async def to_code(config):
),
synchronous=True,
)
async def set_total_action_to_code(config, action_id, template_arg, args):
async def set_total_action_to_code(
config: ConfigType,
action_id: ID,
template_arg: cg.TemplateArguments,
args: TemplateArgsType,
) -> MockObj:
paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren)
template_ = await cg.templatable(config[CONF_VALUE], args, cg.uint32)
+6 -5
View File
@@ -1,6 +1,7 @@
import hashlib
from pathlib import Path
import re
from typing import Any
from esphome import external_files, pins
import esphome.codegen as cg
@@ -66,7 +67,7 @@ KNOWN_FIRMWARE = {
}
def parse_firmware_version(value):
def parse_firmware_version(value: str) -> tuple[int, int]:
match = re.fullmatch(r"(\d+)\.(\d+)", value)
if match is None:
raise ValueError(f"Not a valid version number {value}")
@@ -154,7 +155,7 @@ def _extract_firmware_ref(entry: ConfigType) -> RemoteFile | None:
PREFETCH_FILES = external_files.single_stage_prefetch(_extract_firmware_ref)
def validate_firmware(value):
def validate_firmware(value: ConfigType) -> ConfigType:
config = value.copy()
if CONF_URL not in config:
try:
@@ -167,14 +168,14 @@ def validate_firmware(value):
return config
def validate_sha256(value):
def validate_sha256(value: Any) -> str:
value = cv.string(value)
if not re.fullmatch(r"[0-9a-fA-F]{64}", value):
raise ValueError(f"Not a valid SHA256 hex string: {value}")
return value
def validate_version(value):
def validate_version(value: str) -> str:
parse_firmware_version(value)
return value
@@ -231,7 +232,7 @@ FINAL_VALIDATE_SCHEMA = uart.final_validate_device_schema(
)
async def to_code(config):
async def to_code(config: ConfigType) -> None:
fw_hex = get_firmware(config[CONF_FIRMWARE])
fw_major, fw_minor = parse_firmware_version(config[CONF_FIRMWARE][CONF_VERSION])
+1 -1
View File
@@ -12,7 +12,7 @@ pyserial==3.5
platformio==6.1.19
esptool==5.3.1
click==8.3.3
aioesphomeapi==45.12.0
aioesphomeapi==45.13.1
aiohappyeyeballs==2.7.1 # Happy Eyeballs for requests downloads; already pulled in by aioesphomeapi
zeroconf==0.150.0
puremagic==2.2.0
@@ -0,0 +1,100 @@
"""Tests for emontx sensor tag defaults."""
import pytest
from esphome.components import sensor
from esphome.components.emontx.sensor import CONFIG_SCHEMA, apply_tag_defaults
from esphome.const import (
CONF_ACCURACY_DECIMALS,
CONF_STATE_CLASS,
STATE_CLASS_MEASUREMENT,
STATE_CLASS_TOTAL_INCREASING,
)
def _resolve_via_config_schema(tag: str) -> dict:
"""Run a minimal config through the real CONFIG_SCHEMA pipeline, the
same path a user's YAML goes through."""
return CONFIG_SCHEMA(
{"tag_name": tag, "emontx_id": "my_emontx", "name": f"{tag} sensor"}
)
def test_config_schema_applies_tag_default_state_class():
"""If sensor_schema(state_class=...) is reintroduced, the schema-level
default wins over apply_tag_defaults' per-prefix value, and E1 would
resolve to measurement instead of total_increasing. Driving the real
CONFIG_SCHEMA (not just apply_tag_defaults) catches that, since
sensor_schema() runs before apply_tag_defaults in the cv.All() chain.
"""
result = _resolve_via_config_schema("E1")
assert result[CONF_STATE_CLASS] == sensor.validate_state_class(
STATE_CLASS_TOTAL_INCREASING
)
def test_config_schema_applies_tag_default_accuracy_decimals():
"""Same root cause as the state_class regression: reintroducing
sensor_schema(accuracy_decimals=...) would make V1 resolve to the
schema-level default instead of the prefix-specific value of 2.
"""
result = _resolve_via_config_schema("V1")
assert result[CONF_ACCURACY_DECIMALS] == 2
def _make_config(tag: str) -> dict:
"""Minimal config dict with only tag_name set — no overrides."""
return {"tag_name": tag}
@pytest.mark.parametrize(
("tag", "expected_state_class", "expected_decimals"),
[
# Known numeric-index prefixes
("E1", STATE_CLASS_TOTAL_INCREASING, 0),
("E12", STATE_CLASS_TOTAL_INCREASING, 0),
("P1", STATE_CLASS_MEASUREMENT, 0),
("V1", STATE_CLASS_MEASUREMENT, 2),
("I1", STATE_CLASS_MEASUREMENT, 2),
("T1", STATE_CLASS_MEASUREMENT, 2),
# Known patterns
("PULSE1", STATE_CLASS_TOTAL_INCREASING, 0),
("PULSE12", STATE_CLASS_TOTAL_INCREASING, 0),
("PF1", STATE_CLASS_MEASUREMENT, 2),
# Unknown / free-form tags fall back to generic defaults
("CUSTOM1", STATE_CLASS_MEASUREMENT, 0),
("X", STATE_CLASS_MEASUREMENT, 0),
],
)
def test_apply_tag_defaults(tag, expected_state_class, expected_decimals):
"""apply_tag_defaults must inject the correct state_class and accuracy_decimals
for each tag type when no user overrides are present."""
config = _make_config(tag)
result = apply_tag_defaults(config)
assert result[CONF_STATE_CLASS] == sensor.validate_state_class(expected_state_class)
assert result[CONF_ACCURACY_DECIMALS] == expected_decimals
@pytest.mark.parametrize(
("tag", "user_state_class", "user_decimals"),
[
# User overrides must not be clobbered by defaults
("E1", STATE_CLASS_MEASUREMENT, 3),
("PULSE1", STATE_CLASS_MEASUREMENT, 1),
("V1", STATE_CLASS_TOTAL_INCREASING, 0),
("CUSTOM1", STATE_CLASS_TOTAL_INCREASING, 4),
],
)
def test_apply_tag_defaults_respects_user_overrides(
tag, user_state_class, user_decimals
):
"""apply_tag_defaults must not overwrite values already set by the user."""
config = _make_config(tag)
config[CONF_STATE_CLASS] = sensor.validate_state_class(user_state_class)
config[CONF_ACCURACY_DECIMALS] = user_decimals
result = apply_tag_defaults(config)
assert result[CONF_STATE_CLASS] == sensor.validate_state_class(user_state_class)
assert result[CONF_ACCURACY_DECIMALS] == user_decimals
+1 -2
View File
@@ -1,4 +1,3 @@
packages:
uart_115200: !include ../../test_build_components/common/uart_115200/esp32-idf.yaml
<<: !include common.yaml
emontx: !include common.yaml
@@ -1,4 +1,3 @@
packages:
uart_115200: !include ../../test_build_components/common/uart_115200/esp8266-ard.yaml
<<: !include common.yaml
emontx: !include common.yaml
+1 -2
View File
@@ -1,4 +1,3 @@
packages:
uart_115200: !include ../../test_build_components/common/uart_115200/rp2040-ard.yaml
<<: !include common.yaml
emontx: !include common.yaml
@@ -0,0 +1,73 @@
packages:
uart_115200: !include ../../test_build_components/common/uart_115200/esp32-idf.yaml
emontx: !include common.yaml
# Validate that each sensor type gets the correct default state_class,
# unit_of_measurement, device_class, and accuracy_decimals when NO overrides
# are provided. The values are intentionally omitted so apply_tag_defaults is
# exercised, not the user-override path.
sensor:
# Energy sensor (E prefix): expects state_class=total_increasing, unit=Wh,
# device_class=energy, accuracy_decimals=0
- platform: emontx
tag_name: E1
name: Energy 1
emontx_id: test_emontx
# Power sensor (P prefix): expects state_class=measurement, unit=W,
# device_class=power, accuracy_decimals=0
- platform: emontx
tag_name: P1
name: Power 1
emontx_id: test_emontx
# Voltage sensor (V prefix): expects state_class=measurement, unit=V,
# device_class=voltage, accuracy_decimals=2
- platform: emontx
tag_name: V1
name: Voltage 1
emontx_id: test_emontx
# Current sensor (I prefix): expects state_class=measurement, unit=A,
# device_class=current, accuracy_decimals=2
- platform: emontx
tag_name: I1
name: Current 1
emontx_id: test_emontx
# Temperature sensor (T prefix): expects state_class=measurement, unit=°C,
# device_class=temperature, accuracy_decimals=2
- platform: emontx
tag_name: T1
name: Temperature 1
emontx_id: test_emontx
# Pulse sensor (PULSE pattern): expects state_class=total_increasing,
# unit=pulses, device_class=energy, accuracy_decimals=0
- platform: emontx
tag_name: PULSE1
name: Pulse 1
emontx_id: test_emontx
# Power factor sensor (PF pattern): expects state_class=measurement,
# device_class=power_factor, accuracy_decimals=2
- platform: emontx
tag_name: PF1
name: Power Factor 1
emontx_id: test_emontx
# Unknown tag: no prefix match, falls back to state_class=measurement,
# accuracy_decimals=0
- platform: emontx
tag_name: CUSTOM1
name: Custom sensor
emontx_id: test_emontx
# User override: verify that explicit values are respected and not clobbered
- platform: emontx
tag_name: E2
name: Energy 2 (user override)
emontx_id: test_emontx
state_class: measurement
accuracy_decimals: 3
@@ -6,3 +6,6 @@ update:
type: embedded
path: $component_dir/test_firmware.bin
sha256: de2f256064a0af797747c2b97505dc0b9f3df0de4f489eac731c23ae9ca9cc31
on_update_available:
then:
- logger.log: "Coprocessor update available"
@@ -8,3 +8,6 @@ update:
type: http
source: https://esphome.github.io/esp-hosted-firmware/manifest/esp32c6.json
update_interval: 6h
on_update_available:
then:
- logger.log: "Coprocessor update available"
+1 -1
View File
@@ -1,6 +1,6 @@
# `sendspin.switch` action enables the controller role, so we use a standalone test
packages:
base: !include common.yaml
sendspin: !include common.yaml
wifi:
on_connect:
@@ -0,0 +1,5 @@
packages:
sendspin_hub: !include common-hub.yaml
ethernet:
type: OPENETH
@@ -0,0 +1,6 @@
psram:
mode: quad
sendspin:
id: sendspin_hub_id
task_stack_in_psram: true
@@ -1,4 +1,5 @@
<<: !include common.yaml
packages:
sendspin: !include common.yaml
media_player:
- platform: sendspin
@@ -1,4 +1,5 @@
<<: !include common.yaml
packages:
sendspin: !include common.yaml
media_source:
- platform: sendspin
+2 -1
View File
@@ -1,4 +1,5 @@
<<: !include common.yaml
packages:
sendspin: !include common.yaml
sensor:
- platform: sendspin
@@ -1,4 +1,5 @@
<<: !include common.yaml
packages:
sendspin: !include common.yaml
text_sensor:
- platform: sendspin
+3 -7
View File
@@ -1,9 +1,5 @@
packages:
sendspin_hub: !include common-hub.yaml
wifi:
ap:
psram:
mode: quad
sendspin:
id: sendspin_hub_id
task_stack_in_psram: true
@@ -1 +1,2 @@
<<: !include common-action.yaml
packages:
sendspin: !include common-action.yaml
@@ -1,9 +1,2 @@
ethernet:
type: OPENETH
psram:
mode: quad
sendspin:
id: sendspin_hub_id
task_stack_in_psram: true
packages:
sendspin: !include common-ethernet.yaml
@@ -1 +1,2 @@
<<: !include common-media_player.yaml
packages:
sendspin: !include common-media_player.yaml
@@ -1 +1,2 @@
<<: !include common-media_source.yaml
packages:
sendspin: !include common-media_source.yaml
@@ -1 +1,2 @@
<<: !include common-sensor.yaml
packages:
sendspin: !include common-sensor.yaml
@@ -1 +1,2 @@
<<: !include common-text_sensor.yaml
packages:
sendspin: !include common-text_sensor.yaml
@@ -1 +1,2 @@
<<: !include common.yaml
packages:
sendspin: !include common.yaml