186 files

custom_components/powercalc/strategy/factory.py

@@ -12,10 +12,15 @@ import voluptuous as vol
 
 from custom_components.powercalc.common import SourceEntity
 from custom_components.powercalc.const import (
+    CONF_CALIBRATE,
     CONF_COMPOSITE,
+    CONF_GAMMA_CURVE,
+    CONF_MAX_POWER,
+    CONF_MIN_POWER,
     CONF_MODE,
     CONF_MULTI_SWITCH,
     CONF_POWER,
+    CONF_POWER_CURVE,
     CONF_POWER_OFF,
     CONF_POWER_TEMPLATE,
     CONF_STANDBY_POWER,
@@ -118,7 +123,7 @@ class PowerCalculatorStrategyFactory:
         power_profile: PowerProfile | None,
     ) -> LinearStrategy:
         """Create the linear strategy."""
-        linear_config = self._get_strategy_config(CalculationStrategy.LINEAR, config, power_profile)
+        linear_config = self._get_linear_config(config, power_profile)
 
         return LinearStrategy(
             linear_config,
@@ -127,6 +132,55 @@ class PowerCalculatorStrategyFactory:
             config.get(CONF_STANDBY_POWER),
         )
 
+    @staticmethod
+    def _get_linear_config(config: ConfigType, power_profile: PowerProfile | None) -> ConfigType:
+        """Combine an incomplete profile configuration with user supplied power values."""
+        user_config = cast(ConfigType | None, config.get(CalculationStrategy.LINEAR))
+
+        if power_profile is None or not power_profile.is_strategy_supported(CalculationStrategy.LINEAR):
+            if user_config is not None:
+                return user_config
+            raise StrategyConfigurationError("No linear configuration supplied")
+
+        profile_config = dict(power_profile.linear_config or {})
+
+        # Preserve the existing behavior for complete profiles: an explicitly supplied
+        # user configuration replaces the profile configuration entirely.
+        if not power_profile.needs_linear_config:
+            return user_config if user_config is not None else profile_config
+
+        # Smart dimmer profiles without load values can be used from YAML to measure
+        # only the dimmer's own consumption. Keep that behavior when the profile adds
+        # defaults such as gamma_curve but the user omits the linear configuration.
+        if not user_config:
+            return {
+                CONF_MIN_POWER: 0,
+                CONF_MAX_POWER: 0,
+                **profile_config,
+            }
+
+        linear_config = {
+            **profile_config,
+            **user_config,
+        }
+
+        PowerCalculatorStrategyFactory._remove_overridden_profile_curves(linear_config, user_config)
+
+        return linear_config
+
+    @staticmethod
+    def _remove_overridden_profile_curves(linear_config: ConfigType, user_config: ConfigType) -> None:
+        """Remove profile curve options superseded by explicit user configuration."""
+        if CONF_CALIBRATE in user_config:
+            if CONF_GAMMA_CURVE not in user_config:
+                linear_config.pop(CONF_GAMMA_CURVE, None)
+            if CONF_POWER_CURVE not in user_config:
+                linear_config.pop(CONF_POWER_CURVE, None)
+        elif CONF_GAMMA_CURVE in user_config:
+            linear_config.pop(CONF_POWER_CURVE, None)
+        elif CONF_POWER_CURVE in user_config:
+            linear_config.pop(CONF_GAMMA_CURVE, None)
+
     def _create_fixed(
         self,
         source_entity: SourceEntity,

custom_components/powercalc/strategy/linear.py

@@ -1,3 +1,4 @@
+from bisect import bisect_left, bisect_right
 from decimal import Decimal
 import logging
 from typing import Any
@@ -24,6 +25,7 @@ from custom_components.powercalc.const import (
     CONF_MAX_POWER,
     CONF_MIN_POWER,
     CONF_POWER,
+    CONF_POWER_CURVE,
     CONF_VALUE,
 )
 from custom_components.powercalc.errors import StrategyConfigurationError
@@ -41,6 +43,11 @@ CONFIG_SCHEMA = vol.Schema(
         vol.Optional(CONF_MIN_POWER): vol.Coerce(float),
         vol.Optional(CONF_MAX_POWER): vol.Coerce(float),
         vol.Optional(CONF_GAMMA_CURVE): vol.Coerce(float),
+        vol.Optional(CONF_POWER_CURVE): vol.All(
+            cv.ensure_list,
+            [vol.Match(r"^(?:0(?:\.\d+)?|1(?:\.0+)?) -> (?:0(?:\.\d+)?|1(?:\.0+)?)$")],
+            vol.Length(min=2),
+        ),
         vol.Optional(CONF_ATTRIBUTE): cv.string,
     },
 )
