[core] Add type annotations to component Python (1/11) (#18338)

esphome/components/bl0906/sensor.py

@@ -32,6 +32,9 @@ from esphome.const import (
     UNIT_VOLT,
     UNIT_WATT,
 )
+from esphome.core import ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 # Import ICONS not included in esphome's const.py, from the local components const.py
 from .const import ICON_ENERGY, ICON_FREQUENCY, ICON_VOLTAGE
@@ -145,13 +148,18 @@ FINAL_VALIDATE_SCHEMA = uart.final_validate_device_schema(
     ),
     synchronous=True,
 )
-async def reset_energy_to_code(config, action_id, template_arg, args):
+async def reset_energy_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)
     await uart.register_uart_device(var, config)

esphome/components/datetime/__init__.py

@@ -21,13 +21,14 @@ from esphome.const import (
     CONF_WEB_SERVER,
     CONF_YEAR,
 )
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import (
     entity_duplicate_validator,
     queue_entity_register,
     setup_entity,
 )
-from esphome.cpp_generator import MockObjClass
+from esphome.cpp_generator import MockObj, MockObjClass, TemplateArgsType
+from esphome.types import ConfigType, SafeExpType
 
 CODEOWNERS = ["@rfdarter", "@jesserockz"]
 
@@ -65,7 +66,7 @@ DATETIME_MODES = [
 ]
 
 
-def _validate_time_present(config):
+def _validate_time_present(config: ConfigType) -> ConfigType:
     config = config.copy()
     if CONF_ON_TIME in config and CONF_TIME_ID not in config:
         time_id = cv.use_id(time.RealTimeClock)(None)
@@ -139,7 +140,7 @@ def datetime_schema(class_: MockObjClass) -> cv.Schema:
 
 
 @setup_entity("datetime")
-async def setup_datetime_core_(var, config):
+async def setup_datetime_core_(var: MockObj, config: ConfigType) -> None:
     if (mqtt_id := config.get(CONF_MQTT_ID)) is not None:
         mqtt_ = cg.new_Pvariable(mqtt_id, var)
         await mqtt.register_mqtt_component(mqtt_, config)
@@ -160,7 +161,7 @@ async def setup_datetime_core_(var, config):
         await cg.register_parented(trigger, var)
 
 
-async def register_datetime(var, config):
+async def register_datetime(var: MockObj, config: ConfigType) -> None:
     if not CORE.has_id(config[CONF_ID]):
         var = cg.Pvariable(config[CONF_ID], var)
     entity_type = config[CONF_TYPE].lower()
@@ -169,14 +170,14 @@ async def register_datetime(var, config):
     await setup_datetime_core_(var, config)
 
 
-async def new_datetime(config, *args):
+async def new_datetime(config: ConfigType, *args: SafeExpType) -> MockObj:
     var = cg.new_Pvariable(config[CONF_ID], *args)
     await register_datetime(var, config)
     return var
 
 
 @coroutine_with_priority(CoroPriority.CORE)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     cg.add_global(datetime_ns.using)
 
 
@@ -193,7 +194,12 @@ async def to_code(config):
     ),
     synchronous=True,
 )
-async def datetime_date_set_to_code(config, action_id, template_arg, args):
+async def datetime_date_set_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     action_var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(action_var, config[CONF_ID])
 
@@ -226,7 +232,12 @@ async def datetime_date_set_to_code(config, action_id, template_arg, args):
     ),
     synchronous=True,
 )
-async def datetime_time_set_to_code(config, action_id, template_arg, args):
+async def datetime_time_set_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     action_var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(action_var, config[CONF_ID])
 
@@ -259,7 +270,12 @@ async def datetime_time_set_to_code(config, action_id, template_arg, args):
     ),
     synchronous=True,
 )
-async def datetime_datetime_set_to_code(config, action_id, template_arg, args):
+async def datetime_datetime_set_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     action_var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(action_var, config[CONF_ID])
 

esphome/components/esp32_ble/__init__.py

@@ -31,7 +31,8 @@ from esphome.const import (
     CONF_NAME,
     CONF_NAME_ADD_MAC_SUFFIX,
 )
-from esphome.core import CORE, TimePeriod
+from esphome.core import CORE, ID, TimePeriod
+from esphome.cpp_generator import MockObj, TemplateArgsType
 import esphome.final_validate as fv
 from esphome.types import ConfigType
 
@@ -383,7 +384,7 @@ def _validate_key_sizes(config: ConfigType) -> ConfigType:
 CONFIG_SCHEMA = cv.All(CONFIG_SCHEMA, _validate_key_sizes)
 
 
