esphome/esphome/loader.py

import logging
import typing
from typing import Callable, List, Optional, Dict, Any, ContextManager
from types import ModuleType
import importlib
import importlib.util
import importlib.resources
import importlib.abc
import sys
from pathlib import Path

from esphome.const import ESP_PLATFORMS, SOURCE_FILE_EXTENSIONS
import esphome.core.config
from esphome.core import CORE
from esphome.types import ConfigType

_LOGGER = logging.getLogger(__name__)


class SourceFile:
    def __init__(
        self,
        package: importlib.resources.Package,
        resource: importlib.resources.Resource,
    ) -> None:
        self._package = package
        self._resource = resource

    def open_binary(self) -> typing.BinaryIO:
        return importlib.resources.open_binary(self._package, self._resource)

    def path(self) -> ContextManager[Path]:
        return importlib.resources.path(self._package, self._resource)


class ComponentManifest:
    def __init__(self, module: ModuleType):
        self.module = module

    @property
    def package(self) -> str:
        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 config_schema(self) -> Optional[Any]:
        return getattr(self.module, "CONFIG_SCHEMA", None)

    @property
    def multi_conf(self) -> bool:
        return getattr(self.module, "MULTI_CONF", False)

    @property
    def to_code(self) -> Optional[Callable[[Any], None]]:
        return getattr(self.module, "to_code", None)

    @property
    def esp_platforms(self) -> List[str]:
        return getattr(self.module, "ESP_PLATFORMS", ESP_PLATFORMS)

    @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]:
        return getattr(self.module, "AUTO_LOAD", [])

    @property
    def codeowners(self) -> List[str]:
        return getattr(self.module, "CODEOWNERS", [])

    @property
    def final_validate_schema(self) -> Optional[Callable[[ConfigType], 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 source_files(self) -> Dict[Path, SourceFile]:
        ret = {}
        for resource in importlib.resources.contents(self.package):
            if Path(resource).suffix not in SOURCE_FILE_EXTENSIONS:
                continue
            if not importlib.resources.is_resource(self.package, resource):
                # Not a resource = this is a directory (yeah this is confusing)
                continue
            # Always use / for C++ include names
            target_path = Path(*self.package.split(".")) / resource
            ret[target_path] = SourceFile(self.package, resource)
        return ret


class ComponentMetaFinder(importlib.abc.MetaPathFinder):
    def __init__(
        self, components_path: Path, allowed_components: Optional[List[str]] = 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: Optional[List[str]], 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: Optional[List[str]] = None
):
    sys.meta_path.insert(0, ComponentMetaFinder(components_path, allowed_components))


def install_custom_components_meta_finder():
    custom_components_dir = (Path(CORE.config_dir) / "custom_components").resolve()
    install_meta_finder(custom_components_dir)


def _lookup_module(domain):
    if domain in _COMPONENT_CACHE:
        return _COMPONENT_CACHE[domain]

    try:
        module = importlib.import_module(f"esphome.components.{domain}")
    except ImportError as e:
        if "No module named" not in str(e):
            _LOGGER.error("Unable to import component %s:", domain, exc_info=True)
        return None
    except Exception:  # pylint: disable=broad-except
        _LOGGER.error("Unable to load component %s:", domain, exc_info=True)
        return None
    else:
        manif = ComponentManifest(module)
        _COMPONENT_CACHE[domain] = manif
        return manif


def get_component(domain):
    assert "." not in domain
    return _lookup_module(domain)


def get_platform(domain, platform):
    full = f"{platform}.{domain}"
    return _lookup_module(full)


_COMPONENT_CACHE = {}
CORE_COMPONENTS_PATH = (Path(__file__).parent / "components").resolve()
_COMPONENT_CACHE["esphome"] = ComponentManifest(esphome.core.config)