[core] Add type annotations to component Python (10/11) (#18347)

Co-authored-by: Jonathan Swoboda <154711427+swoboda1337@users.noreply.github.com>

esphome/components/adc/__init__.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from esphome import pins
 import esphome.codegen as cg
 from esphome.components.esp32 import (
@@ -16,6 +18,7 @@ from esphome.components.esp32 import (
 import esphome.config_validation as cv
 from esphome.const import CONF_ANALOG, CONF_INPUT, CONF_NUMBER, PLATFORM_ESP8266
 from esphome.core import CORE
+from esphome.types import ConfigType
 
 CODEOWNERS = ["@esphome/core"]
 
@@ -225,7 +228,7 @@ ESP32_VARIANT_ADC2_PIN_TO_CHANNEL = {
 }
 
 
-def validate_adc_pin(value):
+def validate_adc_pin(value: Any) -> ConfigType | str:
     if str(value).upper() == "VCC":
         if CORE.is_rp2:
             return pins.internal_gpio_input_pin_schema(29)

esphome/components/adc/sensor.py

@@ -52,7 +52,7 @@ _attenuation = cv.enum(ATTENUATION_MODES, lower=True)
 _sampling_mode = cv.enum(SAMPLING_MODES, lower=True)
 
 
-def validate_config(config):
+def validate_config(config: ConfigType) -> ConfigType:
     if config[CONF_RAW] and config.get(CONF_ATTENUATION, None) == "auto":
         raise cv.Invalid("Automatic attenuation cannot be used when raw output is set")
 
@@ -120,7 +120,7 @@ CONFIG_SCHEMA = cv.All(
 CONF_ADC_CHANNEL_ID = "adc_channel_id"
 
 
-def _overlay_io_channels():
+def _overlay_io_channels() -> str:
     channel_count = CORE.data[CONF_ADC_CHANNEL_ID]
     entries = ", ".join(f"<&adc {channel_id}>" for channel_id in range(channel_count))
     return f"""
@@ -132,7 +132,7 @@ def _overlay_io_channels():
             """
 
 
-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 sensor.register_sensor(var, config)

esphome/components/api/__init__.py

@@ -1,5 +1,6 @@
 import base64
 import logging
+from typing import Any
 
 from esphome import automation
 from esphome.automation import Condition
@@ -129,7 +130,7 @@ def _register_provisioning_source(config: ConfigType) -> ConfigType:
     return config
 
 
-def validate_encryption_key(value):
+def validate_encryption_key(value: Any) -> str:
     value = cv.string_strict(value)
     try:
         decoded = base64.b64decode(value, validate=True)
@@ -217,7 +218,7 @@ def _auto_detect_supports_response(config: ConfigType) -> ConfigType:
     return config
 
 
-def _validate_supports_response(value):
+def _validate_supports_response(value: Any) -> str:
     """Validate supports_response after auto-detection has set the value."""
     return cv.enum(SUPPORTS_RESPONSE_OPTIONS, lower=True)(value)
 
@@ -256,7 +257,7 @@ ENCRYPTION_SCHEMA = cv.Schema(
 )
 
 
-def _encryption_schema(config):
+def _encryption_schema(config: ConfigType | None) -> ConfigType:
     if config is None:
         config = {}
     return ENCRYPTION_SCHEMA(config)
@@ -393,7 +394,7 @@ async def to_code(config: ConfigType) -> None:
 
     if actions := config.get(CONF_ACTIONS, []):
         # Collect all triggers first, then register all at once with initializer_list
-        triggers: list[cg.Pvariable] = []
+        triggers: list[cg.MockObj] = []
         for conf in actions:
             func_args: list[tuple[MockObj, str]] = []
             service_template_args: list[MockObj] = []  # User service argument types
@@ -581,7 +582,7 @@ async def homeassistant_service_to_code(
     action_id: ID,
     template_arg: cg.TemplateArguments,
     args: TemplateArgsType,
-):
+) -> MockObj:
     cg.add_define("USE_API_HOMEASSISTANT_SERVICES")
     serv = await cg.get_variable(config[CONF_ID])
     var = cg.new_Pvariable(action_id, template_arg, serv, False)
@@ -647,7 +648,7 @@ async def homeassistant_service_to_code(
     return var
 
 
-def validate_homeassistant_event(value):
+def validate_homeassistant_event(value: Any) -> str:
     value = cv.string(value)
     if not value.startswith("esphome."):
         raise cv.Invalid(
@@ -676,7 +677,12 @@ HOMEASSISTANT_EVENT_ACTION_SCHEMA = cv.Schema(
     HOMEASSISTANT_EVENT_ACTION_SCHEMA,
     synchronous=True,
 )
-async def homeassistant_event_to_code(config, action_id, template_arg, args):
+async def homeassistant_event_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     cg.add_define("USE_API_HOMEASSISTANT_SERVICES")
     serv = await cg.get_variable(config[CONF_ID])
     var = cg.new_Pvariable(action_id, template_arg, serv, True)
@@ -724,7 +730,12 @@ HOMEASSISTANT_TAG_SCANNED_ACTION_SCHEMA = cv.maybe_simple_value(
     HOMEASSISTANT_TAG_SCANNED_ACTION_SCHEMA,
     synchronous=True,
 )
-async def homeassistant_tag_scanned_to_code(config, action_id, template_arg, args):
+async def homeassistant_tag_scanned_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     cg.add_define("USE_API_HOMEASSISTANT_SERVICES")
     serv = await cg.get_variable(config[CONF_ID])
     var = cg.new_Pvariable(action_id, template_arg, serv, True)
@@ -740,7 +751,7 @@ CONF_SUCCESS = "success"
 CONF_ERROR_MESSAGE = "error_message"
 
 
-def _validate_api_respond_data(config):
+def _validate_api_respond_data(config: ConfigType) -> ConfigType:
     """Set flag during validation so AUTO_LOAD can include json component."""
     if CONF_DATA in config:
         CORE.data.setdefault(DOMAIN, {})[CONF_CAPTURE_RESPONSE] = True
@@ -824,7 +835,12 @@ API_CONNECTED_CONDITION_SCHEMA = cv.Schema(
 @automation.register_condition(
     "api.connected", APIConnectedCondition, API_CONNECTED_CONDITION_SCHEMA
 )
-async def api_connected_to_code(config, condition_id, template_arg, args):
+async def api_connected_to_code(
+    config: ConfigType,
+    condition_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(condition_id, template_arg)
     templ = await cg.templatable(config[CONF_STATE_SUBSCRIPTION_ONLY], args, cg.bool_)
     cg.add(var.set_state_subscription_only(templ))

esphome/components/button/__init__.py

@@ -16,14 +16,15 @@ from esphome.const import (
     DEVICE_CLASS_RESTART,
     DEVICE_CLASS_UPDATE,
 )
-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_device_class,
     setup_entity,
 )
-from esphome.cpp_generator import MockObjClass
+from esphome.cpp_generator import MockObj, MockObjClass, TemplateArgsType
+from esphome.types import ConfigType, SafeExpType
 
 CODEOWNERS = ["@esphome/core"]
 IS_PLATFORM_COMPONENT = True
@@ -88,7 +89,7 @@ _CALLBACK_AUTOMATIONS = (
 
 
 @setup_entity("button")
-async def setup_button_core_(var, config):
+async def setup_button_core_(var: MockObj, config: ConfigType) -> None:
     await automation.build_callback_automations(var, config, _CALLBACK_AUTOMATIONS)
 
     setup_device_class(config)
@@ -101,7 +102,7 @@ async def setup_button_core_(var, config):
         await web_server.add_entity_config(var, web_server_config)
 
 
-async def register_button(var, config):
+async def register_button(var: MockObj, config: ConfigType) -> None:
     if not CORE.has_id(config[CONF_ID]):
         var = cg.Pvariable(config[CONF_ID], var)
     queue_entity_register("button", config)
@@ -109,7 +110,7 @@ async def register_button(var, config):
     await setup_button_core_(var, config)
 
 
-async def new_button(config, *args):
+async def new_button(config: ConfigType, *args: SafeExpType) -> MockObj:
     var = cg.new_Pvariable(config[CONF_ID], *args)
     await register_button(var, config)
     return var
@@ -125,11 +126,16 @@ BUTTON_PRESS_SCHEMA = maybe_simple_id(
 @automation.register_action(
     "button.press", PressAction, BUTTON_PRESS_SCHEMA, synchronous=True
 )
-async def button_press_to_code(config, action_id, template_arg, args):
+async def button_press_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)
 
 
 @coroutine_with_priority(CoroPriority.CORE)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     cg.add_global(button_ns.using)

esphome/components/climate/__init__.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from esphome import automation
 import esphome.codegen as cg
 from esphome.components import mqtt, web_server
@@ -48,13 +50,19 @@ from esphome.const import (
     CONF_VISUAL,
     CONF_WEB_SERVER,
 )
-from esphome.core import CORE, CoroPriority, Lambda, coroutine_with_priority
+from esphome.core import CORE, ID, CoroPriority, Lambda, coroutine_with_priority
 from esphome.core.entity_helpers import (
     entity_duplicate_validator,
     queue_entity_register,
     setup_entity,
 )
-from esphome.cpp_generator import LambdaExpression, MockObjClass
+from esphome.cpp_generator import (
+    LambdaExpression,
+    MockObj,
+    MockObjClass,
+    TemplateArgsType,
+)
+from esphome.types import ConfigType, SafeExpType
 
 IS_PLATFORM_COMPONENT = True
 
@@ -132,7 +140,7 @@ VISUAL_TEMPERATURE_STEP_SCHEMA = cv.Schema(
 )
 
 
-def visual_temperature_step(value):
+def visual_temperature_step(value: Any) -> ConfigType:
     # Allow defining target/current temperature steps separately
     if isinstance(value, dict):
         return VISUAL_TEMPERATURE_STEP_SCHEMA(value)
@@ -273,7 +281,7 @@ def climate_schema(
 
 
 @setup_entity("climate")
-async def setup_climate_core_(var, config):
+async def setup_climate_core_(var: MockObj, config: ConfigType) -> None:
     visual = config.get(CONF_VISUAL, {})
     if (min_temp := visual.get(CONF_MIN_TEMPERATURE)) is not None:
         cg.add_define("USE_CLIMATE_VISUAL_OVERRIDES")
@@ -443,7 +451,7 @@ async def setup_climate_core_(var, config):
         await web_server.add_entity_config(var, web_server_config)
 
 
-async def register_climate(var, config):
+async def register_climate(var: MockObj, config: ConfigType) -> None:
     if not CORE.has_id(config[CONF_ID]):
         var = cg.Pvariable(config[CONF_ID], var)
     queue_entity_register("climate", config)
@@ -451,7 +459,7 @@ async def register_climate(var, config):
     await setup_climate_core_(var, config)
 
 
-async def new_climate(config, *args):
+async def new_climate(config: ConfigType, *args: SafeExpType) -> MockObj:
     var = cg.new_Pvariable(config[CONF_ID], *args)
     await register_climate(var, config)
     return var
@@ -485,7 +493,12 @@ CLIMATE_CONTROL_ACTION_SCHEMA = cv.Schema(
     CLIMATE_CONTROL_ACTION_SCHEMA,
     synchronous=True,
 )
-async def climate_control_to_code(config, action_id, template_arg, args):
+async def climate_control_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     paren = await cg.get_variable(config[CONF_ID])
 
     # All configured fields are folded into a single stateless lambda whose
@@ -549,5 +562,5 @@ async def climate_control_to_code(config, action_id, template_arg, args):
 
 
 @coroutine_with_priority(CoroPriority.CORE)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     cg.add_global(climate_ns.using)

esphome/components/cover/__init__.py

@@ -46,7 +46,7 @@ from esphome.core.entity_helpers import (
     setup_entity,
 )
 from esphome.cpp_generator import LambdaExpression, MockObj, MockObjClass
-from esphome.types import ConfigType, TemplateArgsType
+from esphome.types import ConfigType, SafeExpType, TemplateArgsType
 
 IS_PLATFORM_COMPONENT = True
 
@@ -162,7 +162,7 @@ _COVER_SCHEMA = (
 _COVER_SCHEMA.add_extra(entity_duplicate_validator("cover"))
 
 
-def _validate_mqtt_state_topics(config):
+def _validate_mqtt_state_topics(config: ConfigType) -> ConfigType:
     if config.get(CONF_MQTT_JSON_STATE_PAYLOAD):
         if CONF_POSITION_STATE_TOPIC in config:
             raise cv.Invalid(
@@ -201,7 +201,7 @@ def cover_schema(
 
 
 @setup_entity("cover")
-async def setup_cover_core_(var, config):
+async def setup_cover_core_(var: MockObj, config: ConfigType) -> None:
     setup_device_class(config)
 
     if CONF_ON_OPEN in config:
@@ -235,7 +235,7 @@ async def setup_cover_core_(var, config):
         await web_server.add_entity_config(var, web_server_config)
 
 
-async def register_cover(var, config):
+async def register_cover(var: MockObj, config: ConfigType) -> None:
     if not CORE.has_id(config[CONF_ID]):
         var = cg.Pvariable(config[CONF_ID], var)
     queue_entity_register("cover", config)
@@ -243,7 +243,7 @@ async def register_cover(var, config):
     await setup_cover_core_(var, config)
 
 
-async def new_cover(config, *args):
+async def new_cover(config: ConfigType, *args: SafeExpType) -> MockObj:
     var = cg.new_Pvariable(config[CONF_ID], *args)
     await register_cover(var, config)
     return var
@@ -259,7 +259,12 @@ COVER_ACTION_SCHEMA = maybe_simple_id(
 @automation.register_action(
     "cover.open", OpenAction, COVER_ACTION_SCHEMA, synchronous=True
 )
-async def cover_open_to_code(config, action_id, template_arg, args):
+async def cover_open_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)
 
@@ -267,7 +272,12 @@ async def cover_open_to_code(config, action_id, template_arg, args):
 @automation.register_action(
     "cover.close", CloseAction, COVER_ACTION_SCHEMA, synchronous=True
 )
-async def cover_close_to_code(config, action_id, template_arg, args):
+async def cover_close_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)
 
@@ -275,7 +285,12 @@ async def cover_close_to_code(config, action_id, template_arg, args):
 @automation.register_action(
     "cover.stop", StopAction, COVER_ACTION_SCHEMA, synchronous=True
 )
-async def cover_stop_to_code(config, action_id, template_arg, args):
+async def cover_stop_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)
 
@@ -283,7 +298,12 @@ async def cover_stop_to_code(config, action_id, template_arg, args):
 @automation.register_action(
     "cover.toggle", ToggleAction, COVER_ACTION_SCHEMA, synchronous=True
 )
-async def cover_toggle_to_code(config, action_id, template_arg, args):
+async def cover_toggle_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)
 
@@ -421,5 +441,5 @@ automation.register_condition(
 
 
 @coroutine_with_priority(CoroPriority.CORE)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     cg.add_global(cover_ns.using)

esphome/components/dashboard_import/__init__.py

@@ -2,6 +2,7 @@ import base64
 from pathlib import Path
 import re
 import secrets
+from typing import Any
 
 import requests
 from ruamel.yaml import YAML
@@ -13,6 +14,7 @@ import esphome.config_validation as cv
 from esphome.const import CONF_ESPHOME, CONF_PROJECT, CONF_REF, CONF_WIFI
 import esphome.final_validate as fv
 from esphome.happy_eyeballs import ensure_happy_eyeballs
+from esphome.types import ConfigType
 from esphome.yaml_util import dump
 
 dashboard_import_ns = cg.esphome_ns.namespace("dashboard_import")
@@ -23,14 +25,14 @@ DEPENDENCIES = ["api"]
 CODEOWNERS = ["@esphome/core"]
 
 
-def validate_import_url(value):
+def validate_import_url(value: Any) -> str:
     value = cv.string_strict(value)
     value = cv.Length(max=255)(value)
     validate_source_shorthand(value)
     return value
 
 
-def validate_full_url(config):
+def validate_full_url(config: ConfigType) -> ConfigType:
     if not config[CONF_IMPORT_FULL_CONFIG]:
         return config
     source = validate_source_shorthand(config[CONF_PACKAGE_IMPORT_URL])
@@ -55,7 +57,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-def _final_validate(config):
+def _final_validate(config: ConfigType) -> None:
     full_config = fv.full_config.get()[CONF_ESPHOME]
     if CONF_PROJECT not in full_config:
         raise cv.Invalid(
@@ -73,7 +75,7 @@ wifi:
 """
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     cg.add_define("USE_DASHBOARD_IMPORT")
     url = config[CONF_PACKAGE_IMPORT_URL]
     if config[CONF_IMPORT_FULL_CONFIG]:

esphome/components/debug/__init__.py

@@ -12,6 +12,7 @@ from esphome.const import (
     PlatformFramework,
 )
 from esphome.core import CORE
+from esphome.types import ConfigType
 
 CODEOWNERS = ["@esphome/core"]
 DEPENDENCIES = ["logger"]
@@ -45,7 +46,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     if CORE.using_zephyr:
         zephyr_add_prj_conf("HWINFO", True)
         # gdb thread support

esphome/components/debug/sensor.py

@@ -21,6 +21,7 @@ from esphome.const import (
     UNIT_MILLISECOND,
     UNIT_PERCENT,
 )
+from esphome.types import ConfigType
 
 from . import (  # noqa: F401  pylint: disable=unused-import
     CONF_DEBUG_ID,
@@ -111,7 +112,7 @@ CONFIG_SCHEMA = {
 }
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     debug_component = await cg.get_variable(config[CONF_DEBUG_ID])
 
     if free_conf := config.get(CONF_FREE):

esphome/components/debug/text_sensor.py

@@ -7,6 +7,7 @@ from esphome.const import (
     ICON_CHIP,
     ICON_RESTART,
 )
+from esphome.types import ConfigType
 
 from . import (  # noqa: F401  pylint: disable=unused-import
     CONF_DEBUG_ID,
@@ -33,7 +34,7 @@ CONFIG_SCHEMA = cv.Schema(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     debug_component = await cg.get_variable(config[CONF_DEBUG_ID])
 
     if CONF_DEVICE in config:

esphome/components/esp8266/__init__.py

@@ -3,6 +3,7 @@ from pathlib import Path
 import platform
 import re
 import subprocess
+from typing import Any
 
 import esphome.codegen as cg
 import esphome.config_validation as cv
@@ -31,6 +32,7 @@ from esphome.core import (
 from esphome.core.config import BOARD_MAX_LENGTH
 from esphome.helpers import IS_MACOS, copy_file_if_changed
 from esphome.platformio.toolchain import copy_ccache_script
+from esphome.storage_json import StorageJSON
 from esphome.types import ConfigType
 
 from .boards import BOARDS, ESP8266_LD_SCRIPTS
@@ -88,7 +90,7 @@ def lambdas_use_scanf_float(config: ConfigType) -> bool:
     return False
 
 
-def set_core_data(config):
+def set_core_data(config: ConfigType) -> ConfigType:
     CORE.data[KEY_ESP8266] = {}
     CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] = PLATFORM_ESP8266
     CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] = "arduino"
@@ -102,7 +104,7 @@ def set_core_data(config):
     return config
 
 
-def get_download_types(storage_json):
+def get_download_types(storage_json: StorageJSON) -> list[dict[str, str]]:
     """Binary-download entries for a built ESP8266 firmware.
 
     Used by device-builder (esphome/device-builder), via
@@ -157,7 +159,7 @@ ARDUINO_3_PLATFORM_VERSION = cv.Version(3, 2, 0)
 ARDUINO_4_PLATFORM_VERSION = cv.Version(4, 2, 1)
 
 
-def _arduino_check_versions(value):
+def _arduino_check_versions(value: ConfigType) -> ConfigType:
     value = value.copy()
     lookups = {
         "dev": (cv.Version(3, 1, 2), "https://github.com/esp8266/Arduino.git"),
@@ -200,7 +202,7 @@ def _arduino_check_versions(value):
     return value
 
 
-def _parse_platform_version(value):
+def _parse_platform_version(value: Any) -> str:
     try:
         # if platform version is a valid version constraint, prefix the default package
         cv.platformio_version_constraint(value)
@@ -275,7 +277,7 @@ def check_rosetta() -> None:
 
 
 @coroutine_with_priority(CoroPriority.PLATFORM)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     cg.add(esp8266_ns.setup_preferences())
 
     cg.add_platformio_option("lib_ldf_mode", "off")
@@ -504,7 +506,7 @@ ESP8266_EXCEPTION_CODES = {
 }
 
 
-def _decode_pc(config, addr):
+def _decode_pc(config: ConfigType, addr: str) -> None:
     from esphome.platformio import toolchain
 
     idedata = toolchain.get_idedata(config)
@@ -525,7 +527,7 @@ def _decode_pc(config, addr):
     _LOGGER.warning("Decoded %s", translation)
 
 
-def _parse_register(config, regex, line):
+def _parse_register(config: ConfigType, regex: re.Pattern[str], line: str) -> None:
     match = regex.match(line)
     if match is not None:
         _decode_pc(config, match.group(1))
@@ -549,7 +551,7 @@ STACKTRACE_BAD_ALLOC_RE = re.compile(
 STACKTRACE_ESP8266_BACKTRACE_PC_RE = re.compile(r"4[0-9a-f]{7}")
 
 
-def process_stacktrace(config, line, backtrace_state):
+def process_stacktrace(config: ConfigType, line: str, backtrace_state: bool) -> bool:
     line = line.strip()
     # ESP8266 Exception type
     match = re.match(STACKTRACE_ESP8266_EXCEPTION_TYPE_RE, line)

esphome/components/esp8266/gpio.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass
 import logging
+from typing import Any
 
 from esphome import pins
 import esphome.codegen as cg
@@ -18,6 +19,8 @@ from esphome.const import (
     PLATFORM_ESP8266,
 )
 from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType
 
 from . import boards
 from .const import KEY_BOARD, KEY_ESP8266, KEY_PIN_INITIAL_STATES, esp8266_ns
@@ -27,7 +30,7 @@ _LOGGER = logging.getLogger(__name__)
 ESP8266GPIOPin = esp8266_ns.class_("ESP8266GPIOPin", cg.InternalGPIOPin)
 
 
-def _lookup_pin(value):
+def _lookup_pin(value: str) -> int:
     board = CORE.data[KEY_ESP8266][KEY_BOARD]
     board_pins = boards.ESP8266_BOARD_PINS.get(board, {})
 
@@ -42,7 +45,7 @@ def _lookup_pin(value):
     raise cv.Invalid(f"Cannot resolve pin name '{value}' for board {board}.")
 
 
-def _translate_pin(value):
+def _translate_pin(value: Any) -> int:
     if isinstance(value, dict) or value is None:
         raise cv.Invalid(
             "This variable only supports pin numbers, not full pin schemas "
@@ -69,7 +72,7 @@ _ESP_SDIO_PINS = {
 }
 
 
-def validate_gpio_pin(value):
+def validate_gpio_pin(value: Any) -> int:
     value = _translate_pin(value)
     if value < 0 or value > 17:
         raise cv.Invalid(f"ESP8266: Invalid pin number: {value}")
@@ -86,7 +89,7 @@ def validate_gpio_pin(value):
     return value
 
 
-def validate_supports(value):
+def validate_supports(value: ConfigType) -> ConfigType:
     num = value[CONF_NUMBER]
     mode = value[CONF_MODE]
     is_input = mode[CONF_INPUT]
@@ -160,7 +163,7 @@ class PinInitialState:
 
 
 @pins.PIN_SCHEMA_REGISTRY.register(PLATFORM_ESP8266, ESP8266_PIN_SCHEMA)
-async def esp8266_pin_to_code(config):
+async def esp8266_pin_to_code(config: ConfigType) -> MockObj:
     var = cg.new_Pvariable(config[CONF_ID])
     num = config[CONF_NUMBER]
     mode = config[CONF_MODE]
@@ -192,7 +195,7 @@ async def esp8266_pin_to_code(config):
 
 
 @coroutine_with_priority(CoroPriority.WORKAROUNDS)
-async def add_pin_initial_states_array():
+async def add_pin_initial_states_array() -> None:
     # Add includes at the very end, so that they override everything
     initial_states: list[PinInitialState] = CORE.data[KEY_ESP8266][
         KEY_PIN_INITIAL_STATES

esphome/components/file/image.py

@@ -5,6 +5,7 @@ import io
 import logging
 from pathlib import Path
 import re
+from typing import Any
 
 from PIL import Image, UnidentifiedImageError
 
@@ -75,12 +76,12 @@ def compute_local_image_path(value: str | ConfigType) -> Path:
     return external_files.compute_local_file_path(DOMAIN, url)
 
 
-def local_path(value):
+def local_path(value: str | ConfigType) -> str:
     value = value[CONF_PATH] if isinstance(value, dict) else value
     return str(CORE.relative_config_path(value))
 
 
-def download_file(url, path):
+def download_file(url: str, path: Path) -> str:
     # The shared NETWORK_TIMEOUT applies; a per-caller timeout would be
     # silently ignored on a per-run memo hit anyway (memos key by path).
     external_files.download_content(url, path)
@@ -98,7 +99,7 @@ def download_gh_svg(value: str | ConfigType, source: str) -> str:
     return download_file(url, path)
 
 
-def download_image(value):
+def download_image(value: str | ConfigType) -> str:
     value = value[CONF_URL] if isinstance(value, dict) else value
     return download_file(value, compute_local_image_path(value))
 
@@ -146,7 +147,7 @@ def _extract_entry_ref(entry: ConfigType) -> RemoteFile | None:
 PREFETCH_FILES = external_files.single_stage_prefetch(_extract_entry_ref)
 
 
-def validate_file_shorthand(value):
+def validate_file_shorthand(value: Any) -> str:
     value = cv.string_strict(value)
     if (remote := _parse_remote_shorthand(value)) is not None:
         return download_file(remote.url, remote.path)
@@ -163,8 +164,8 @@ LOCAL_SCHEMA = cv.All(
 )
 
 
-def mdi_schema(source):
+def mdi_schema(source: str) -> cv.All:
-    def validate_mdi(value):
+    def validate_mdi(value: ConfigType) -> str:
         return download_gh_svg(value, source)
 
     return cv.All(
@@ -259,7 +260,9 @@ async def new_image(config: ConfigType) -> MockObj:
     return var
 
 
-async def write_image(config, all_frames=False):
+async def write_image(
+    config: ConfigType, all_frames: bool = False
+) -> tuple[MockObj, int, int, MockObj, MockObj, int]:
     path = Path(config[CONF_FILE])
     if not path.is_file():
         raise core.EsphomeError(f"Could not load image file {path}")

esphome/components/globals/__init__.py

@@ -8,7 +8,8 @@ from esphome.const import (
     CONF_TYPE,
     CONF_VALUE,
 )
-from esphome.core import CoroPriority, coroutine_with_priority
+from esphome.core import ID, CoroPriority, coroutine_with_priority
+from esphome.cpp_generator import MockObj, TemplateArgsType
 from esphome.types import ConfigType
 
 CODEOWNERS = ["@esphome/core"]
@@ -62,7 +63,7 @@ CONFIG_SCHEMA = _globals_schema
 
 # Run with low priority so that namespaces are registered first
 @coroutine_with_priority(CoroPriority.LATE)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     type_ = cg.RawExpression(config[CONF_TYPE])
     restore = config[CONF_RESTORE_VALUE]
 
@@ -104,7 +105,12 @@ async def to_code(config):
     ),
     synchronous=True,
 )
-async def globals_set_to_code(config, action_id, template_arg, args):
+async def globals_set_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     full_id, paren = await cg.get_variable_with_full_id(config[CONF_ID])
     template_arg = cg.TemplateArguments(full_id.type, *template_arg)
     var = cg.new_Pvariable(action_id, template_arg, paren)

esphome/components/gpio/binary_sensor/__init__.py

@@ -12,6 +12,7 @@ from esphome.const import (
     CONF_PIN,
 )
 from esphome.core import CORE
+from esphome.types import ConfigType
 
 from .. import gpio_ns
 
@@ -68,7 +69,7 @@ def _pin_shared_only_with_deep_sleep(pin_num: int) -> bool:
     return any(path and path[0] == "deep_sleep" for path, _, _ in pin_users)
 
 
-def _final_validate(config) -> None:
+def _final_validate(config: ConfigType) -> None:
     use_interrupt = config[CONF_USE_INTERRUPT]
     if not use_interrupt:
         return
@@ -124,7 +125,7 @@ def _final_validate(config) -> None:
 FINAL_VALIDATE_SCHEMA = _final_validate
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = await binary_sensor.new_binary_sensor(config)
     await cg.register_component(var, config)
 

esphome/components/gpio/one_wire/__init__.py

@@ -3,6 +3,7 @@ import esphome.codegen as cg
 from esphome.components.one_wire import OneWireBus
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_PIN
+from esphome.types import ConfigType
 
 from .. import gpio_ns
 
@@ -18,7 +19,7 @@ CONFIG_SCHEMA = cv.Schema(
 ).extend(cv.COMPONENT_SCHEMA)
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
 

esphome/components/gpio/output/__init__.py

@@ -3,6 +3,7 @@ import esphome.codegen as cg
 from esphome.components import output
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_PIN
+from esphome.types import ConfigType
 
 from .. import gpio_ns
 
@@ -16,7 +17,7 @@ CONFIG_SCHEMA = output.BINARY_OUTPUT_SCHEMA.extend(
 ).extend(cv.COMPONENT_SCHEMA)
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     await output.register_output(var, config)
     await cg.register_component(var, config)

esphome/components/gpio/switch/__init__.py

@@ -3,6 +3,7 @@ import esphome.codegen as cg
 from esphome.components import switch
 import esphome.config_validation as cv
 from esphome.const import CONF_INTERLOCK, CONF_PIN
+from esphome.types import ConfigType
 
 from .. import gpio_ns
 
@@ -24,7 +25,7 @@ CONFIG_SCHEMA = (
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = await switch.new_switch(config)
     await cg.register_component(var, config)
 

esphome/components/homeassistant/__init__.py

@@ -1,13 +1,19 @@
+from collections.abc import Callable, Iterable
+
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.const import CONF_ATTRIBUTE, CONF_ENTITY_ID, CONF_INTERNAL
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType
 
 CODEOWNERS = ["@OttoWinter", "@esphome/core"]
 homeassistant_ns = cg.esphome_ns.namespace("homeassistant")
 
 
-def validate_entity_domain(platform, supported_domains):
+def validate_entity_domain(
-    def validator(config):
+    platform: str, supported_domains: Iterable[str]
+) -> Callable[[ConfigType], ConfigType]:
+    def validator(config: ConfigType) -> ConfigType:
         domain = config[CONF_ENTITY_ID].split(".", 1)[0]
         if domain not in supported_domains:
             raise cv.Invalid(
@@ -34,7 +40,7 @@ HOME_ASSISTANT_IMPORT_CONTROL_SCHEMA = cv.Schema(
 )
 
 
-def setup_home_assistant_entity(var, config):
+def setup_home_assistant_entity(var: MockObj, config: ConfigType) -> None:
     cg.add(var.set_entity_id(config[CONF_ENTITY_ID]))
     if CONF_ATTRIBUTE in config:
         cg.add(var.set_attribute(config[CONF_ATTRIBUTE]))

esphome/components/homeassistant/binary_sensor/__init__.py

@@ -1,5 +1,6 @@
 import esphome.codegen as cg
 from esphome.components import binary_sensor
+from esphome.types import ConfigType
 
 from .. import (
     HOME_ASSISTANT_IMPORT_SCHEMA,
@@ -18,7 +19,7 @@ CONFIG_SCHEMA = binary_sensor.binary_sensor_schema(HomeassistantBinarySensor).ex
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = await binary_sensor.new_binary_sensor(config)
     await cg.register_component(var, config)
     setup_home_assistant_entity(var, config)

esphome/components/homeassistant/number/__init__.py

@@ -1,6 +1,7 @@
 import esphome.codegen as cg
 from esphome.components import number
 import esphome.config_validation as cv
+from esphome.types import ConfigType
 
 from .. import (
     HOME_ASSISTANT_IMPORT_CONTROL_SCHEMA,
@@ -22,7 +23,7 @@ CONFIG_SCHEMA = (
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     cg.add_define("USE_API_HOMEASSISTANT_SERVICES")
     var = await number.new_number(
         config,

esphome/components/homeassistant/sensor/__init__.py

@@ -1,5 +1,6 @@
 import esphome.codegen as cg
 from esphome.components import sensor
+from esphome.types import ConfigType
 
 from .. import (
     HOME_ASSISTANT_IMPORT_SCHEMA,
@@ -18,7 +19,7 @@ CONFIG_SCHEMA = sensor.sensor_schema(HomeassistantSensor, accuracy_decimals=1).e
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = await sensor.new_sensor(config)
     await cg.register_component(var, config)
     setup_home_assistant_entity(var, config)

esphome/components/homeassistant/switch/__init__.py

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import switch
 import esphome.config_validation as cv
 from esphome.const import CONF_ID
+from esphome.types import ConfigType
 
 from .. import (
     HOME_ASSISTANT_IMPORT_CONTROL_SCHEMA,
@@ -36,7 +37,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     cg.add_define("USE_API_HOMEASSISTANT_SERVICES")
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)

esphome/components/homeassistant/text_sensor/__init__.py

@@ -1,5 +1,6 @@
 import esphome.codegen as cg
 from esphome.components import text_sensor
+from esphome.types import ConfigType
 
 from .. import (
     HOME_ASSISTANT_IMPORT_SCHEMA,
@@ -18,7 +19,7 @@ CONFIG_SCHEMA = text_sensor.text_sensor_schema(HomeassistantTextSensor).extend(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = await text_sensor.new_text_sensor(config)
     await cg.register_component(var, config)
     setup_home_assistant_entity(var, config)

esphome/components/homeassistant/time/__init__.py

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import time as time_
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_TIMEZONE
+from esphome.types import ConfigType
 
 from .. import homeassistant_ns
 
@@ -16,7 +17,7 @@ CONFIG_SCHEMA = time_.TIME_SCHEMA.extend(
 ).extend(cv.COMPONENT_SCHEMA)
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     await time_.register_time(var, config)
     await cg.register_component(var, config)

esphome/components/host/__init__.py

@@ -11,6 +11,7 @@ from esphome.const import (
 )
 from esphome.core import CORE
 from esphome.platformio.toolchain import copy_ccache_script
+from esphome.types import ConfigType
 
 from .const import KEY_HOST
 
@@ -22,7 +23,7 @@ AUTO_LOAD = ["network", "preferences"]
 IS_TARGET_PLATFORM = True
 
 
-def set_core_data(config):
+def set_core_data(config: ConfigType) -> ConfigType:
     CORE.data[KEY_HOST] = {}
     CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] = PLATFORM_HOST
     CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] = "host"
@@ -40,7 +41,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     cg.add_build_flag("-DUSE_HOST")
     cg.add_define("USE_NATIVE_64BIT_TIME")
     # The prefs file finds stored preferences by key, so key migration is possible

esphome/components/host/gpio.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Any
 
 from esphome import pins
 import esphome.codegen as cg
@@ -14,6 +15,8 @@ from esphome.const import (
     CONF_PULLDOWN,
     CONF_PULLUP,
 )
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType
 
 from .const import host_ns
 
@@ -22,7 +25,7 @@ _LOGGER = logging.getLogger(__name__)
 HostGPIOPin = host_ns.class_("HostGPIOPin", cg.InternalGPIOPin)
 
 
-def _translate_pin(value):
+def _translate_pin(value: Any) -> int | str:
     if isinstance(value, dict) or value is None:
         raise cv.Invalid(
             "This variable only supports pin numbers, not full pin schemas "
@@ -41,7 +44,7 @@ def _translate_pin(value):
     return value
 
 
-def validate_gpio_pin(value):
+def validate_gpio_pin(value: Any) -> int | str:
     return _translate_pin(value)
 
 
@@ -53,7 +56,7 @@ HOST_PIN_SCHEMA = pins.gpio_base_schema(
 
 
 @pins.PIN_SCHEMA_REGISTRY.register("host", HOST_PIN_SCHEMA)
-async def host_pin_to_code(config):
+async def host_pin_to_code(config: ConfigType) -> MockObj:
     var = cg.new_Pvariable(config[CONF_ID])
     num = config[CONF_NUMBER]
     cg.add(var.set_pin(num))

esphome/components/host/time/__init__.py

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import time as time_
 import esphome.config_validation as cv
 from esphome.const import CONF_ID
+from esphome.types import ConfigType
 
 CODEOWNERS = ["@clydebarrow"]
 
@@ -14,7 +15,7 @@ CONFIG_SCHEMA = time_.TIME_SCHEMA.extend(
 ).extend(cv.COMPONENT_SCHEMA)
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
     await time_.register_time(var, config)

esphome/components/i2c/__init__.py

@@ -1,6 +1,7 @@
 import logging
 import re
 import sys
+from typing import Any
 
 from esphome import pins
 import esphome.codegen as cg
@@ -52,9 +53,10 @@ from esphome.const import (
     PLATFORM_RP2,
     PlatformFramework,
 )
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
 from esphome.cpp_generator import MockObj
 import esphome.final_validate as fv
+from esphome.types import ConfigType
 
 LOGGER = logging.getLogger(__name__)
 CODEOWNERS = ["@esphome/core"]
@@ -96,13 +98,13 @@ CONF_SCL_PULLUP_ENABLED = "scl_pullup_enabled"
 MULTI_CONF = True
 
 
-def validate_device(value):
+def validate_device(value: str) -> str:
     if not re.match(r"^/(?:[^/]+/)*[^/]+$", value):
         raise cv.Invalid("Device must be an absolute device path (e.g., /dev/i2c-0)")
     return value
 
 
-def _bus_declare_type(value):
+def _bus_declare_type(value: Any) -> ID:
     if CORE.is_esp32:
         return cv.declare_id(IDFI2CBus)(value)
     if CORE.using_arduino:
@@ -114,7 +116,7 @@ def _bus_declare_type(value):
     raise NotImplementedError
 
 
-def _rp2040_i2c_controller(pin):
+def _rp2040_i2c_controller(pin: int) -> int:
     """Return the I2C controller number (0 or 1) for a given RP2040/RP2350 GPIO pin.
 
     See RP2040 datasheet Table 2 (section 1.4.3, "GPIO Functions"):
@@ -125,7 +127,7 @@ def _rp2040_i2c_controller(pin):
     return (pin // 2) % 2
 
 
-def validate_config(config):
+def validate_config(config: ConfigType) -> ConfigType:
     if CORE.is_esp32:
         return cv.require_framework_version(
             esp_idf=cv.Version(5, 4, 2), esp32_arduino=cv.Version(3, 2, 1)
@@ -142,7 +144,7 @@ def validate_config(config):
     return config
 
 
-def validate_host_config(config):
+def validate_host_config(config: ConfigType) -> ConfigType:
     if CORE.is_host:
         # Host I2C is currently only supported on Linux
         if not sys.platform.lower().startswith("linux"):
@@ -229,7 +231,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-def _final_validate(config):
+def _final_validate(config: ConfigType) -> None:
     full_config = fv.full_config.get()[CONF_I2C]
     if CORE.using_zephyr and len(full_config) > 1:
         raise cv.Invalid("Second i2c is not implemented on Zephyr yet")
@@ -281,7 +283,7 @@ FINAL_VALIDATE_SCHEMA = _final_validate
 
 
 @coroutine_with_priority(CoroPriority.BUS)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     cg.add_global(i2c_ns.using)
     cg.add_define("USE_I2C")
     if CORE.is_esp32:
@@ -358,7 +360,7 @@ async def to_code(config):
             cg.add(var.set_lp_mode(bool(config[CONF_LOW_POWER_MODE])))
 
 
-def i2c_device_schema(default_address):
+def i2c_device_schema(default_address: int | None) -> cv.Schema:
     """Create a schema for a i2c device.
 
     :param default_address: The default address of the i2c device, can be None to represent
@@ -375,7 +377,7 @@ def i2c_device_schema(default_address):
     return cv.Schema(schema)
 
 
-async def register_i2c_device(var, config):
+async def register_i2c_device(var: MockObj, config: ConfigType) -> None:
     """Register an i2c device with the given config.
 
     Sets the i2c bus to use and the i2c address.
@@ -390,11 +392,11 @@ async def register_i2c_device(var, config):
 def final_validate_device_schema(
     name: str,
     *,
-    min_frequency: cv.frequency = None,
+    min_frequency: Any = None,
-    max_frequency: cv.frequency = None,
+    max_frequency: Any = None,
-    min_timeout: cv.time_period = None,
+    min_timeout: Any = None,
-    max_timeout: cv.time_period = None,
+    max_timeout: Any = None,
-):
+) -> cv.Schema:
     hub_schema = {}
     if (min_frequency is not None) and (max_frequency is not None):
         hub_schema[cv.Required(CONF_FREQUENCY)] = cv.Range(

esphome/components/lock/__init__.py

@@ -12,13 +12,14 @@ from esphome.const import (
     CONF_ON_UNLOCK,
     CONF_WEB_SERVER,
 )
-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 = ["@esphome/core"]
 IS_PLATFORM_COMPONENT = True
@@ -102,7 +103,7 @@ _CALLBACK_AUTOMATIONS = (
 
 
 @setup_entity("lock")
-async def _setup_lock_core(var, config):
+async def _setup_lock_core(var: MockObj, config: ConfigType) -> None:
     await automation.build_callback_automations(var, config, _CALLBACK_AUTOMATIONS)
 
     if mqtt_id := config.get(CONF_MQTT_ID):
@@ -113,7 +114,7 @@ async def _setup_lock_core(var, config):
         await web_server.add_entity_config(var, web_server_config)
 
 
-async def register_lock(var, config):
+async def register_lock(var: MockObj, config: ConfigType) -> None:
     if not CORE.has_id(config[CONF_ID]):
         var = cg.Pvariable(config[CONF_ID], var)
     queue_entity_register("lock", config)
@@ -121,7 +122,7 @@ async def register_lock(var, config):
     await _setup_lock_core(var, config)
 
 
-async def new_lock(config, *args):
+async def new_lock(config: ConfigType, *args: SafeExpType) -> MockObj:
     var = cg.new_Pvariable(config[CONF_ID], *args)
     await register_lock(var, config)
     return var
@@ -143,23 +144,38 @@ LOCK_ACTION_SCHEMA = maybe_simple_id(
 @automation.register_action(
     "lock.open", OpenAction, LOCK_ACTION_SCHEMA, synchronous=True
 )
-async def lock_action_to_code(config, action_id, template_arg, args):
+async def lock_action_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)
 
 
 @automation.register_condition("lock.is_locked", LockCondition, LOCK_ACTION_SCHEMA)
-async def lock_is_on_to_code(config, condition_id, template_arg, args):
+async def lock_is_on_to_code(
+    config: ConfigType,
+    condition_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(condition_id, template_arg, paren, True)
 
 
 @automation.register_condition("lock.is_unlocked", LockCondition, LOCK_ACTION_SCHEMA)
-async def lock_is_off_to_code(config, condition_id, template_arg, args):
+async def lock_is_off_to_code(
+    config: ConfigType,
+    condition_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(condition_id, template_arg, paren, False)
 
 
 @coroutine_with_priority(CoroPriority.CORE)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     cg.add_global(lock_ns.using)