-def validate_variant(_):
+def validate_variant(_: ConfigType) -> None:
     variant = get_esp32_variant()
     if variant in NO_BLUETOOTH_VARIANTS:
         raise cv.Invalid(f"{variant} does not support Bluetooth")
@@ -443,7 +444,7 @@ def validate_connection_slots(max_connections: int) -> None:
     )
 
 
-def final_validation(config) -> None:
+def final_validation(config: ConfigType) -> None:
     validate_variant(config)
     if (name := config.get(CONF_NAME)) is not None:
         full_config = fv.full_config.get()
@@ -518,7 +519,7 @@ def final_validation(config) -> None:
 FINAL_VALIDATE_SCHEMA = final_validation
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     cg.add(var.set_enable_on_boot(config[CONF_ENABLE_ON_BOOT]))
     cg.add(var.set_io_capability(config[CONF_IO_CAPABILITY]))
@@ -605,19 +606,34 @@ async def to_code(config):
 
 
 @automation.register_condition("ble.enabled", BLEEnabledCondition, cv.Schema({}))
-async def ble_enabled_to_code(config, condition_id, template_arg, args):
+async def ble_enabled_to_code(
+    config: ConfigType,
+    condition_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     return cg.new_Pvariable(condition_id, template_arg)
 
 
 @automation.register_action(
     "ble.enable", BLEEnableAction, cv.Schema({}), synchronous=True
 )
-async def ble_enable_to_code(config, action_id, template_arg, args):
+async def ble_enable_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     return cg.new_Pvariable(action_id, template_arg)
 
 
 @automation.register_action(
     "ble.disable", BLEDisableAction, cv.Schema({}), synchronous=True
 )
-async def ble_disable_to_code(config, action_id, template_arg, args):
+async def ble_disable_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     return cg.new_Pvariable(action_id, template_arg)

esphome/components/esp32_rmt/__init__.py

@@ -1,17 +1,23 @@
+from collections.abc import Callable, Iterable
+from typing import Any
+
 from esphome.components import esp32
 import esphome.config_validation as cv
 from esphome.core import CORE
+from esphome.types import ConfigType
 
 CODEOWNERS = ["@jesserockz"]
 
 VARIANTS_NO_RMT = {esp32.VARIANT_ESP32C2, esp32.VARIANT_ESP32C61}
 
 
-def validate_rmt_not_supported(rmt_only_keys):
+def validate_rmt_not_supported(
+    rmt_only_keys: Iterable[str],
+) -> Callable[[ConfigType], ConfigType]:
     """Validate that RMT-only config keys are not used on variants without RMT hardware."""
     rmt_only_keys = set(rmt_only_keys)
 
-    def _validator(config):
+    def _validator(config: ConfigType) -> ConfigType:
         if CORE.is_esp32:
             variant = esp32.get_esp32_variant()
             if variant in VARIANTS_NO_RMT:
@@ -26,8 +32,8 @@ def validate_rmt_not_supported(rmt_only_keys):
     return _validator
 
 
-def validate_clock_resolution():
+def validate_clock_resolution() -> Callable[[Any], int]:
-    def _validator(value):
+    def _validator(value: Any) -> int:
         cv.only_on_esp32(value)
         value = cv.int_(value)
         variant = esp32.get_esp32_variant()

esphome/components/espnow/__init__.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from esphome import automation, core
 import esphome.codegen as cg
 from esphome.components import wifi
@@ -14,6 +16,7 @@ from esphome.const import (
     CONF_WIFI,
 )
 from esphome.core import CORE, HexInt
+from esphome.cpp_generator import MockObj, TemplateArgsType
 from esphome.types import ConfigType
 
 CODEOWNERS = ["@jesserockz"]
@@ -78,7 +81,7 @@ CONF_CONTINUE_ON_ERROR = "continue_on_error"
 CONF_WAIT_FOR_SENT = "wait_for_sent"
 
 
-def _validate_max_payload_size(value: int) -> int:
+def _validate_max_payload_size(value: Any) -> int:
     if value > ESPNOW_PAYLOAD_V1:
         return cv.require_framework_version(
             esp_idf=cv.Version(5, 4, 0),
@@ -88,7 +91,7 @@ def _validate_max_payload_size(value: int) -> int:
     return value
 
 
-def validate_channel(value):
+def validate_channel(value: Any) -> int:
     if value is None:
         raise cv.Invalid("channel is required if wifi is not configured")
     return wifi.validate_channel(value)
@@ -129,7 +132,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def _trigger_to_code(config):
+async def _trigger_to_code(config: ConfigType) -> MockObj:
     if address := config.get(CONF_ADDRESS):
         address = address.parts
     trigger = cg.new_Pvariable(config[CONF_TRIGGER_ID], address)
@@ -145,7 +148,7 @@ async def _trigger_to_code(config):
     return trigger
 
 
-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)
 
@@ -180,13 +183,13 @@ async def to_code(config):
 # ========================================== A C T I O N S ================================================
 
 
-def validate_peer(value):
+def validate_peer(value: Any) -> Any:
     if isinstance(value, cv.Lambda):
         return cv.returning_lambda(value)
     return cv.mac_address(value)
 
 
-def _validate_raw_data(value):
+def _validate_raw_data(value: Any) -> str | list:
     if isinstance(value, str):
         if len(value) > MAX_ESPNOW_PACKET_SIZE:
             raise cv.Invalid(
@@ -204,7 +207,9 @@ def _validate_raw_data(value):
     )
 
 
-async def register_peer(var, config, args):
+async def register_peer(
+    var: MockObj, config: ConfigType, args: TemplateArgsType
+) -> None:
     peer = config[CONF_ADDRESS]
     if isinstance(peer, core.MACAddress):
         peer = [HexInt(p) for p in peer.parts]
@@ -231,7 +236,7 @@ SEND_SCHEMA = PEER_SCHEMA.extend(
 )
 
 
-def _validate_send_action(config):
+def _validate_send_action(config: ConfigType) -> ConfigType:
     if not config[CONF_WAIT_FOR_SENT] and not config[CONF_CONTINUE_ON_ERROR]:
         raise cv.Invalid(
             f"'{CONF_CONTINUE_ON_ERROR}' cannot be false if '{CONF_WAIT_FOR_SENT}' is false as the automation will not wait for the failed result.",
@@ -267,7 +272,7 @@ async def send_action(
     action_id: core.ID,
     template_arg: cg.TemplateArguments,
     args: list[tuple],
-):
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
 
@@ -316,7 +321,7 @@ async def peer_action(
     action_id: core.ID,
     template_arg: cg.TemplateArguments,
     args: list[tuple],
-):
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     await register_peer(var, config, args)
@@ -341,7 +346,7 @@ async def channel_action(
     action_id: core.ID,
     template_arg: cg.TemplateArguments,
     args: list[tuple],
-):
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     template_ = await cg.templatable(config[CONF_CHANNEL], args, cg.uint8)

esphome/components/espnow/packet_transport/__init__.py

@@ -9,6 +9,7 @@ from esphome.components.packet_transport import (
 import esphome.config_validation as cv
 from esphome.core import HexInt
 from esphome.cpp_types import PollingComponent
+from esphome.types import ConfigType
 
 from .. import ESPNowComponent, espnow_ns
 
@@ -28,7 +29,7 @@ CONFIG_SCHEMA = transport_schema(ESPNowTransport).extend(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     """Set up the ESP-NOW transport component."""
     var, _ = await new_packet_transport(config)
 

esphome/components/http_request/__init__.py

@@ -1,4 +1,5 @@
 from pathlib import Path
+from typing import Any
 
 from esphome import automation
 import esphome.codegen as cg
@@ -20,8 +21,10 @@ from esphome.const import (
     PlatformFramework,
     __version__,
 )
-from esphome.core import CORE, Lambda
+from esphome.core import CORE, ID, Lambda
+from esphome.cpp_generator import MockObj, TemplateArgsType
 from esphome.helpers import IS_MACOS
+from esphome.types import ConfigType
 
 DEPENDENCIES = ["network"]
 AUTO_LOAD = ["json", "watchdog"]
@@ -63,14 +66,14 @@ CONF_BODY = "body"
 CONF_JSON = "json"
 
 
-def validate_url(value):
+def validate_url(value: Any) -> str:
     value = cv.url(value)
     if value.startswith(("http://", "https://")):
         return value
     raise cv.Invalid("URL must start with 'http://' or 'https://'")
 
 
-def validate_ssl_verification(config):
+def validate_ssl_verification(config: ConfigType) -> ConfigType:
     error_message = ""
 
     if CORE.is_rp2 and config[CONF_VERIFY_SSL]:
@@ -91,7 +94,7 @@ def validate_ssl_verification(config):
     return config
 
 
-def _declare_request_class(value):
+def _declare_request_class(value: Any) -> ID:
     if CORE.is_host:
         return cv.declare_id(HttpRequestHost)(value)
     if CORE.is_esp32:
@@ -151,7 +154,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     cg.add(var.set_timeout(config[CONF_TIMEOUT]))
     cg.add(var.set_useragent(config[CONF_USERAGENT]))
@@ -298,7 +301,12 @@ HTTP_REQUEST_SEND_ACTION_SCHEMA = HTTP_REQUEST_ACTION_SCHEMA.extend(
     HTTP_REQUEST_SEND_ACTION_SCHEMA,
     synchronous=True,
 )
-async def http_request_action_to_code(config, action_id, template_arg, args):
+async def http_request_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)
 

esphome/components/http_request/ota/__init__.py

@@ -3,8 +3,10 @@ import esphome.codegen as cg
 from esphome.components.ota import BASE_OTA_SCHEMA, OTAComponent, ota_to_code
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_PASSWORD, CONF_URL, CONF_USERNAME
-from esphome.core import coroutine_with_priority
+from esphome.core import ID, coroutine_with_priority
 from esphome.coroutine import CoroPriority
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 from .. import CONF_HTTP_REQUEST_ID, HttpRequestComponent, http_request_ns
 
@@ -42,7 +44,7 @@ CONFIG_SCHEMA = cv.All(
 
 
 @coroutine_with_priority(CoroPriority.OTA_UPDATES)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     await ota_to_code(var, config)
     await cg.register_component(var, config)
@@ -72,7 +74,12 @@ OTA_HTTP_REQUEST_FLASH_ACTION_SCHEMA = cv.All(
     OTA_HTTP_REQUEST_FLASH_ACTION_SCHEMA,
     synchronous=True,
 )
-async def ota_http_request_action_to_code(config, action_id, template_arg, args):
+async def ota_http_request_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)
 

esphome/components/http_request/update/__init__.py

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import ota, update
 import esphome.config_validation as cv
 from esphome.const import CONF_SOURCE
+from esphome.types import ConfigType
 
 from .. import CONF_HTTP_REQUEST_ID, HttpRequestComponent, http_request_ns
 from ..ota import OtaHttpRequestComponent
@@ -29,7 +30,7 @@ CONFIG_SCHEMA = (
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = await update.new_update(config)
     ota_parent = await cg.get_variable(config[CONF_OTA_ID])
     cg.add(var.set_ota_parent(ota_parent))

esphome/components/i2s_audio/__init__.py

@@ -21,8 +21,9 @@ from esphome.components.esp32.const import (
 import esphome.config_validation as cv
 from esphome.const import CONF_BITS_PER_SAMPLE, CONF_CHANNEL, CONF_ID, CONF_SAMPLE_RATE
 from esphome.core import CORE
-from esphome.cpp_generator import MockObjClass
+from esphome.cpp_generator import MockObj, MockObjClass
 import esphome.final_validate as fv
+from esphome.types import ConfigType
 
 CODEOWNERS = ["@jesserockz"]
 DEPENDENCIES = ["esp32"]
@@ -145,7 +146,7 @@ I2S_MCLK_MULTIPLE = {
 _validate_bits = cv.float_with_unit("bits", "bit")
 
 
-def validate_mclk_divisible_by_3(config):
+def validate_mclk_divisible_by_3(config: ConfigType) -> ConfigType:
     if config[CONF_BITS_PER_SAMPLE] == 24 and config[CONF_MCLK_MULTIPLE] % 3 != 0:
         raise cv.Invalid(
             f"{CONF_MCLK_MULTIPLE} must be divisible by 3 when bits per sample is 24"
@@ -159,7 +160,7 @@ def i2s_audio_component_schema(
     default_sample_rate: int,
     default_channel: str,
     default_bits_per_sample: str,
-):
+) -> cv.Schema:
     return cv.Schema(
         {
             cv.GenerateID(): cv.declare_id(class_),
@@ -182,7 +183,7 @@ def i2s_audio_component_schema(
     )
 
 
-async def register_i2s_audio_component(var, config):
+async def register_i2s_audio_component(var: MockObj, config: ConfigType) -> None:
     await cg.register_parented(var, config[CONF_I2S_AUDIO_ID])
     cg.add(var.set_i2s_role(I2S_ROLE_OPTIONS[config[CONF_I2S_MODE]]))
     slot_mode = config[CONF_CHANNEL]
@@ -260,7 +261,7 @@ def _assign_ports() -> None:
             next_port += 1
 
 
-def _final_validate(_):
+def _final_validate(_: ConfigType) -> None:
     i2s_audio_configs = fv.full_config.get()[CONF_I2S_AUDIO]
     variant = get_esp32_variant()
     if variant not in I2S_PORTS:
@@ -275,7 +276,7 @@ def _final_validate(_):
 FINAL_VALIDATE_SCHEMA = _final_validate
 
 
-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)
 

esphome/components/i2s_audio/microphone/__init__.py

@@ -10,6 +10,7 @@ from esphome.const import (
     CONF_NUM_CHANNELS,
     CONF_SAMPLE_RATE,
 )
+from esphome.types import ConfigType
 
 from .. import (
     CONF_ADC_TYPE,
@@ -46,7 +47,7 @@ I2S_PDM_DSR = {
 }
 
 
-def _validate_esp32_variant(config):
+def _validate_esp32_variant(config: ConfigType) -> ConfigType:
     variant = esp32.get_esp32_variant()
     if config[CONF_ADC_TYPE] == "external":
         if config[CONF_PDM] and variant not in PDM_VARIANTS:
@@ -65,13 +66,13 @@ def _validate_esp32_variant(config):
     raise NotImplementedError
 
 
-def _validate_channel(config):
+def _validate_channel(config: ConfigType) -> ConfigType:
     if config[CONF_CHANNEL] == CONF_MONO:
         raise cv.Invalid(f"I2S microphone does not support {CONF_MONO}.")
     return config
 
 
-def _set_num_channels_from_config(config):
+def _set_num_channels_from_config(config: ConfigType) -> ConfigType:
     if config[CONF_CHANNEL] in (CONF_LEFT, CONF_RIGHT):
         config[CONF_NUM_CHANNELS] = 1
     else:
@@ -80,7 +81,7 @@ def _set_num_channels_from_config(config):
     return config
 
 
-def _set_stream_limits(config):
+def _set_stream_limits(config: ConfigType) -> ConfigType:
     audio.set_stream_limits(
         min_bits_per_sample=config.get(CONF_BITS_PER_SAMPLE),
         max_bits_per_sample=config.get(CONF_BITS_PER_SAMPLE),
@@ -134,7 +135,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-def _final_validate(config):
+def _final_validate(config: ConfigType) -> None:
     if config[CONF_ADC_TYPE] == "internal":
         raise cv.Invalid(
             "Internal ADC is no longer supported. Use an external I2S microphone instead."
@@ -144,7 +145,7 @@ def _final_validate(config):
 FINAL_VALIDATE_SCHEMA = _final_validate
 
 
-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 register_i2s_audio_component(var, config)

esphome/components/i2s_audio/speaker/__init__.py

@@ -13,6 +13,7 @@ from esphome.const import (
     CONF_SAMPLE_RATE,
     CONF_TIMEOUT,
 )
+from esphome.types import ConfigType
 
 from .. import (
     CONF_I2S_DOUT_PIN,
@@ -78,7 +79,7 @@ I2C_COMM_FMT_OPTIONS = {
 INTERNAL_DAC_VARIANTS = [esp32.VARIANT_ESP32]
 
 
-def _set_num_channels_from_config(config):
+def _set_num_channels_from_config(config: ConfigType) -> ConfigType:
     if config[CONF_CHANNEL] in (CONF_MONO, CONF_LEFT, CONF_RIGHT):
         config[CONF_NUM_CHANNELS] = 1
     else:
@@ -87,7 +88,7 @@ def _set_num_channels_from_config(config):
     return config
 
 
-def _set_stream_limits(config):
+def _set_stream_limits(config: ConfigType) -> ConfigType:
     if config.get(CONF_SPDIF_MODE, False):
         # SPDIF mode: 16/24/32-bit audio and stereo at configured sample rate
         audio.set_stream_limits(
@@ -133,14 +134,14 @@ def _set_stream_limits(config):
     return config
 
 
-def _select_speaker_class(config):
+def _select_speaker_class(config: ConfigType) -> ConfigType:
     """Override ID type when SPDIF mode is enabled."""
     if config.get(CONF_SPDIF_MODE, False):
         config[CONF_ID].type = I2SAudioSpeakerSPDIF
     return config
 
 
-def _validate_esp32_variant(config):
+def _validate_esp32_variant(config: ConfigType) -> ConfigType:
     variant = esp32.get_esp32_variant()
     if config[CONF_DAC_TYPE] == "internal":
         if variant not in INTERNAL_DAC_VARIANTS:
@@ -207,7 +208,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-def _final_validate(config):
+def _final_validate(config: ConfigType) -> None:
     if config[CONF_DAC_TYPE] == "internal":
         raise cv.Invalid(
             "Internal DAC is no longer supported. Use an external I2S DAC instead."
@@ -238,7 +239,7 @@ def _final_validate(config):
 FINAL_VALIDATE_SCHEMA = _final_validate
 
 
-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 register_i2s_audio_component(var, config)

esphome/components/mcp23xxx_base/__init__.py

@@ -15,6 +15,8 @@ from esphome.const import (
     CONF_PULLUP,
 )
 from esphome.core import CORE, ID, coroutine
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType
 
 AUTO_LOAD = ["gpio_expander"]
 CODEOWNERS = ["@jesserockz"]
@@ -41,7 +43,7 @@ MCP23XXX_CONFIG_SCHEMA = cv.Schema(
 
 
 @coroutine
-async def register_mcp23xxx(config, num_pins):
+async def register_mcp23xxx(config: ConfigType, num_pins: int) -> MockObj:
     id: ID = config[CONF_ID]
     var = cg.new_Pvariable(id)
     await cg.register_component(var, config)
@@ -52,7 +54,7 @@ async def register_mcp23xxx(config, num_pins):
     return var
 
 
-def validate_mode(value):
+def validate_mode(value: ConfigType) -> ConfigType:
     if not (value[CONF_INPUT] or value[CONF_OUTPUT]):
         raise cv.Invalid("Mode must be either input or output")
     if value[CONF_INPUT] and value[CONF_OUTPUT]:
@@ -81,7 +83,7 @@ MCP23XXX_PIN_SCHEMA = pins.gpio_base_schema(
 
 
 @pins.PIN_SCHEMA_REGISTRY.register(CONF_MCP23XXX, MCP23XXX_PIN_SCHEMA)
-async def mcp23xxx_pin_to_code(config):
+async def mcp23xxx_pin_to_code(config: ConfigType) -> MockObj:
     parent_id: ID = config[CONF_MCP23XXX]
     parent = await cg.get_variable(parent_id)
 

esphome/components/mcp4461/__init__.py

@@ -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
 
 CODEOWNERS = ["@p1ngb4ck"]
 DEPENDENCIES = ["i2c"]
@@ -30,7 +31,7 @@ CONFIG_SCHEMA = (
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(
         config[CONF_ID],
         config[CONF_DISABLE_WIPER_0],

esphome/components/mcp4461/output/__init__.py

@@ -3,6 +3,9 @@ import esphome.codegen as cg
 from esphome.components import output
 import esphome.config_validation as cv
 from esphome.const import CONF_CHANNEL, CONF_ID, CONF_INITIAL_VALUE
+from esphome.core import ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 from .. import CONF_MCP4461_ID, Mcp4461Component, mcp4461_ns
 
@@ -34,7 +37,7 @@ CONF_NONVOLATILE_WRITE_DELAY = "nonvolatile_write_delay"
 VOLATILE_CHANNELS = ("A", "B", "C", "D")
 
 
-def _validate_nonvolatile(config) -> None:
+def _validate_nonvolatile(config: ConfigType) -> None:
     channel = str(config[CONF_CHANNEL])
 
     # Channels E-H address the nonvolatile registers directly — the mirroring options only
@@ -89,7 +92,7 @@ CONFIG_SCHEMA = output.FLOAT_OUTPUT_SCHEMA.extend(
 FINAL_VALIDATE_SCHEMA = _validate_nonvolatile
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     parent = await cg.get_variable(config[CONF_MCP4461_ID])
     var = cg.new_Pvariable(
         config[CONF_ID],
@@ -147,7 +150,12 @@ TERMINAL_ACTION_SCHEMA = cv.Schema(
 @automation.register_action(
     "mcp4461.wiper.decrease", WiperDecreaseAction, WIPER_ACTION_SCHEMA, synchronous=True
 )
-async def mcp4461_wiper_step_to_code(config, action_id, template_arg, args):
+async def mcp4461_wiper_step_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     wiper = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, wiper)
 
@@ -158,7 +166,12 @@ async def mcp4461_wiper_step_to_code(config, action_id, template_arg, args):
     WIPER_ACTION_SCHEMA,
     synchronous=True,
 )
-async def mcp4461_wiper_store_to_code(config, action_id, template_arg, args):
+async def mcp4461_wiper_store_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     wiper = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, wiper)
 
@@ -169,7 +182,12 @@ async def mcp4461_wiper_store_to_code(config, action_id, template_arg, args):
     TERMINAL_ACTION_SCHEMA,
     synchronous=True,
 )
-async def mcp4461_wiper_terminal_to_code(config, action_id, template_arg, args):
+async def mcp4461_wiper_terminal_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     wiper = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(
         action_id, template_arg, wiper, ord(config[CONF_TERMINAL]), config[CONF_ENABLE]

esphome/components/microphone/__init__.py

@@ -1,3 +1,5 @@
+from collections.abc import Callable
+
 from esphome import automation
 from esphome.automation import maybe_simple_id
 import esphome.codegen as cg
@@ -12,8 +14,10 @@ from esphome.const import (
     CONF_ON_DATA,
     CONF_TRIGGER_ID,
 )
-from esphome.core import CORE
+from esphome.core import CORE, ID
 from esphome.coroutine import CoroPriority, coroutine_with_priority
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 AUTO_LOAD = ["audio"]
 CODEOWNERS = ["@jesserockz", "@kahrendt"]
@@ -50,7 +54,7 @@ IsCapturingCondition = microphone_ns.class_(
 IsMutedCondition = microphone_ns.class_("IsMutedCondition", automation.Condition)
 
 
-async def setup_microphone_core_(var, config):
+async def setup_microphone_core_(var: MockObj, config: ConfigType) -> None:
     for conf in config.get(CONF_ON_DATA, []):
         trigger = cg.new_Pvariable(conf[CONF_TRIGGER_ID], var)
         await automation.build_automation(
@@ -60,7 +64,7 @@ async def setup_microphone_core_(var, config):
         )
 
 
-async def register_microphone(var, config):
+async def register_microphone(var: MockObj, config: ConfigType) -> None:
     if not CORE.has_id(config[CONF_ID]):
         var = cg.Pvariable(config[CONF_ID], var)
     await setup_microphone_core_(var, config)
@@ -85,7 +89,7 @@ def microphone_source_schema(
     max_bits_per_sample: int = 16,
     min_channels: int = 1,
     max_channels: int = 1,
-):
+) -> cv.All:
     """Schema for a microphone source
 
     Components requesting microphone data should use this schema instead of accessing a microphone directly.
@@ -97,7 +101,7 @@ def microphone_source_schema(
       max_channels (int, optional): Maximum number of channels the requesting component supports. Defaults to 1.
     """
 
-    def _validate_unique_channels(config):
+    def _validate_unique_channels(config: list[int]) -> list[int]:
         if len(config) != len(set(config)):
             raise cv.Invalid("Channels must be unique")
         return config
@@ -124,7 +128,7 @@ def microphone_source_schema(
 
 def final_validate_microphone_source_schema(
     component_name: str, sample_rate: int = cv.UNDEFINED
-):
+) -> Callable[[ConfigType], ConfigType]:
     """Validates that the microphone source can provide audio in the correct format. In particular it validates the sample rate and the enabled channels.
 
     Note that:
@@ -136,7 +140,7 @@ def final_validate_microphone_source_schema(
         sample_rate (int, optional): The sample rate the component requesting mic audio requires
     """
 
-    def _validate_audio_compatability(config):
+    def _validate_audio_compatability(config: ConfigType) -> ConfigType:
         if sample_rate is not cv.UNDEFINED:
             # Issues require changing the microphone configuration
             #  - Verifies sample rates match
@@ -161,7 +165,9 @@ def final_validate_microphone_source_schema(
     return _validate_audio_compatability
 
 
-async def microphone_source_to_code(config, passive=False):
+async def microphone_source_to_code(
+    config: ConfigType, passive: bool = False
+) -> MockObj:
     """Creates a MicrophoneSource variable for codegen.
 
     Setting passive to true makes the MicrophoneSource never start/stop the microphone, but only receives audio when another component has actively started the Microphone. If false, then the microphone needs to be explicitly started/stopped.
@@ -183,7 +189,12 @@ async def microphone_source_to_code(config, passive=False):
     return mic_source
 
 
-async def microphone_action(config, action_id, template_arg, args):
+async def microphone_action(
+    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
@@ -219,6 +230,6 @@ automation.register_condition(
 
 
 @coroutine_with_priority(CoroPriority.CORE)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     cg.add_global(microphone_ns.using)
     cg.add_define("USE_MICROPHONE")

esphome/components/pn532/__init__.py

@@ -9,6 +9,9 @@ from esphome.const import (
     CONF_ON_TAG_REMOVED,
     CONF_TRIGGER_ID,
 )
+from esphome.core import ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 CODEOWNERS = ["@OttoWinter", "@jesserockz"]
 AUTO_LOAD = ["binary_sensor", "nfc"]
@@ -41,7 +44,7 @@ PN532_SCHEMA = cv.Schema(
 ).extend(cv.polling_component_schema("1s"))
 
 
-def CONFIG_SCHEMA(conf):
+def CONFIG_SCHEMA(conf: ConfigType) -> None:
     if conf:
         raise cv.Invalid(
             "This component has been moved in 1.16, please see the docs for updated "
@@ -56,7 +59,7 @@ _CALLBACK_AUTOMATIONS = (
 )
 
 
-async def setup_pn532(var, config):
+async def setup_pn532(var: MockObj, config: ConfigType) -> None:
     await cg.register_component(var, config)
 
     for conf in config.get(CONF_ON_TAG, []):
@@ -85,7 +88,12 @@ async def setup_pn532(var, config):
         }
     ),
 )
-async def pn532_is_writing_to_code(config, condition_id, template_arg, args):
+async def pn532_is_writing_to_code(
+    config: ConfigType,
+    condition_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(condition_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     return var

esphome/components/pn532/binary_sensor.py

@@ -1,15 +1,18 @@
+from typing import Any
+
 import esphome.codegen as cg
 from esphome.components import binary_sensor
 import esphome.config_validation as cv
 from esphome.const import CONF_UID
 from esphome.core import HexInt
+from esphome.types import ConfigType
 
 from . import CONF_PN532_ID, PN532, pn532_ns
 
 DEPENDENCIES = ["pn532"]
 
 
-def validate_uid(value):
+def validate_uid(value: Any) -> str:
     value = cv.string_strict(value)
     for x in value.split("-"):
         if len(x) != 2:
@@ -39,7 +42,7 @@ CONFIG_SCHEMA = binary_sensor.binary_sensor_schema(PN532BinarySensor).extend(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = await binary_sensor.new_binary_sensor(config)
 
     hub = await cg.get_variable(config[CONF_PN532_ID])

esphome/components/pn7150/__init__.py

@@ -12,6 +12,9 @@ from esphome.const import (
     CONF_ON_TAG_REMOVED,
     CONF_TRIGGER_ID,
 )
+from esphome.core import ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 AUTO_LOAD = ["binary_sensor", "nfc"]
 CODEOWNERS = ["@kbx81", "@jesserockz"]
@@ -107,7 +110,12 @@ PN7150_SCHEMA = cv.Schema(
     SET_MESSAGE_ACTION_SCHEMA,
     synchronous=True,
 )
-async def pn7150_set_message_to_code(config, action_id, template_arg, args):
+async def pn7150_set_message_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_MESSAGE], args, cg.std_string)
@@ -158,7 +166,12 @@ async def pn7150_set_message_to_code(config, action_id, template_arg, args):
     SIMPLE_ACTION_SCHEMA,
     synchronous=True,
 )
-async def pn7150_simple_action_to_code(config, action_id, template_arg, args):
+async def pn7150_simple_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
@@ -174,7 +187,7 @@ _CALLBACK_AUTOMATIONS = (
 )
 
 
-async def setup_pn7150(var, config):
+async def setup_pn7150(var: MockObj, config: ConfigType) -> None:
     await cg.register_component(var, config)
 
     pin = await cg.gpio_pin_expression(config[CONF_IRQ_PIN])
@@ -216,7 +229,12 @@ async def setup_pn7150(var, config):
         }
     ),
 )
-async def pn7150_is_writing_to_code(config, condition_id, template_arg, args):
+async def pn7150_is_writing_to_code(
+    config: ConfigType,
+    condition_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(condition_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     return var

esphome/components/pn7160/__init__.py

@@ -12,6 +12,9 @@ from esphome.const import (
     CONF_ON_TAG_REMOVED,
     CONF_TRIGGER_ID,
 )
+from esphome.core import ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 AUTO_LOAD = ["binary_sensor", "nfc"]
 CODEOWNERS = ["@kbx81", "@jesserockz"]
@@ -111,7 +114,12 @@ PN7160_SCHEMA = cv.Schema(
     SET_MESSAGE_ACTION_SCHEMA,
     synchronous=True,
 )
-async def pn7160_set_message_to_code(config, action_id, template_arg, args):
+async def pn7160_set_message_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_MESSAGE], args, cg.std_string)
@@ -162,7 +170,12 @@ async def pn7160_set_message_to_code(config, action_id, template_arg, args):
     SIMPLE_ACTION_SCHEMA,
     synchronous=True,
 )
-async def pn7160_simple_action_to_code(config, action_id, template_arg, args):
+async def pn7160_simple_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
@@ -178,7 +191,7 @@ _CALLBACK_AUTOMATIONS = (
 )
 
 
-async def setup_pn7160(var, config):
+async def setup_pn7160(var: MockObj, config: ConfigType) -> None:
     await cg.register_component(var, config)
 
     if dwl_req_pin_config := config.get(CONF_DWL_REQ_PIN):
@@ -228,7 +241,12 @@ async def setup_pn7160(var, config):
         }
     ),
 )
-async def pn7160_is_writing_to_code(config, condition_id, template_arg, args):
+async def pn7160_is_writing_to_code(
+    config: ConfigType,
+    condition_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(condition_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     return var