@@ -71,11 +78,13 @@ class LinearStrategy(PowerCalculationStrategyInterface):
         self._initialized: bool = False
         self._missing_attribute_warned: bool = False
         self._calibration: list[tuple[int, float]] | None = None
+        self._power_curve: list[tuple[float, float]] | None = None
 
     async def initialize(self) -> None:
         """Initialize the strategy, called once on creation."""
         self._value_entity = await self.get_value_entity()
         self._calibration = self.create_calibrate_list()
+        self._power_curve = self.create_power_curve_list()
 
     async def calculate(self, entity_state: State) -> Decimal | None:
         """Calculate the current power consumption."""
@@ -89,8 +98,7 @@ class LinearStrategy(PowerCalculationStrategyInterface):
         if value is None:
             return None
 
-        min_calibrate = self.get_min_calibrate(value)
-        max_calibrate = self.get_max_calibrate(value)
+        min_calibrate, max_calibrate = self.get_calibration_segment(value)
         min_value = min_calibrate[0]
         max_value = max_calibrate[0]
 
@@ -108,11 +116,8 @@ class LinearStrategy(PowerCalculationStrategyInterface):
         value_range = max_value - min_value
         power_range = max_power - min_power
 
-        gamma_curve = self._config.get(CONF_GAMMA_CURVE) or 1
-
         relative_value = (value - min_value) / value_range
-
-        power = power_range * relative_value**gamma_curve + min_power
+        power = power_range * self.apply_curve(relative_value) + min_power
 
         return Decimal(power)
 
@@ -120,13 +125,22 @@ class LinearStrategy(PowerCalculationStrategyInterface):
         """Return if this strategy is enabled based on entity state."""
         return not (self._source_entity.domain == media_player.DOMAIN and entity_state.state != STATE_PLAYING)
 
-    def get_min_calibrate(self, value: int) -> tuple[int, float]:
-        """Get closest lower value from calibration table."""
-        return min(self._calibration or (), key=lambda v: (v[0] > value, value - v[0]))
+    def get_calibration_segment(self, value: int) -> tuple[tuple[int, float], tuple[int, float]]:
+        """Get the two calibration points to interpolate between, in ascending order.
 
-    def get_max_calibrate(self, value: int) -> tuple[int, float]:
-        """Get closest higher value from calibration table."""
-        return max(self._calibration or (), key=lambda v: (v[0] > value, value - v[0]))
+        Values inside the table use the segment they fall in. Values outside it are
+        extrapolated along the chord between the first and the last point.
+        """
+        calibration = self._calibration
+        if not calibration:
+            raise StrategyConfigurationError("Linear strategy has not been initialized")
+
+        if value < calibration[0][0] or value > calibration[-1][0]:
+            return calibration[0], calibration[-1]
+
+        index = bisect_right(calibration, value, key=lambda point: point[0])
+        index = min(max(index, 1), len(calibration) - 1)
+        return calibration[index - 1], calibration[index]
 
     def create_calibrate_list(self) -> list[tuple[int, float]]:
         """Build a table of calibration values."""
@@ -156,6 +170,43 @@ class LinearStrategy(PowerCalculationStrategyInterface):
 
         return sorted(calibration_list, key=lambda tup: tup[0])
 
+    def create_power_curve_list(self) -> list[tuple[float, float]] | None:
+        """Build a table of normalized power curve values."""
+        power_curve = self._config.get(CONF_POWER_CURVE)
+        if not power_curve:
+            return None
+
+        points = []
+        for line in power_curve:
+            value, power = line.split(" -> ")
+            points.append((float(value), float(power)))
+        return sorted(points, key=lambda point: point[0])
+
+    def apply_curve(self, relative_value: float) -> float:
+        """Apply a configured gamma or normalized power curve."""
+        gamma_curve = self._config.get(CONF_GAMMA_CURVE)
+        if gamma_curve:
+            if relative_value < 0:
+                # A negative base raised to a fractional exponent is complex. Below the
+                # calibrated range there is no curve to apply, so stay linear.
+                return relative_value
+            return float(relative_value ** float(gamma_curve))
+
+        if self._power_curve:
+            if relative_value <= self._power_curve[0][0]:
+                return self._power_curve[0][1]
+            if relative_value >= self._power_curve[-1][0]:
+                return self._power_curve[-1][1]
+
+            max_index = bisect_left(self._power_curve, relative_value, key=lambda point: point[0])
+            min_point = self._power_curve[max_index - 1]
+            max_point = self._power_curve[max_index]
+            value_range = max_point[0] - min_point[0]
+            curve_range = max_point[1] - min_point[1]
+            return curve_range * ((relative_value - min_point[0]) / value_range) + min_point[1]
+
+        return relative_value
+
     def get_entity_value_range(self) -> tuple[int, int]:
         """Get the min/max range for a given entity domain."""
         if self.get_initialized_value_entity().domain == light.DOMAIN:

