Merge branch 'bramstroker:master' into master

Author: xxddff

.github/workflows/release-drafter.yml

@@ -37,9 +37,5 @@ jobs:
         with:
           footer: ${{ steps.supporters.outputs.value }}
           config-name: release-drafter.yml
-
-          # Mark beta as prerelease and auto-increment beta.N
-          prerelease: ${{ github.ref_name == 'beta' }}
-          prerelease-identifier: beta
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

.gitignore

@@ -18,3 +18,5 @@ custom_components/test
 
 # Visual Studio settings
 /.vs
+
+.omx

AGENTS.md

@@ -75,6 +75,7 @@ Configured via `pyproject.toml`:
 - Test files: `test_*.py`, async functions: `async def test_*`
 - Fixtures in `conftest.py` files
 - Use `@pytest.mark.parametrize` for test variants
+- We strive to 100% test coverage in the project
 
 ## Pull Requests
 

custom_components/powercalc/common.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import math
 import re
 from typing import NamedTuple
 
@@ -8,7 +9,6 @@
 from homeassistant.core import HomeAssistant, split_entity_id
 import homeassistant.helpers.device_registry as dr
 import homeassistant.helpers.entity_registry as er
-from homeassistant.helpers.template import is_number
 import voluptuous as vol
 
 from .const import (
@@ -37,6 +37,15 @@ class SourceEntity(NamedTuple):
     device_entry: dr.DeviceEntry | None = None
 
 
+def is_number(value: str) -> bool:
+    """Return whether the value can be converted to a finite float."""
+    try:
+        fvalue = float(value)
+    except (TypeError, ValueError):
+        return False
+    return math.isfinite(fvalue)
+
+
 async def create_source_entity(entity_id: str, hass: HomeAssistant) -> SourceEntity:
     """Create object containing all information about the source entity."""
 
@@ -162,6 +171,6 @@ def validate_name_pattern(value: str) -> str:
 
 def validate_is_number(value: str) -> str:
     """Validate value is a number."""
-    if is_number(value):  # type: ignore[no-untyped-call]
+    if is_number(value):
         return value
     raise vol.Invalid("Value is not a number")

custom_components/powercalc/flow_helper/flows/library.py

@@ -21,6 +21,7 @@
     CONF_SELF_USAGE_INCLUDED,
     CONF_SUB_PROFILE,
     CONF_VARIABLES,
+    DUMMY_ENTITY_ID,
     LIBRARY_URL,
     CalculationStrategy,
 )
