148 files

custom_components/powercalc/config_flow.py

@@ -280,7 +280,7 @@ class PowercalcCommonFlow(ABC, ConfigEntryBaseFlow):
             return user_input
 
         validated_input = form_step.validate_user_input(user_input)
-        return await validated_input if isawaitable(validated_input) else validated_input  # ty: ignore[invalid-return-type]
+        return await validated_input if isawaitable(validated_input) else validated_input
 
     def _store_form_step_input(
         self,

custom_components/powercalc/flow_helper/flows/group.py

@@ -206,7 +206,8 @@ def create_schema_group_custom(
     member_sensors = [
         selector.SelectOptionDict(value=config_entry.entry_id, label=config_entry.title)
         for config_entry in hass.config_entries.async_entries(DOMAIN)
-        if config_entry.data.get(CONF_SENSOR_TYPE) in [SensorType.VIRTUAL_POWER, SensorType.REAL_POWER]
+        if config_entry.data.get(CONF_SENSOR_TYPE)
+        in [SensorType.VIRTUAL_POWER, SensorType.REAL_POWER, SensorType.DAILY_ENERGY]
         and config_entry.unique_id is not None
         and config_entry.title is not None
     ]

custom_components/powercalc/manifest.json

@@ -22,5 +22,5 @@
     "requirements": [
         "numpy>=1.21.1"
     ],
-    "version": "v1.25.4"
+    "version": "v1.25.5"
 }

custom_components/powercalc/power_profile/library.py

@@ -20,6 +20,7 @@ from custom_components.powercalc.helpers import (
 from .error import LibraryError
 from .loader.composite import CompositeLoader
 from .loader.local import LocalLoader
+from .loader.profile_cache import CACHE_DIRECTORY
 from .loader.protocol import Loader, ModelMetadata
 from .loader.remote import RemoteLoader
 from .power_profile import DeviceType, DiscoveryBy, PowerProfile
@@ -33,6 +34,8 @@ def load_sub_profile_data(base_dir: str) -> list[tuple[str, dict[str, Any]]]:
     sub_dirs = next(os.walk(base_dir))[1]
     result = []
     for sub_dir in sub_dirs:
+        if sub_dir == CACHE_DIRECTORY:
+            continue
         json_path = os.path.join(base_dir, sub_dir, "model.json")
         if os.path.isfile(json_path):
             with open(json_path, encoding="utf-8") as f:

custom_components/powercalc/power_profile/loader/remote.py

@@ -7,7 +7,6 @@ import logging
 import os
 from pathlib import Path
 import shutil
-import tempfile
 from typing import Any, NotRequired, TypedDict, cast
 from urllib.parse import SplitResult, unquote, urlsplit
 
@@ -27,6 +26,13 @@ from custom_components.powercalc.const import (
 )
 from custom_components.powercalc.helpers import async_cache, clear_async_cache
 from custom_components.powercalc.power_profile.error import LibraryLoadingError, ProfileDownloadError
+from custom_components.powercalc.power_profile.loader.profile_cache import (
+    InstalledProfile,
+    create_staging_directory,
+    install_profile,
+    read_installed_profile,
+    save_resource,
+)
 from custom_components.powercalc.power_profile.loader.protocol import Loader, ModelMetadata
 from custom_components.powercalc.power_profile.power_profile import DeviceType, DiscoveryBy
 
@@ -36,7 +42,6 @@ ENDPOINT_LIBRARY = f"{API_URL}/library"
 ENDPOINT_DOWNLOAD = f"{API_URL}/download"
 
 TIMEOUT_SECONDS = 30
-MODEL_JSON_RETRY_LIMIT = 2
 
 ALLOWED_RESOURCE_HOSTS = frozenset({"github.com", "raw.githubusercontent.com"})
 # Profile resources are only ever served from the profile library of the Powercalc repository.
@@ -183,49 +188,10 @@ def _validate_resources(resources: object, storage_path: str) -> list[tuple[str,
     ]
 
 
-def _sync_directory(directory: Path) -> None:
-    """Persist a directory entry, so a completed rename survives an unclean shutdown."""
-    try:
-        directory_descriptor = os.open(directory, os.O_RDONLY)
-    except OSError:  # pragma: no cover - directories cannot be opened on all platforms
-        return
-    try:
-        os.fsync(directory_descriptor)
-    except OSError:  # pragma: no cover - directory fsync is not supported on all platforms
-        pass
-    finally:
-        os.close(directory_descriptor)
-
-
-def _save_resource(data: bytes, path: Path) -> None:
-    """Atomically save a downloaded resource to the local profile storage directory.
-
-    The contents are flushed to disk before the rename, and the directory entry is flushed
-    after it. Without both, a power loss shortly after an update can leave the new file name
-    pointing at unwritten data, which is how a cached profile ends up as invalid JSON.
-    """
-    os.makedirs(path.parent, exist_ok=True)
-    file_descriptor, temporary_name = tempfile.mkstemp(
-        dir=path.parent,
-        prefix=f".{path.name}.",
-        suffix=".tmp",
-    )
-    temporary_path = Path(temporary_name)
-    try:
-        with os.fdopen(file_descriptor, "wb") as file_handle:
-            file_handle.write(data)
-            file_handle.flush()
-            os.fsync(file_handle.fileno())
-        os.replace(temporary_path, path)
-        _sync_directory(path.parent)
-    finally:
-        temporary_path.unlink(missing_ok=True)
-
-
 def _save_resources(resources: list[tuple[bytes, Path]]) -> None:
     """Save all downloaded resources after every response has completed successfully."""
     for data, path in resources:
-        _save_resource(data, path)
+        save_resource(data, path)
 
 
 class LibraryModel(TypedDict):
@@ -257,6 +223,9 @@ class RemoteLoader(Loader):
         self.model_lookup: dict[str, dict[str, list[LibraryModel]]] = {}
         self.manufacturer_lookup: dict[str, set[str]] = {}
         self.profile_hashes: dict[str, str] = {}
+        self.installed_profiles: dict[str, InstalledProfile] = {}
+        self.powercalc_version = AwesomeVersion("0.0.0")
+        self._fallback_models: set[str] = set()
         self._model_load_locks: dict[tuple[str, str], asyncio.Lock] = {}
 
     async def initialize(self, prefer_cached: bool = False) -> None:
@@ -267,12 +236,14 @@ class RemoteLoader(Loader):
         """
 
         integration = await async_get_integration(self.hass, DOMAIN)
-        powercalc_version = AwesomeVersion(str(integration.version))
+        self.powercalc_version = AwesomeVersion(str(integration.version))
 
         self._clear_caches()
         self.library_contents = _validate_library_contents(await self.load_library_json(prefer_cached))
         self.profile_hashes = await self.hass.async_add_executor_job(self._load_profile_hashes)
 
+        self.installed_profiles = await self.hass.async_add_executor_job(self._load_installed_profiles)
+        self._fallback_models.clear()
         self.model_infos.clear()
         self.model_lookup.clear()
         self.manufacturer_models.clear()
@@ -281,7 +252,32 @@ class RemoteLoader(Loader):
         manufacturers: list[LibraryManufacturer] = self.library_contents.get("manufacturers", [])
 
         for manufacturer in manufacturers:
-            self._index_manufacturer(manufacturer, powercalc_version)
+            self._index_manufacturer(manufacturer, self.powercalc_version)
+
+    def _load_installed_profiles(self) -> dict[str, InstalledProfile]:
+        """Read cached revisions before applying remote compatibility filters."""
+        models = {
+            (manufacturer["dir_name"], model["id"])
+            for manufacturer in self.library_contents["manufacturers"]
+            for model in manufacturer.get("models", [])
+        }
+        storage_root = Path(self.hass.config.path(STORAGE_DIR, BUILT_IN_LIBRARY_DIR))
+        installed = {}
+        # Most library models are never downloaded. Only inspect directories present on this installation.
+        for directory in storage_root.glob("*/*"):
+            manufacturer, model = directory.parent.name, directory.name
+            if (manufacturer, model) not in models:
+                continue
+            key = f"{manufacturer}/{model}"
+            profile = read_installed_profile(
+                Path(self.get_storage_path(manufacturer, model)),
+                model,
+                self.profile_hashes.get(key),
+                self.powercalc_version,
+            )
+            if profile is not None:
+                installed[key] = profile
+        return installed
 
     def get_discovery_low_priority_domains(self) -> set[str]:
         """Get the low priority discovery integration domains declared by library metadata."""
@@ -303,10 +299,16 @@ class RemoteLoader(Loader):
 
         for model in models:
             model_id = str(model.get("id"))
-            self.model_infos[f"{manufacturer_name}/{model_id}"] = model
-
+            key = f"{manufacturer_name}/{model_id}"
+            self.model_infos[key] = model
             if self._is_unsupported_version(manufacturer_name, model_id, model, powercalc_version):
-                continue
+                installed = self.installed_profiles.get(key)
+                if installed is None:
+                    continue
+                model = cast(LibraryModel, installed.metadata)
+                self.model_infos[key] = model
+                self._fallback_models.add(key)
+                _LOGGER.debug("Using installed compatible profile for %s", key)
 
             kept_models.append(model)
             self._add_model_to_lookup(lookup, model, model_id.lower())
@@ -430,7 +432,7 @@ class RemoteLoader(Loader):
             raise ProfileDownloadError(f"Failed to download library.json: {err}") from err
 
         library_contents = _decode_library_json(data, "Remote library")
-        await self.hass.async_add_executor_job(_save_resource, data, Path(local_path))
+        await self.hass.async_add_executor_job(save_resource, data, Path(local_path))
 
         return library_contents
 
@@ -545,54 +547,37 @@ class RemoteLoader(Loader):
         manufacturer: str,
         model: str,
         force_update: bool = False,
-        retry_count: int = 0,
     ) -> tuple[dict[str, Any], str] | None:
         """Load a model, downloading it if necessary, with retry logic."""
         lock = self._model_load_locks.setdefault((manufacturer, model), asyncio.Lock())
         async with lock:
-            return await self._load_model_locked(manufacturer, model, force_update, retry_count)
+            return await self._load_model_locked(manufacturer, model, force_update)
 
     async def _load_model_locked(
         self,
         manufacturer: str,
         model: str,
         force_update: bool,
-        retry_count: int,
     ) -> tuple[dict[str, Any], str] | None:
         """Load a model while holding its per-profile lock."""
         model_info = self._get_library_model(manufacturer, model)
+        key = f"{manufacturer}/{model}"
+        if self._is_unsupported_version(manufacturer, model, model_info, self.powercalc_version):
+            raise LibraryLoadingError(f"Profile {key} requires Powercalc {model_info['min_version']}")
         storage_path = self.get_storage_path(manufacturer, model)
-        model_path = os.path.join(storage_path, "model.json")
-
-        while True:
-            if await self._needs_update(model_info, manufacturer, model, model_path, force_update):
-                await self._download_profile_with_retry(manufacturer, model, storage_path, model_path)
-
-            try:
-                json_data = await self._load_model_json(model_path)
-            except JSONDecodeError as error:
-                if retry_count >= MODEL_JSON_RETRY_LIMIT:
-                    _LOGGER.error(
-                        "model.json remains invalid after %d redownload attempts for manufacturer: %s, model: %s",
-                        MODEL_JSON_RETRY_LIMIT,
-                        manufacturer,
-                        model,
-                    )
-                    raise LibraryLoadingError("Failed to load model.json file") from error
-
-                retry_count += 1
-                force_update = True
-                _LOGGER.warning(
-                    "model.json is not valid JSON for manufacturer: %s, model: %s; redownloading profile "
-                    "(attempt %d of %d)",
-                    manufacturer,
-                    model,
-                    retry_count,
-                    MODEL_JSON_RETRY_LIMIT,
-                )
-                continue
-
-            return json_data, storage_path
+        installed = await self.hass.async_add_executor_job(
+            read_installed_profile,
+            Path(storage_path),
+            model,
+            self.profile_hashes.get(key),
+            self.powercalc_version,
+        )
+        if key in self._fallback_models:
+            if installed is None:
+                raise LibraryLoadingError(f"No compatible installed profile for {key}")
+        elif force_update or installed is None or installed.metadata.get("hash") != model_info.get("hash"):
+            installed = await self._download_profile_with_retry(manufacturer, model, storage_path, installed)
+        return await self._load_model_json(str(installed.directory / "model.json")), str(installed.directory)
 
     def _get_library_model(self, manufacturer: str, model: str) -> LibraryModel:
         """Retrieve model info, or raise an error if not found."""
@@ -601,57 +586,52 @@ class RemoteLoader(Loader):
             raise LibraryLoadingError(f"Model not found in library: {manufacturer}/{model}")
         return model_info
 
-    async def _needs_update(
-        self,
-        model_info: LibraryModel,
-        manufacturer: str,
-        model: str,
-        model_path: str,
-        force_update: bool,
-    ) -> bool:
-        """Check if the model needs to be updated."""
-        if force_update:
-            return True
-
-        path_exists = await self.hass.async_add_executor_job(os.path.exists, model_path)
-        if not path_exists:
-            return True
-
-        existing_hash = self.profile_hashes.get(f"{manufacturer}/{model}")
-        new_hash = model_info.get("hash")
-        return existing_hash != new_hash
-
     async def _download_profile_with_retry(
         self,
         manufacturer: str,
         model: str,
         storage_path: str,
-        model_path: str,
-    ) -> None:
-        """Attempt to download the profile, with retry logic and error handling."""
+        installed: InstalledProfile | None,
+    ) -> InstalledProfile:
+        """Update a profile, falling back only to a validated compatible installed copy."""
         try:
             model_info = self._get_library_model(manufacturer, model)
-            model_hash = str(model_info.get("hash"))
-            callback = partial(self.download_profile, manufacturer, model, storage_path, model_hash)
+            callback = partial(self._download_and_install_profile, manufacturer, model, storage_path, model_info)
             await self.download_with_retry(callback)
-            self.profile_hashes[f"{manufacturer}/{model}"] = model_hash
+            self.profile_hashes[f"{manufacturer}/{model}"] = str(model_info.get("hash"))
             await self.hass.async_add_executor_job(self._write_profile_hashes, dict(self.profile_hashes))
-        except ProfileDownloadError as e:
-            path_exists, storage_path_exists = await self.hass.async_add_executor_job(
-                self._profile_paths_exist,
-                model_path,
-                storage_path,
-            )
-            if not path_exists:
-                if storage_path_exists:
-                    await self.hass.async_add_executor_job(shutil.rmtree, storage_path)  # pragma: no cover
-                raise e
+            return self.installed_profiles[f"{manufacturer}/{model}"]
+        except ProfileDownloadError:
+            if installed is None:
+                raise
             _LOGGER.debug("Failed to download profile, falling back to local profile")
+            return installed
 
-    @staticmethod
-    def _profile_paths_exist(model_path: str, storage_path: str) -> tuple[bool, bool]:
-        """Check profile paths from the executor."""
-        return os.path.exists(model_path), os.path.exists(storage_path)
+    async def _download_and_install_profile(
+        self,
+        manufacturer: str,
+        model: str,
+        storage_path: str,
+        model_info: LibraryModel,
+    ) -> None:
+        """Stage files and metadata so failed updates leave the installed revision untouched."""
+        staging: Path | None = None
+        try:
+            staging = await self.hass.async_add_executor_job(create_staging_directory, Path(storage_path))
+            await self.download_profile(manufacturer, model, str(staging), str(model_info.get("hash")))
+            installed = await self.hass.async_add_executor_job(
+                install_profile,
+                Path(storage_path),
+                staging,
+                dict(model_info),
+                self.powercalc_version,
+            )
+            self.installed_profiles[f"{manufacturer}/{model}"] = installed
+        except OSError as err:
+            raise ProfileDownloadError(f"Failed to install profile: {manufacturer}/{model}") from err
+        finally:
+            if staging is not None:
+                await self.hass.async_add_executor_job(shutil.rmtree, staging, True)
 
     async def _load_model_json(self, model_path: str) -> dict[str, Any]:
         """Load the JSON data from the model file."""
@@ -790,4 +770,4 @@ class RemoteLoader(Loader):
         """Write profile hashes to local storage, atomically."""
 
         path = self._get_profile_hashes_path()
-        _save_resource(json.dumps(hashes, indent=4).encode(), Path(path))
+        save_resource(json.dumps(hashes, indent=4).encode(), Path(path))

custom_components/powercalc/strategy/composite.py

@@ -24,6 +24,7 @@ from custom_components.powercalc.const import (
     CONF_PLAYBOOK,
     CONF_STRATEGIES,
     CONF_WLED,
+    UNAVAILABLE_STATES,
 )
 from custom_components.powercalc.strategy.fixed import CONFIG_SCHEMA as FIXED_SCHEMA
 from custom_components.powercalc.strategy.linear import CONFIG_SCHEMA as LINEAR_SCHEMA
@@ -142,6 +143,7 @@ LUT_SCHEMA = vol.Schema({})
 
 ITEM_SCHEMA = vol.Schema(
     {
+        vol.Optional(CONF_ENTITY_ID): cv.entity_id,
         vol.Optional(CONF_CONDITION): CONDITION_SCHEMA,
         vol.Optional(CONF_FIXED): FIXED_SCHEMA,
         vol.Optional(CONF_LINEAR): LINEAR_SCHEMA,
@@ -188,7 +190,12 @@ class CompositeStrategy(PowerCalculationStrategyInterface):
 
         total = Decimal(0)
         for sub_strategy in self.strategies:
-            value = await self._calculate_sub_strategy(sub_strategy, entity_state)
+            sub_state = entity_state
+            if sub_strategy.entity_id is not None:
+                sub_state = self.hass.states.get(sub_strategy.entity_id)
+                if sub_state is None or sub_state.state in UNAVAILABLE_STATES:
+                    return None
+            value = await self._calculate_sub_strategy(sub_strategy, sub_state)
             if value is None:
                 continue
             if self.mode == CompositeMode.STOP_AT_FIRST:
@@ -247,18 +254,14 @@ class CompositeStrategy(PowerCalculationStrategyInterface):
 
     def get_entities_to_track(self) -> list[str | TrackTemplate]:
         """Return entities that should be tracked."""
-        track_templates: list[str | TrackTemplate] = []
+        entities: list[str | TrackTemplate] = []
         for sub_strategy in self.strategies:
             if sub_strategy.condition_config:
-                self.resolve_track_templates_from_condition(
-                    sub_strategy.condition_config,
-                    track_templates,
-                )
-
-        track_entities = [
-            entity for sub_strategy in self.strategies for entity in sub_strategy.strategy.get_entities_to_track()
-        ]
-        return track_templates + track_entities
+                self.resolve_track_templates_from_condition(sub_strategy.condition_config, entities)
+            entities.extend(sub_strategy.strategy.get_entities_to_track())
+            if sub_strategy.entity_id is not None:
+                entities.append(sub_strategy.entity_id)
+        return entities
 
     def can_calculate_standby(self) -> bool:
         """Return if this strategy can calculate standby power."""
@@ -291,3 +294,4 @@ class SubStrategy:
     condition_config: ConfigType | None
     condition: ConditionCheckerType | None
     strategy: PowerCalculationStrategyInterface
+    entity_id: str | None = None

custom_components/powercalc/strategy/factory.py

@@ -10,7 +10,7 @@ from homeassistant.helpers.template import Template
 from homeassistant.helpers.typing import ConfigType
 import voluptuous as vol
 
-from custom_components.powercalc.common import SourceEntity
+from custom_components.powercalc.common import SourceEntity, create_source_entity
 from custom_components.powercalc.const import (
     CONF_CALIBRATE,
     CONF_COMPOSITE,
@@ -258,10 +258,12 @@ class PowerCalculatorStrategyFactory:
             sub_strategies = composite_config
 
         async def _create_sub_strategy(strategy_config: ConfigType) -> SubStrategy:
+            entity_id = strategy_config.get(CONF_ENTITY_ID)
+            sub_source_entity = create_source_entity(entity_id, self._hass) if entity_id is not None else source_entity
             condition_instance = None
             condition_config = strategy_config.get(CONF_CONDITION)
             if condition_config:
-                condition_config = resolve_condition_entity_ids(condition_config, source_entity)
+                condition_config = resolve_condition_entity_ids(condition_config, sub_source_entity)
                 condition_config = await condition.async_validate_condition_config(self._hass, condition_config)
                 condition_instance = await condition.async_from_config(
                     self._hass,
@@ -273,9 +275,9 @@ class PowerCalculatorStrategyFactory:
                 strategy_config,
                 strategy,
                 power_profile,
-                source_entity,
+                sub_source_entity,
             )
-            return SubStrategy(condition_config, condition_instance, strategy_instance)  # type: ignore
+            return SubStrategy(condition_config, condition_instance, strategy_instance, entity_id)  # type: ignore
 
         if not sub_strategies:
             raise StrategyConfigurationError("No strategies configured for composite strategy")