mirror of
https://github.com/esphome/esphome.git
synced 2026-06-24 13:09:12 +00:00
277 lines
9.5 KiB
Python
277 lines
9.5 KiB
Python
from collections.abc import Callable
|
|
from contextlib import AbstractContextManager
|
|
from dataclasses import dataclass
|
|
import importlib
|
|
import importlib.abc
|
|
import importlib.resources
|
|
import importlib.util
|
|
import logging
|
|
from pathlib import Path
|
|
import sys
|
|
from types import ModuleType
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from esphome.const import SOURCE_FILE_EXTENSIONS
|
|
from esphome.core import CORE
|
|
from esphome.types import ConfigType
|
|
|
|
if TYPE_CHECKING:
|
|
from esphome.cpp_generator import MockObjClass
|
|
|
|
# `esphome.core.config` is imported lazily in `_lookup_module` when the
|
|
# "esphome" pseudo-component is first resolved. It pulls in
|
|
# `esphome.automation` and `esphome.config_validation`, which together
|
|
# dominate `esphome.__main__` startup cost when loaded eagerly.
|
|
# `esphome.cpp_generator` is similarly avoided at module scope; it pulls
|
|
# in `esphome.yaml_util` and is only needed for the `MockObjClass` type
|
|
# annotation, which is resolved lazily via `TYPE_CHECKING`.
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass(frozen=True, order=True)
|
|
class FileResource:
|
|
package: str
|
|
resource: str
|
|
|
|
def path(self) -> AbstractContextManager[Path]:
|
|
return importlib.resources.as_file(
|
|
importlib.resources.files(self.package) / self.resource
|
|
)
|
|
|
|
|
|
class ComponentManifest:
|
|
def __init__(self, module: ModuleType, recursive_sources: bool = False):
|
|
self.module = module
|
|
self.recursive_sources = recursive_sources
|
|
|
|
@property
|
|
def package(self) -> str:
|
|
"""Return the package name the module is contained in.
|
|
|
|
Examples:
|
|
- esphome/components/gpio/__init__.py -> esphome.components.gpio
|
|
- esphome/components/gpio/switch/__init__.py -> esphome.components.gpio.switch
|
|
- esphome/components/a4988/stepper.py -> esphome.components.a4988
|
|
"""
|
|
return self.module.__package__
|
|
|
|
@property
|
|
def is_platform(self) -> bool:
|
|
return len(self.module.__name__.split(".")) == 4
|
|
|
|
@property
|
|
def is_platform_component(self) -> bool:
|
|
return getattr(self.module, "IS_PLATFORM_COMPONENT", False)
|
|
|
|
@property
|
|
def is_target_platform(self) -> bool:
|
|
return getattr(self.module, "IS_TARGET_PLATFORM", False)
|
|
|
|
@property
|
|
def config_schema(self) -> Any | None:
|
|
return getattr(self.module, "CONFIG_SCHEMA", None)
|
|
|
|
@property
|
|
def multi_conf(self) -> bool:
|
|
return getattr(self.module, "MULTI_CONF", False)
|
|
|
|
@property
|
|
def multi_conf_no_default(self) -> bool:
|
|
return getattr(self.module, "MULTI_CONF_NO_DEFAULT", False)
|
|
|
|
@property
|
|
def to_code(self) -> Callable[[Any], None] | None:
|
|
return getattr(self.module, "to_code", None)
|
|
|
|
@property
|
|
def dependencies(self) -> list[str]:
|
|
return getattr(self.module, "DEPENDENCIES", [])
|
|
|
|
@property
|
|
def conflicts_with(self) -> list[str]:
|
|
return getattr(self.module, "CONFLICTS_WITH", [])
|
|
|
|
@property
|
|
def auto_load(
|
|
self,
|
|
) -> list[str] | Callable[[], list[str]] | Callable[[ConfigType], list[str]]:
|
|
return getattr(self.module, "AUTO_LOAD", [])
|
|
|
|
@property
|
|
def codeowners(self) -> list[str]:
|
|
return getattr(self.module, "CODEOWNERS", [])
|
|
|
|
@property
|
|
def instance_type(self) -> "MockObjClass | None":
|
|
return getattr(self.module, "INSTANCE_TYPE", None)
|
|
|
|
@property
|
|
def final_validate_schema(self) -> Callable[[ConfigType], None] | None:
|
|
"""Components can declare a `FINAL_VALIDATE_SCHEMA` cv.Schema that gets called
|
|
after the main validation. In that function checks across components can be made.
|
|
|
|
Note that the function can't mutate the configuration - no changes are saved
|
|
"""
|
|
return getattr(self.module, "FINAL_VALIDATE_SCHEMA", None)
|
|
|
|
@property
|
|
def resources(self) -> list[FileResource]:
|
|
"""Return a list of all file resources defined in the package of this component.
|
|
|
|
By default only files directly in the package directory are returned. Manifests
|
|
constructed with ``recursive_sources=True`` also descend into non-subpackage
|
|
subdirectories (subdirectories without an ``__init__.py``), so core code can
|
|
live under ``esphome/core/<group>/`` without every component paying the cost.
|
|
"""
|
|
ret: list[FileResource] = []
|
|
|
|
# Get filter function for source files
|
|
filter_source_files_func = getattr(self.module, "FILTER_SOURCE_FILES", None)
|
|
|
|
# Get list of files to exclude
|
|
excluded_files = (
|
|
set(filter_source_files_func()) if filter_source_files_func else set()
|
|
)
|
|
|
|
root = importlib.resources.files(self.package)
|
|
|
|
for child in root.iterdir():
|
|
name = child.name
|
|
if child.is_file():
|
|
if Path(name).suffix not in SOURCE_FILE_EXTENSIONS:
|
|
continue
|
|
if name in excluded_files:
|
|
continue
|
|
ret.append(FileResource(self.package, name))
|
|
elif self.recursive_sources and child.is_dir() and name != "__pycache__":
|
|
# Skip Python subpackages — they load as their own components.
|
|
if child.joinpath("__init__.py").is_file():
|
|
continue
|
|
for sub in child.iterdir():
|
|
if not sub.is_file():
|
|
continue
|
|
if Path(sub.name).suffix not in SOURCE_FILE_EXTENSIONS:
|
|
continue
|
|
resource = f"{name}/{sub.name}"
|
|
if resource in excluded_files:
|
|
continue
|
|
ret.append(FileResource(self.package, resource))
|
|
|
|
return ret
|
|
|
|
|
|
class ComponentMetaFinder(importlib.abc.MetaPathFinder):
|
|
def __init__(
|
|
self, components_path: Path, allowed_components: list[str] | None = None
|
|
) -> None:
|
|
self._allowed_components = allowed_components
|
|
self._finders = []
|
|
for hook in sys.path_hooks:
|
|
try:
|
|
finder = hook(str(components_path))
|
|
except ImportError:
|
|
continue
|
|
self._finders.append(finder)
|
|
|
|
def find_spec(self, fullname: str, path: list[str] | None, target=None):
|
|
if not fullname.startswith("esphome.components."):
|
|
return None
|
|
parts = fullname.split(".")
|
|
if len(parts) != 3:
|
|
# only handle direct components, not platforms
|
|
# platforms are handled automatically when parent is imported
|
|
return None
|
|
component = parts[2]
|
|
if (
|
|
self._allowed_components is not None
|
|
and component not in self._allowed_components
|
|
):
|
|
return None
|
|
|
|
for finder in self._finders:
|
|
spec = finder.find_spec(fullname, target=target)
|
|
if spec is not None:
|
|
return spec
|
|
return None
|
|
|
|
|
|
def clear_component_meta_finders():
|
|
sys.meta_path = [x for x in sys.meta_path if not isinstance(x, ComponentMetaFinder)]
|
|
|
|
|
|
def install_meta_finder(
|
|
components_path: Path, allowed_components: list[str] | None = None
|
|
):
|
|
sys.meta_path.insert(0, ComponentMetaFinder(components_path, allowed_components))
|
|
|
|
|
|
def install_custom_components_meta_finder():
|
|
# Remove before 2026.6.0
|
|
custom_components_dir = (Path(CORE.config_dir) / "custom_components").resolve()
|
|
if custom_components_dir.is_dir() and any(custom_components_dir.iterdir()):
|
|
_LOGGER.warning(
|
|
"The 'custom_components' folder is deprecated and will be removed in 2026.6.0. "
|
|
"Please use 'external_components' instead. "
|
|
"See https://esphome.io/components/external_components.html for more information."
|
|
)
|
|
install_meta_finder(custom_components_dir)
|
|
|
|
|
|
def _lookup_module(domain: str, exception: bool) -> ComponentManifest | None:
|
|
if domain in _COMPONENT_CACHE:
|
|
return _COMPONENT_CACHE[domain]
|
|
|
|
if domain == "esphome":
|
|
import esphome.core.config
|
|
|
|
manif = ComponentManifest(esphome.core.config, recursive_sources=True)
|
|
_COMPONENT_CACHE[domain] = manif
|
|
return manif
|
|
|
|
try:
|
|
module = importlib.import_module(f"esphome.components.{domain}")
|
|
except ImportError as e:
|
|
if exception:
|
|
raise
|
|
if "No module named" in str(e):
|
|
_LOGGER.info(
|
|
"Unable to import component %s: %s", domain, str(e), exc_info=False
|
|
)
|
|
else:
|
|
_LOGGER.error("Unable to import component %s:", domain, exc_info=True)
|
|
return None
|
|
except Exception: # pylint: disable=broad-except
|
|
if exception:
|
|
raise
|
|
_LOGGER.error("Unable to load component %s:", domain, exc_info=True)
|
|
return None
|
|
|
|
manif = ComponentManifest(module)
|
|
_COMPONENT_CACHE[domain] = manif
|
|
return manif
|
|
|
|
|
|
def get_component(domain: str, exception: bool = False) -> ComponentManifest | None:
|
|
assert "." not in domain
|
|
return _lookup_module(domain, exception)
|
|
|
|
|
|
def get_platform(domain: str, platform: str) -> ComponentManifest | None:
|
|
full = f"{platform}.{domain}"
|
|
return _lookup_module(full, False)
|
|
|
|
|
|
_COMPONENT_CACHE: dict[str, ComponentManifest] = {}
|
|
CORE_COMPONENTS_PATH = (Path(__file__).parent / "components").resolve()
|
|
|
|
|
|
def _replace_component_manifest(domain: str, manifest: ComponentManifest) -> None:
|
|
"""Replace the cached manifest for a component.
|
|
|
|
This is an intentionally-supported hook for the C++ test infrastructure
|
|
to install ``ComponentManifestOverride`` wrappers. Normal application
|
|
code should never call this.
|
|
"""
|
|
_COMPONENT_CACHE[domain] = manifest
|