[core] Add type annotations to component Python (6/11) (#18343)

This commit is contained in:
Jesse Hills
2026-08-21 11:11:52 -05:00
committed by GitHub
parent 409d74a48d
commit aa944456e0
31 changed files with 246 additions and 97 deletions
+16 -3
View File
@@ -17,6 +17,9 @@ from esphome.const import (
UNIT_OHM, UNIT_OHM,
UNIT_PARTS_PER_BILLION, UNIT_PARTS_PER_BILLION,
) )
from esphome.core import ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
CONF_RESISTANCE = "resistance" CONF_RESISTANCE = "resistance"
@@ -62,7 +65,7 @@ CONFIG_SCHEMA = (
FINAL_VALIDATE_SCHEMA = i2c.final_validate_device_schema("ags10", max_frequency="15khz") 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]) var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config) await cg.register_component(var, config)
await i2c.register_i2c_device(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, AGS10_NEW_I2C_ADDRESS_SCHEMA,
synchronous=True, 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) var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID]) await cg.register_parented(var, config[CONF_ID])
address = await cg.templatable(config[CONF_ADDRESS], args, cg.uint8) 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, AGS10_SET_ZERO_POINT_SCHEMA,
synchronous=True, 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) var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID]) await cg.register_parented(var, config[CONF_ID])
mode = await cg.templatable( mode = await cg.templatable(
+16 -3
View File
@@ -4,6 +4,9 @@ import esphome.codegen as cg
from esphome.components import i2c from esphome.components import i2c
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.const import CONF_FREQUENCY, CONF_ID 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"] CODEOWNERS = ["@X-Ryl669"]
DEPENDENCIES = ["i2c"] 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]) var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config) await cg.register_component(var, config)
await i2c.register_i2c_device(var, config) await i2c.register_i2c_device(var, config)
@@ -91,7 +94,12 @@ AT581XSettingsAction = at581x_ns.class_("AT581XSettingsAction", automation.Actio
), ),
synchronous=True, 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) var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID]) await cg.register_parented(var, config[CONF_ID])
@@ -163,7 +171,12 @@ RADAR_SETTINGS_SCHEMA = cv.Schema(
RADAR_SETTINGS_SCHEMA, RADAR_SETTINGS_SCHEMA,
synchronous=True, 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) var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID]) 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 from esphome.components import switch
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.const import DEVICE_CLASS_SWITCH, ICON_WIFI from esphome.const import DEVICE_CLASS_SWITCH, ICON_WIFI
from esphome.types import ConfigType
from .. import CONF_AT581X_ID, AT581XComponent, at581x_ns 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]) at581x_component = await cg.get_variable(config[CONF_AT581X_ID])
s = await switch.new_switch(config) s = await switch.new_switch(config)
await cg.register_parented(s, config[CONF_AT581X_ID]) await cg.register_parented(s, config[CONF_AT581X_ID])
+14 -6
View File
@@ -1,10 +1,13 @@
import re import re
from typing import Any
from esphome import automation from esphome import automation
import esphome.codegen as cg import esphome.codegen as cg
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.const import CONF_DATA, CONF_ID, CONF_TRIGGER_ID from esphome.const import CONF_DATA, CONF_ID, CONF_TRIGGER_ID
from esphome.core import CORE, ID from esphome.core import CORE, ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
CODEOWNERS = ["@mvturnho", "@danielschramm"] CODEOWNERS = ["@mvturnho", "@danielschramm"]
IS_PLATFORM_COMPONENT = True IS_PLATFORM_COMPONENT = True
@@ -18,7 +21,7 @@ CONF_BIT_RATE = "bit_rate"
CONF_ON_FRAME = "on_frame" CONF_ON_FRAME = "on_frame"
def validate_id(config): def validate_id(config: ConfigType) -> ConfigType:
if CONF_CAN_ID in config: if CONF_CAN_ID in config:
can_id = config[CONF_CAN_ID] can_id = config[CONF_CAN_ID]
id_ext = config[CONF_USE_EXTENDED_ID] id_ext = config[CONF_USE_EXTENDED_ID]
@@ -27,7 +30,7 @@ def validate_id(config):
return config return config
def validate_raw_data(value): def validate_raw_data(value: Any) -> bytes | list:
if isinstance(value, str): if isinstance(value, str):
return value.encode("utf-8") return value.encode("utf-8")
if isinstance(value, list): 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) match = re.match(r"(\d+)(?:K(\d+)?)?BPS", value, re.IGNORECASE)
if not match: if not match:
raise ValueError(f"Invalid rate format: {value}") raise ValueError(f"Invalid rate format: {value}")
@@ -103,7 +106,7 @@ CANBUS_SCHEMA = cv.Schema(
CANBUS_SCHEMA.add_extra(validate_id) 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) await cg.register_component(var, config)
cg.add(var.set_can_id([config[CONF_CAN_ID]])) cg.add(var.set_can_id([config[CONF_CAN_ID]]))
cg.add(var.set_use_extended_id([config[CONF_USE_EXTENDED_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]): if not CORE.has_id(config[CONF_ID]):
var = cg.new_Pvariable(config[CONF_ID], var) var = cg.new_Pvariable(config[CONF_ID], var)
await setup_canbus_core_(var, config) await setup_canbus_core_(var, config)
@@ -157,7 +160,12 @@ async def register_canbus(var, config):
), ),
synchronous=True, 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) var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_CANBUS_ID]) 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 from esphome.components import uart
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.const import CONF_ADDRESS, CONF_ID from esphome.const import CONF_ADDRESS, CONF_ID
from esphome.types import ConfigType
CODEOWNERS = ["@s1lvi0"] CODEOWNERS = ["@s1lvi0"]
MULTI_CONF = True 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]) var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config) await cg.register_component(var, config)
await uart.register_uart_device(var, config) await uart.register_uart_device(var, config)
+4 -2
View File
@@ -1,6 +1,8 @@
import esphome.codegen as cg import esphome.codegen as cg
from esphome.components import binary_sensor from esphome.components import binary_sensor
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
from . import CONF_BMS_DALY_ID, DalyBmsComponent 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): if sensor_config := config.get(key):
var = await binary_sensor.new_binary_sensor(sensor_config) var = await binary_sensor.new_binary_sensor(sensor_config)
cg.add(getattr(hub, f"set_{key}_binary_sensor")(var)) 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]) hub = await cg.get_variable(config[CONF_BMS_DALY_ID])
for key in TYPES: for key in TYPES:
await setup_conf(config, key, hub) await setup_conf(config, key, hub)
+4 -2
View File
@@ -23,6 +23,8 @@ from esphome.const import (
UNIT_PERCENT, UNIT_PERCENT,
UNIT_VOLT, UNIT_VOLT,
) )
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
from . import CONF_BMS_DALY_ID, DalyBmsComponent 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): if sensor_config := config.get(key):
sens = await sensor.new_sensor(sensor_config) sens = await sensor.new_sensor(sensor_config)
cg.add(getattr(hub, f"set_{key}_sensor")(sens)) 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]) hub = await cg.get_variable(config[CONF_BMS_DALY_ID])
for key in TYPES: for key in TYPES:
await setup_conf(config, key, hub) 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 from esphome.components import text_sensor
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.const import CONF_STATUS from esphome.const import CONF_STATUS
from esphome.cpp_generator import MockObj
from esphome.types import ConfigType
from . import CONF_BMS_DALY_ID, DalyBmsComponent 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): if sensor_config := config.get(key):
sens = await text_sensor.new_text_sensor(sensor_config) sens = await text_sensor.new_text_sensor(sensor_config)
cg.add(getattr(hub, f"set_{key}_text_sensor")(sens)) 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]) hub = await cg.get_variable(config[CONF_BMS_DALY_ID])
for key in TYPES: for key in TYPES:
await setup_conf(config, key, hub) await setup_conf(config, key, hub)
+16 -5
View File
@@ -38,7 +38,8 @@ from esphome.const import (
PLATFORM_NRF52, PLATFORM_NRF52,
PlatformFramework, PlatformFramework,
) )
from esphome.core import CORE from esphome.core import CORE, ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType from esphome.types import ConfigType
WAKEUP_PINS = { WAKEUP_PINS = {
@@ -174,7 +175,7 @@ def validate_config(config: ConfigType) -> ConfigType:
return config return config
def _validate_ex1_wakeup_mode(value): def _validate_ex1_wakeup_mode(value: str) -> str:
if value == "ALL_LOW": if value == "ALL_LOW":
esp32.only_on_variant(supported=[VARIANT_ESP32], msg_prefix="ALL_LOW")(value) esp32.only_on_variant(supported=[VARIANT_ESP32], msg_prefix="ALL_LOW")(value)
if value == "ANY_LOW": 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]) var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config) await cg.register_component(var, config)
@@ -458,7 +459,12 @@ DEEP_SLEEP_ENTER_SCHEMA = cv.All(
DEEP_SLEEP_ENTER_SCHEMA, DEEP_SLEEP_ENTER_SCHEMA,
synchronous=True, 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]) paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren) var = cg.new_Pvariable(action_id, template_arg, paren)
if CONF_SLEEP_DURATION in config: 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), automation.maybe_simple_id(DEEP_SLEEP_ACTION_SCHEMA),
synchronous=True, 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) var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID]) await cg.register_parented(var, config[CONF_ID])
return var return var
+16 -3
View File
@@ -3,6 +3,9 @@ import esphome.codegen as cg
from esphome.components import i2c, time from esphome.components import i2c, time
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.const import CONF_ID 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"] CODEOWNERS = ["@badbadc0ffee"]
DEPENDENCIES = ["i2c"] DEPENDENCIES = ["i2c"]
@@ -29,7 +32,12 @@ CONFIG_SCHEMA = time.TIME_SCHEMA.extend(
), ),
synchronous=True, 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) var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID]) await cg.register_parented(var, config[CONF_ID])
return var return var
@@ -45,13 +53,18 @@ async def ds1307_write_time_to_code(config, action_id, template_arg, args):
), ),
synchronous=True, 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) var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID]) await cg.register_parented(var, config[CONF_ID])
return var return var
async def to_code(config): async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID]) var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config) await cg.register_component(var, config)
@@ -38,7 +38,8 @@ from esphome.const import (
CONF_SERVICE_UUID, CONF_SERVICE_UUID,
CONF_TRIGGER_ID, 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.enum import StrEnum
from esphome.types import ConfigType 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 # Register the loggers this component needs
esp32_ble.register_bt_logger(BTLoggers.BLE_SCAN) 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 # chance to call register_ble_tracker and register_client before the list is checked
# and added to the global defines list. # and added to the global defines list.
@coroutine_with_priority(CoroPriority.FINAL) @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 # Add feature-specific defines based on what's needed
required_features = _get_required_features() required_features = _get_required_features()
# Sensors registered through the neutral ble_device_base path (BLEHub) need # 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, synchronous=True,
) )
async def esp32_ble_tracker_start_scan_action_to_code( 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]) paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren) var = cg.new_Pvariable(action_id, template_arg, paren)
template_ = await cg.templatable(config[CONF_CONTINUOUS], args, cg.bool_) 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, synchronous=True,
) )
async def esp32_ble_tracker_stop_scan_action_to_code( 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) var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID]) await cg.register_parented(var, config[CONF_ID])
return var return var
+17 -10
View File
@@ -48,10 +48,12 @@ from esphome.const import (
) )
from esphome.core import ( from esphome.core import (
CORE, CORE,
ID,
CoroPriority, CoroPriority,
TimePeriodMilliseconds, TimePeriodMilliseconds,
coroutine_with_priority, coroutine_with_priority,
) )
from esphome.cpp_generator import MockObj, TemplateArgsType
import esphome.final_validate as fv import esphome.final_validate as fv
from esphome.types import ConfigType from esphome.types import ConfigType
@@ -276,7 +278,7 @@ def _validate_spi_interface(config: ConfigType) -> ConfigType:
return config return config
def _validate(config): def _validate(config: ConfigType) -> ConfigType:
if CONF_USE_ADDRESS not in config: if CONF_USE_ADDRESS not in config:
if CONF_MANUAL_IP in config: if CONF_MANUAL_IP in config:
use_address = str(config[CONF_MANUAL_IP][CONF_STATIC_IP]) 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( return cv.All(
BASE_SCHEMA.extend( BASE_SCHEMA.extend(
cv.Schema( 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: if not CORE.is_esp32:
return # SPI interface validation is ESP32-only return # SPI interface validation is ESP32-only
if config[CONF_TYPE] not in SPI_ETHERNET_TYPES: 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( return cg.StructInitializer(
ManualIP, ManualIP,
("static_ip", ip_address_literal(config[CONF_STATIC_IP])), ("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( return cg.StructInitializer(
PHYRegister, PHYRegister,
("address", address), ("address", address),
@@ -558,7 +560,7 @@ def phy_register(address: int, value: int, page: int):
@coroutine_with_priority(CoroPriority.COMMUNICATION) @coroutine_with_priority(CoroPriority.COMMUNICATION)
async def to_code(config): async def to_code(config: ConfigType) -> None:
var = cg.new_Pvariable(config[CONF_ID]) var = cg.new_Pvariable(config[CONF_ID])
# Apply network priority before register_component (which emits the user's # 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) 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 ( from esphome.components.esp32 import (
add_idf_component, add_idf_component,
add_idf_sdkconfig_option, 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) 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_clk_pin(config[CONF_CLK_PIN]))
cg.add(var.set_miso_pin(config[CONF_MISO_PIN])) cg.add(var.set_miso_pin(config[CONF_MISO_PIN]))
cg.add(var.set_mosi_pin(config[CONF_MOSI_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) @coroutine_with_priority(CoroPriority.FINAL)
async def final_step(): async def final_step() -> None:
"""Final code generation step to configure optional Ethernet features.""" """Final code generation step to configure optional Ethernet features."""
if ip_state_count := CORE.data.get(ETHERNET_IP_STATE_LISTENERS_KEY, 0): if ip_state_count := CORE.data.get(ETHERNET_IP_STATE_LISTENERS_KEY, 0):
cg.add_define("USE_ETHERNET_IP_STATE_LISTENERS") cg.add_define("USE_ETHERNET_IP_STATE_LISTENERS")
@@ -845,7 +847,12 @@ def _filter_source_files() -> list[str]:
FILTER_SOURCE_FILES = _filter_source_files 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) 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 import automation
from esphome.automation import maybe_simple_id from esphome.automation import maybe_simple_id
import esphome.codegen as cg import esphome.codegen as cg
@@ -16,6 +18,9 @@ from esphome.const import (
UNIT_CELSIUS, UNIT_CELSIUS,
UNIT_PERCENT, UNIT_PERCENT,
) )
from esphome.core import ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
DEPENDENCIES = ["i2c"] 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]) var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config) await cg.register_component(var, config)
await i2c.register_i2c_device(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""" """Accept enum names or raw uint16 values"""
if isinstance(value, cv.Lambda): if isinstance(value, cv.Lambda):
return value return value
@@ -119,7 +124,12 @@ HDC302X_HEATER_ON_ACTION_SCHEMA = maybe_simple_id(
HDC302X_HEATER_ON_ACTION_SCHEMA, HDC302X_HEATER_ON_ACTION_SCHEMA,
synchronous=True, 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) var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID]) await cg.register_parented(var, config[CONF_ID])
template_ = await cg.templatable(config[CONF_POWER], args, cg.uint16) 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, HDC302X_ACTION_SCHEMA,
synchronous=True, 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) var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID]) await cg.register_parented(var, config[CONF_ID])
return var return var
+16 -3
View File
@@ -17,6 +17,9 @@ from esphome.const import (
UNIT_EMPTY, UNIT_EMPTY,
UNIT_PERCENT, UNIT_PERCENT,
) )
from esphome.core import ID
from esphome.cpp_generator import MockObj, TemplateArgsType
from esphome.types import ConfigType
DEPENDENCIES = ["i2c"] 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]) var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config) await cg.register_component(var, config)
await i2c.register_i2c_device(var, config) await i2c.register_i2c_device(var, config)
@@ -95,7 +98,12 @@ async def to_code(config):
), ),
synchronous=True, 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) var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID]) await cg.register_parented(var, config[CONF_ID])
level_ = await cg.templatable(config[CONF_LEVEL], args, cg.uint8) 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, 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) var = cg.new_Pvariable(action_id, template_arg)
await cg.register_parented(var, config[CONF_ID]) await cg.register_parented(var, config[CONF_ID])
status_ = await cg.templatable(config[CONF_STATUS], args, cg.bool_) 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]) var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config) await cg.register_component(var, config)
await uart.register_uart_device(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 from esphome.components import binary_sensor
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.const import CONF_TARGET, DEVICE_CLASS_OCCUPANCY from esphome.const import CONF_TARGET, DEVICE_CLASS_OCCUPANCY
from esphome.types import ConfigType
from . import LD6002BComponent from . import LD6002BComponent
from .const import AREA_COUNT, CONF_LD6002B_ID, MAX_TARGETS 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]) hub = await cg.get_variable(config[CONF_LD6002B_ID])
if target_config := config.get(CONF_TARGET): 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(): for key, button_type in BUTTON_MAP.items():
if button_config := config.get(key): if button_config := config.get(key):
b = cg.new_Pvariable(button_config[CONF_ID], button_type) 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 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]) hub = await cg.get_variable(config[CONF_LD6002B_ID])
for key, number_type, setter, min_value, max_value, step in ( 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 from esphome.components import select
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.const import CONF_AREA_ID, CONF_SENSITIVITY, ENTITY_CATEGORY_CONFIG from esphome.const import CONF_AREA_ID, CONF_SENSITIVITY, ENTITY_CATEGORY_CONFIG
from esphome.types import ConfigType
from .. import LD6002BComponent, ld6002b_ns from .. import LD6002BComponent, ld6002b_ns
from ..const import CONF_INSTALLATION_MODE, CONF_LD6002B_ID, CONF_TRIGGER_SPEED 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]) hub = await cg.get_variable(config[CONF_LD6002B_ID])
for key, select_type, setter, options in SELECT_MAP: for key, select_type, setter, options in SELECT_MAP:
+2 -1
View File
@@ -9,6 +9,7 @@ from esphome.const import (
STATE_CLASS_MEASUREMENT, STATE_CLASS_MEASUREMENT,
UNIT_METER, UNIT_METER,
) )
from esphome.types import ConfigType
from . import LD6002BComponent from . import LD6002BComponent
from .const import ( 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]) hub = await cg.get_variable(config[CONF_LD6002B_ID])
if target_count_config := config.get(CONF_TARGET_COUNT): if target_count_config := config.get(CONF_TARGET_COUNT):
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import switch from esphome.components import switch
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.const import DEVICE_CLASS_SWITCH, ENTITY_CATEGORY_CONFIG from esphome.const import DEVICE_CLASS_SWITCH, ENTITY_CATEGORY_CONFIG
from esphome.types import ConfigType
from .. import LD6002BComponent, ld6002b_ns from .. import LD6002BComponent, ld6002b_ns
from ..const import ( 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]) hub = await cg.get_variable(config[CONF_LD6002B_ID])
for key, switch_type, setter in ( 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 from esphome.components import text_sensor
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.const import ENTITY_CATEGORY_DIAGNOSTIC from esphome.const import ENTITY_CATEGORY_DIAGNOSTIC
from esphome.types import ConfigType
from . import LD6002BComponent from . import LD6002BComponent
from .const import CONF_LD6002B_ID, CONF_OTA_VERSION, CONF_WORK_MODE 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]) hub = await cg.get_variable(config[CONF_LD6002B_ID])
if work_mode_config := config.get(CONF_WORK_MODE): if work_mode_config := config.get(CONF_WORK_MODE):
sens = await text_sensor.new_text_sensor(work_mode_config) sens = await text_sensor.new_text_sensor(work_mode_config)
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import i2c from esphome.components import i2c
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.const import CONF_ID from esphome.const import CONF_ID
from esphome.types import ConfigType
DEPENDENCIES = ["i2c"] DEPENDENCIES = ["i2c"]
CODEOWNERS = ["@rnauber"] CODEOWNERS = ["@rnauber"]
@@ -26,7 +27,7 @@ CONFIG_SCHEMA = cv.Schema(
).extend(i2c.i2c_device_schema(0x43)) ).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]) var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config) await cg.register_component(var, config)
await i2c.register_i2c_device(var, config) await i2c.register_i2c_device(var, config)
@@ -1,6 +1,7 @@
import esphome.codegen as cg import esphome.codegen as cg
from esphome.components import binary_sensor from esphome.components import binary_sensor
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.types import ConfigType
from .. import CONF_M5STACK_8ANGLE_ID, M5Stack8AngleComponent, m5stack_8angle_ns 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]) hub = await cg.get_variable(config[CONF_M5STACK_8ANGLE_ID])
sens = await binary_sensor.new_binary_sensor(config) sens = await binary_sensor.new_binary_sensor(config)
cg.add(sens.set_parent(hub)) cg.add(sens.set_parent(hub))
@@ -2,6 +2,7 @@ import esphome.codegen as cg
from esphome.components import light from esphome.components import light
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.const import CONF_OUTPUT_ID from esphome.const import CONF_OUTPUT_ID
from esphome.types import ConfigType
from .. import CONF_M5STACK_8ANGLE_ID, M5Stack8AngleComponent, m5stack_8angle_ns 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]) hub = await cg.get_variable(config[CONF_M5STACK_8ANGLE_ID])
lights = cg.new_Pvariable(config[CONF_OUTPUT_ID]) lights = cg.new_Pvariable(config[CONF_OUTPUT_ID])
await light.register_light(lights, config) await light.register_light(lights, config)
@@ -8,6 +8,7 @@ from esphome.const import (
ICON_ROTATE_RIGHT, ICON_ROTATE_RIGHT,
STATE_CLASS_MEASUREMENT, STATE_CLASS_MEASUREMENT,
) )
from esphome.types import ConfigType
from .. import ( from .. import (
CONF_M5STACK_8ANGLE_ID, 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) var = await sensor.new_sensor(config)
await cg.register_component(var, config) await cg.register_component(var, config)
await cg.register_parented(var, config[CONF_M5STACK_8ANGLE_ID]) 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 from esphome.components import uart
import esphome.config_validation as cv import esphome.config_validation as cv
from esphome.const import CONF_ADDRESS, CONF_DISABLE_CRC, CONF_FLOW_CONTROL_PIN, CONF_ID 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 from esphome.cpp_helpers import gpio_pin_expression
import esphome.final_validate as fv import esphome.final_validate as fv
from esphome.types import ConfigType
_LOGGER = logging.getLogger(__name__) _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) cg.add_global(modbus_ns.using)
var = cg.new_Pvariable(config[CONF_ID]) var = cg.new_Pvariable(config[CONF_ID])
await cg.register_component(var, config) await cg.register_component(var, config)
@@ -112,7 +114,9 @@ def _validate_server_address(value: Any) -> int:
return address 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 hub_type = ModbusClient if role == "client" else ModbusServer
address_validator = _validate_server_address if role == "server" else cv.hex_uint8_t address_validator = _validate_server_address if role == "server" else cv.hex_uint8_t
schema = { schema = {
@@ -127,14 +131,14 @@ def modbus_device_schema(default_address, role: Literal["client", "server"] = "c
def final_validate_modbus_device( def final_validate_modbus_device(
name: str, *, role: Literal["server", "client"] | None = None name: str, *, role: Literal["server", "client"] | None = None
): ) -> cv.Schema:
def validate_role(value): def validate_role(value: str) -> str:
assert role in MODBUS_ROLES assert role in MODBUS_ROLES
if value != role: if value != role:
raise cv.Invalid(f"Component {name} requires role to be {role}") raise cv.Invalid(f"Component {name} requires role to be {role}")
return value return value
def validate_hub(hub_config): def validate_hub(hub_config: ConfigType) -> ConfigType:
hub_schema = {} hub_schema = {}
if role is not None: if role is not None:
hub_schema[cv.Required(CONF_ROLE)] = validate_role 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]) parent = await cg.get_variable(config[CONF_MODBUS_ID])
cg.add(var.set_parent(parent)) cg.add(var.set_parent(parent))
cg.add(var.set_address(config[CONF_ADDRESS])) 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]) parent = await cg.get_variable(config[CONF_MODBUS_ID])
cg.add(var.set_address(config[CONF_ADDRESS])) cg.add(var.set_address(config[CONF_ADDRESS]))
cg.add(parent.register_device(var)) 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 # Remove before 2026.12.0
_LOGGER.warning( _LOGGER.warning(
"'register_modbus_device' is deprecated, use 'register_modbus_client_device' " "'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 from esphome import automation
import esphome.codegen as cg import esphome.codegen as cg
from esphome.components.esp32 import ( from esphome.components.esp32 import (
@@ -31,10 +33,12 @@ from esphome.const import (
) )
from esphome.core import ( from esphome.core import (
CORE, CORE,
ID,
CoroPriority, CoroPriority,
TimePeriodMilliseconds, TimePeriodMilliseconds,
coroutine_with_priority, coroutine_with_priority,
) )
from esphome.cpp_generator import MockObj, TemplateArgsType
import esphome.final_validate as fv import esphome.final_validate as fv
from esphome.types import ConfigType 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: if CORE.is_esp32:
variant = get_esp32_variant() variant = get_esp32_variant()
@@ -90,7 +94,7 @@ def _validate_txpower(value):
return value # Unsupported, fail later with clear error 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 # and expose options for using SPI/UART RCPs
add_idf_sdkconfig_option("CONFIG_IEEE802154_ENABLED", True) add_idf_sdkconfig_option("CONFIG_IEEE802154_ENABLED", True)
add_idf_sdkconfig_option("CONFIG_OPENTHREAD_RADIO_NATIVE", True) add_idf_sdkconfig_option("CONFIG_OPENTHREAD_RADIO_NATIVE", True)
@@ -180,7 +184,7 @@ def _validate(config: ConfigType) -> ConfigType:
return config return config
def _require_vfs_select(config): def _require_vfs_select(config: ConfigType) -> ConfigType:
"""Register VFS select requirement during config validation.""" """Register VFS select requirement during config validation."""
# OpenThread uses esp_vfs_eventfd which requires VFS select support (ESP32 only) # OpenThread uses esp_vfs_eventfd which requires VFS select support (ESP32 only)
if CORE.is_esp32: if CORE.is_esp32:
@@ -188,7 +192,7 @@ def _require_vfs_select(config):
return config return config
def _validate_platform(config): def _validate_platform(config: ConfigType) -> ConfigType:
if CORE.using_zephyr: if CORE.using_zephyr:
return config return config
return only_on_variant( return only_on_variant(
@@ -203,7 +207,7 @@ def _validate_platform(config):
)(config) )(config)
def _validate_tlv_hex(value): def _validate_tlv_hex(value: Any) -> str:
s = cv.string_strict(value) s = cv.string_strict(value)
if len(s) % 2 != 0: if len(s) % 2 != 0:
raise cv.Invalid("TLV must have an even number of hex characters") 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() full_config = fv.full_config.get()
network_config = full_config.get("network", {}) network_config = full_config.get("network", {})
if not network_config.get(CONF_ENABLE_IPV6, False): 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) @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) # Re-enable openthread IDF component (excluded by default)
if CORE.is_esp32: if CORE.is_esp32:
include_builtin_idf_component("openthread") include_builtin_idf_component("openthread")
@@ -339,7 +343,12 @@ POLL_PERIOD_ACTION_SCHEMA = automation.maybe_conf(
POLL_PERIOD_ACTION_SCHEMA, POLL_PERIOD_ACTION_SCHEMA,
synchronous=True, 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]) paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren) var = cg.new_Pvariable(action_id, template_arg, paren)
template_ = await cg.templatable(config[CONF_POLL_PERIOD], args, cg.uint32) 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 from esphome import automation, pins
import esphome.codegen as cg import esphome.codegen as cg
from esphome.components import sensor from esphome.components import sensor
@@ -19,7 +21,9 @@ from esphome.const import (
UNIT_PULSES, UNIT_PULSES,
UNIT_PULSES_PER_MINUTE, 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" 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) use_pcnt = value.get(CONF_USE_PCNT)
if CORE.is_esp8266 and use_pcnt: if CORE.is_esp8266 and use_pcnt:
raise cv.Invalid( raise cv.Invalid(
@@ -63,7 +67,7 @@ def validate_internal_filter(value):
return 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) value = pins.internal_gpio_input_pin_schema(value)
if CORE.is_esp8266 and value[CONF_NUMBER] >= 16: if CORE.is_esp8266 and value[CONF_NUMBER] >= 16:
raise cv.Invalid( raise cv.Invalid(
@@ -72,7 +76,7 @@ def validate_pulse_counter_pin(value):
return value return value
def validate_count_mode(value): def validate_count_mode(value: ConfigType) -> ConfigType:
rising_edge = value[CONF_RISING_EDGE] rising_edge = value[CONF_RISING_EDGE]
falling_edge = value[CONF_FALLING_EDGE] falling_edge = value[CONF_FALLING_EDGE]
if rising_edge == "DISABLE" and falling_edge == "DISABLE": 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) use_pcnt = config.get(CONF_USE_PCNT)
if CORE.is_esp32 and use_pcnt: if CORE.is_esp32 and use_pcnt:
include_builtin_idf_component("esp_driver_pcnt") include_builtin_idf_component("esp_driver_pcnt")
@@ -157,7 +161,12 @@ async def to_code(config):
), ),
synchronous=True, 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]) paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren) var = cg.new_Pvariable(action_id, template_arg, paren)
template_ = await cg.templatable(config[CONF_VALUE], args, cg.uint32) 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 from esphome import automation, pins
import esphome.codegen as cg import esphome.codegen as cg
from esphome.components import sensor from esphome.components import sensor
@@ -17,7 +19,9 @@ from esphome.const import (
UNIT_PULSES, UNIT_PULSES,
UNIT_PULSES_PER_MINUTE, 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"] CODEOWNERS = ["@stevebaxter", "@cstaahl", "@TrentHouliston"]
@@ -37,18 +41,18 @@ FILTER_MODES = {
SetTotalPulsesAction = pulse_meter_ns.class_("SetTotalPulsesAction", automation.Action) 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) return cv.positive_time_period_microseconds(value)
def validate_timeout(value): def validate_timeout(value: Any) -> TimePeriodMicroseconds:
value = cv.positive_time_period_microseconds(value) value = cv.positive_time_period_microseconds(value)
if value.total_minutes > 70: if value.total_minutes > 70:
raise cv.Invalid("Maximum timeout is 70 minutes") raise cv.Invalid("Maximum timeout is 70 minutes")
return value return value
def validate_pulse_meter_pin(value): def validate_pulse_meter_pin(value: Any) -> ConfigType:
value = pins.internal_gpio_input_pin_schema(value) value = pins.internal_gpio_input_pin_schema(value)
if CORE.is_esp8266 and value[CONF_NUMBER] >= 16: if CORE.is_esp8266 and value[CONF_NUMBER] >= 16:
raise cv.Invalid( 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) var = await sensor.new_sensor(config)
await cg.register_component(var, config) await cg.register_component(var, config)
@@ -107,7 +111,12 @@ async def to_code(config):
), ),
synchronous=True, 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]) paren = await cg.get_variable(config[CONF_ID])
var = cg.new_Pvariable(action_id, template_arg, paren) var = cg.new_Pvariable(action_id, template_arg, paren)
template_ = await cg.templatable(config[CONF_VALUE], args, cg.uint32) template_ = await cg.templatable(config[CONF_VALUE], args, cg.uint32)
+6 -5
View File
@@ -1,6 +1,7 @@
import hashlib import hashlib
from pathlib import Path from pathlib import Path
import re import re
from typing import Any
from esphome import external_files, pins from esphome import external_files, pins
import esphome.codegen as cg 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) match = re.fullmatch(r"(\d+)\.(\d+)", value)
if match is None: if match is None:
raise ValueError(f"Not a valid version number {value}") 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) PREFETCH_FILES = external_files.single_stage_prefetch(_extract_firmware_ref)
def validate_firmware(value): def validate_firmware(value: ConfigType) -> ConfigType:
config = value.copy() config = value.copy()
if CONF_URL not in config: if CONF_URL not in config:
try: try:
@@ -167,14 +168,14 @@ def validate_firmware(value):
return config return config
def validate_sha256(value): def validate_sha256(value: Any) -> str:
value = cv.string(value) value = cv.string(value)
if not re.fullmatch(r"[0-9a-fA-F]{64}", value): if not re.fullmatch(r"[0-9a-fA-F]{64}", value):
raise ValueError(f"Not a valid SHA256 hex string: {value}") raise ValueError(f"Not a valid SHA256 hex string: {value}")
return value return value
def validate_version(value): def validate_version(value: str) -> str:
parse_firmware_version(value) parse_firmware_version(value)
return 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_hex = get_firmware(config[CONF_FIRMWARE])
fw_major, fw_minor = parse_firmware_version(config[CONF_FIRMWARE][CONF_VERSION]) fw_major, fw_minor = parse_firmware_version(config[CONF_FIRMWARE][CONF_VERSION])