francop/HomeAssistantVS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

217 files

Browse Source
This commit is contained in:
Home Assistant Version Control
2026-07-30 23:59:38 +00:00
parent d43a63ad29
commit 7b5e46e702
217 changed files with 15978 additions and 3912 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/powercalc/power_profile/factory.py
+2 -1
View File
@@ -4,6 +4,7 @@ import logging
import os
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType
from custom_components.powercalc.common import SourceEntity
from custom_components.powercalc.const import (
@@ -24,7 +25,7 @@ _LOGGER = logging.getLogger(__name__)
async def get_power_profile(
hass: HomeAssistant,
config: dict,
config: ConfigType,
source_entity: SourceEntity | None = None,
model_info: ModelInfo | None = None,
log_errors: bool = True,
custom_components/powercalc/power_profile/library.py
+8 -4
View File
@@ -50,7 +50,6 @@ class ProfileLibrary:
self._loader = loader
self._profiles: dict[str, list[PowerProfile]] = {}
self._manufacturer_models: dict[str, set[tuple[str, str]]] = {}
self._manufacturer_device_types: dict[str, list] = {}
async def initialize(self) -> None:
await self._loader.initialize()
@@ -301,7 +300,12 @@ class ProfileLibrary:
return next(iter(matches))
async def _load_model_data(self, manufacturer: str, model: str, custom_directory: str | None) -> tuple[dict, str]:
async def _load_model_data(
self,
manufacturer: str,
model: str,
custom_directory: str | None,
) -> tuple[dict[str, Any], str]:
"""Load the model data from the appropriate directory."""
loader = (
LocalLoader(self._hass, custom_directory, is_custom_directory=True) if custom_directory else self._loader
@@ -317,8 +321,8 @@ class ProfileLibrary:
manufacturer: str,
model: str,
directory: str,
json_data: dict,
sub_profiles: list[tuple[str, dict]] | None = None,
json_data: dict[str, Any],
sub_profiles: list[tuple[str, dict[str, Any]]] | None = None,
) -> PowerProfile:
"""Create and initialize the PowerProfile object."""
profile = PowerProfile(
custom_components/powercalc/power_profile/loader/composite.py
+2 -1
View File
@@ -1,4 +1,5 @@
import logging
from typing import Any
from custom_components.powercalc.power_profile.loader.protocol import Loader
from custom_components.powercalc.power_profile.power_profile import DeviceType, DiscoveryBy
@@ -53,7 +54,7 @@ class CompositeLoader(Loader):
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:
async def load_model(self, manufacturer: str, model: str) -> tuple[dict[str, Any], str] | None:
for loader in self.loaders:
result = await loader.load_model(manufacturer, model)
if result:
custom_components/powercalc/power_profile/loader/local.py
+1 -1
View File
@@ -88,7 +88,7 @@ class LocalLoader(Loader):
return found_models
async def load_model(self, manufacturer: str, model: str) -> tuple[dict, str] | None:
async def load_model(self, manufacturer: str, model: str) -> tuple[dict[str, Any], str] | None:
"""Load a model.json file from disk for a given manufacturer.lower() and model.lower()
by querying the custom library.
If self._is_custom_directory == true model.json will be loaded directly from there.
custom_components/powercalc/power_profile/loader/protocol.py
+2 -2
View File
@@ -1,4 +1,4 @@
from typing import Protocol
from typing import Any, Protocol
from custom_components.powercalc.power_profile.power_profile import DeviceType, DiscoveryBy
@@ -25,7 +25,7 @@ class Loader(Protocol):
) -> 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:
async def load_model(self, manufacturer: str, model: str) -> tuple[dict[str, Any], str] | None:
"""Load and optionally download a model profile."""
async def find_model(self, manufacturer: str, search: set[str]) -> list[str]:
custom_components/powercalc/power_profile/loader/remote.py
+58 -40
View File
@@ -53,7 +53,7 @@ class RemoteLoader(Loader):
def __init__(self, hass: HomeAssistant) -> None:
self.hass = hass
self.library_contents: dict = {}
self.library_contents: dict[str, Any] = {}
self.model_infos: dict[str, LibraryModel] = {}
self.manufacturer_models: dict[str, list[LibraryModel]] = {}
self.model_lookup: dict[str, dict[str, list[LibraryModel]]] = {}
@@ -78,51 +78,69 @@ class RemoteLoader(Loader):
manufacturers: list[LibraryManufacturer] = self.library_contents.get("manufacturers", [])
for manufacturer in manufacturers:
manufacturer_name = str(manufacturer.get("dir_name"))
models: list[LibraryModel] = manufacturer.get("models", []) or []
self._index_manufacturer(manufacturer, powercalc_version)
# manufacturer alias map (alias -> {canonical manufacturer_name})
self.manufacturer_lookup.setdefault(manufacturer_name.lower(), set()).add(manufacturer_name)
for alias in manufacturer.get("aliases", []) or []:
self.manufacturer_lookup.setdefault(str(alias).lower(), set()).add(manufacturer_name)
def _index_manufacturer(self, manufacturer: LibraryManufacturer, powercalc_version: AwesomeVersion) -> None:
"""Register a manufacturer, its aliases and all of its supported models in the lookup tables."""
manufacturer_name = str(manufacturer.get("dir_name"))
models: list[LibraryModel] = manufacturer.get("models", []) or []
# per-manufacturer model lookup
kept_models: list[LibraryModel] = []
lookup: dict[str, list[LibraryModel]] = {}
# manufacturer alias map (alias -> {canonical manufacturer_name})
self.manufacturer_lookup.setdefault(manufacturer_name.lower(), set()).add(manufacturer_name)
for alias in manufacturer.get("aliases", []) or []:
self.manufacturer_lookup.setdefault(str(alias).lower(), set()).add(manufacturer_name)
for model in models:
min_version = model.get("min_version")
model_id = str(model.get("id"))
model_id_lower = model_id.lower()
# per-manufacturer model lookup
kept_models: list[LibraryModel] = []
lookup: dict[str, list[LibraryModel]] = {}
self.model_infos[f"{manufacturer_name}/{model_id!s}"] = model
for model in models:
model_id = str(model.get("id"))
self.model_infos[f"{manufacturer_name}/{model_id}"] = model
if min_version and powercalc_version < AwesomeVersion(min_version):
_LOGGER.debug(
"Skipping model %s/%s as it requires powercalc version %s (current: %s)",
manufacturer_name,
model_id,
min_version,
powercalc_version,
)
continue
if self._is_unsupported_version(manufacturer_name, model_id, model, powercalc_version):
continue
kept_models.append(model)
kept_models.append(model)
self._add_model_to_lookup(lookup, model, model_id.lower())
# Exact id bucket first (highest priority)
bucket = lookup.setdefault(model_id_lower, [])
bucket.insert(0, model)
self.manufacturer_models[manufacturer_name] = kept_models
self.model_lookup[manufacturer_name] = lookup
# Alias buckets afterwards (lower priority)
for alias in model.get("aliases", []) or []:
alias_lower = str(alias).lower()
if alias_lower == model_id_lower:
continue
# Append to the end to ensure aliased models are always last
lookup.setdefault(alias_lower, []).append(model)
@staticmethod
def _is_unsupported_version(
manufacturer_name: str,
model_id: str,
model: LibraryModel,
powercalc_version: AwesomeVersion,
) -> bool:
"""Check whether the model requires a newer powercalc version than the one installed."""
min_version = model.get("min_version")
if not min_version or powercalc_version >= AwesomeVersion(min_version):
return False
self.manufacturer_models[manufacturer_name] = kept_models
self.model_lookup[manufacturer_name] = lookup
_LOGGER.debug(
"Skipping model %s/%s as it requires powercalc version %s (current: %s)",
manufacturer_name,
model_id,
min_version,
powercalc_version,
)
return True
@staticmethod
def _add_model_to_lookup(lookup: dict[str, list[LibraryModel]], model: LibraryModel, model_id_lower: str) -> None:
"""Bucket a model by its id and aliases. Exact ids take priority over aliases."""
# Exact id bucket first (highest priority)
lookup.setdefault(model_id_lower, []).insert(0, model)
# Alias buckets afterwards (lower priority)
for alias in model.get("aliases", []) or []:
alias_lower = str(alias).lower()
if alias_lower == model_id_lower:
continue
# Append to the end to ensure aliased models are always last
lookup.setdefault(alias_lower, []).append(model)
def _clear_caches(self) -> None:
"""Clear cached lookups backed by mutable library state."""
@@ -269,7 +287,7 @@ class RemoteLoader(Loader):
model: str,
force_update: bool = False,
retry_count: int = 0,
) -> tuple[dict, str] | None:
) -> tuple[dict[str, Any], str] | None:
"""Load a model, downloading it if necessary, with retry logic."""
model_info = self._get_library_model(manufacturer, model)
storage_path = self.get_storage_path(manufacturer, model)
@@ -344,7 +362,7 @@ class RemoteLoader(Loader):
"""Check profile paths from the executor."""
return os.path.exists(model_path), os.path.exists(storage_path)
async def _load_model_json(self, model_path: str) -> dict:
async def _load_model_json(self, model_path: str) -> dict[str, Any]:
"""Load the JSON data from the model file."""
def _load_json() -> dict[str, Any]:
@@ -359,7 +377,7 @@ class RemoteLoader(Loader):
manufacturer: str,
model: str,
retry_count: int,
) -> tuple[dict, str] | None:
) -> tuple[dict[str, Any], str] | None:
"""Handle JSON decode errors with retry logic."""
_LOGGER.error("model.json file is not valid JSON for manufacturer: %s, model: %s", manufacturer, model)
if retry_count < 2:
custom_components/powercalc/power_profile/power_profile.py
+4 -4
View File
@@ -126,7 +126,7 @@ class PowerProfile:
model: str,
directory: str,
json_data: ConfigType,
sub_profiles: list[tuple[str, dict]] | None = None,
sub_profiles: list[tuple[str, dict[str, Any]]] | None = None,
) -> None:
self._manufacturer = manufacturer
self._model = model.replace("#slash#", "/")
@@ -230,9 +230,9 @@ class PowerProfile:
return config
@property
def composite_config(self) -> list | None:
def composite_config(self) -> list[ConfigType] | None:
"""Get configuration to set up composite strategy."""
return cast(list, self._json_data.get("composite_config"))
return cast(list[ConfigType], self._json_data.get("composite_config"))
@property
def playbook_config(self) -> ConfigType | None:
@@ -356,7 +356,7 @@ class PowerProfile:
return "remarks_smart_dimmer"
return None
async def get_sub_profiles(self) -> list[tuple[str, dict]]:
async def get_sub_profiles(self) -> list[tuple[str, dict[str, Any]]]:
"""Get listing of possible sub profiles and their corresponding JSON data."""
return self._sub_profiles
custom_components/powercalc/power_profile/sub_profile_selector.py
+3 -3
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
from enum import StrEnum
import re
from typing import NamedTuple, Protocol
from typing import Any, NamedTuple, Protocol
from homeassistant.core import HomeAssistant, State
@@ -50,7 +50,7 @@ class SubProfileSelector:
"""Get additional list of entities to track for state changes."""
return [entity_id for matcher in self._matchers for entity_id in matcher.get_tracking_entities()]
def _create_matcher(self, matcher_config: dict) -> SubProfileMatcher:
def _create_matcher(self, matcher_config: dict[str, Any]) -> SubProfileMatcher:
"""Create a matcher from json config. Can be extended for more matchers in the future."""
matcher_type: SubProfileMatcherType = matcher_config["type"]
match matcher_type:
@@ -81,7 +81,7 @@ class SubProfileSelector:
class SubProfileSelectConfig(NamedTuple):
default: str
matchers: list[dict] | None = None
matchers: list[dict[str, Any]] | None = None
class SubProfileMatcher(Protocol):