@@ -72,10 +73,12 @@ async def async_step_manufacturer(
         async def _create_schema() -> vol.Schema:
             """Create manufacturer schema."""
             library = await ProfileLibrary.factory(self.flow.hass)
-            device_types = DOMAIN_DEVICE_TYPE_MAPPING.get(self.flow.source_entity.domain, set()) if self.flow.source_entity else None
             manufacturers = [
                 selector.SelectOptionDict(value=manufacturer[0], label=manufacturer[1])
-                for manufacturer in await library.get_manufacturer_listing(device_types)
+                for manufacturer in await library.get_manufacturer_listing(
+                    self._get_library_device_types(),
+                    self._get_library_discovery_by(),
+                )
             ]
             return vol.Schema(
                 {
@@ -104,6 +107,11 @@ async def async_step_model(
     ) -> FlowResult:
         """Ask the user to select the model."""
 
+        def _build_model_label(model_id: str, model_name: str) -> str:
+            if not model_name or model_name == model_id:
+                return model_id
+            return f"{model_id} ({model_name})"
+
         async def _validate(user_input: dict[str, Any]) -> dict[str, str]:
             library = await ProfileLibrary.factory(self.flow.hass)
             profile = await library.get_profile(
@@ -123,8 +131,14 @@ async def _create_schema() -> vol.Schema:
             """Create model schema."""
             manufacturer = str(self.flow.sensor_config.get(CONF_MANUFACTURER))
             library = await ProfileLibrary.factory(self.flow.hass)
-            device_types = DOMAIN_DEVICE_TYPE_MAPPING.get(self.flow.source_entity.domain, set()) if self.flow.source_entity else None
-            models = [selector.SelectOptionDict(value=model, label=model) for model in await library.get_model_listing(manufacturer, device_types)]
+            models = [
+                selector.SelectOptionDict(value=model_id, label=_build_model_label(model_id, model_name))
+                for model_id, model_name in await library.get_model_listing(
+                    manufacturer,
+                    self._get_library_device_types(),
+                    self._get_library_discovery_by(),
+                )
+            ]
             model = self.flow.selected_profile.model if self.flow.selected_profile else self.flow.sensor_config.get(CONF_MODEL)
             return vol.Schema(
                 {
@@ -338,6 +352,22 @@ def _resolve_availability_entity(self) -> str | None:
                 return entity
         return None
 
+    def _get_library_device_types(self) -> set[DeviceType] | None:
+        """Determine which device types should be shown in the library selectors."""
+        if self._get_library_discovery_by() == DiscoveryBy.DEVICE:
+            return None
+
+        if self.flow.source_entity:
+            return DOMAIN_DEVICE_TYPE_MAPPING.get(self.flow.source_entity.domain, set())
+
+        return None  # pragma: no cover
+
+    def _get_library_discovery_by(self) -> DiscoveryBy | None:
+        """Determine whether listing should be filtered by discovery mode."""
+        if self.flow.source_entity and self.flow.source_entity.entity_id == DUMMY_ENTITY_ID:
+            return DiscoveryBy.DEVICE
+        return None
+
 
 class LibraryConfigFlow(LibraryFlow):
     def __init__(self, flow: PowercalcConfigFlow) -> None:

custom_components/powercalc/power_profile/library.py

@@ -22,7 +22,7 @@
 from .loader.local import LocalLoader
 from .loader.protocol import Loader
 from .loader.remote import RemoteLoader
-from .power_profile import DeviceType, PowerProfile
+from .power_profile import DeviceType, DiscoveryBy, PowerProfile
 
 LEGACY_CUSTOM_DATA_DIRECTORY = "powercalc-custom-models"
 CUSTOM_DATA_DIRECTORY = "powercalc/profiles"
@@ -33,7 +33,7 @@ def __init__(self, hass: HomeAssistant, loader: Loader) -> None:
         self._hass = hass
         self._loader = loader
         self._profiles: dict[str, list[PowerProfile]] = {}
-        self._manufacturer_models: dict[str, set[str]] = {}
+        self._manufacturer_models: dict[str, set[tuple[str, str]]] = {}
         self._manufacturer_device_types: dict[str, list] = {}
 
     async def initialize(self) -> None:
@@ -70,29 +70,39 @@ def create_loader(hass: HomeAssistant, skip_remote_loader: bool = False) -> Load
 
         return CompositeLoader(loaders)
 
-    async def get_manufacturer_listing(self, device_types: set[DeviceType] | None = None) -> list[tuple[str, str]]:
+    async def get_manufacturer_listing(
+        self,
+        device_types: set[DeviceType] | None = None,
+        discovery_by: DiscoveryBy | None = None,
+    ) -> list[tuple[str, str]]:
         """Get listing of available manufacturers."""
-        manufacturers = await self._loader.get_manufacturer_listing(device_types)
+        manufacturers = await self._loader.get_manufacturer_listing(device_types, discovery_by)
         return sorted(manufacturers)
 
-    async def get_model_listing(self, manufacturer: str, device_types: set[DeviceType] | None = None) -> list[str]:
-        """Get listing of available models for a given manufacturer."""
+    async def get_model_listing(
+        self,
+        manufacturer: str,
+        device_types: set[DeviceType] | None = None,
+        discovery_by: DiscoveryBy | None = None,
+    ) -> list[tuple[str, str]]:
+        """Get listing of available models and display names for a given manufacturer."""
 
         resolved_manufacturers = await self._loader.find_manufacturers(manufacturer)
         if not resolved_manufacturers:
             return []
-        all_models: list[str] = []
+
+        all_models: list[tuple[str, str]] = []
         for manufacturer in resolved_manufacturers:
-            cache_key = f"{manufacturer}/{device_types}"
+            cache_key = f"{manufacturer}/{device_types}/{discovery_by}"
             cached_models = self._manufacturer_models.get(cache_key)
             if cached_models:
-                all_models.extend(cached_models)
+                all_models.extend(sorted(cached_models))
                 continue
-            models = await self._loader.get_model_listing(manufacturer, device_types)
+            models = await self._loader.get_model_listing(manufacturer, device_types, discovery_by)
             self._manufacturer_models[cache_key] = models
-            all_models.extend(models)
+            all_models.extend(sorted(models))
 
-        return sorted(all_models)
+        return sorted(all_models, key=lambda model: model[0])
 
     async def get_profile(
         self,

custom_components/powercalc/power_profile/loader/composite.py

@@ -1,7 +1,7 @@
 import logging
 
 from custom_components.powercalc.power_profile.loader.protocol import Loader
-from custom_components.powercalc.power_profile.power_profile import DeviceType
+from custom_components.powercalc.power_profile.power_profile import DeviceType, DiscoveryBy
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -13,10 +13,14 @@ def __init__(self, loaders: list[Loader]) -> None:
     async def initialize(self) -> None:
         [await loader.initialize() for loader in self.loaders]  # type: ignore[func-returns-value]
 
-    async def get_manufacturer_listing(self, device_types: set[DeviceType] | None) -> set[tuple[str, str]]:
+    async def get_manufacturer_listing(
+        self,
+        device_types: set[DeviceType] | None,
+        discovery_by: DiscoveryBy | None = None,
+    ) -> set[tuple[str, str]]:
         """Get listing of available manufacturers."""
 
-        return {manufacturer for loader in self.loaders for manufacturer in await loader.get_manufacturer_listing(device_types)}
+        return {manufacturer for loader in self.loaders for manufacturer in await loader.get_manufacturer_listing(device_types, discovery_by)}
 
     async def find_manufacturers(self, search: str) -> set[str]:
         """Check if a manufacturer is available. Also must check aliases."""
@@ -30,10 +34,15 @@ async def find_manufacturers(self, search: str) -> set[str]:
 
         return found_manufacturers
 
-    async def get_model_listing(self, manufacturer: str, device_types: set[DeviceType] | None) -> set[str]:
-        """Get listing of available models for a given manufacturer."""
+    async def get_model_listing(
+        self,
+        manufacturer: str,
+        device_types: set[DeviceType] | None,
+        discovery_by: DiscoveryBy | None = None,
+    ) -> set[tuple[str, str]]:
+        """Get listing of available models and display names for a given manufacturer."""
 
-        return {model for loader in self.loaders for model in await loader.get_model_listing(manufacturer, device_types)}
+        return {model for loader in self.loaders for model in await loader.get_model_listing(manufacturer, device_types, discovery_by)}
 
     async def load_model(self, manufacturer: str, model: str) -> tuple[dict, str] | None:
         for loader in self.loaders:

custom_components/powercalc/power_profile/loader/local.py

@@ -8,7 +8,7 @@
 
 from custom_components.powercalc.power_profile.error import LibraryLoadingError
 from custom_components.powercalc.power_profile.loader.protocol import Loader
-from custom_components.powercalc.power_profile.power_profile import DeviceType, PowerProfile
+from custom_components.powercalc.power_profile.power_profile import DeviceType, DiscoveryBy, PowerProfile
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -25,14 +25,24 @@ async def initialize(self) -> None:
         if not self._is_custom_directory:
             await self._hass.async_add_executor_job(self._load_custom_library)
 
-    async def get_manufacturer_listing(self, device_types: set[DeviceType] | None) -> set[tuple[str, str]]:
+    async def get_manufacturer_listing(
+        self,
+        device_types: set[DeviceType] | None,
+        discovery_by: DiscoveryBy | None = None,
+    ) -> set[tuple[str, str]]:
         """Get listing of all available manufacturers or filtered by model device_type."""
         if device_types is None:
-            return {(manufacturer, manufacturer) for manufacturer in self._manufacturer_model_listing}
+            if discovery_by is None:
+                return {(manufacturer, manufacturer) for manufacturer in self._manufacturer_model_listing}
+            return {
+                (manufacturer, manufacturer)
+                for manufacturer, profiles in self._manufacturer_model_listing.items()
+                if any(profile.discovery_by == discovery_by for profile in profiles.values())
+            }
 
         manufacturers: set[tuple[str, str]] = set()
         for manufacturer in self._manufacturer_model_listing:
-            models = await self.get_model_listing(manufacturer, device_types)
+            models = await self.get_model_listing(manufacturer, device_types, discovery_by)
             if not models:
                 continue
             manufacturers.add((manufacturer, manufacturer))
@@ -49,25 +59,32 @@ async def find_manufacturers(self, search: str) -> set[str]:
 
         return set()
 
-    async def get_model_listing(self, manufacturer: str, device_types: set[DeviceType] | None) -> set[str]:
+    async def get_model_listing(
+        self,
+        manufacturer: str,
+        device_types: set[DeviceType] | None,
+        discovery_by: DiscoveryBy | None = None,
+    ) -> set[tuple[str, str]]:
         """Get listing of available models for a given manufacturer.
 
         param manufacturer: manufacturer always handled in lower case
         param device_type:  models of the manufacturer will be filtered by DeviceType, models
                             without assigned device_type will be handled as DeviceType.LIGHT.
                             None will return all models of a manufacturer.
-        returns:            Set[str] of models
+        returns:            Set[tuple[str, str]] of (model_id, model_name)
         """
 
-        found_models: set[str] = set()
+        found_models: set[tuple[str, str]] = set()
         models = self._manufacturer_model_listing.get(manufacturer.lower())
         if not models:
             return found_models
 
         for profile in models.values():
             if device_types and profile.device_type not in device_types:
                 continue
-            found_models.add(profile.model)
+            if discovery_by and profile.discovery_by != discovery_by:
+                continue
+            found_models.add((profile.model, profile.name or profile.model))
 
         return found_models
 

custom_components/powercalc/power_profile/loader/protocol.py

@@ -1,20 +1,29 @@
 from typing import Protocol
 
-from custom_components.powercalc.power_profile.power_profile import DeviceType
+from custom_components.powercalc.power_profile.power_profile import DeviceType, DiscoveryBy
 
 
 class Loader(Protocol):
     async def initialize(self) -> None:
         """Initialize the loader."""
 
-    async def get_manufacturer_listing(self, device_types: set[DeviceType] | None) -> set[tuple[str, str]]:
+    async def get_manufacturer_listing(
+        self,
+        device_types: set[DeviceType] | None,
+        discovery_by: DiscoveryBy | None = None,
+    ) -> set[tuple[str, str]]:
         """Get listing of possible manufacturers."""
 
     async def find_manufacturers(self, search: str) -> set[str]:
         """Check if a manufacturer is available. Also must check aliases."""
 
-    async def get_model_listing(self, manufacturer: str, device_types: set[DeviceType] | None) -> set[str]:
-        """Get listing of available models for a given manufacturer."""
+    async def get_model_listing(
+        self,
+        manufacturer: str,
+        device_types: set[DeviceType] | None,
+        discovery_by: DiscoveryBy | None = None,
+    ) -> set[tuple[str, str]]:
+        """Get listing of available models and display names for a given manufacturer."""
 
     async def load_model(self, manufacturer: str, model: str) -> tuple[dict, str] | None:
         """Load and optionally download a model profile."""