Merge branch 'binary-sensor-lazy-integration-imports' into sensor-lazy-integration-imports

# Conflicts:
#	tests/benchmarks/python/test_component_import_bench.py

esphome/components/binary_sensor/__init__.py

@@ -3,7 +3,6 @@ from logging import getLogger
 from esphome import automation, core
 from esphome.automation import maybe_simple_id
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server, zigbee
 from esphome.components.const import CONF_ON_STATE_CHANGE
 from esphome.config_helpers import filter_source_files_from_defines
 import esphome.config_validation as cv
@@ -29,7 +28,6 @@ from esphome.const import (
     CONF_STATE,
     CONF_TIMING,
     CONF_TRIGGER_ID,
-    CONF_WEB_SERVER,
     DEVICE_CLASS_BATTERY,
     DEVICE_CLASS_BATTERY_CHARGING,
     DEVICE_CLASS_CARBON_MONOXIDE,
@@ -61,10 +59,12 @@ from esphome.const import (
     DEVICE_CLASS_VIBRATION,
     DEVICE_CLASS_WINDOW,
 )
-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,
+    lazy_load_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_device_class,
     setup_entity,
@@ -433,14 +433,14 @@ def validate_publish_initial_state(value):
 
 
 _BINARY_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(zigbee.BINARY_SENSOR_SCHEMA)
+    .extend(entity_helpers.ZIGBEE_BINARY_SENSOR_SCHEMA)
     .extend(
         {
             cv.GenerateID(): cv.declare_id(BinarySensor),
             cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
-                mqtt.MQTTBinarySensorComponent
+                mqtt_component_class("MQTTBinarySensorComponent")
             ),
             cv.Exclusive(
                 CONF_PUBLISH_INITIAL_STATE, CONF_TRIGGER_ON_INITIAL_STATE
@@ -508,7 +508,7 @@ _BINARY_SENSOR_SCHEMA = (
 
 
 _BINARY_SENSOR_SCHEMA.add_extra(entity_duplicate_validator("binary_sensor"))
-_BINARY_SENSOR_SCHEMA.add_extra(zigbee.validate_binary_sensor)
+_BINARY_SENSOR_SCHEMA.add_extra(lazy_load_validator("zigbee", "validate_binary_sensor"))
 
 
 def binary_sensor_schema(
@@ -619,14 +619,12 @@ async def setup_binary_sensor_core_(var, config):
 
     CORE.add_job(_build_binary_sensor_automations, var, 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)
 
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
+    if "zigbee" in CORE.loaded_integrations:
+        from esphome.components import zigbee
 
-    await zigbee.setup_binary_sensor(var, config)
+        await zigbee.setup_binary_sensor(var, config)
 
 
 async def register_binary_sensor(var, config):

esphome/components/sensor/__init__.py

@@ -4,7 +4,6 @@ from typing import Any
 
 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server, zigbee
 from esphome.components.const import CONF_B_CONSTANT
 from esphome.config_helpers import filter_source_files_from_defines
 import esphome.config_validation as cv
@@ -47,7 +46,6 @@ from esphome.const import (
     CONF_TYPE_ID,
     CONF_UNIT_OF_MEASUREMENT,
     CONF_VALUE,
-    CONF_WEB_SERVER,
     CONF_WINDOW_SIZE,
     DEVICE_CLASS_ABSOLUTE_HUMIDITY,
     DEVICE_CLASS_APPARENT_POWER,
@@ -113,11 +111,20 @@ from esphome.const import (
     DEVICE_CLASS_WIND_SPEED,
     ENTITY_CATEGORY_CONFIG,
 )
-from esphome.core import CORE, ID, CoroPriority, Lambda, coroutine_with_priority
+from esphome.core import (
+    CORE,
+    ID,
+    CoroPriority,
+    Lambda,
+    coroutine_with_priority,
+    entity_helpers,
+)
 from esphome.core.config import UNIT_OF_MEASUREMENT_MAX_LENGTH
 from esphome.core.entity_helpers import (
     SubEntities,
     entity_duplicate_validator,
+    lazy_load_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_device_class,
     setup_entity,
@@ -327,12 +334,14 @@ validate_icon = cv.icon
 validate_device_class = cv.one_of(*DEVICE_CLASSES, lower=True, space="_")
 
 _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(zigbee.SENSOR_SCHEMA)
+    .extend(entity_helpers.ZIGBEE_SENSOR_SCHEMA)
     .extend(
         {
-            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(mqtt.MQTTSensorComponent),
+            cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
+                mqtt_component_class("MQTTSensorComponent")
+            ),
             cv.GenerateID(): cv.declare_id(Sensor),
             cv.Optional(
                 CONF_UNIT_OF_MEASUREMENT, visibility=cv.Visibility.ADVANCED
@@ -374,7 +383,7 @@ _SENSOR_SCHEMA = (
 )
 
 _SENSOR_SCHEMA.add_extra(entity_duplicate_validator("sensor"))
-_SENSOR_SCHEMA.add_extra(zigbee.validate_sensor)
+_SENSOR_SCHEMA.add_extra(lazy_load_validator("zigbee", "validate_sensor"))
 
 
 def sensor_schema(
@@ -1061,22 +1070,20 @@ async def setup_sensor_core_(var, config):
 
     CORE.add_job(_build_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)
+    mqtt_ = await entity_helpers.setup_entity_integrations(var, config)
+    if mqtt_ is not None and (
+        (expire_after := config.get(CONF_EXPIRE_AFTER, cv.UNDEFINED))
+        is not cv.UNDEFINED
+    ):
+        if expire_after is None:
+            cg.add(mqtt_.disable_expire_after())
+        else:
+            cg.add(mqtt_.set_expire_after(expire_after))
 
-        if (
-            expire_after := config.get(CONF_EXPIRE_AFTER, cv.UNDEFINED)
-        ) is not cv.UNDEFINED:
-            if expire_after is None:
-                cg.add(mqtt_.disable_expire_after())
-            else:
-                cg.add(mqtt_.set_expire_after(expire_after))
+    if "zigbee" in CORE.loaded_integrations:
+        from esphome.components import zigbee
 
-    if web_server_config := config.get(CONF_WEB_SERVER):
-        await web_server.add_entity_config(var, web_server_config)
-
-    await zigbee.setup_sensor(var, config)
+        await zigbee.setup_sensor(var, config)
 
 
 async def register_sensor(var, config):

esphome/components/web_server/__init__.py

@@ -12,7 +12,11 @@ from esphome.components.json import enable_arena
 from esphome.components.logger import request_log_listener
 from esphome.components.web_server_base import CONF_WEB_SERVER_BASE_ID
 import esphome.config_validation as cv
-from esphome.const import (
+
+# Re-exported so entity components and external components keep importing
+# these from web_server; defined outside this package so entity base
+# schemas do not need to import it.
+from esphome.const import (  # noqa: F401
     CONF_AP,
     CONF_AUTH,
     CONF_COMPRESSION,
@@ -31,6 +35,8 @@ from esphome.const import (
     CONF_OTA,
     CONF_PASSWORD,
     CONF_PORT,
+    CONF_SORTING_GROUP_ID,
+    CONF_SORTING_WEIGHT,
     CONF_STATIC_IP,
     CONF_TYPE,
     CONF_USERNAME,
@@ -46,6 +52,7 @@ from esphome.const import (
     PLATFORM_RTL87XX,
 )
 from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core.entity_helpers import WEBSERVER_SORTING_SCHEMA  # noqa: F401
 from esphome.cpp_generator import MockObj
 import esphome.final_validate as fv
 from esphome.types import ConfigType
@@ -72,9 +79,7 @@ DOMAIN = "web_server"
 AUTH_TYPE_BASIC = "basic"
 AUTH_TYPE_DIGEST = "digest"
 
-CONF_SORTING_GROUP_ID = "sorting_group_id"
 CONF_SORTING_GROUPS = "sorting_groups"
-CONF_SORTING_WEIGHT = "sorting_weight"
 CONF_ALLOWED_ORIGINS = "allowed_origins"
 
 # Schema default that also matches the C++ initializer in web_server_base.h; codegen
@@ -248,27 +253,6 @@ sorting_group = {
     cv.Optional(CONF_SORTING_WEIGHT): cv.float_,
 }
 
-WEBSERVER_SORTING_SCHEMA = cv.Schema(
-    {
-        # The per-entity web_server block is cosmetic dashboard ordering —
-        # mark the whole block advanced; the children inherit via the cascade.
-        cv.Optional(CONF_WEB_SERVER, visibility=cv.Visibility.ADVANCED): cv.Schema(
-            {
-                cv.OnlyWith(CONF_WEB_SERVER_ID, "web_server"): cv.use_id(WebServer),
-                cv.Optional(CONF_SORTING_WEIGHT): cv.All(
-                    cv.requires_component(DOMAIN),
-                    cv.float_,
-                ),
-                cv.Optional(CONF_SORTING_GROUP_ID): cv.All(
-                    cv.requires_component(DOMAIN),
-                    cv.use_id(cg.int_),
-                ),
-            }
-        )
-    }
-)
-
-
 CONFIG_SCHEMA = cv.All(
     cv.Schema(
         {

esphome/components/zigbee/__init__.py

@@ -26,11 +26,14 @@ from esphome.const import (
     CONF_VERSION,
     __version__ as ESPHOME_VERSION,
 )
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority, entity_helpers
 import esphome.final_validate as fv
 from esphome.types import ConfigType
 
-from .const import (
+# CONF_CLUSTER, CONF_ENDPOINT, CONF_MAX_EP_NUMBER, CONF_REPORT,
+# CONF_USE_DEVICE_TYPE and REPORT are re-exported for existing consumers of
+# this package's namespace.
+from .const import (  # noqa: F401
     CONF_CLUSTER,
     CONF_ENDPOINT,
     CONF_MAX_EP_NUMBER,
@@ -62,12 +65,7 @@ from .zigbee_esp32 import (
     validate_switch_esp32,
     zigbee_require_vfs_select,
 )
-from .zigbee_zephyr import (
-    zephyr_binary_sensor,
-    zephyr_number,
-    zephyr_sensor,
-    zephyr_switch,
-)
+from .zigbee_zephyr import zephyr_number, zephyr_switch
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -77,68 +75,20 @@ DOMAIN = "zigbee"
 CONFLICTS_WITH = ["openthread"]
 
 
-def _check_report_deprecation(value: str) -> str:
-    if str(value).lower() == "coordinator":
-        _LOGGER.warning(
-            "Report option 'coordinator' is deprecated and will be removed in a future release. Use 'default' instead."
-        )
-    return value
-
-
-def _get_base_schema(cluster_options: list[str] | None = None) -> cv.Schema:
-    schema = cv.Schema(
-        {
-            cv.Optional(CONF_REPORT): cv.All(
-                cv.requires_component(DOMAIN),
-                cv.requires_component(ESP32_DOMAIN),
-                _check_report_deprecation,
-                cv.enum(REPORT, lower=True),
-            ),
-            cv.Optional(CONF_ENDPOINT): cv.All(
-                cv.requires_component(DOMAIN),
-                cv.requires_component(ESP32_DOMAIN),
-                cv.int_range(1, CONF_MAX_EP_NUMBER),
-            ),
-            cv.Optional(CONF_USE_DEVICE_TYPE): cv.All(
-                cv.requires_component(DOMAIN),
-                cv.requires_component(ESP32_DOMAIN),
-                cv.boolean,
-            ),
-        }
-    )
-    if cluster_options:
-        schema = cv.Schema(
-            {
-                cv.OnlyWith(
-                    CONF_CLUSTER, ["esp32", "zigbee"], default=cluster_options[0]
-                ): cv.All(
-                    cv.requires_component(DOMAIN),
-                    cv.requires_component(ESP32_DOMAIN),
-                    cv.one_of(*cluster_options, lower=True),
-                ),
-            }
-        ).extend(schema)
-    return schema
-
-
-# set BASE_SCHEMA for CI and backwards compatibility
-BASE_SCHEMA = _get_base_schema()
-
-BINARY_SENSOR_SCHEMA = (
-    cv.Schema({})
-    .extend(_get_base_schema(["generic", "device_class"]))
-    .extend(zephyr_binary_sensor)
-)
-SENSOR_SCHEMA = (
-    cv.Schema({})
-    .extend(_get_base_schema(["generic", "device_class"]))
-    .extend(zephyr_sensor)
-)
+# Defined in esphome.core.entity_helpers so entity base schemas can reference
+# them without importing this package; re-exported here for existing consumers.
+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(_get_base_schema(["generic", "on_off"])).extend(zephyr_switch)
+    cv.Schema({})
+    .extend(entity_helpers.zigbee_base_entity_schema(["generic", "on_off"]))
+    .extend(zephyr_switch)
 )
 NUMBER_SCHEMA = (
-    cv.Schema({}).extend(_get_base_schema(["generic"])).extend(zephyr_number)
+    cv.Schema({})
+    .extend(entity_helpers.zigbee_base_entity_schema(["generic"]))
+    .extend(zephyr_number)
 )
 
 

esphome/components/zigbee/const.py

@@ -1,7 +1,15 @@
 from enum import IntEnum
 
 import esphome.codegen as cg
-from esphome.const import (
+
+# The entity schema keys and report enum live outside this package so
+# entity base schemas can use them without importing it; re-imported here
+# so zigbee code keeps its existing import paths.
+from esphome.const import (  # noqa: F401  # pylint: disable=unused-import
+    CONF_CLUSTER,
+    CONF_ENDPOINT,
+    CONF_REPORT,
+    CONF_USE_DEVICE_TYPE,
     DEVICE_CLASS_CURRENT,
     DEVICE_CLASS_DURATION,
     DEVICE_CLASS_ENERGY,
@@ -48,6 +56,10 @@ from esphome.const import (
     UNIT_WATT,
     UNIT_WATT_HOURS,
 )
+from esphome.core.entity_helpers import (  # noqa: F401  # pylint: disable=unused-import
+    ZIGBEE_MAX_EP_NUMBER as CONF_MAX_EP_NUMBER,
+    ZIGBEE_REPORT as REPORT,
+)
 
 zigbee_ns = cg.esphome_ns.namespace("zigbee")
 ZigbeeComponent = zigbee_ns.class_("ZigbeeComponent", cg.Component)
@@ -56,24 +68,11 @@ BinaryAttrs = zigbee_ns.struct("BinaryAttrs")
 AnalogAttrs = zigbee_ns.struct("AnalogAttrs")
 AnalogAttrsOutput = zigbee_ns.struct("AnalogAttrsOutput")
 
-report = zigbee_ns.enum("ZigbeeReportT")
-REPORT = {
-    "coordinator": report.ZIGBEE_REPORT_COORDINATOR,
-    "enable": report.ZIGBEE_REPORT_ENABLE,
-    "force": report.ZIGBEE_REPORT_FORCE,
-    "default": report.ZIGBEE_REPORT_DEFAULT,
-}
-
-CONF_CLUSTER = "cluster"
-CONF_ENDPOINT = "endpoint"
-CONF_MAX_EP_NUMBER = 239
 CONF_ON_JOIN = "on_join"
 CONF_WIPE_ON_BOOT = "wipe_on_boot"
-CONF_REPORT = "report"
 CONF_ROUTER = "router"
 CONF_POWER_SOURCE = "power_source"
 CONF_SLEEPY = "sleepy"
-CONF_USE_DEVICE_TYPE = "use_device_type"
 POWER_SOURCE = {
     "UNKNOWN": 0x00,  # ZB_ZCL_BASIC_POWER_SOURCE_UNKNOWN
     "MAINS_SINGLE_PHASE": 0x01,  # ZB_ZCL_BASIC_POWER_SOURCE_MAINS_SINGLE_PHASE

esphome/components/zigbee/const_zephyr.py

@@ -1,7 +1,8 @@
+# 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
+
 CONF_MAX_EP_NUMBER_ZEPHYR = 8
-CONF_ZIGBEE_ID = "zigbee_id"
-CONF_ZIGBEE_BINARY_SENSOR = "zigbee_binary_sensor"
-CONF_ZIGBEE_SENSOR = "zigbee_sensor"
 CONF_ZIGBEE_SWITCH = "zigbee_switch"
 CONF_ZIGBEE_NUMBER = "zigbee_number"
 CONF_IEEE802154_VENDOR_OUI = "ieee802154_vendor_oui"

esphome/components/zigbee/zigbee_zephyr.py

@@ -60,24 +60,6 @@ ZigbeeSensor = zigbee_ns.class_("ZigbeeSensor", cg.Component)
 ZigbeeSwitch = zigbee_ns.class_("ZigbeeSwitch", cg.Component)
 ZigbeeNumber = zigbee_ns.class_("ZigbeeNumber", cg.Component)
 
-zephyr_binary_sensor = cv.Schema(
-    {
-        cv.OnlyWith(CONF_ZIGBEE_ID, ["nrf52", "zigbee"]): cv.use_id(ZigbeeComponent),
-        cv.OnlyWith(CONF_ZIGBEE_BINARY_SENSOR, ["nrf52", "zigbee"]): cv.declare_id(
-            ZigbeeBinarySensor
-        ),
-    }
-)
-
-zephyr_sensor = cv.Schema(
-    {
-        cv.OnlyWith(CONF_ZIGBEE_ID, ["nrf52", "zigbee"]): cv.use_id(ZigbeeComponent),
-        cv.OnlyWith(CONF_ZIGBEE_SENSOR, ["nrf52", "zigbee"]): cv.declare_id(
-            ZigbeeSensor
-        ),
-    }
-)
-
 zephyr_switch = cv.Schema(
     {
         cv.OnlyWith(CONF_ZIGBEE_ID, ["nrf52", "zigbee"]): cv.use_id(ZigbeeComponent),

esphome/const.py

@@ -267,6 +267,7 @@ CONF_CLOCK_RESOLUTION = "clock_resolution"
 CONF_CLOSE_ACTION = "close_action"
 CONF_CLOSE_DURATION = "close_duration"
 CONF_CLOSE_ENDSTOP = "close_endstop"
+CONF_CLUSTER = "cluster"
 CONF_CO2 = "co2"
 CONF_CODE = "code"
 CONF_COL = "col"
@@ -405,6 +406,7 @@ CONF_ENABLE_PRIVATE_NETWORK_ACCESS = "enable_private_network_access"
 CONF_ENABLE_RRM = "enable_rrm"
 CONF_ENABLE_TIME = "enable_time"
 CONF_ENCRYPTION = "encryption"
+CONF_ENDPOINT = "endpoint"
 CONF_ENERGY = "energy"
 CONF_ENTITY_CATEGORY = "entity_category"
 CONF_ENTITY_ID = "entity_id"
@@ -903,6 +905,7 @@ CONF_REFERENCE_VOLTAGE = "reference_voltage"
 CONF_REFRESH = "refresh"
 CONF_RELABEL = "relabel"
 CONF_REPEAT = "repeat"
+CONF_REPORT = "report"
 CONF_REPOSITORY = "repository"
 CONF_RESET = "reset"
 CONF_RESET_DURATION = "reset_duration"
@@ -980,6 +983,8 @@ CONF_SLEEP_DURATION = "sleep_duration"
 CONF_SLEEP_PIN = "sleep_pin"
 CONF_SLEEP_WHEN_DONE = "sleep_when_done"
 CONF_SONY = "sony"
+CONF_SORTING_GROUP_ID = "sorting_group_id"
+CONF_SORTING_WEIGHT = "sorting_weight"
 CONF_SOURCE = "source"
 CONF_SOURCE_ID = "source_id"
 CONF_SPEAKER = "speaker"
@@ -1111,6 +1116,7 @@ CONF_UPDATE_ON_BOOT = "update_on_boot"
 CONF_URL = "url"
 CONF_USE_ABBREVIATIONS = "use_abbreviations"
 CONF_USE_ADDRESS = "use_address"
+CONF_USE_DEVICE_TYPE = "use_device_type"
 CONF_USE_DMA = "use_dma"
 CONF_USE_FAHRENHEIT = "use_fahrenheit"
 CONF_USERNAME = "username"
@@ -1162,6 +1168,9 @@ CONF_Y = "y"
 CONF_Y_GRID = "y_grid"
 CONF_YEAR = "year"
 CONF_ZERO = "zero"
+CONF_ZIGBEE_BINARY_SENSOR = "zigbee_binary_sensor"
+CONF_ZIGBEE_ID = "zigbee_id"
+CONF_ZIGBEE_SENSOR = "zigbee_sensor"
 
 TYPE_GIT = "git"
 TYPE_LOCAL = "local"

esphome/core/entity_helpers.py

@@ -1,21 +1,35 @@
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
 import functools
+from importlib import import_module
+from importlib.util import find_spec
 import logging
 from typing import Any
 
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.const import (
+    CONF_CLUSTER,
     CONF_DEVICE_CLASS,
     CONF_DEVICE_ID,
     CONF_DISABLED_BY_DEFAULT,
+    CONF_ENDPOINT,
     CONF_ENTITY_CATEGORY,
     CONF_ICON,
     CONF_ID,
     CONF_INTERNAL,
+    CONF_MQTT_ID,
     CONF_NAME,
+    CONF_REPORT,
+    CONF_SORTING_GROUP_ID,
+    CONF_SORTING_WEIGHT,
     CONF_UNIT_OF_MEASUREMENT,
+    CONF_USE_DEVICE_TYPE,
+    CONF_WEB_SERVER,
+    CONF_WEB_SERVER_ID,
+    CONF_ZIGBEE_BINARY_SENSOR,
+    CONF_ZIGBEE_ID,
+    CONF_ZIGBEE_SENSOR,
 )
 from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
 from esphome.core.config import (
@@ -23,7 +37,7 @@ from esphome.core.config import (
     ICON_MAX_LENGTH,
     UNIT_OF_MEASUREMENT_MAX_LENGTH,
 )
-from esphome.cpp_generator import MockObj, RawStatement, add, get_variable
+from esphome.cpp_generator import MockObj, MockObjClass, RawStatement, add, get_variable
 from esphome.cpp_types import App
 import esphome.final_validate as fv
 from esphome.helpers import (
@@ -664,3 +678,179 @@ def entity_duplicate_validator(platform: str) -> Callable[[ConfigType], ConfigTy
         return config
 
     return validator
+
+
+# ---------------------------------------------------------------------------
+# Cross-integration entity schema fragments
+# ---------------------------------------------------------------------------
+#
+# Entity base schemas offer mqtt/web_server/zigbee options, but importing
+# those packages pulls their full dependency chains (mqtt and zigbee both
+# import the esp32 package) into every entity component import. The
+# fragments below are built from cheap primitives instead: MockObjClass
+# identity is string-based, so the class handles here are interchangeable
+# with the ones the integrations declare, and every key is guarded by
+# ``cv.requires_component``/``cv.OnlyWith``, which consult
+# ``CORE.loaded_integrations`` at validation time without importing. The
+# owning integrations re-export the shared schema names and key strings so
+# each stays defined once.
+
+_WebServer = cg.esphome_ns.namespace("web_server").class_("WebServer", cg.Component)
+
+WEBSERVER_SORTING_SCHEMA = cv.Schema(
+    {
+        # The per-entity web_server block is cosmetic dashboard ordering —
+        # mark the whole block advanced; the children inherit via the cascade.
+        cv.Optional(CONF_WEB_SERVER, visibility=cv.Visibility.ADVANCED): cv.Schema(
+            {
+                cv.OnlyWith(CONF_WEB_SERVER_ID, "web_server"): cv.use_id(_WebServer),
+                cv.Optional(CONF_SORTING_WEIGHT): cv.All(
+                    cv.requires_component("web_server"),
+                    cv.float_,
+                ),
+                cv.Optional(CONF_SORTING_GROUP_ID): cv.All(
+                    cv.requires_component("web_server"),
+                    cv.use_id(cg.int_),
+                ),
+            }
+        )
+    }
+)
+
+_mqtt_ns = cg.esphome_ns.namespace("mqtt")
+_MQTTComponent = _mqtt_ns.class_("MQTTComponent", cg.Component)
+
+
+def mqtt_component_class(name: str) -> MockObjClass:
+    """Handle for a per-entity mqtt::<name> companion class."""
+    return _mqtt_ns.class_(name, _MQTTComponent)
+
+
+ZIGBEE_MAX_EP_NUMBER = 239
+
+_zigbee_ns = cg.esphome_ns.namespace("zigbee")
+_ZigbeeComponent = _zigbee_ns.class_("ZigbeeComponent", cg.Component)
+_zigbee_report = _zigbee_ns.enum("ZigbeeReportT")
+ZIGBEE_REPORT = {
+    "coordinator": _zigbee_report.ZIGBEE_REPORT_COORDINATOR,
+    "enable": _zigbee_report.ZIGBEE_REPORT_ENABLE,
+    "force": _zigbee_report.ZIGBEE_REPORT_FORCE,
+    "default": _zigbee_report.ZIGBEE_REPORT_DEFAULT,
+}
+
+
+def _check_report_deprecation(value: str) -> str:
+    if str(value).lower() == "coordinator":
+        _LOGGER.warning(
+            "Report option 'coordinator' is deprecated and will be removed in a future release. Use 'default' instead."
+        )
+    return value
+
+
+def zigbee_base_entity_schema(cluster_options: list[str] | None = None) -> cv.Schema:
+    """Zigbee options every entity offers; ``cluster_options`` adds the esp32 cluster key."""
+    schema = cv.Schema(
+        {
+            cv.Optional(CONF_REPORT): cv.All(
+                cv.requires_component("zigbee"),
+                cv.requires_component("esp32"),
+                _check_report_deprecation,
+                cv.enum(ZIGBEE_REPORT, lower=True),
+            ),
+            cv.Optional(CONF_ENDPOINT): cv.All(
+                cv.requires_component("zigbee"),
+                cv.requires_component("esp32"),
+                cv.int_range(1, ZIGBEE_MAX_EP_NUMBER),
+            ),
+            cv.Optional(CONF_USE_DEVICE_TYPE): cv.All(
+                cv.requires_component("zigbee"),
+                cv.requires_component("esp32"),
+                cv.boolean,
+            ),
+        }
+    )
+    if cluster_options:
+        schema = cv.Schema(
+            {
+                cv.OnlyWith(
+                    CONF_CLUSTER, ["esp32", "zigbee"], default=cluster_options[0]
+                ): cv.All(
+                    cv.requires_component("zigbee"),
+                    cv.requires_component("esp32"),
+                    cv.one_of(*cluster_options, lower=True),
+                ),
+            }
+        ).extend(schema)
+    return schema
+
+
+ZIGBEE_BASE_ENTITY_SCHEMA = zigbee_base_entity_schema()
+
+# Entity platform -> (config key, zigbee C++ class, esp32 cluster options). A
+# unit test checks each class against the owning declaration in zigbee_zephyr.
+_ZIGBEE_ENTITY_CLASSES = {
+    "binary_sensor": (
+        CONF_ZIGBEE_BINARY_SENSOR,
+        "ZigbeeBinarySensor",
+        ["generic", "device_class"],
+    ),
+    "sensor": (CONF_ZIGBEE_SENSOR, "ZigbeeSensor", ["generic", "device_class"]),
+}
+
+
+def _zigbee_entity_schema(platform: str) -> cv.Schema:
+    conf_key, class_name, cluster_options = _ZIGBEE_ENTITY_CLASSES[platform]
+    return zigbee_base_entity_schema(cluster_options).extend(
+        {
+            cv.OnlyWith(CONF_ZIGBEE_ID, ["nrf52", "zigbee"]): cv.use_id(
+                _ZigbeeComponent
+            ),
+            cv.OnlyWith(conf_key, ["nrf52", "zigbee"]): cv.declare_id(
+                _zigbee_ns.class_(class_name, cg.Component)
+            ),
+        }
+    )
+
+
+ZIGBEE_BINARY_SENSOR_SCHEMA = _zigbee_entity_schema("binary_sensor")
+
+ZIGBEE_SENSOR_SCHEMA = _zigbee_entity_schema("sensor")
+
+
+def lazy_load_validator(
+    component: str, name: str
+) -> Callable[[ConfigType], ConfigType]:
+    """Schema extra delegating to ``components.<component>.<name>`` when loaded."""
+    if find_spec(f"esphome.components.{component}") is None:
+        raise ValueError(f"No such component {component!r}")
+
+    def validator(config: ConfigType) -> ConfigType:
+        if component not in CORE.loaded_integrations:
+            return config
+        module = import_module(f"esphome.components.{component}")
+        if (delegate := getattr(module, name, None)) is None:
+            raise ValueError(f"{component} has no validator {name!r}")
+        return delegate(config)
+
+    return validator
+
+
+async def setup_entity_integrations(var: MockObj, config: ConfigType) -> MockObj | None:
+    """Register the mqtt companion and web_server entry for an entity.
+
+    Imports the integrations lazily; returns the mqtt companion (or None)
+    so callers can apply integration specific options to it.
+    """
+    mqtt_ = None
+    if (mqtt_id := config.get(CONF_MQTT_ID)) is not None:
+        from esphome.components import mqtt
+
+        mqtt_ = cg.new_Pvariable(mqtt_id, var)
+        await mqtt.register_mqtt_component(mqtt_, config)
+
+    if web_server_config := config.get(CONF_WEB_SERVER):
+        from esphome.components import web_server
+
+        await web_server.add_entity_config(var, web_server_config)
+
+    return mqtt_

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 = 1016
+CONST_PY_MAX_CONF = 1025
 
 
 @lint_content_check(include=["esphome/const.py"])

tests/benchmarks/python/test_component_import_bench.py

@@ -0,0 +1,71 @@
+"""CodSpeed benchmarks for importing entity components.
+
+Every ``esphome config`` or ``esphome compile`` imports the Python module of
+each component the YAML uses. A component that imports its optional
+integrations (mqtt, web_server, zigbee) at module scope drags their whole
+import chains in for every user, so the cost of a fresh import is what these
+benchmarks pin down.
+
+The core of esphome is imported once up front and stays loaded; each run
+drops only the component modules from ``sys.modules`` and imports the
+component again, so the import machinery and the module bodies of the
+component and everything it pulls in run every time, as they do in a new
+process, while a change to the core moves no number here. The modules the
+session had before a case are put back afterwards, so later tests keep the
+objects they already hold.
+
+Skipped when ``pytest-codspeed`` isn't installed so the regular unit-test
+suite keeps working unchanged.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+import importlib
+import sys
+
+import pytest
+
+pytest.importorskip("pytest_codspeed")
+
+# Load the core before the first case so no case pays for it.
+import esphome.codegen  # noqa: E402, F401
+import esphome.config_validation  # noqa: E402, F401
+
+COMPONENTS = ("binary_sensor", "sensor", "logger")
+COMPONENT_PACKAGE = "esphome.components"
+COMPONENT_PREFIX = COMPONENT_PACKAGE + "."
+
+
+def _component_modules() -> dict[str, object]:
+    # The package itself goes too: ``from esphome.components import x`` is
+    # answered from its attributes, so a stale package would hide the import.
+    return {
+        n: m
+        for n, m in sys.modules.items()
+        if n == COMPONENT_PACKAGE or n.startswith(COMPONENT_PREFIX)
+    }
+
+
+def _forget_component_modules() -> None:
+    for name in _component_modules():
+        del sys.modules[name]
+
+
+def _import_component(name: str) -> object:
+    _forget_component_modules()
+    return importlib.import_module(COMPONENT_PREFIX + name)
+
+
+@pytest.fixture
+def restore_component_modules() -> Generator[None]:
+    saved = _component_modules()
+    yield
+    _forget_component_modules()
+    sys.modules.update(saved)
+
+
+@pytest.mark.usefixtures("restore_component_modules")
+@pytest.mark.parametrize("component", COMPONENTS)
+def test_import_component(benchmark, component: str) -> None:
+    benchmark(_import_component, component)

tests/unit_tests/core/test_entity_helpers.py

@@ -1,14 +1,17 @@
 """Test get_base_entity_object_id function matches C++ behavior."""
 
 from collections.abc import Callable, Generator
+import logging
 from pathlib import Path
 import re
 from typing import Any
+from unittest.mock import patch
 
 import pytest
 
 from esphome.config_validation import Invalid
 from esphome.const import (
+    CONF_CLUSTER,
     CONF_DEVICE_CLASS,
     CONF_DEVICE_ID,
     CONF_DISABLED_BY_DEFAULT,
@@ -21,11 +24,14 @@ from esphome.const import (
 )
 from esphome.core import CORE, ID, entity_helpers
 from esphome.core.entity_helpers import (
+    _check_report_deprecation,
     _register_string,
     _setup_entity_impl,
     entity_duplicate_validator,
     finalize_entity_strings,
     get_base_entity_object_id,
+    lazy_load_validator,
+    mqtt_component_class,
     register_device_class,
     register_icon,
     register_unit_of_measurement,
@@ -1238,6 +1244,111 @@ async def test_finalize_comment_sanitization(
     assert "INJECTED_CODE" in comment_line  # still visible but safe in comment
 
 
+@pytest.mark.parametrize(
+    ("value", "warns"),
+    [
+        ("coordinator", True),
+        ("enable", False),
+        ("force", False),
+        ("default", False),
+    ],
+)
+def test_check_report_deprecation(
+    value: str, warns: bool, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Deprecated zigbee report options warn; the value always passes through."""
+    with caplog.at_level(logging.WARNING):
+        assert _check_report_deprecation(value) == value
+    assert ("deprecated" in caplog.text) is warns
+
+
+def test_lazy_load_validator_defers_import() -> None:
+    """The validator no-ops without importing unless the component is loaded."""
+    validator = lazy_load_validator("zigbee", "validate_binary_sensor")
+    config = {CONF_NAME: "test"}
+
+    with (
+        patch.object(CORE, "loaded_integrations", set()),
+        patch("esphome.core.entity_helpers.import_module") as import_mock,
+    ):
+        assert validator(config) is config
+        import_mock.assert_not_called()
+
+        CORE.loaded_integrations.add("zigbee")
+        delegate = import_mock.return_value.validate_binary_sensor
+        delegate.return_value = {CONF_NAME: "validated"}
+        assert validator(config) == {CONF_NAME: "validated"}
+        import_mock.assert_called_once_with("esphome.components.zigbee")
+        delegate.assert_called_once_with(config)
+
+
+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"):
+        lazy_load_validator("no_such_component", "validate_binary_sensor")
+
+
+def test_lazy_load_validator_names_missing_hook() -> None:
+    """A missing hook raises a clear error naming the component and hook."""
+    validator = lazy_load_validator("zigbee", "no_such_hook")
+
+    with (
+        patch.object(CORE, "loaded_integrations", {"zigbee"}),
+        patch("esphome.core.entity_helpers.import_module") as import_mock,
+        pytest.raises(ValueError, match="no_such_hook"),
+    ):
+        del import_mock.return_value.no_such_hook
+        validator({})
+
+
+def test_integration_class_handles_match_owning_definitions() -> None:
+    """The cheap class handles must stay string-equal to the integrations'
+    own declarations, or use_id/declare_id resolution silently drifts."""
+    from esphome.components import mqtt, web_server
+    from esphome.components.zigbee import zigbee_zephyr
+    from esphome.components.zigbee.const import ZigbeeComponent
+
+    mqtt_handle = mqtt_component_class("MQTTBinarySensorComponent")
+    assert str(mqtt_handle) == str(mqtt.MQTTBinarySensorComponent)
+    assert mqtt_handle.inherits_from(mqtt.MQTTComponent)
+    assert mqtt.MQTTBinarySensorComponent.inherits_from(entity_helpers._MQTTComponent)
+
+    assert str(entity_helpers._WebServer) == str(web_server.WebServer)
+    assert entity_helpers._WebServer.inherits_from(web_server.WebServer)
+    assert web_server.WebServer.inherits_from(entity_helpers._WebServer)
+
+    assert str(entity_helpers._ZigbeeComponent) == str(ZigbeeComponent)
+    for (
+        _conf_key,
+        class_name,
+        _cluster_options,
+    ) in entity_helpers._ZIGBEE_ENTITY_CLASSES.values():
+        owning = getattr(zigbee_zephyr, class_name)
+        assert str(owning) == f"zigbee::{class_name}"
+
+
+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."""
+    schema = entity_helpers.ZIGBEE_BINARY_SENSOR_SCHEMA
+
+    with patch.object(CORE, "loaded_integrations", set()):
+        assert schema({}) == {}
+        with pytest.raises(Invalid, match="requires component zigbee"):
+            schema({CONF_CLUSTER: "generic"})
+
+    with patch.object(CORE, "loaded_integrations", {"esp32", "zigbee"}):
+        assert schema({}) == {CONF_CLUSTER: "generic"}
+        assert schema({CONF_CLUSTER: "Device_Class"}) == {CONF_CLUSTER: "device_class"}
+        with pytest.raises(Invalid, match="Unknown value 'on_off'"):
+            schema({CONF_CLUSTER: "on_off"})
+
+    with patch.object(CORE, "loaded_integrations", {"nrf52", "zigbee"}):
+        assert CONF_CLUSTER not in schema({})
+        with pytest.raises(Invalid, match="requires component esp32"):
+            schema({CONF_CLUSTER: "generic"})
+
+
 @pytest.mark.asyncio
 async def test_sub_entities_creates_and_sets(
     monkeypatch: pytest.MonkeyPatch, setup_test_environment: list[str]

tests/unit_tests/test_lazy_imports.py

@@ -19,6 +19,8 @@ from pathlib import Path
 import subprocess
 import sys
 
+import pytest
+
 # Modules that must only load for the commands that actually use them
 # (compile/config validation, shell completion), never from a bare
 # ``import esphome.__main__``.
@@ -218,6 +220,41 @@ def test_api_client_does_not_import_heavy_modules() -> None:
     )
 
 
+@pytest.mark.parametrize("component", ["binary_sensor", "sensor"])
+def test_entity_component_does_not_import_integrations(component: str) -> None:
+    """An entity component must not drag in its optional integrations.
+
+    The mqtt/web_server/zigbee schema fragments live in
+    ``esphome.core.entity_helpers``; the integration packages (and the
+    esp32/logger chains mqtt and zigbee pull in) must only load when the
+    user's config actually uses them.
+    """
+    allowed = (
+        "esphome.components",
+        f"esphome.components.{component}",
+        "esphome.components.const",
+    )
+    check = (
+        f"import sys; import esphome.components.{component}; "
+        f"leaked = [m for m in sys.modules "
+        f"if m.startswith('esphome.components') and m not in {allowed!r}]; "
+        "print(','.join(leaked))"
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", check],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    leaked = result.stdout.strip()
+    assert not leaked, (
+        f"esphome.components.{component} imports integration packages at "
+        f"top level: {leaked}. Keep the shared schema fragments in "
+        "esphome.core.entity_helpers and import the integration inside "
+        "to_code instead."
+    )
+
+
 def test_stacktrace_does_not_import_heavy_modules() -> None:
     """``esphome.stacktrace`` guards its own docstring's contract.