Keep track of top level components (#115586)

* Keep track of top level components

Currently we have to do a set comp for icons, translations,
and integration platforms every time to split the top level
components from the platforms. Keep track of the top level
components in a seperate set so avoid having to do the setcomp
every time.

* remove impossible paths

* remove unused code

* preen

* preen

* fix

* coverage and fixes

* Update homeassistant/core.py

* Update homeassistant/core.py

* Update tests/test_core.py

homeassistant/core.py

@@ -2670,6 +2670,41 @@ class ServiceRegistry:
         return await self._hass.async_add_executor_job(target, service_call)
 
 
+class _ComponentSet(set[str]):
+    """Set of loaded components.
+
+    This set contains both top level components and platforms.
+
+    Examples:
+    `light`, `switch`, `hue`, `mjpeg.camera`, `universal.media_player`,
+    `homeassistant.scene`
+
+    The top level components set only contains the top level components.
+
+    """
+
+    def __init__(self, top_level_components: set[str]) -> None:
+        """Initialize the component set."""
+        self._top_level_components = top_level_components
+
+    def add(self, component: str) -> None:
+        """Add a component to the store."""
+        if "." not in component:
+            self._top_level_components.add(component)
+        return super().add(component)
+
+    def remove(self, component: str) -> None:
+        """Remove a component from the store."""
+        if "." in component:
+            raise ValueError("_ComponentSet does not support removing sub-components")
+        self._top_level_components.remove(component)
+        return super().remove(component)
+
+    def discard(self, component: str) -> None:
+        """Remove a component from the store."""
+        raise NotImplementedError("_ComponentSet does not support discard, use remove")
+
+
 class Config:
     """Configuration settings for Home Assistant."""
 
@@ -2702,8 +2737,13 @@ class Config:
         # List of packages to skip when installing requirements on startup
         self.skip_pip_packages: list[str] = []
 
-        # List of loaded components
+        # Set of loaded top level components
-        self.components: set[str] = set()
+        # This set is updated by _ComponentSet
+        # and should not be modified directly
+        self.top_level_components: set[str] = set()
+
+        # Set of loaded components
+        self.components: _ComponentSet = _ComponentSet(self.top_level_components)
 
         # API (HTTP) server configuration
         self.api: ApiConfig | None = None

homeassistant/helpers/icon.py

@@ -6,6 +6,7 @@ import asyncio
 from collections.abc import Iterable
 from functools import lru_cache
 import logging
+import pathlib
 from typing import Any
 
 from homeassistant.core import HomeAssistant, callback
@@ -20,23 +21,17 @@ _LOGGER = logging.getLogger(__name__)
 
 
 @callback
-def _component_icons_path(component: str, integration: Integration) -> str | None:
+def _component_icons_path(integration: Integration) -> pathlib.Path:
     """Return the icons json file location for a component.
 
     Ex: components/hue/icons.json
-    If component is just a single file, will return None.
     """
-    domain = component.rpartition(".")[-1]
+    return integration.file_path / "icons.json"
-
-    # If it's a component that is just one file, we don't support icons
-    # Example custom_components/my_component.py
-    if integration.file_path.name != domain:
-        return None
-
-    return str(integration.file_path / "icons.json")
 
 
-def _load_icons_files(icons_files: dict[str, str]) -> dict[str, dict[str, Any]]:
+def _load_icons_files(
+    icons_files: dict[str, pathlib.Path],
+) -> dict[str, dict[str, Any]]:
     """Load and parse icons.json files."""
     return {
         component: load_json_object(icons_file)
@@ -53,19 +48,15 @@ async def _async_get_component_icons(
     icons: dict[str, Any] = {}
 
     # Determine files to load
-    files_to_load = {}
+    files_to_load = {
-    for loaded in components:
+        comp: _component_icons_path(integrations[comp]) for comp in components
-        domain = loaded.rpartition(".")[-1]
+    }
-        if (path := _component_icons_path(loaded, integrations[domain])) is None:
-            icons[loaded] = {}
-        else:
-            files_to_load[loaded] = path
 
     # Load files
-    if files_to_load and (
+    if files_to_load:
-        load_icons_job := hass.async_add_executor_job(_load_icons_files, files_to_load)
+        icons.update(
-    ):
+            await hass.async_add_executor_job(_load_icons_files, files_to_load)
-        icons |= await load_icons_job
+        )
 
     return icons
 
@@ -108,8 +99,7 @@ class _IconsCache:
         _LOGGER.debug("Cache miss for: %s", components)
 
         integrations: dict[str, Integration] = {}
-        domains = {loaded.rpartition(".")[-1] for loaded in components}
+        ints_or_excs = await async_get_integrations(self._hass, components)
-        ints_or_excs = await async_get_integrations(self._hass, domains)
         for domain, int_or_exc in ints_or_excs.items():
             if isinstance(int_or_exc, Exception):
                 raise int_or_exc
@@ -127,11 +117,9 @@ class _IconsCache:
         icons: dict[str, dict[str, Any]],
     ) -> None:
         """Extract resources into the cache."""
-        categories: set[str] = set()
+        categories = {
-
+            category for component in icons.values() for category in component
-        for resource in icons.values():
+        }
-            categories.update(resource)
-
         for category in categories:
             self._cache.setdefault(category, {}).update(
                 build_resources(icons, components, category)
@@ -151,9 +139,7 @@ async def async_get_icons(
     if integrations:
         components = set(integrations)
     else:
-        components = {
+        components = hass.config.top_level_components
-            component for component in hass.config.components if "." not in component
-        }
 
     if ICON_CACHE in hass.data:
         cache: _IconsCache = hass.data[ICON_CACHE]

homeassistant/helpers/integration_platform.py

@@ -174,7 +174,7 @@ async def async_process_integration_platforms(
         integration_platforms = hass.data[DATA_INTEGRATION_PLATFORMS]
 
     async_register_preload_platform(hass, platform_name)
-    top_level_components = {comp for comp in hass.config.components if "." not in comp}
+    top_level_components = hass.config.top_level_components.copy()
     process_job = HassJob(
         catch_log_exception(
             process_platform,

homeassistant/helpers/translation.py

@@ -212,8 +212,7 @@ class _TranslationCache:
         languages = [LOCALE_EN] if language == LOCALE_EN else [LOCALE_EN, language]
 
         integrations: dict[str, Integration] = {}
-        domains = {loaded.partition(".")[0] for loaded in components}
+        ints_or_excs = await async_get_integrations(self.hass, components)
-        ints_or_excs = await async_get_integrations(self.hass, domains)
         for domain, int_or_exc in ints_or_excs.items():
             if isinstance(int_or_exc, Exception):
                 _LOGGER.warning(
@@ -345,7 +344,7 @@ async def async_get_translations(
     elif integrations is not None:
         components = set(integrations)
     else:
-        components = {comp for comp in hass.config.components if "." not in comp}
+        components = hass.config.top_level_components
 
     return await _async_get_translations_cache(hass).async_fetch(
         language, category, components
@@ -364,11 +363,7 @@ def async_get_cached_translations(
     If integration is specified, return translations for it.
     Otherwise, default to all loaded integrations.
     """
-    if integration is not None:
+    components = {integration} if integration else hass.config.top_level_components
-        components = {integration}
-    else:
-        components = {comp for comp in hass.config.components if "." not in comp}
-
     return _async_get_translations_cache(hass).get_cached(
         language, category, components
     )

tests/helpers/test_icon.py

@@ -106,8 +106,8 @@ async def test_get_icons(hass: HomeAssistant) -> None:
     # Ensure icons file for platform isn't loaded, as that isn't supported
     icons = await icon.async_get_icons(hass, "entity")
     assert icons == {}
-    icons = await icon.async_get_icons(hass, "entity", ["test.switch"])
+    with pytest.raises(ValueError, match="test.switch"):
-    assert icons == {}
+        await icon.async_get_icons(hass, "entity", ["test.switch"])
 
     # Load up an custom integration
     hass.config.components.add("test_package")

tests/test_core.py

@@ -3411,3 +3411,20 @@ async def test_async_listen_with_run_immediately_deprecated(
         f"Detected code that calls `{method}` with run_immediately, which is "
         "deprecated and will be removed in Home Assistant 2025.5."
     ) in caplog.text
+
+
+async def test_top_level_components(hass: HomeAssistant) -> None:
+    """Test top level components are updated when components change."""
+    hass.config.components.add("homeassistant")
+    assert hass.config.components == {"homeassistant"}
+    assert hass.config.top_level_components == {"homeassistant"}
+    hass.config.components.add("homeassistant.scene")
+    assert hass.config.components == {"homeassistant", "homeassistant.scene"}
+    assert hass.config.top_level_components == {"homeassistant"}
+    hass.config.components.remove("homeassistant")
+    assert hass.config.components == {"homeassistant.scene"}
+    assert hass.config.top_level_components == set()
+    with pytest.raises(ValueError):
+        hass.config.components.remove("homeassistant.scene")
+    with pytest.raises(NotImplementedError):
+        hass.config.components.discard("homeassistant")