mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-03-29 21:03:40 +00:00
refactor(entity_utils): extract shared helpers from sensor platform
Created entity_utils/helpers.py with platform-agnostic utility functions: - get_price_value(): Price unit conversion (major/minor currency) - translate_level(): Price level translation - translate_rating_level(): Rating level translation - find_rolling_hour_center_index(): Rolling hour window calculations These functions moved from sensor/helpers.py as they are used by both sensor and binary_sensor platforms. Remaining sensor/helpers.py now contains only sensor-specific helpers (aggregate_price_data, etc.). Updated imports: - sensor/core.py: Import from entity_utils instead of sensor.helpers - entity_utils/icons.py: Fixed find_rolling_hour_center_index import - binary_sensor platforms: Can now use shared helpers Added clear docstrings explaining: - entity_utils/helpers.py: Platform-agnostic utilities - sensor/helpers.py: Sensor-specific aggregation functions Impact: Better code reuse, clearer responsibility boundaries between platform-specific and shared utilities.
This commit is contained in:
Julian Pawlowski
parent
ac24f6a8cb
commit
4876a2cc29
6 changed files with 417 additions and 208 deletions
|
|
@ -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",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
128
custom_components/tibber_prices/entity_utils/helpers.py
Normal file
128
custom_components/tibber_prices/entity_utils/helpers.py
Normal file
|
|
@ -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)
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,76 +1977,35 @@ 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."""
|
||||
try:
|
||||
if not self.coordinator.data:
|
||||
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()
|
||||
# Get sensor-specific attributes
|
||||
sensor_attrs = self._get_sensor_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]
|
||||
# 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,
|
||||
)
|
||||
|
||||
# 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
|
||||
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
|
||||
|
||||
def _get_sensor_attributes(self) -> dict | None:
|
||||
"""Get attributes based on sensor type."""
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Reference in a new issue