[core] Defer integration imports in ir_rf_base and media_player

infrared and radio_frequency loaded remote_base and api through
ir_rf_base, and media_player loaded the esp32 chain through audio, on
import. Move those imports into the functions that use them and drop
the known-leak table from the lazy import test.

esphome/components/alarm_control_panel/__init__.py

@@ -1,7 +1,6 @@
 from esphome import automation
 from esphome.automation import maybe_simple_id
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_CODE,
@@ -10,11 +9,11 @@ from esphome.const import (
     CONF_ID,
     CONF_MQTT_ID,
     CONF_ON_STATE,
-    CONF_WEB_SERVER,
 )
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     entity_duplicate_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_entity,
 )
@@ -44,12 +43,12 @@ AlarmControlPanelState = alarm_control_panel_ns.enum("AlarmControlPanelState")
 
 
 _ALARM_CONTROL_PANEL_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA)
     .extend(
         {
             cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
-                mqtt.MQTTAlarmControlPanelComponent
+                mqtt_component_class("MQTTAlarmControlPanelComponent")
             ),
             cv.Optional(CONF_ON_STATE): automation.validate_automation({}),
             cv.Optional(CONF_ON_TRIGGERED): automation.validate_automation({}),
@@ -166,11 +165,7 @@ _CALLBACK_AUTOMATIONS = (
 @setup_entity("alarm_control_panel")
 async def setup_alarm_control_panel_core_(var, config):
     await automation.build_callback_automations(var, config, _CALLBACK_AUTOMATIONS)
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
-    if mqtt_id := config.get(CONF_MQTT_ID):
-        mqtt_ = cg.new_Pvariable(mqtt_id, var)
-        await mqtt.register_mqtt_component(mqtt_, config)
+    await entity_helpers.setup_entity_integrations(var, config)
 
 
 async def register_alarm_control_panel(var, config):

esphome/components/binary_sensor/__init__.py

@@ -63,6 +63,7 @@ from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority, entity
 from esphome.core.entity_helpers import (
     SubEntities,
     entity_duplicate_validator,
+    lazy_load_setup,
     lazy_load_validator,
     mqtt_component_class,
     queue_entity_register,
@@ -621,10 +622,7 @@ async def setup_binary_sensor_core_(var, config):
 
     await entity_helpers.setup_entity_integrations(var, config)
 
-    if "zigbee" in CORE.loaded_integrations:
-        from esphome.components import zigbee
-
-        await zigbee.setup_binary_sensor(var, config)
+    await lazy_load_setup("zigbee", "setup_binary_sensor", var, config)
 
 
 async def register_binary_sensor(var, config):

esphome/components/button/__init__.py

@@ -1,7 +1,6 @@
 from esphome import automation
 from esphome.automation import maybe_simple_id
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_DEVICE_CLASS,
@@ -10,7 +9,6 @@ from esphome.const import (
     CONF_ID,
     CONF_MQTT_ID,
     CONF_ON_PRESS,
-    CONF_WEB_SERVER,
     DEVICE_CLASS_EMPTY,
     DEVICE_CLASS_IDENTIFY,
     DEVICE_CLASS_RESTART,
@@ -21,11 +19,12 @@ from esphome.const import (
     ICON_RESTART,
     ICON_RESTART_ALERT,
 )
-from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     ApplyEntity,
     SubEntities,
     entity_duplicate_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_device_class,
     setup_entity,
@@ -53,11 +52,13 @@ validate_device_class = cv.one_of(*DEVICE_CLASSES, lower=True, space="_")
 
 
 _BUTTON_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA)
     .extend(
         {
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTButtonComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTButtonComponent")
+            ),
             cv.Optional(
                 CONF_DEVICE_CLASS, visibility=cv.Visibility.ADVANCED
             ): validate_device_class,
@@ -135,12 +136,7 @@ async def setup_button_core_(var: MockObj, config: ConfigType) -> None:
 
     setup_device_class(config)
 
-    if mqtt_id := config.get(CONF_MQTT_ID):
-        mqtt_ = cg.new_Pvariable(mqtt_id, var)
-        await mqtt.register_mqtt_component(mqtt_, config)
-
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
+    await entity_helpers.setup_entity_integrations(var, config)
 
 
 async def register_button(var: MockObj, config: ConfigType) -> None:

esphome/components/climate/__init__.py

@@ -3,7 +3,6 @@ from typing import Any
 
 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_ACTION_STATE_TOPIC,
@@ -49,15 +48,15 @@ from esphome.const import (
     CONF_TEMPERATURE_STEP,
     CONF_UNIT_OF_MEASUREMENT,
     CONF_VISUAL,
-    CONF_WEB_SERVER,
     UNIT_KELVIN,
 )
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     TemperatureUnitResolver,
     convert_visual_temperatures,
     declare_fixed_temperature_unit,
     entity_duplicate_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_entity,
     setup_entity_temperature_unit,
@@ -177,11 +176,13 @@ _visual_temperatures_in_entity_unit = partial(
 
 
 _CLIMATE_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA)
     .extend(
         {
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTClimateComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTClimateComponent")
+            ),
             cv.Optional(CONF_VISUAL, default={}): cv.Schema(
                 {
                     cv.Optional(CONF_MIN_TEMPERATURE): cv.temperature_with_unit,
@@ -353,10 +354,8 @@ async def setup_climate_core_(var: MockObj, config: ConfigType) -> None:
         cg.add(var.set_visual_max_humidity_override(max_humidity))
     setup_entity_temperature_unit(var, config)
 
-    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)
-
+    mqtt_ = await entity_helpers.setup_entity_integrations(var, config)
+    if mqtt_ is not None:
         if (action_state_topic := config.get(CONF_ACTION_STATE_TOPIC)) is not None:
             cg.add(mqtt_.set_custom_action_state_topic(action_state_topic))
         if (away_command_topic := config.get(CONF_AWAY_COMMAND_TOPIC)) is not None:
@@ -486,9 +485,6 @@ async def setup_climate_core_(var: MockObj, config: ConfigType) -> None:
 
     await automation.build_callback_automations(var, config, _CALLBACK_AUTOMATIONS)
 
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
-
 
 def set_custom_fan_modes(var: MockObj, modes: list[str]) -> None:
     """Point the climate at a shared flash table of custom fan modes."""

esphome/components/cover/__init__.py

@@ -3,7 +3,6 @@ import logging
 from esphome import automation
 from esphome.automation import maybe_simple_id
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_DEVICE_CLASS,
@@ -23,7 +22,6 @@ from esphome.const import (
     CONF_TILT_COMMAND_TOPIC,
     CONF_TILT_STATE_TOPIC,
     CONF_TRIGGER_ID,
-    CONF_WEB_SERVER,
     DEVICE_CLASS_AWNING,
     DEVICE_CLASS_BLIND,
     DEVICE_CLASS_CURTAIN,
@@ -36,9 +34,10 @@ from esphome.const import (
     DEVICE_CLASS_SHUTTER,
     DEVICE_CLASS_WINDOW,
 )
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     entity_duplicate_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_device_class,
     setup_entity,
@@ -112,11 +111,13 @@ TRIGGERS = {
 
 
 _COVER_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA)
     .extend(
         {
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTCoverComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTCoverComponent")
+            ),
             cv.Optional(CONF_MQTT_JSON_STATE_PAYLOAD): cv.All(
                 cv.requires_component("mqtt"), cv.boolean
             ),
@@ -202,10 +203,8 @@ async def setup_cover_core_(var: MockObj, config: ConfigType) -> None:
             trigger = cg.new_Pvariable(conf[CONF_TRIGGER_ID], var)
             await automation.build_automation(trigger, [], conf)
 
-    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)
-
+    mqtt_ = await entity_helpers.setup_entity_integrations(var, config)
+    if mqtt_ is not None:
         if (position_state_topic := config.get(CONF_POSITION_STATE_TOPIC)) is not None:
             cg.add(mqtt_.set_custom_position_state_topic(position_state_topic))
         if (
@@ -220,9 +219,6 @@ async def setup_cover_core_(var: MockObj, config: ConfigType) -> None:
         if (tilt_command_topic := config.get(CONF_TILT_COMMAND_TOPIC)) is not None:
             cg.add(mqtt_.set_custom_tilt_command_topic(tilt_command_topic))
 
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
-
 
 async def register_cover(var: MockObj, config: ConfigType) -> None:
     if not CORE.has_id(config[CONF_ID]):

esphome/components/datetime/__init__.py

@@ -2,7 +2,6 @@ from collections.abc import Callable
 
 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import mqtt, time, web_server
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_DATE,
@@ -20,12 +19,12 @@ from esphome.const import (
     CONF_TIME_ID,
     CONF_TRIGGER_ID,
     CONF_TYPE,
-    CONF_WEB_SERVER,
     CONF_YEAR,
 )
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     entity_duplicate_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_entity,
 )
@@ -38,6 +37,10 @@ DOMAIN = "datetime"
 IS_PLATFORM_COMPONENT = True
 
 datetime_ns = cg.esphome_ns.namespace("datetime")
+# Matches time.RealTimeClock by name without importing the time component
+_RealTimeClock = cg.esphome_ns.namespace("time").class_(
+    "RealTimeClock", cg.PollingComponent
+)
 DateTimeBase = datetime_ns.class_("DateTimeBase", cg.EntityBase)
 DateEntity = datetime_ns.class_("DateEntity", DateTimeBase)
 TimeEntity = datetime_ns.class_("TimeEntity", DateTimeBase)
@@ -64,7 +67,7 @@ DATETIME_MODES = [
 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)
+        time_id = cv.use_id(_RealTimeClock)(None)
         config[CONF_TIME_ID] = time_id
     return config
 
@@ -73,10 +76,10 @@ _DATETIME_SCHEMA = cv.ENTITY_BASE_SCHEMA.extend(
     cv.Schema(
         {
             cv.Optional(CONF_ON_VALUE): automation.validate_automation({}),
-            cv.Optional(CONF_TIME_ID): cv.use_id(time.RealTimeClock),
+            cv.Optional(CONF_TIME_ID): cv.use_id(_RealTimeClock),
         }
     )
-    .extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    .extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA)
 ).add_extra(_validate_time_present)
 
@@ -87,7 +90,9 @@ def date_schema(class_: MockObjClass) -> cv.Schema:
     schema = cv.Schema(
         {
             cv.GenerateID(): cv.declare_id(class_),
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTDateComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTDateComponent")
+            ),
             cv.Optional(CONF_TYPE, default="DATE"): cv.one_of("DATE", upper=True),
         }
     )
@@ -98,7 +103,9 @@ def time_schema(class_: MockObjClass) -> cv.Schema:
     schema = cv.Schema(
         {
             cv.GenerateID(): cv.declare_id(class_),
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTTimeComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTTimeComponent")
+            ),
             cv.Optional(CONF_TYPE, default="TIME"): cv.one_of("TIME", upper=True),
             cv.Optional(CONF_ON_TIME): automation.validate_automation(
                 {
@@ -115,7 +122,7 @@ def datetime_schema(class_: MockObjClass) -> cv.Schema:
         {
             cv.GenerateID(): cv.declare_id(class_),
             cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
-                mqtt.MQTTDateTimeComponent
+                mqtt_component_class("MQTTDateTimeComponent")
             ),
             cv.Optional(CONF_TYPE, default="DATETIME"): cv.one_of(
                 "DATETIME", upper=True
@@ -132,11 +139,7 @@ def datetime_schema(class_: MockObjClass) -> cv.Schema:
 
 @setup_entity("datetime")
 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)
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
+    await entity_helpers.setup_entity_integrations(var, config)
     # The state callback carries nothing; the automation gets the value as an ESPTime.
     for conf in config.get(CONF_ON_VALUE, []):
         await automation.build_callback_automation(

esphome/components/event/__init__.py

@@ -1,6 +1,5 @@
 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_DEVICE_CLASS,
@@ -10,15 +9,15 @@ from esphome.const import (
     CONF_ID,
     CONF_MQTT_ID,
     CONF_ON_EVENT,
-    CONF_WEB_SERVER,
     DEVICE_CLASS_BUTTON,
     DEVICE_CLASS_DOORBELL,
     DEVICE_CLASS_EMPTY,
     DEVICE_CLASS_MOTION,
 )
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     entity_duplicate_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_device_class,
     setup_entity,
@@ -44,11 +43,13 @@ EventPtr = Event.operator("ptr")
 validate_device_class = cv.one_of(*DEVICE_CLASSES, lower=True, space="_")
 
 _EVENT_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMPONENT_SCHEMA)
     .extend(
         {
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTEventComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTEventComponent")
+            ),
             cv.GenerateID(): cv.declare_id(Event),
             cv.Optional(
                 CONF_DEVICE_CLASS, visibility=cv.Visibility.ADVANCED
@@ -104,12 +105,7 @@ async def setup_event_core_(
 
     setup_device_class(config)
 
-    if mqtt_id := config.get(CONF_MQTT_ID):
-        mqtt_ = cg.new_Pvariable(mqtt_id, var)
-        await mqtt.register_mqtt_component(mqtt_, config)
-
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
+    await entity_helpers.setup_entity_integrations(var, config)
 
 
 async def register_event(

esphome/components/fan/__init__.py

@@ -1,7 +1,6 @@
 from esphome import automation
 from esphome.automation import maybe_simple_id
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_DIRECTION,
@@ -29,11 +28,11 @@ from esphome.const import (
     CONF_SPEED_LEVEL_STATE_TOPIC,
     CONF_SPEED_STATE_TOPIC,
     CONF_TRIGGER_ID,
-    CONF_WEB_SERVER,
 )
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     entity_duplicate_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_entity,
 )
@@ -78,14 +77,16 @@ FanPresetSetTrigger = fan_ns.class_(
 )
 
 _FAN_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA)
     .extend(
         {
             cv.Optional(CONF_RESTORE_MODE, default="ALWAYS_OFF"): cv.enum(
                 RESTORE_MODES, upper=True, space="_"
             ),
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTFanComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTFanComponent")
+            ),
             cv.Optional(CONF_DIRECTION_STATE_TOPIC): cv.All(
                 cv.requires_component("mqtt"), cv.publish_topic
             ),
@@ -236,10 +237,8 @@ _TRIGGER_AUTOMATIONS = (
 async def setup_fan_core_(var, config):
     cg.add(var.set_restore_mode(config[CONF_RESTORE_MODE]))
 
-    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)
-
+    mqtt_ = await entity_helpers.setup_entity_integrations(var, config)
+    if mqtt_ is not None:
         if (
             direction_state_topic := config.get(CONF_DIRECTION_STATE_TOPIC)
         ) is not None:
@@ -273,9 +272,6 @@ async def setup_fan_core_(var, config):
         if (speed_command_topic := config.get(CONF_SPEED_COMMAND_TOPIC)) is not None:
             cg.add(mqtt_.set_custom_speed_command_topic(speed_command_topic))
 
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
-
     for conf in config.get(CONF_ON_STATE, []):
         await automation.build_parent_callback_automation(
             var, "add_on_state_callback", (Fan.operator("ptr"), "x"), conf

esphome/components/ir_rf_base/__init__.py

@@ -1,8 +1,7 @@
 """Shared base for the infrared and radio_frequency entity components."""
 
 import esphome.codegen as cg
-from esphome.components import remote_base
-from esphome.components.api import DOMAIN as API_DOMAIN
+from esphome.const import CONF_API
 from esphome.core import CORE
 from esphome.core.entity_helpers import queue_entity_register
 from esphome.types import ConfigType
@@ -24,8 +23,10 @@ async def attach_transmitter(var: cg.MockObj, config: ConfigType, key: str) -> N
     does. Without this call anywhere in the build a request is answered as
     soon as the frame is handed over.
     """
+    from esphome.components import remote_base
+
     await remote_base.register_transmittable(var, config, key)
-    if API_DOMAIN in CORE.config:
+    if CONF_API in CORE.config:
         cg.add_define("USE_IR_RF_TRANSMIT_COMPLETE")
 
 

esphome/components/light/__init__.py

@@ -5,7 +5,7 @@ import logging
 
 import esphome.automation as auto
 import esphome.codegen as cg
-from esphome.components import mqtt, power_supply, web_server
+from esphome.components import power_supply
 from esphome.components.const import CONF_CHANNEL_COLORS, CONF_IS_WRGB
 from esphome.config_helpers import filter_source_files_from_defines
 import esphome.config_validation as cv
@@ -35,7 +35,6 @@ from esphome.const import (
     CONF_STATE,
     CONF_TRIGGER_ID,
     CONF_WARM_WHITE_COLOR_TEMPERATURE,
-    CONF_WEB_SERVER,
 )
 from esphome.core import (
     CORE,
@@ -44,10 +43,12 @@ from esphome.core import (
     EsphomeError,
     HexInt,
     coroutine_with_priority,
+    entity_helpers,
 )
 from esphome.core.entity_helpers import (
     RESTORE_STATE_NONE,
     entity_duplicate_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_entity,
 )
@@ -428,14 +429,14 @@ INITIAL_STATE_SCHEMA = LIGHT_STATE_SCHEMA.extend(
 )
 
 LIGHT_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA)
     .extend(
         {
             cv.GenerateID(): cv.declare_id(LightState),
             cv.GenerateID(CONF_GAMMA_TABLE_ID): cv.declare_id(GammaTable),
             cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
-                mqtt.MQTTJSONLightComponent
+                mqtt_component_class("MQTTJSONLightComponent")
             ),
             cv.Exclusive(CONF_RESTORE_MODE, "restore"): cv.one_of(
                 *LEGACY_RESTORE_MODES, upper=True, space="_"
@@ -722,12 +723,7 @@ async def setup_light_core_(light_var, config, output_var):
         if (max_current_ma := power_estimation.get(CONF_MAX_CURRENT_MA)) is not None:
             cg.add(output_var.set_max_current_ma(max_current_ma))
 
-    if (mqtt_id := config.get(CONF_MQTT_ID)) is not None:
-        mqtt_ = cg.new_Pvariable(mqtt_id, light_var)
-        await mqtt.register_mqtt_component(mqtt_, config)
-
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(light_var, web_server_config)
+    await entity_helpers.setup_entity_integrations(light_var, config)
 
 
 async def register_light(output_var, config):

esphome/components/light/automation.py

@@ -5,7 +5,6 @@ from typing import Any, NamedTuple
 from esphome import automation
 import esphome.codegen as cg
 from esphome.components.const.css_colors import CSS_COLORS
-from esphome.config import path_context
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_BLUE,
@@ -212,6 +211,8 @@ def _record_effect_ref(config: ConfigType) -> ConfigType:
     if effect.lower() == "none":
         return config  # "None" is always valid
 
+    from esphome.config import path_context
+
     from . import EffectRef, _get_data
 
     _get_data().effect_refs.append(
@@ -313,6 +314,8 @@ automation.register_apply_action(
 
 def _record_effect_cycle_ref(config: ConfigType) -> ConfigType:
     """Record a cycle-action reference for later validation against the target light."""
+    from esphome.config import path_context
+
     from . import EffectCycleRef, _get_data
 
     _get_data().effect_cycle_refs.append(

esphome/components/lock/__init__.py

@@ -1,7 +1,6 @@
 from esphome import automation
 from esphome.automation import maybe_simple_id
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_ENTITY_CATEGORY,
@@ -10,11 +9,11 @@ from esphome.const import (
     CONF_MQTT_ID,
     CONF_ON_LOCK,
     CONF_ON_UNLOCK,
-    CONF_WEB_SERVER,
 )
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     entity_duplicate_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_entity,
 )
@@ -48,11 +47,13 @@ LOCK_STATES = LockState.options(
 validate_lock_state = cv.enum(LOCK_STATES, upper=True)
 
 _LOCK_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA)
     .extend(
         {
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTLockComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTLockComponent")
+            ),
             cv.Optional(CONF_ON_LOCK): automation.validate_automation({}),
             cv.Optional(CONF_ON_UNLOCK): automation.validate_automation({}),
         }
@@ -106,12 +107,7 @@ _CALLBACK_AUTOMATIONS = (
 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):
-        mqtt_ = cg.new_Pvariable(mqtt_id, var)
-        await mqtt.register_mqtt_component(mqtt_, config)
-
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
+    await entity_helpers.setup_entity_integrations(var, config)
 
 
 async def register_lock(var: MockObj, config: ConfigType) -> None:

esphome/components/media_player/__init__.py

@@ -2,7 +2,6 @@ from collections.abc import Callable
 
 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import audio
 from esphome.components.const import CONF_BITRATE
 import esphome.config_validation as cv
 from esphome.const import (
@@ -116,6 +115,8 @@ def validate_preferred_format(
     """
 
     def validator(config: ConfigType) -> ConfigType:
+        from esphome.components import audio
+
         # Inherit settings from audio device if not manually set
         inherit_property_from(CONF_NUM_CHANNELS, audio_device_key)(config)
         inherit_property_from(CONF_SAMPLE_RATE, audio_device_key)(config)
@@ -155,6 +156,8 @@ def request_codecs_for_format_configs(
 
     Public API for external components. Do not remove.
     """
+    from esphome.components import audio
+
     needed_formats: set[str] = set()
     need_all = False
 

esphome/components/number/__init__.py

@@ -1,6 +1,5 @@
 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server, zigbee
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_ABOVE,
@@ -18,7 +17,6 @@ from esphome.const import (
     CONF_TRIGGER_ID,
     CONF_UNIT_OF_MEASUREMENT,
     CONF_VALUE,
-    CONF_WEB_SERVER,
     DEVICE_CLASS_ABSOLUTE_HUMIDITY,
     DEVICE_CLASS_APPARENT_POWER,
     DEVICE_CLASS_AQI,
@@ -79,12 +77,15 @@ from esphome.const import (
     DEVICE_CLASS_WIND_DIRECTION,
     DEVICE_CLASS_WIND_SPEED,
 )
-from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.config import UNIT_OF_MEASUREMENT_MAX_LENGTH
 from esphome.core.entity_helpers import (
     ApplyEntity,
     SubEntities,
     entity_duplicate_validator,
+    lazy_load_setup,
+    lazy_load_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_device_class,
     setup_entity,
@@ -195,12 +196,14 @@ validate_unit_of_measurement = cv.All(
 )
 
 _NUMBER_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA)
-    .extend(zigbee.NUMBER_SCHEMA)
+    .extend(entity_helpers.ZIGBEE_NUMBER_SCHEMA)
     .extend(
         {
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTNumberComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTNumberComponent")
+            ),
             cv.Optional(CONF_ON_VALUE): automation.validate_automation({}),
             cv.Optional(CONF_ON_VALUE_RANGE): automation.validate_automation(
                 {
@@ -225,7 +228,7 @@ _NUMBER_SCHEMA = (
 
 
 _NUMBER_SCHEMA.add_extra(entity_duplicate_validator("number"))
-_NUMBER_SCHEMA.add_extra(zigbee.validate_number)
+_NUMBER_SCHEMA.add_extra(lazy_load_validator("zigbee", "validate_number"))
 
 
 def number_schema(
@@ -294,13 +297,11 @@ async def setup_number_core_(
     setup_device_class(config)
     setup_unit_of_measurement(config)
 
-    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)
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
+    await entity_helpers.setup_entity_integrations(var, config)
 
-    await zigbee.setup_number(var, config, min_value, max_value, step)
+    await lazy_load_setup(
+        "zigbee", "setup_number", var, config, min_value, max_value, step
+    )
 
 
 async def register_number(

esphome/components/select/__init__.py

@@ -1,6 +1,5 @@
 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_CYCLE,
@@ -15,13 +14,13 @@ from esphome.const import (
     CONF_OPERATION,
     CONF_OPTION,
     CONF_OPTIONS,
-    CONF_WEB_SERVER,
 )
-from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     ApplyEntity,
     SubEntities,
     entity_duplicate_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_entity,
 )
@@ -49,11 +48,13 @@ SELECT_OPERATION_OPTIONS = SelectOperation.options(
 
 
 _SELECT_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA)
     .extend(
         {
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTSelectComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTSelectComponent")
+            ),
             cv.GenerateID(): cv.declare_id(Select),
             cv.Optional(CONF_ON_VALUE): automation.validate_automation(),
         }
@@ -101,12 +102,7 @@ async def setup_select_core_(var, config, *, options: list[str]):
             forward=[cg.StringRef(parent.option_at(index)), index],
         )
 
-    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)
-
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
+    await entity_helpers.setup_entity_integrations(var, config)
 
 
 async def register_select(var, config, *, options: list[str]):

esphome/components/sensor/__init__.py

@@ -126,6 +126,7 @@ from esphome.core.config import UNIT_OF_MEASUREMENT_MAX_LENGTH
 from esphome.core.entity_helpers import (
     SubEntities,
     entity_duplicate_validator,
+    lazy_load_setup,
     lazy_load_validator,
     mqtt_component_class,
     queue_entity_register,
@@ -1082,10 +1083,7 @@ async def setup_sensor_core_(var, config):
         else:
             cg.add(mqtt_.set_expire_after(expire_after))
 
-    if "zigbee" in CORE.loaded_integrations:
-        from esphome.components import zigbee
-
-        await zigbee.setup_sensor(var, config)
+    await lazy_load_setup("zigbee", "setup_sensor", var, config)
 
 
 async def register_sensor(var, config):

esphome/components/switch/__init__.py

@@ -3,7 +3,6 @@ from collections.abc import Callable
 from esphome import automation, preferences
 from esphome.automation import maybe_simple_id
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server, zigbee
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_DEVICE_CLASS,
@@ -20,12 +19,11 @@ from esphome.const import (
     CONF_RESTORE_STATE,
     CONF_STATE,
     CONF_STORAGE,
-    CONF_WEB_SERVER,
     DEVICE_CLASS_EMPTY,
     DEVICE_CLASS_OUTLET,
     DEVICE_CLASS_SWITCH,
 )
-from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     RESTORE_STATE_INITIAL,
     RESTORE_STATE_INVERT,
@@ -33,6 +31,9 @@ from esphome.core.entity_helpers import (
     ApplyEntity,
     SubEntities,
     entity_duplicate_validator,
+    lazy_load_setup,
+    lazy_load_validator,
+    mqtt_component_class,
     queue_entity_register,
     restore_state_schema,
     setup_device_class,
@@ -183,12 +184,14 @@ def _resolve_restore_mode(config: ConfigType) -> MockObj | str:
 
 
 _SWITCH_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA)
-    .extend(zigbee.SWITCH_SCHEMA)
+    .extend(entity_helpers.ZIGBEE_SWITCH_SCHEMA)
     .extend(
         {
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTSwitchComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTSwitchComponent")
+            ),
             cv.Optional(CONF_INVERTED): cv.boolean,
             cv.Exclusive(CONF_RESTORE_MODE, "restore"): validate_restore_mode,
             cv.Exclusive(CONF_RESTORE_STATE, "restore"): RESTORE_STATE_SCHEMA,
@@ -205,7 +208,7 @@ _SWITCH_SCHEMA = (
 
 
 _SWITCH_SCHEMA.add_extra(entity_duplicate_validator("switch"))
-_SWITCH_SCHEMA.add_extra(zigbee.validate_switch)
+_SWITCH_SCHEMA.add_extra(lazy_load_validator("zigbee", "validate_switch"))
 _SWITCH_SCHEMA.add_extra(_validate_initial_state)
 
 
@@ -272,17 +275,12 @@ async def setup_switch_core_(var, config):
 
     CORE.add_job(_build_switch_automations, var, config)
 
-    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)
-
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
+    await entity_helpers.setup_entity_integrations(var, config)
 
     setup_device_class(config)
 
     cg.add(var.set_restore_mode(_resolve_restore_mode(config)))
-    await zigbee.setup_switch(var, config)
+    await lazy_load_setup("zigbee", "setup_switch", var, config)
 
 
 async def register_switch(var, config):

esphome/components/text/__init__.py

@@ -1,6 +1,5 @@
 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_ENTITY_CATEGORY,
@@ -10,11 +9,11 @@ from esphome.const import (
     CONF_MQTT_ID,
     CONF_ON_VALUE,
     CONF_VALUE,
-    CONF_WEB_SERVER,
 )
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     entity_duplicate_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_entity,
 )
@@ -38,11 +37,13 @@ TEXT_MODES = {
 }
 
 _TEXT_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMPONENT_SCHEMA)
     .extend(
         {
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTTextComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTTextComponent")
+            ),
             cv.GenerateID(): cv.declare_id(Text),
             cv.Optional(CONF_ON_VALUE): automation.validate_automation(),
             cv.Required(CONF_MODE): cv.enum(TEXT_MODES, upper=True),
@@ -98,12 +99,7 @@ async def setup_text_core_(
             var, "add_on_state_callback", [(cg.std_string, "x")], conf
         )
 
-    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)
-
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
+    await entity_helpers.setup_entity_integrations(var, config)
 
 
 async def register_text(

esphome/components/text_sensor/__init__.py

@@ -1,6 +1,5 @@
 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server
 from esphome.config_helpers import filter_source_files_from_defines
 import esphome.config_validation as cv
 from esphome.const import (
@@ -15,15 +14,15 @@ from esphome.const import (
     CONF_ON_VALUE,
     CONF_STATE,
     CONF_TO,
-    CONF_WEB_SERVER,
     DEVICE_CLASS_DATE,
     DEVICE_CLASS_EMPTY,
     DEVICE_CLASS_TIMESTAMP,
 )
-from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     SubEntities,
     entity_duplicate_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_device_class,
     setup_entity,
@@ -131,11 +130,13 @@ async def substitution_filter_to_code(
 validate_device_class = cv.one_of(*DEVICE_CLASSES, lower=True, space="_")
 
 _TEXT_SENSOR_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMPONENT_SCHEMA)
     .extend(
         {
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTTextSensor),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTTextSensor")
+            ),
             cv.GenerateID(): cv.declare_id(TextSensor),
             cv.Optional(
                 CONF_DEVICE_CLASS, visibility=cv.Visibility.ADVANCED
@@ -209,12 +210,7 @@ async def setup_text_sensor_core_(var, config):
 
     CORE.add_job(_build_text_sensor_automations, var, config)
 
-    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)
-
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
+    await entity_helpers.setup_entity_integrations(var, config)
 
 
 async def register_text_sensor(var, config):

esphome/components/update/__init__.py

@@ -1,6 +1,5 @@
 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_DEVICE_CLASS,
@@ -9,14 +8,14 @@ from esphome.const import (
     CONF_ICON,
     CONF_ID,
     CONF_MQTT_ID,
-    CONF_WEB_SERVER,
     DEVICE_CLASS_EMPTY,
     DEVICE_CLASS_FIRMWARE,
     ENTITY_CATEGORY_CONFIG,
 )
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     entity_duplicate_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_device_class,
     setup_entity,
@@ -41,11 +40,13 @@ DEVICE_CLASSES = [
 CONF_ON_UPDATE_AVAILABLE = "on_update_available"
 
 _UPDATE_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA)
     .extend(
         {
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTUpdateComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTUpdateComponent")
+            ),
             cv.Optional(
                 CONF_DEVICE_CLASS, visibility=cv.Visibility.ADVANCED
             ): cv.one_of(*DEVICE_CLASSES, lower=True),
@@ -97,12 +98,7 @@ async def setup_update_core_(var: MockObj, config: ConfigType) -> None:
             on_update_available,
         )
 
-    if mqtt_id_config := config.get(CONF_MQTT_ID):
-        mqtt_ = cg.new_Pvariable(mqtt_id_config, var)
-        await mqtt.register_mqtt_component(mqtt_, config)
-
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
+    await entity_helpers.setup_entity_integrations(var, config)
 
 
 async def register_update(var: MockObj, config: ConfigType) -> None:

esphome/components/valve/__init__.py

@@ -1,7 +1,6 @@
 from esphome import automation
 from esphome.automation import maybe_simple_id
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_DEVICE_CLASS,
@@ -15,14 +14,14 @@ from esphome.const import (
     CONF_POSITION_STATE_TOPIC,
     CONF_STATE,
     CONF_STOP,
-    CONF_WEB_SERVER,
     DEVICE_CLASS_EMPTY,
     DEVICE_CLASS_GAS,
     DEVICE_CLASS_WATER,
 )
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     entity_duplicate_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_device_class,
     setup_entity,
@@ -72,12 +71,14 @@ _CALLBACK_AUTOMATIONS = (
 )
 
 _VALVE_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA)
     .extend(
         {
             cv.GenerateID(): cv.declare_id(Valve),
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTValveComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTValveComponent")
+            ),
             cv.Optional(
                 CONF_DEVICE_CLASS, visibility=cv.Visibility.ADVANCED
             ): cv.one_of(*DEVICE_CLASSES, lower=True),
@@ -126,10 +127,8 @@ async def _setup_valve_core(var, config):
 
     await automation.build_callback_automations(var, config, _CALLBACK_AUTOMATIONS)
 
-    if mqtt_id_config := config.get(CONF_MQTT_ID):
-        mqtt_ = cg.new_Pvariable(mqtt_id_config, var)
-        await mqtt.register_mqtt_component(mqtt_, config)
-
+    mqtt_ = await entity_helpers.setup_entity_integrations(var, config)
+    if mqtt_ is not None:
         if position_state_topic_config := config.get(CONF_POSITION_STATE_TOPIC):
             cg.add(mqtt_.set_custom_position_state_topic(position_state_topic_config))
         if position_command_topic_config := config.get(CONF_POSITION_COMMAND_TOPIC):
@@ -137,9 +136,6 @@ async def _setup_valve_core(var, config):
                 mqtt_.set_custom_position_command_topic(position_command_topic_config)
             )
 
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
-
 
 async def register_valve(var, config):
     if not CORE.has_id(config[CONF_ID]):

esphome/components/zigbee/__init__.py

@@ -65,7 +65,6 @@ from .zigbee_esp32 import (
     validate_switch_esp32,
     zigbee_require_vfs_select,
 )
-from .zigbee_zephyr import zephyr_number, zephyr_switch
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -80,16 +79,8 @@ CONFLICTS_WITH = ["openthread"]
 BASE_SCHEMA = entity_helpers.ZIGBEE_BASE_ENTITY_SCHEMA
 BINARY_SENSOR_SCHEMA = entity_helpers.ZIGBEE_BINARY_SENSOR_SCHEMA
 SENSOR_SCHEMA = entity_helpers.ZIGBEE_SENSOR_SCHEMA
-SWITCH_SCHEMA = (
-    cv.Schema({})
-    .extend(entity_helpers.zigbee_base_entity_schema(["generic", "on_off"]))
-    .extend(zephyr_switch)
-)
-NUMBER_SCHEMA = (
-    cv.Schema({})
-    .extend(entity_helpers.zigbee_base_entity_schema(["generic"]))
-    .extend(zephyr_number)
-)
+SWITCH_SCHEMA = entity_helpers.ZIGBEE_SWITCH_SCHEMA
+NUMBER_SCHEMA = entity_helpers.ZIGBEE_NUMBER_SCHEMA
 
 
 def _validate_router_sleepy(config: ConfigType) -> ConfigType:

esphome/components/zigbee/const_zephyr.py

@@ -1,10 +1,14 @@
 # These schema keys live in esphome.const so entity base
 # schemas can use them without importing this package.
-from esphome.const import CONF_ZIGBEE_BINARY_SENSOR, CONF_ZIGBEE_ID, CONF_ZIGBEE_SENSOR  # noqa: F401  # pylint: disable=unused-import
+from esphome.const import (  # noqa: F401  # pylint: disable=unused-import
+    CONF_ZIGBEE_BINARY_SENSOR,
+    CONF_ZIGBEE_ID,
+    CONF_ZIGBEE_NUMBER,
+    CONF_ZIGBEE_SENSOR,
+    CONF_ZIGBEE_SWITCH,
+)
 
 CONF_MAX_EP_NUMBER_ZEPHYR = 8
-CONF_ZIGBEE_SWITCH = "zigbee_switch"
-CONF_ZIGBEE_NUMBER = "zigbee_number"
 CONF_IEEE802154_VENDOR_OUI = "ieee802154_vendor_oui"
 
 # Keys for CORE.data storage

esphome/components/zigbee/time/__init__.py

@@ -6,13 +6,12 @@ from esphome.core import CORE
 from esphome.types import ConfigType
 
 from .. import consume_endpoint
-from ..const import zigbee_ns
+from ..const import ZigbeeComponent, zigbee_ns
 from ..const_esp32 import ROLE
 from ..const_zephyr import CONF_ZIGBEE_ID
 from ..zigbee_ep_esp32 import add_clusters_to_first_ep, get_first_ep_num
 from ..zigbee_zephyr import (
     ZigbeeClusterDesc,
-    ZigbeeComponent,
     get_slot_index,
     zigbee_assign,
     zigbee_new_attr_list,

esphome/components/zigbee/zigbee_zephyr.py

@@ -34,7 +34,6 @@ from .const import (
     AnalogAttrs,
     AnalogAttrsOutput,
     BinaryAttrs,
-    ZigbeeComponent,
     zigbee_ns,
 )
 from .const_zephyr import (
@@ -60,24 +59,6 @@ ZigbeeSensor = zigbee_ns.class_("ZigbeeSensor", cg.Component)
 ZigbeeSwitch = zigbee_ns.class_("ZigbeeSwitch", cg.Component)
 ZigbeeNumber = zigbee_ns.class_("ZigbeeNumber", cg.Component)
 
-zephyr_switch = cv.Schema(
-    {
-        cv.OnlyWith(CONF_ZIGBEE_ID, ["nrf52", "zigbee"]): cv.use_id(ZigbeeComponent),
-        cv.OnlyWith(CONF_ZIGBEE_SWITCH, ["nrf52", "zigbee"]): cv.declare_id(
-            ZigbeeSwitch
-        ),
-    }
-)
-
-zephyr_number = cv.Schema(
-    {
-        cv.OnlyWith(CONF_ZIGBEE_ID, ["nrf52", "zigbee"]): cv.use_id(ZigbeeComponent),
-        cv.OnlyWith(CONF_ZIGBEE_NUMBER, ["nrf52", "zigbee"]): cv.declare_id(
-            ZigbeeNumber
-        ),
-    }
-)
-
 
 async def zephyr_to_code(config: ConfigType) -> "MockObj":
     zephyr_add_prj_conf("ZIGBEE", True)

esphome/const.py

@@ -1170,7 +1170,9 @@ CONF_YEAR = "year"
 CONF_ZERO = "zero"
 CONF_ZIGBEE_BINARY_SENSOR = "zigbee_binary_sensor"
 CONF_ZIGBEE_ID = "zigbee_id"
+CONF_ZIGBEE_NUMBER = "zigbee_number"
 CONF_ZIGBEE_SENSOR = "zigbee_sensor"
+CONF_ZIGBEE_SWITCH = "zigbee_switch"
 
 TYPE_GIT = "git"
 TYPE_LOCAL = "local"

esphome/core/entity_helpers.py

@@ -35,7 +35,9 @@ from esphome.const import (
     CONF_WEB_SERVER_ID,
     CONF_ZIGBEE_BINARY_SENSOR,
     CONF_ZIGBEE_ID,
+    CONF_ZIGBEE_NUMBER,
     CONF_ZIGBEE_SENSOR,
+    CONF_ZIGBEE_SWITCH,
     UNIT_CELSIUS,
 )
 from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
@@ -1101,6 +1103,8 @@ _ZIGBEE_ENTITY_CLASSES = {
         ["generic", "device_class"],
     ),
     "sensor": (CONF_ZIGBEE_SENSOR, "ZigbeeSensor", ["generic", "device_class"]),
+    "switch": (CONF_ZIGBEE_SWITCH, "ZigbeeSwitch", ["generic", "on_off"]),
+    "number": (CONF_ZIGBEE_NUMBER, "ZigbeeNumber", ["generic"]),
 }
 
 
@@ -1119,8 +1123,9 @@ def _zigbee_entity_schema(platform: str) -> cv.Schema:
 
 
 ZIGBEE_BINARY_SENSOR_SCHEMA = _zigbee_entity_schema("binary_sensor")
-
 ZIGBEE_SENSOR_SCHEMA = _zigbee_entity_schema("sensor")
+ZIGBEE_SWITCH_SCHEMA = _zigbee_entity_schema("switch")
+ZIGBEE_NUMBER_SCHEMA = _zigbee_entity_schema("number")
 
 
 def lazy_load_validator(
@@ -1141,6 +1146,16 @@ def lazy_load_validator(
     return validator
 
 
+async def lazy_load_setup(component: str, name: str, *args: Any) -> None:
+    """Await ``components.<component>.<name>(*args)`` when the component is loaded."""
+    if component not in CORE.loaded_integrations:
+        return
+    module = import_module(f"esphome.components.{component}")
+    if (hook := getattr(module, name, None)) is None:
+        raise ValueError(f"{component} has no setup hook {name!r}")
+    await hook(*args)
+
+
 async def setup_entity_integrations(var: MockObj, config: ConfigType) -> MockObj | None:
     """Register the mqtt companion and web_server entry for an entity.
 

script/ci-custom.py

@@ -792,7 +792,7 @@ def lint_unit_constants_usage():
 # Maximum allowed CONF_ constants in esphome/const.py.
 # This file is frozen — new constants go in esphome/components/const/__init__.py.
 # Decrease this number when constants are moved out of const.py.
-CONST_PY_MAX_CONF = 1025
+CONST_PY_MAX_CONF = 1027
 
 
 @lint_content_check(include=["esphome/const.py"])

tests/unit_tests/core/test_entity_helpers.py

@@ -1,11 +1,14 @@
 """Test get_base_entity_object_id function matches C++ behavior."""
 
+import ast
 from collections.abc import Callable, Generator
+from functools import cache
+from importlib import import_module
 import logging
 from pathlib import Path
 import re
 from typing import Any
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
 
@@ -40,6 +43,7 @@ from esphome.core.entity_helpers import (
     finalize_entity_strings,
     get_base_entity_object_id,
     inherit_property_from,
+    lazy_load_setup,
     lazy_load_validator,
     mqtt_component_class,
     register_device_class,
@@ -1294,6 +1298,34 @@ def test_lazy_load_validator_defers_import() -> None:
         delegate.assert_called_once_with(config)
 
 
+@pytest.mark.asyncio
+async def test_lazy_load_setup_defers_import() -> None:
+    """The setup hook no-ops without importing unless the component is loaded."""
+    with (
+        patch.object(CORE, "loaded_integrations", set()),
+        patch("esphome.core.entity_helpers.import_module") as import_mock,
+    ):
+        await lazy_load_setup("zigbee", "setup_number", "var", {}, 0, 10, 1)
+        import_mock.assert_not_called()
+
+        CORE.loaded_integrations.add("zigbee")
+        import_mock.return_value.setup_number = AsyncMock()
+        await lazy_load_setup("zigbee", "setup_number", "var", {}, 0, 10, 1)
+        import_mock.assert_called_once_with("esphome.components.zigbee")
+        import_mock.return_value.setup_number.assert_awaited_once_with(
+            "var", {}, 0, 10, 1
+        )
+
+
+@pytest.mark.asyncio
+async def test_lazy_load_setup_names_missing_hook() -> None:
+    with (
+        patch.object(CORE, "loaded_integrations", {"zigbee"}),
+        pytest.raises(ValueError, match="no_such_hook"),
+    ):
+        await lazy_load_setup("zigbee", "no_such_hook", "var", {})
+
+
 def test_lazy_load_validator_rejects_unknown_component() -> None:
     """A typo in the component name fails at schema construction."""
     with pytest.raises(ValueError, match="no_such_component"):
@@ -1339,6 +1371,57 @@ def test_integration_class_handles_match_owning_definitions() -> None:
         assert str(owning) == f"zigbee::{class_name}"
 
 
+@cache
+def _component_calls_by_name() -> dict[str, list[tuple[Path, ast.Call]]]:
+    components = Path(entity_helpers.__file__).parent.parent / "components"
+    calls: dict[str, list[tuple[Path, ast.Call]]] = {}
+    for path in components.rglob("*.py"):
+        for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))):
+            if isinstance(node, ast.Call) and (name := getattr(node.func, "id", None)):
+                calls.setdefault(name, []).append((path, node))
+    return calls
+
+
+def _component_calls(func: str, nargs: int) -> set[tuple[str, ...]]:
+    """The leading string arguments of every call to ``func`` in the component sources."""
+    calls: set[tuple[str, ...]] = set()
+    for path, node in _component_calls_by_name().get(func, []):
+        args = node.args[:nargs]
+        # A computed name would escape the checks below
+        assert all(
+            isinstance(arg, ast.Constant) and isinstance(arg.value, str) for arg in args
+        ), f"{path}:{node.lineno} {func} needs string literal arguments"
+        calls.add(tuple(arg.value for arg in args))
+    return calls
+
+
+def test_mqtt_component_class_names_exist() -> None:
+    """Every mqtt_component_class("...") call names a class mqtt declares."""
+    from esphome.components import mqtt
+
+    names = _component_calls("mqtt_component_class", 1)
+    assert ("MQTTSensorComponent",) in names
+    for (name,) in names:
+        assert str(mqtt_component_class(name)) == str(getattr(mqtt, name)), name
+
+
+def test_lazy_load_targets_exist() -> None:
+    """Every lazy_load_validator/lazy_load_setup call names a real component hook."""
+    calls = _component_calls("lazy_load_validator", 2) | _component_calls(
+        "lazy_load_setup", 2
+    )
+    assert ("zigbee", "setup_switch") in calls
+    for component, name in calls:
+        module = import_module(f"esphome.components.{component}")
+        assert getattr(module, name, None) is not None, (component, name)
+
+
+def test_datetime_time_handle_matches_time_component() -> None:
+    from esphome.components import datetime, time
+
+    assert str(datetime._RealTimeClock) == str(time.RealTimeClock)
+
+
 def test_zigbee_entity_schema_cluster_options() -> None:
     """The cluster key only exists for esp32 + zigbee and is limited to the
     platform's options, with the first option as its default."""

tests/unit_tests/test_lazy_imports.py

@@ -220,7 +220,33 @@ def test_api_client_does_not_import_heavy_modules() -> None:
     )
 
 
-@pytest.mark.parametrize("component", ["binary_sensor", "sensor"])
+# Packages a domain needs for its own schema; cheap and never an integration
+_ENTITY_COMPONENT_DEPENDENCIES = {
+    "infrared": ("esphome.components.ir_rf_base",),
+    "light": ("esphome.components.power_supply", "esphome.components.const.css_colors"),
+    "radio_frequency": ("esphome.components.ir_rf_base",),
+}
+
+
+def _entity_components() -> list[str]:
+    """Every platform component that registers entities, found from source."""
+    components = Path(__file__).parents[2] / "esphome" / "components"
+    return sorted(
+        init.parent.name
+        for init in components.glob("*/__init__.py")
+        if "IS_PLATFORM_COMPONENT = True" in (text := init.read_text())
+        and ("setup_entity(" in text or "queue_entity_register(" in text)
+    )
+
+
+def test_entity_component_tables_match_discovery() -> None:
+    """Catch stale table entries and a discovery that silently finds nothing."""
+    found = set(_entity_components())
+    stale = _ENTITY_COMPONENT_DEPENDENCIES.keys() - found
+    assert not stale, f"not discovered as entity components: {sorted(stale)}"
+
+
+@pytest.mark.parametrize("component", _entity_components())
 def test_entity_component_does_not_import_integrations(component: str) -> None:
     """An entity component must not drag in its optional integrations.
 
@@ -229,15 +255,18 @@ def test_entity_component_does_not_import_integrations(component: str) -> None:
     esp32/logger chains mqtt and zigbee pull in) must only load when the
     user's config actually uses them.
     """
+    own = f"esphome.components.{component}"
     allowed = (
         "esphome.components",
-        f"esphome.components.{component}",
+        own,
         "esphome.components.const",
+        *_ENTITY_COMPONENT_DEPENDENCIES.get(component, ()),
     )
     check = (
-        f"import sys; import esphome.components.{component}; "
+        f"import sys; import {own}; "
         f"leaked = [m for m in sys.modules "
-        f"if m.startswith('esphome.components') and m not in {allowed!r}]; "
+        f"if m.startswith('esphome.components') and m not in {allowed!r} "
+        f"and not m.startswith('{own}.')]; "
         "print(','.join(leaked))"
     )
     result = subprocess.run(