diff --git a/custom_components/tibber_prices/entity_utils/__init__.py b/custom_components/tibber_prices/entity_utils/__init__.py index 800bdad..24e2f4e 100644 --- a/custom_components/tibber_prices/entity_utils/__init__.py +++ b/custom_components/tibber_prices/entity_utils/__init__.py @@ -1,9 +1,36 @@ -"""Entity utilities for Tibber Prices integration.""" +""" +Home Assistant entity-specific utilities for Tibber Prices integration. + +This package contains HA entity integration logic: +- Dynamic icon selection based on state/price levels +- Icon color mapping for visual feedback +- Attribute builders (timestamps, descriptions, periods) +- Translation-aware formatting + +These functions depend on Home Assistant concepts: +- Entity states and attributes +- Translation systems (custom_translations/) +- Configuration entries and coordinator data +- User-configurable options (CONF_EXTENDED_DESCRIPTIONS, etc.) + +For pure data transformation (no HA dependencies), see utils/ package. +""" from __future__ import annotations -from .attributes import build_period_attributes, build_timestamp_attribute +from .attributes import ( + add_description_attributes, + async_add_description_attributes, + build_period_attributes, + build_timestamp_attribute, +) from .colors import add_icon_color_attribute, get_icon_color +from .helpers import ( + find_rolling_hour_center_index, + get_price_value, + translate_level, + translate_rating_level, +) from .icons import ( get_binary_sensor_icon, get_dynamic_icon, @@ -16,16 +43,22 @@ from .icons import ( ) __all__ = [ + "add_description_attributes", "add_icon_color_attribute", + "async_add_description_attributes", "build_period_attributes", "build_timestamp_attribute", + "find_rolling_hour_center_index", "get_binary_sensor_icon", "get_dynamic_icon", "get_icon_color", "get_level_sensor_icon", "get_price_level_for_icon", "get_price_sensor_icon", + "get_price_value", "get_rating_sensor_icon", "get_trend_icon", "get_volatility_sensor_icon", + "translate_level", + "translate_rating_level", ] diff --git a/custom_components/tibber_prices/entity_utils/attributes.py b/custom_components/tibber_prices/entity_utils/attributes.py index 1835676..a26d2e9 100644 --- a/custom_components/tibber_prices/entity_utils/attributes.py +++ b/custom_components/tibber_prices/entity_utils/attributes.py @@ -2,6 +2,13 @@ from __future__ import annotations +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from homeassistant.core import HomeAssistant + + from ..data import TibberPricesConfigEntry # noqa: TID252 + def build_timestamp_attribute(interval_data: dict | None) -> str | None: """ @@ -40,3 +47,191 @@ def build_period_attributes(period_data: dict) -> dict: "duration_minutes": period_data.get("duration_minutes"), "timestamp": period_data.get("start"), # Timestamp = period start } + + +def add_description_attributes( # noqa: PLR0913 + attributes: dict, + platform: str, + translation_key: str | None, + hass: HomeAssistant, + config_entry: TibberPricesConfigEntry, + *, + position: str = "end", +) -> None: + """ + Add description attributes from custom translations to an existing attributes dict. + + Adds description (always), and optionally long_description and usage_tips if + CONF_EXTENDED_DESCRIPTIONS is enabled in config. + + This function modifies the attributes dict in-place. By default, descriptions are + added at the END of the dict (after all other attributes). For special cases like + chart_data_export, use position="before_service_data" to add descriptions before + service data attributes. + + Args: + attributes: Existing attributes dict to modify (in-place) + platform: Platform name ("sensor" or "binary_sensor") + translation_key: Translation key for entity + hass: Home Assistant instance + config_entry: Config entry with options + position: Where to add descriptions: + - "end" (default): Add at the very end + - "before_service_data": Add before service data (for chart_data_export) + + """ + if not translation_key or not hass: + return + + # Import here to avoid circular dependency + from ..const import ( # noqa: PLC0415, TID252 + CONF_EXTENDED_DESCRIPTIONS, + DEFAULT_EXTENDED_DESCRIPTIONS, + get_entity_description, + ) + + language = hass.config.language if hass.config.language else "en" + + # Build description dict + desc_attrs: dict[str, str] = {} + + description = get_entity_description(platform, translation_key, language, "description") + if description: + desc_attrs["description"] = description + + extended_descriptions = config_entry.options.get( + CONF_EXTENDED_DESCRIPTIONS, + config_entry.data.get(CONF_EXTENDED_DESCRIPTIONS, DEFAULT_EXTENDED_DESCRIPTIONS), + ) + + if extended_descriptions: + long_desc = get_entity_description(platform, translation_key, language, "long_description") + if long_desc: + desc_attrs["long_description"] = long_desc + + usage_tips = get_entity_description(platform, translation_key, language, "usage_tips") + if usage_tips: + desc_attrs["usage_tips"] = usage_tips + + # Add descriptions at appropriate position + if position == "end": + # Default: Add at the very end + attributes.update(desc_attrs) + elif position == "before_service_data": + # Special case: Insert before service data + # This is used by chart_data_export to keep our attributes before foreign data + # We need to rebuild the dict to maintain order + temp_attrs = dict(attributes) + attributes.clear() + + # Add everything except service data + for key, value in temp_attrs.items(): + if key not in ("timestamp", "error"): + continue + attributes[key] = value + + # Add descriptions here (before service data) + attributes.update(desc_attrs) + + # Add service data last + for key, value in temp_attrs.items(): + if key in ("timestamp", "error"): + continue + attributes[key] = value + + +async def async_add_description_attributes( # noqa: PLR0913 + attributes: dict, + platform: str, + translation_key: str | None, + hass: HomeAssistant, + config_entry: TibberPricesConfigEntry, + *, + position: str = "end", +) -> None: + """ + Async version of add_description_attributes. + + Adds description attributes from custom translations to an existing attributes dict. + Uses async translation loading (calls async_get_entity_description). + + Args: + attributes: Existing attributes dict to modify (in-place) + platform: Platform name ("sensor" or "binary_sensor") + translation_key: Translation key for entity + hass: Home Assistant instance + config_entry: Config entry with options + position: Where to add descriptions ("end" or "before_service_data") + + """ + if not translation_key or not hass: + return + + # Import here to avoid circular dependency + from ..const import ( # noqa: PLC0415, TID252 + CONF_EXTENDED_DESCRIPTIONS, + DEFAULT_EXTENDED_DESCRIPTIONS, + async_get_entity_description, + ) + + language = hass.config.language if hass.config.language else "en" + + # Build description dict + desc_attrs: dict[str, str] = {} + + description = await async_get_entity_description( + hass, + platform, + translation_key, + language, + "description", + ) + if description: + desc_attrs["description"] = description + + extended_descriptions = config_entry.options.get( + CONF_EXTENDED_DESCRIPTIONS, + config_entry.data.get(CONF_EXTENDED_DESCRIPTIONS, DEFAULT_EXTENDED_DESCRIPTIONS), + ) + + if extended_descriptions: + long_desc = await async_get_entity_description( + hass, + platform, + translation_key, + language, + "long_description", + ) + if long_desc: + desc_attrs["long_description"] = long_desc + + usage_tips = await async_get_entity_description( + hass, + platform, + translation_key, + language, + "usage_tips", + ) + if usage_tips: + desc_attrs["usage_tips"] = usage_tips + + # Add descriptions at appropriate position + if position == "end": + # Default: Add at the very end + attributes.update(desc_attrs) + elif position == "before_service_data": + # Special case: Insert before service data (same logic as sync version) + temp_attrs = dict(attributes) + attributes.clear() + + for key, value in temp_attrs.items(): + if key not in ("timestamp", "error"): + continue + attributes[key] = value + + attributes.update(desc_attrs) + + for key, value in temp_attrs.items(): + if key in ("timestamp", "error"): + continue + attributes[key] = value diff --git a/custom_components/tibber_prices/entity_utils/helpers.py b/custom_components/tibber_prices/entity_utils/helpers.py new file mode 100644 index 0000000..478fc7d --- /dev/null +++ b/custom_components/tibber_prices/entity_utils/helpers.py @@ -0,0 +1,128 @@ +""" +Common helper functions for entities across platforms. + +This module provides utility functions used by both sensor and binary_sensor platforms: +- Price value conversion (major/minor currency units) +- Translation helpers (price levels, ratings) +- Time-based calculations (rolling hour center index) + +These functions operate on entity-level concepts (states, translations) but are +platform-agnostic and can be used by both sensor and binary_sensor platforms. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from custom_components.tibber_prices.const import get_price_level_translation +from custom_components.tibber_prices.utils.average import ( + round_to_nearest_quarter_hour, +) +from homeassistant.util import dt as dt_util + +if TYPE_CHECKING: + from datetime import datetime + + from homeassistant.core import HomeAssistant + + +def get_price_value(price: float, *, in_euro: bool) -> float: + """ + Convert price based on unit. + + Args: + price: Price value to convert + in_euro: If True, return price in euros; if False, return in cents/øre + + Returns: + Price in requested unit (euros or minor currency units) + + """ + return price if in_euro else round((price * 100), 2) + + +def translate_level(hass: HomeAssistant, level: str) -> str: + """ + Translate price level to the user's language. + + Args: + hass: HomeAssistant instance for language configuration + level: Price level to translate (e.g., VERY_CHEAP, NORMAL, etc.) + + Returns: + Translated level string, or original level if translation not found + + """ + if not hass: + return level + + language = hass.config.language or "en" + translated = get_price_level_translation(level, language) + if translated: + return translated + + if language != "en": + fallback = get_price_level_translation(level, "en") + if fallback: + return fallback + + return level + + +def translate_rating_level(rating: str) -> str: + """ + Translate price rating level to the user's language. + + Args: + rating: Price rating to translate (e.g., LOW, NORMAL, HIGH) + + Returns: + Translated rating string, or original rating if translation not found + + Note: + Currently returns the rating as-is. Translation mapping for ratings + can be added here when needed, similar to translate_level(). + + """ + # For now, ratings are returned as-is + # Add translation mapping here when needed + return rating + + +def find_rolling_hour_center_index( + all_prices: list[dict], + current_time: datetime, + hour_offset: int, +) -> int | None: + """ + Find the center index for the rolling hour window. + + Args: + all_prices: List of all price interval dictionaries with 'startsAt' key + current_time: Current datetime to find the current interval + hour_offset: Number of hours to offset from current interval (can be negative) + + Returns: + Index of the center interval for the rolling hour window, or None if not found + + """ + # Round to nearest interval boundary to handle edge cases where HA schedules + # us slightly before the boundary (e.g., 14:59:59.999 → 15:00:00) + target_time = round_to_nearest_quarter_hour(current_time) + current_idx = None + + for idx, price_data in enumerate(all_prices): + starts_at = dt_util.parse_datetime(price_data["startsAt"]) + if starts_at is None: + continue + starts_at = dt_util.as_local(starts_at) + + # Exact match after rounding + if starts_at == target_time: + current_idx = idx + break + + if current_idx is None: + return None + + return current_idx + (hour_offset * 4) diff --git a/custom_components/tibber_prices/entity_utils/icons.py b/custom_components/tibber_prices/entity_utils/icons.py index a3aee5f..4f0c336 100644 --- a/custom_components/tibber_prices/entity_utils/icons.py +++ b/custom_components/tibber_prices/entity_utils/icons.py @@ -9,16 +9,15 @@ from typing import TYPE_CHECKING, Any from custom_components.tibber_prices.const import ( BINARY_SENSOR_ICON_MAPPING, + MINUTES_PER_INTERVAL, PRICE_LEVEL_CASH_ICON_MAPPING, PRICE_LEVEL_ICON_MAPPING, PRICE_RATING_ICON_MAPPING, VOLATILITY_ICON_MAPPING, ) -from custom_components.tibber_prices.price_utils import find_price_data_for_interval -from custom_components.tibber_prices.sensor.helpers import ( - aggregate_level_data, - find_rolling_hour_center_index, -) +from custom_components.tibber_prices.entity_utils.helpers import find_rolling_hour_center_index +from custom_components.tibber_prices.sensor.helpers import aggregate_level_data +from custom_components.tibber_prices.utils.price import find_price_data_for_interval from homeassistant.util import dt as dt_util @@ -35,9 +34,6 @@ class IconContext: if TYPE_CHECKING: from collections.abc import Callable -# Constants imported from price_utils -MINUTES_PER_INTERVAL = 15 - # Timing sensor icon thresholds (in minutes) TIMING_URGENT_THRESHOLD = 15 # ≤15 min: Alert icon TIMING_SOON_THRESHOLD = 60 # ≤1 hour: Timer icon diff --git a/custom_components/tibber_prices/sensor/core.py b/custom_components/tibber_prices/sensor/core.py index e36f7e9..1f27ece 100644 --- a/custom_components/tibber_prices/sensor/core.py +++ b/custom_components/tibber_prices/sensor/core.py @@ -7,28 +7,17 @@ from typing import TYPE_CHECKING, Any import yaml -from custom_components.tibber_prices.average_utils import ( - calculate_current_leading_avg, - calculate_current_leading_max, - calculate_current_leading_min, - calculate_current_trailing_avg, - calculate_current_trailing_max, - calculate_current_trailing_min, - calculate_next_n_hours_avg, -) from custom_components.tibber_prices.binary_sensor.attributes import ( get_price_intervals_attributes, ) from custom_components.tibber_prices.const import ( CONF_CHART_DATA_CONFIG, - CONF_EXTENDED_DESCRIPTIONS, CONF_PRICE_RATING_THRESHOLD_HIGH, CONF_PRICE_RATING_THRESHOLD_LOW, CONF_PRICE_TREND_THRESHOLD_FALLING, CONF_PRICE_TREND_THRESHOLD_RISING, CONF_VOLATILITY_THRESHOLD_HIGH, CONF_VOLATILITY_THRESHOLD_MODERATE, - DEFAULT_EXTENDED_DESCRIPTIONS, DEFAULT_PRICE_RATING_THRESHOLD_HIGH, DEFAULT_PRICE_RATING_THRESHOLD_LOW, DEFAULT_PRICE_TREND_THRESHOLD_FALLING, @@ -36,9 +25,9 @@ from custom_components.tibber_prices.const import ( DEFAULT_VOLATILITY_THRESHOLD_HIGH, DEFAULT_VOLATILITY_THRESHOLD_MODERATE, DOMAIN, + MINUTES_PER_INTERVAL, format_price_unit_major, format_price_unit_minor, - get_entity_description, ) from custom_components.tibber_prices.coordinator import ( MINUTE_UPDATE_ENTITY_KEYS, @@ -47,11 +36,21 @@ from custom_components.tibber_prices.coordinator import ( from custom_components.tibber_prices.entity import TibberPricesEntity from custom_components.tibber_prices.entity_utils import ( add_icon_color_attribute, + find_rolling_hour_center_index, get_dynamic_icon, + get_price_value, ) from custom_components.tibber_prices.entity_utils.icons import IconContext -from custom_components.tibber_prices.price_utils import ( - MINUTES_PER_INTERVAL, +from custom_components.tibber_prices.utils.average import ( + calculate_current_leading_avg, + calculate_current_leading_max, + calculate_current_leading_min, + calculate_current_trailing_avg, + calculate_current_trailing_max, + calculate_current_trailing_min, + calculate_next_n_hours_avg, +) +from custom_components.tibber_prices.utils.price import ( calculate_price_trend, calculate_volatility_level, find_price_data_for_interval, @@ -67,6 +66,7 @@ from homeassistant.util import dt as dt_util from .attributes import ( add_volatility_type_attributes, + build_extra_state_attributes, build_sensor_attributes, get_future_prices, get_prices_for_volatility, @@ -75,8 +75,6 @@ from .helpers import ( aggregate_level_data, aggregate_price_data, aggregate_rating_data, - find_rolling_hour_center_index, - get_price_value, ) if TYPE_CHECKING: @@ -1979,77 +1977,36 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity): # Fall back to static icon from entity description return icon or self.entity_description.icon - def _get_description_attributes(self) -> dict[str, str]: - """Get description/long_description/usage_tips attributes.""" - attributes = {} - - if not self.entity_description.translation_key or not self.hass: - return attributes - - language = self.hass.config.language if self.hass.config.language else "en" - - # Add basic description (from cache, synchronous) - description = get_entity_description("sensor", self.entity_description.translation_key, language, "description") - if description: - attributes["description"] = description - - # Check if extended descriptions are enabled - extended_descriptions = self.coordinator.config_entry.options.get( - CONF_EXTENDED_DESCRIPTIONS, - self.coordinator.config_entry.data.get(CONF_EXTENDED_DESCRIPTIONS, DEFAULT_EXTENDED_DESCRIPTIONS), - ) - - if extended_descriptions: - long_desc = get_entity_description( - "sensor", self.entity_description.translation_key, language, "long_description" - ) - if long_desc: - attributes["long_description"] = long_desc - - usage_tips = get_entity_description( - "sensor", self.entity_description.translation_key, language, "usage_tips" - ) - if usage_tips: - attributes["usage_tips"] = usage_tips - - return attributes - @property def extra_state_attributes(self) -> dict[str, Any] | None: """Return additional state attributes.""" - if not self.coordinator.data: + try: + if not self.coordinator.data: + return None + + # Get sensor-specific attributes + sensor_attrs = self._get_sensor_attributes() + + # Build complete attributes using unified builder + return build_extra_state_attributes( + entity_key=self.entity_description.key, + translation_key=self.entity_description.translation_key, + hass=self.hass, + config_entry=self.coordinator.config_entry, + coordinator_data=self.coordinator.data, + sensor_attrs=sensor_attrs, + ) + + except (KeyError, ValueError, TypeError) as ex: + self.coordinator.logger.exception( + "Error getting sensor attributes", + extra={ + "error": str(ex), + "entity": self.entity_description.key, + }, + ) return None - # For chart_data_export: special ordering (metadata → descriptions → service data) - if self.entity_description.key == "chart_data_export": - attributes: dict[str, Any] = {} - chart_attrs = self._get_chart_data_export_attributes() - - # Step 1: Add metadata (timestamp, error) - if chart_attrs: - for key in ("timestamp", "error"): - if key in chart_attrs: - attributes[key] = chart_attrs[key] - - # Step 2: Add descriptions - description_attrs = self._get_description_attributes() - attributes.update(description_attrs) - - # Step 3: Add service data (everything except metadata) - if chart_attrs: - attributes.update({k: v for k, v in chart_attrs.items() if k not in ("timestamp", "error")}) - - return attributes if attributes else None - - # For all other sensors: standard behavior - attributes: dict[str, Any] = self._get_sensor_attributes() or {} - description_attrs = self._get_description_attributes() - # Merge description attributes - for key, value in description_attrs.items(): - attributes[key] = value - - return attributes if attributes else None - def _get_sensor_attributes(self) -> dict | None: """Get attributes based on sensor type.""" key = self.entity_description.key diff --git a/custom_components/tibber_prices/sensor/helpers.py b/custom_components/tibber_prices/sensor/helpers.py index 276e8ab..447294e 100644 --- a/custom_components/tibber_prices/sensor/helpers.py +++ b/custom_components/tibber_prices/sensor/helpers.py @@ -1,23 +1,25 @@ -"""Helper functions for sensor platform.""" +""" +Sensor platform-specific helper functions. + +This module contains helper functions specific to the sensor platform: +- aggregate_price_data: Calculate average price from window data +- aggregate_level_data: Aggregate price levels from intervals +- aggregate_rating_data: Aggregate price ratings from intervals + +For shared helper functions (used by both sensor and binary_sensor platforms), +see entity_utils/helpers.py: +- get_price_value: Price unit conversion +- translate_level: Price level translation +- translate_rating_level: Rating level translation +- find_rolling_hour_center_index: Rolling hour window calculations +""" from __future__ import annotations -from typing import TYPE_CHECKING - -from custom_components.tibber_prices.average_utils import ( - round_to_nearest_quarter_hour, -) -from custom_components.tibber_prices.const import get_price_level_translation -from custom_components.tibber_prices.price_utils import ( +from custom_components.tibber_prices.utils.price import ( aggregate_price_levels, aggregate_price_rating, ) -from homeassistant.util import dt as dt_util - -if TYPE_CHECKING: - from datetime import datetime - - from homeassistant.core import HomeAssistant def aggregate_price_data(window_data: list[dict]) -> float | None: @@ -79,105 +81,3 @@ def aggregate_rating_data( aggregated, _ = aggregate_price_rating(differences, threshold_low, threshold_high) return aggregated.lower() if aggregated else None - - -def find_rolling_hour_center_index( - all_prices: list[dict], - current_time: datetime, - hour_offset: int, -) -> int | None: - """ - Find the center index for the rolling hour window. - - Args: - all_prices: List of all price interval dictionaries with 'startsAt' key - current_time: Current datetime to find the current interval - hour_offset: Number of hours to offset from current interval (can be negative) - - Returns: - Index of the center interval for the rolling hour window, or None if not found - - """ - # Round to nearest interval boundary to handle edge cases where HA schedules - # us slightly before the boundary (e.g., 14:59:59.999 → 15:00:00) - target_time = round_to_nearest_quarter_hour(current_time) - current_idx = None - - for idx, price_data in enumerate(all_prices): - starts_at = dt_util.parse_datetime(price_data["startsAt"]) - if starts_at is None: - continue - starts_at = dt_util.as_local(starts_at) - - # Exact match after rounding - if starts_at == target_time: - current_idx = idx - break - - if current_idx is None: - return None - - return current_idx + (hour_offset * 4) - - -def translate_level(hass: HomeAssistant, level: str) -> str: - """ - Translate price level to the user's language. - - Args: - hass: HomeAssistant instance for language configuration - level: Price level to translate (e.g., VERY_CHEAP, NORMAL, etc.) - - Returns: - Translated level string, or original level if translation not found - - """ - if not hass: - return level - - language = hass.config.language or "en" - translated = get_price_level_translation(level, language) - if translated: - return translated - - if language != "en": - fallback = get_price_level_translation(level, "en") - if fallback: - return fallback - - return level - - -def translate_rating_level(rating: str) -> str: - """ - Translate price rating level to the user's language. - - Args: - rating: Price rating to translate (e.g., LOW, NORMAL, HIGH) - - Returns: - Translated rating string, or original rating if translation not found - - Note: - Currently returns the rating as-is. Translation mapping for ratings - can be added here when needed, similar to translate_level(). - - """ - # For now, ratings are returned as-is - # Add translation mapping here when needed - return rating - - -def get_price_value(price: float, *, in_euro: bool) -> float: - """ - Convert price based on unit. - - Args: - price: Price value to convert - in_euro: If True, return price in euros; if False, return in cents/øre - - Returns: - Price in requested unit (euros or minor currency units) - - """ - return price if in_euro else round((price * 100), 2)