custom_components/powercalc/strategy/lut.py

@@ -5,7 +5,6 @@ from dataclasses import dataclass
 from decimal import Decimal
 from enum import StrEnum
 from functools import partial
-import gzip
 import logging
 import os
 from typing import Any, TextIO, cast
@@ -30,6 +29,7 @@ from custom_components.powercalc.errors import (
 )
 from custom_components.powercalc.power_profile.power_profile import PowerProfile
 
+from .profile_data import open_profile_csv
 from .strategy_interface import PowerCalculationStrategyInterface
 
 _LOGGER = logging.getLogger(__name__)
@@ -78,7 +78,7 @@ class LutRegistry:
         self._hass = hass
         self._lut_entries: dict[_CacheKey, _LutEntry] = {}
         self._effect_entries: dict[_CacheKey, _EffectEntry] = {}
-        self._supported_modes: dict[tuple[str, str, str], set[LookupMode]] = {}
+        self._supported_modes: dict[tuple[str, str, str | None, str], set[LookupMode]] = {}
 
     async def get_lookup_entry(
         self,
@@ -107,7 +107,7 @@ class LutRegistry:
 
     async def get_supported_modes(self, power_profile: PowerProfile) -> set[LookupMode]:
         """Return the LUT modes supported by the profile."""
-        cache_key = (power_profile.manufacturer, power_profile.model, "supported_modes")
+        cache_key = (power_profile.manufacturer, power_profile.model, power_profile.sub_profile, "supported_modes")
         supported_modes = self._supported_modes.get(cache_key)
         if supported_modes is None:
             supported_modes = set()
@@ -184,11 +184,11 @@ class LutRegistry:
         gzip_path = f"{path}.gz"
         if os.path.exists(gzip_path):
             _LOGGER.debug("Loading LUT data file: %s", gzip_path)
-            return gzip.open(gzip_path, "rt")
+            return open_profile_csv(gzip_path)
 
         if os.path.exists(path):
             _LOGGER.debug("Loading LUT data file: %s", path)
-            return open(path)
+            return open_profile_csv(path)
 
         raise LutFileNotFoundError(f"Data file not found: {path}")
 
@@ -232,7 +232,7 @@ class LutStrategy(PowerCalculationStrategyInterface):
             return None
 
         effect = attrs.get(ATTR_EFFECT)
-        if effect and str(effect).lower() not in ("off", "none", "white"):
+        if effect and str(effect).lower() not in ("off", "none", "white", "default", "mode color"):
             return await self._calculate_effect_power(entity_state, str(effect), brightness)
 
         lut_mode = LookupMode.from_color_mode(color_mode)

custom_components/powercalc/strategy/playbook.py

@@ -4,7 +4,6 @@ import csv
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from decimal import Decimal
-import gzip
 import logging
 import os
 
@@ -25,6 +24,7 @@ from custom_components.powercalc.const import (
 )
 from custom_components.powercalc.errors import StrategyConfigurationError
 
+from .profile_data import open_profile_csv
 from .strategy_interface import PowerCalculationStrategyInterface
 
 CONFIG_SCHEMA = vol.All(
@@ -222,8 +222,7 @@ class PlaybookStrategy(PowerCalculationStrategyInterface):
                     f"Playbook file '{file_path}' does not exist",
                 )
             actual_path = file_path if os.path.exists(file_path) else f"{file_path}.gz"
-            open_func = gzip.open if actual_path.endswith(".gz") else open
-            with open_func(actual_path, mode="rt") as csv_file:
+            with open_profile_csv(actual_path) as csv_file:
                 csv_reader = csv.reader(csv_file)
                 entries = []
                 for row in csv_reader: