refactoring

This commit is contained in:
Julian Pawlowski 2025-05-11 13:21:22 +02:00
parent 94ef6ed4a6
commit a4859a9d2e
No known key found for this signature in database
8 changed files with 1011 additions and 358 deletions

View file

@ -14,6 +14,7 @@ from homeassistant.const import EntityCategory
from homeassistant.util import dt as dt_util
from .entity import TibberPricesEntity
from .sensor import detect_interval_granularity, find_price_data_for_interval
if TYPE_CHECKING:
from collections.abc import Callable
@ -26,15 +27,15 @@ if TYPE_CHECKING:
ENTITY_DESCRIPTIONS = (
BinarySensorEntityDescription(
key="peak_hour",
translation_key="peak_hour",
name="Peak Hour",
key="peak_interval",
translation_key="peak_interval",
name="Peak Price Interval",
icon="mdi:clock-alert",
),
BinarySensorEntityDescription(
key="best_price_hour",
translation_key="best_price_hour",
name="Best Price Hour",
key="best_price_interval",
translation_key="best_price_interval",
name="Best Price Interval",
icon="mdi:clock-check",
),
BinarySensorEntityDescription(
@ -81,9 +82,9 @@ class TibberPricesBinarySensor(TibberPricesEntity, BinarySensorEntity):
"""Return the appropriate state getter method based on the sensor type."""
key = self.entity_description.key
if key == "peak_hour":
if key == "peak_interval":
return lambda: self._get_price_threshold_state(threshold_percentage=0.8, high_is_active=True)
if key == "best_price_hour":
if key == "best_price_interval":
return lambda: self._get_price_threshold_state(threshold_percentage=0.2, high_is_active=False)
if key == "connection":
return lambda: True if self.coordinator.data else None
@ -94,45 +95,40 @@ class TibberPricesBinarySensor(TibberPricesEntity, BinarySensorEntity):
"""Return the appropriate attribute getter method based on the sensor type."""
key = self.entity_description.key
if key == "peak_hour":
return lambda: self._get_price_hours_attributes(attribute_name="peak_hours", reverse_sort=True)
if key == "best_price_hour":
return lambda: self._get_price_hours_attributes(attribute_name="best_price_hours", reverse_sort=False)
if key == "peak_interval":
return lambda: self._get_price_intervals_attributes(attribute_name="peak_intervals", reverse_sort=True)
if key == "best_price_interval":
return lambda: self._get_price_intervals_attributes(
attribute_name="best_price_intervals", reverse_sort=False
)
return None
def _get_current_price_data(self) -> tuple[list[float], float] | None:
"""Get current price data if available."""
if not (
self.coordinator.data
and (
today_prices := self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"][
"priceInfo"
].get("today", [])
)
):
if not self.coordinator.data:
return None
price_info = self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"]["priceInfo"]
today_prices = price_info.get("today", [])
if not today_prices:
return None
now = dt_util.now()
# Find price data for current hour
current_hour_data = None
for price_data in today_prices:
starts_at = dt_util.parse_datetime(price_data["startsAt"])
if starts_at is None:
continue
# Detect interval granularity
interval_minutes = detect_interval_granularity(today_prices)
starts_at = dt_util.as_local(starts_at)
if starts_at.hour == now.hour and starts_at.date() == now.date():
current_hour_data = price_data
break
# Find price data for current interval
current_interval_data = find_price_data_for_interval({"today": today_prices}, now, interval_minutes)
if not current_hour_data:
if not current_interval_data:
return None
prices = [float(price["total"]) for price in today_prices]
prices.sort()
return prices, float(current_hour_data["total"])
return prices, float(current_interval_data["total"])
def _get_price_threshold_state(self, *, threshold_percentage: float, high_is_active: bool) -> bool | None:
"""
@ -155,6 +151,73 @@ class TibberPricesBinarySensor(TibberPricesEntity, BinarySensorEntity):
return current_price <= prices[threshold_index]
def _get_price_intervals_attributes(self, *, attribute_name: str, reverse_sort: bool) -> dict | None:
"""
Get price interval attributes with support for 15-minute intervals.
Args:
attribute_name: The attribute name to use in the result dictionary
reverse_sort: Whether to sort prices in reverse (high to low)
Returns:
Dictionary with interval data or None if not available
"""
if not self.coordinator.data:
return None
price_info = self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"]["priceInfo"]
today_prices = price_info.get("today", [])
if not today_prices:
return None
# Detect the granularity of the data
interval_minutes = detect_interval_granularity(today_prices)
# Build a list of price data with timestamps and values
price_intervals = []
for price_data in today_prices:
starts_at = dt_util.parse_datetime(price_data["startsAt"])
if starts_at is None:
continue
starts_at = dt_util.as_local(starts_at)
price_intervals.append(
{
"starts_at": starts_at,
"price": float(price_data["total"]),
"hour": starts_at.hour,
"minute": starts_at.minute,
}
)
# Sort by price (high to low for peak, low to high for best)
sorted_intervals = sorted(price_intervals, key=lambda x: x["price"], reverse=reverse_sort)[:5]
# Format the result based on granularity
hourly_interval_minutes = 60
result = []
for interval in sorted_intervals:
if interval_minutes < hourly_interval_minutes: # More granular than hourly
result.append(
{
"hour": interval["hour"],
"minute": interval["minute"],
"time": f"{interval['hour']:02d}:{interval['minute']:02d}",
"price": interval["price"],
}
)
else: # Hourly data (for backward compatibility)
result.append(
{
"hour": interval["hour"],
"price": interval["price"],
}
)
return {attribute_name: result}
def _get_price_hours_attributes(self, *, attribute_name: str, reverse_sort: bool) -> dict | None:
"""Get price hours attributes."""
if not self.coordinator.data:

View file

@ -42,8 +42,17 @@ PRICE_LEVEL_MAPPING = {
PRICE_LEVEL_VERY_EXPENSIVE: 2,
}
# Sensor type constants
SENSOR_TYPE_PRICE_LEVEL = "price_level"
# Price rating constants
PRICE_RATING_NORMAL = "NORMAL"
PRICE_RATING_LOW = "LOW"
PRICE_RATING_HIGH = "HIGH"
# Mapping for comparing price ratings (used for sorting)
PRICE_RATING_MAPPING = {
PRICE_RATING_LOW: -1,
PRICE_RATING_NORMAL: 0,
PRICE_RATING_HIGH: 1,
}
LOGGER = logging.getLogger(__package__)

View file

@ -1,83 +1,150 @@
{
"sensor": {
"current_price": {
"description": "Der aktuelle Strompreis inklusive Steuern",
"long_description": "Zeigt den Strompreis für die aktuelle Stunde, einschließlich aller Steuern und Gebühren",
"usage_tips": "Verwende diesen Sensor für Automatisierungen, die auf den aktuellen Preis reagieren sollen"
"name": "Aktueller Strompreis",
"description": "Der aktuelle Strompreis in Euro",
"long_description": "Zeigt den aktuellen Preis pro kWh (in Euro) von deinem Tibber-Abonnement an",
"usage_tips": "Nutze dies, um Preise zu verfolgen oder Automatisierungen zu erstellen, die bei günstigem Strom ausgeführt werden"
},
"next_hour_price": {
"description": "Der Strompreis für die nächste Stunde inklusive Steuern",
"long_description": "Zeigt den Strompreis für die kommende Stunde, einschließlich aller Steuern und Gebühren",
"usage_tips": "Perfekt für die Planung von Geräten, die in der nächsten Stunde basierend auf dem Preis laufen sollen"
"current_price_cents": {
"name": "Aktueller Strompreis",
"description": "Der aktuelle Strompreis in Cent pro kWh",
"long_description": "Zeigt den aktuellen Preis pro kWh (in Cent) von deinem Tibber-Abonnement an",
"usage_tips": "Nutze dies, um Preise zu verfolgen oder Automatisierungen zu erstellen, die bei günstigem Strom ausgeführt werden"
},
"next_interval_price": {
"name": "Strompreis nächstes Intervall",
"description": "Der Strompreis für das nächste 15-Minuten-Intervall in Euro",
"long_description": "Zeigt den Preis für das nächste 15-Minuten-Intervall (in Euro) von deinem Tibber-Abonnement an",
"usage_tips": "Nutze dies, um dich auf kommende Preisänderungen vorzubereiten oder Geräte während günstigerer Intervalle zu planen"
},
"next_interval_price_cents": {
"name": "Strompreis nächstes Intervall",
"description": "Der Strompreis für das nächste 15-Minuten-Intervall in Cent pro kWh",
"long_description": "Zeigt den Preis für das nächste 15-Minuten-Intervall (in Cent) von deinem Tibber-Abonnement an",
"usage_tips": "Nutze dies, um dich auf kommende Preisänderungen vorzubereiten oder Geräte während günstigerer Intervalle zu planen"
},
"price_level": {
"description": "Aktueller Preisstandanzeige (SEHR_GÜNSTIG bis SEHR_TEUER)",
"long_description": "Zeigt das aktuelle Preisniveau auf einer Skala von sehr günstig bis sehr teuer an",
"usage_tips": "Verwende dies für visuelle Anzeigen oder einfache Automatisierungen ohne Schwellenwertberechnung",
"name": "Aktuelles Preisniveau",
"description": "Die aktuelle Preislevelklassifikation",
"long_description": "Zeigt die Klassifizierung von Tibber für den aktuellen Preis im Vergleich zu historischen Preisen an",
"usage_tips": "Nutze dies, um Automatisierungen auf Basis des relativen Preisniveaus anstelle der absoluten Preise zu erstellen",
"price_levels": {
"VERY_CHEAP": "Sehr Günstig",
"VERY_CHEAP": "Sehr günstig",
"CHEAP": "Günstig",
"NORMAL": "Normal",
"EXPENSIVE": "Teuer",
"VERY_EXPENSIVE": "Sehr Teuer"
"VERY_EXPENSIVE": "Sehr teuer"
}
},
"lowest_price_today": {
"description": "Der niedrigste Strompreis für den aktuellen Tag",
"long_description": "Zeigt den niedrigsten Preis des aktuellen Tages an",
"usage_tips": "Nützlich, um die optimale Zeit für den Betrieb energieintensiver Geräte zu finden"
"name": "Niedrigster Preis heute",
"description": "Der niedrigste Strompreis für heute in Euro",
"long_description": "Zeigt den niedrigsten Preis pro kWh (in Euro) für den aktuellen Tag von deinem Tibber-Abonnement an",
"usage_tips": "Nutze dies, um aktuelle Preise mit der günstigsten Zeit des Tages zu vergleichen"
},
"lowest_price_today_cents": {
"name": "Niedrigster Preis heute",
"description": "Der niedrigste Strompreis für heute in Cent pro kWh",
"long_description": "Zeigt den niedrigsten Preis pro kWh (in Cent) für den aktuellen Tag von deinem Tibber-Abonnement an",
"usage_tips": "Nutze dies, um aktuelle Preise mit der günstigsten Zeit des Tages zu vergleichen"
},
"highest_price_today": {
"description": "Der höchste Strompreis für den aktuellen Tag",
"long_description": "Zeigt den höchsten Preis während des aktuellen Tages an",
"usage_tips": "Hilfreich, um Spitzenpreiszeiten zu vermeiden"
"name": "Höchster Preis heute",
"description": "Der höchste Strompreis für heute in Euro",
"long_description": "Zeigt den höchsten Preis pro kWh (in Euro) für den aktuellen Tag von deinem Tibber-Abonnement an",
"usage_tips": "Nutze dies, um den Betrieb von Geräten während Spitzenpreiszeiten zu vermeiden"
},
"highest_price_today_cents": {
"name": "Höchster Preis heute",
"description": "Der höchste Strompreis für heute in Cent pro kWh",
"long_description": "Zeigt den höchsten Preis pro kWh (in Cent) für den aktuellen Tag von deinem Tibber-Abonnement an",
"usage_tips": "Nutze dies, um den Betrieb von Geräten während Spitzenpreiszeiten zu vermeiden"
},
"average_price_today": {
"description": "Der durchschnittliche Strompreis für den aktuellen Tag",
"long_description": "Berechnet den durchschnittlichen Preis über alle Stunden des aktuellen Tages",
"usage_tips": "Als Grundlage für Preisvergleiche verwenden"
"name": "Durchschnittspreis heute",
"description": "Der durchschnittliche Strompreis für heute in Euro",
"long_description": "Zeigt den durchschnittlichen Preis pro kWh (in Euro) für den aktuellen Tag von deinem Tibber-Abonnement an",
"usage_tips": "Nutze dies als Grundlage für den Vergleich mit aktuellen Preisen"
},
"hourly_rating": {
"description": "Preisvergleich mit historischen Daten für die aktuelle Stunde",
"long_description": "Zeigt, wie der Preis der aktuellen Stunde im Vergleich zu historischen Daten als prozentuale Differenz abschneidet",
"usage_tips": "Hilft zu verstehen, ob die aktuellen Preise höher oder niedriger als üblich für diese Zeit sind"
"average_price_today_cents": {
"name": "Durchschnittspreis heute",
"description": "Der durchschnittliche Strompreis für heute in Cent pro kWh",
"long_description": "Zeigt den durchschnittlichen Preis pro kWh (in Cent) für den aktuellen Tag von deinem Tibber-Abonnement an",
"usage_tips": "Nutze dies als Grundlage für den Vergleich mit aktuellen Preisen"
},
"price_rating": {
"name": "Aktuelle Preisbewertung",
"description": "Wie sich der Preis des aktuellen Intervalls mit historischen Daten vergleicht",
"long_description": "Zeigt, wie sich der Preis des aktuellen Intervalls im Vergleich zu historischen Preisdaten als Prozentsatz verhält",
"usage_tips": "Ein positiver Prozentsatz bedeutet, dass der aktuelle Preis überdurchschnittlich ist, negativ bedeutet unterdurchschnittlich",
"price_levels": {
"LOW": "Niedrig",
"NORMAL": "Normal",
"HIGH": "Hoch"
}
},
"daily_rating": {
"description": "Preisvergleich mit historischen Daten für den aktuellen Tag",
"long_description": "Zeigt, wie der heutige Durchschnittspreis im Vergleich zu historischen Daten als prozentuale Differenz abschneidet",
"usage_tips": "Nützlich, um zu verstehen, ob heute generell teuer oder günstig ist"
"name": "Tägliche Preisbewertung",
"description": "Wie sich die heutigen Preise mit historischen Daten vergleichen",
"long_description": "Zeigt, wie sich die heutigen Preise im Vergleich zu historischen Preisdaten als Prozentsatz verhällt",
"usage_tips": "Ein positiver Prozentsatz bedeutet, dass die heutigen Preise überdurchschnittlich sind, negativ bedeutet unterdurchschnittlich"
},
"monthly_rating": {
"description": "Preisvergleich mit historischen Daten für den aktuellen Monat",
"long_description": "Zeigt, wie der durchschnittliche Preis dieses Monats im Vergleich zu historischen Daten als prozentuale Differenz abschneidet",
"usage_tips": "Hilfreich für die langfristige Energiebudgetplanung"
"name": "Monatliche Preisbewertung",
"description": "Wie sich die Preise dieses Monats mit historischen Daten vergleichen",
"long_description": "Zeigt, wie sich die Preise dieses Monats im Vergleich zu historischen Preisdaten als Prozentsatz verhällt",
"usage_tips": "Ein positiver Prozentsatz bedeutet, dass die Preise dieses Monats überdurchschnittlich sind, negativ bedeutet unterdurchschnittlich"
},
"data_timestamp": {
"description": "Zeitstempel der neuesten Preisdaten von Tibber",
"long_description": "Zeigt an, wann die Preisdaten zuletzt von der Tibber API aktualisiert wurden",
"usage_tips": "Überwache dies, um sicherzustellen, dass Ihre Preisdaten aktuell sind"
"name": "Zeitstempel der neuesten Daten",
"description": "Wann die neuesten Preisdaten empfangen wurden",
"long_description": "Zeigt den Zeitstempel des letzten Preisaktualisierung von Tibber",
"usage_tips": "Nutze dies, um zu überprüfen, wann die Preisinformationen zuletzt aktualisiert wurden"
},
"tomorrow_data_available": {
"description": "Zeigt an, ob Preisdaten für morgen verfügbar sind",
"long_description": "Zeigt an, ob vollständige, teilweise oder keine Preisdaten für morgen verfügbar sind",
"usage_tips": "Verwende dies, um zu prüfen, ob Geräte für morgen zuverlässig geplant werden können"
"name": "Datenstatus für morgen",
"description": "Ob Preisdaten für morgen verfügbar sind",
"long_description": "Gibt an, ob Preisdaten für den folgenden Tag von Tibber empfangen wurden",
"usage_tips": "Nutze dies, um zu überprüfen, ob die Preise für morgen für die Planung verfügbar sind"
},
"price_forecast": {
"name": "Preisprognose",
"description": "Prognose der kommenden Strompreise",
"long_description": "Zeigt kommende Strompreise für zukünftige Intervalle in einem Format an, das leicht in Dashboards verwendet werden kann",
"usage_tips": "Verwende die Attribute dieser Entität, um bevorstehende Preise in Diagrammen oder benutzerdefinierten Karten anzuzeigen. Greife entweder auf 'intervals' für alle zukünftigen Intervalle oder auf 'hours' für stündliche Zusammenfassungen zu."
}
},
"binary_sensor": {
"peak_interval": {
"name": "Spitzenpreis-Intervall",
"description": "Ob das aktuelle Intervall zu den teuersten des Tages gehört",
"long_description": "Wird aktiviert, wenn der aktuelle Preis in den oberen 20% der heutigen Preise liegt",
"usage_tips": "Nutze dies, um den Betrieb von Geräten mit hohem Verbrauch während teurer Intervalle zu vermeiden"
},
"best_price_interval": {
"name": "Bestpreis-Intervall",
"description": "Ob das aktuelle Intervall zu den günstigsten des Tages gehört",
"long_description": "Wird aktiviert, wenn der aktuelle Preis in den unteren 20% der heutigen Preise liegt",
"usage_tips": "Nutze dies, um Geräte mit hohem Verbrauch während der günstigsten Intervalle zu betreiben"
},
"peak_hour": {
"description": "Zeigt an, ob die aktuelle Stunde den höchsten Preis des Tages hat",
"long_description": "Wird während Stunden aktiv, die zu den teuersten des Tages gehören",
"usage_tips": "In Automatisierungen verwenden, um den Betrieb von Geräten mit hohem Verbrauch während Spitzenzeiten zu vermeiden"
"name": "Spitzenstunde",
"description": "Ob die aktuelle Stunde zu den teuersten des Tages gehört",
"long_description": "Wird aktiviert, wenn der aktuelle Preis in den oberen 20% der heutigen Preise liegt",
"usage_tips": "Nutze dies, um den Betrieb von Geräten mit hohem Verbrauch während teurer Stunden zu vermeiden"
},
"best_price_hour": {
"description": "Zeigt an, ob die aktuelle Stunde den niedrigsten Preis des Tages hat",
"long_description": "Wird während Stunden aktiv, die zu den günstigsten des Tages gehören",
"usage_tips": "Perfekt, um energieintensive Geräte zu optimalen Zeiten zu aktivieren"
"name": "Beste-Preis-Stunde",
"description": "Ob die aktuelle Stunde zu den günstigsten des Tages gehört",
"long_description": "Wird aktiviert, wenn der aktuelle Preis in den unteren 20% der heutigen Preise liegt",
"usage_tips": "Nutze dies, um Geräte mit hohem Verbrauch während der günstigsten Stunden zu betreiben"
},
"connection": {
"description": "Zeigt den Verbindungsstatus zur Tibber API an",
"long_description": "Zeigt an, ob die Komponente erfolgreich eine Verbindung zur Tibber API herstellt",
"usage_tips": "Überwache dies, um sicherzustellen, dass die Preisdaten korrekt aktualisiert werden"
"name": "Tibber API-Verbindung",
"description": "Ob die Verbindung zur Tibber API funktioniert",
"long_description": "Zeigt an, ob die Integration erfolgreich eine Verbindung zur Tibber API herstellen kann",
"usage_tips": "Nutze dies, um den Verbindungsstatus zur Tibber API zu überwachen"
}
}
}

View file

@ -1,19 +1,34 @@
{
"sensor": {
"current_price": {
"description": "The current hour's electricity price including taxes",
"long_description": "Shows the electricity price for the current hour, including all taxes and fees",
"usage_tips": "Use this sensor for automations that should react to the current price"
"name": "Current Electricity Price",
"description": "The current electricity price in Euro",
"long_description": "Shows the current price per kWh (in Euro) from your Tibber subscription",
"usage_tips": "Use this to track prices or to create automations that run when electricity is cheap"
},
"next_hour_price": {
"description": "The next hour's electricity price including taxes",
"long_description": "Shows the electricity price for the upcoming hour, including all taxes and fees",
"usage_tips": "Perfect for scheduling devices to run in the next hour based on price"
"current_price_cents": {
"name": "Current Electricity Price",
"description": "The current electricity price in cents per kWh",
"long_description": "Shows the current price per kWh (in cents) from your Tibber subscription",
"usage_tips": "Use this to track prices or to create automations that run when electricity is cheap"
},
"next_interval_price": {
"name": "Next Interval Electricity Price",
"description": "The next interval electricity price in Euro",
"long_description": "Shows the price for the next 15-minute interval (in Euro) from your Tibber subscription",
"usage_tips": "Use this to prepare for upcoming price changes or to schedule devices to run during cheaper intervals"
},
"next_interval_price_cents": {
"name": "Next Interval Electricity Price",
"description": "The next interval electricity price in cents per kWh",
"long_description": "Shows the price for the next 15-minute interval (in cents) from your Tibber subscription",
"usage_tips": "Use this to prepare for upcoming price changes or to schedule devices to run during cheaper intervals"
},
"price_level": {
"description": "Current price level indicator (VERY_CHEAP to VERY_EXPENSIVE)",
"long_description": "Indicates the current price level on a scale from very cheap to very expensive",
"usage_tips": "Use this for visual indicators or simple automations without needing to calculate thresholds",
"name": "Current Price Level",
"description": "The current price level classification",
"long_description": "Shows Tibber's classification of the current price compared to historical prices",
"usage_tips": "Use this to create automations based on relative price levels rather than absolute prices",
"price_levels": {
"VERY_CHEAP": "Very Cheap",
"CHEAP": "Cheap",
@ -23,61 +38,113 @@
}
},
"lowest_price_today": {
"description": "The lowest electricity price for the current day",
"long_description": "Shows the lowest price point available during the current day",
"usage_tips": "Useful to find the optimal time to run energy-intensive appliances"
"name": "Today's Lowest Price",
"description": "The lowest electricity price for today in Euro",
"long_description": "Shows the lowest price per kWh (in Euro) for the current day from your Tibber subscription",
"usage_tips": "Use this to compare current prices to the cheapest time of the day"
},
"lowest_price_today_cents": {
"name": "Today's Lowest Price",
"description": "The lowest electricity price for today in cents per kWh",
"long_description": "Shows the lowest price per kWh (in cents) for the current day from your Tibber subscription",
"usage_tips": "Use this to compare current prices to the cheapest time of the day"
},
"highest_price_today": {
"description": "The highest electricity price for the current day",
"long_description": "Shows the highest price point during the current day",
"usage_tips": "Helpful for avoiding peak price periods"
"name": "Today's Highest Price",
"description": "The highest electricity price for today in Euro",
"long_description": "Shows the highest price per kWh (in Euro) for the current day from your Tibber subscription",
"usage_tips": "Use this to avoid running appliances during peak price times"
},
"highest_price_today_cents": {
"name": "Today's Highest Price",
"description": "The highest electricity price for today in cents per kWh",
"long_description": "Shows the highest price per kWh (in cents) for the current day from your Tibber subscription",
"usage_tips": "Use this to avoid running appliances during peak price times"
},
"average_price_today": {
"description": "The average electricity price for the current day",
"long_description": "Calculates the average price across all hours of the current day",
"usage_tips": "Use as a baseline for price comparison"
"name": "Today's Average Price",
"description": "The average electricity price for today in Euro",
"long_description": "Shows the average price per kWh (in Euro) for the current day from your Tibber subscription",
"usage_tips": "Use this as a baseline for comparing current prices"
},
"hourly_rating": {
"description": "Price comparison with historical data for the current hour",
"long_description": "Shows how the current hour's price compares to historical data as a percentage difference",
"usage_tips": "Helps understand if current prices are higher or lower than usual for this time"
"average_price_today_cents": {
"name": "Today's Average Price",
"description": "The average electricity price for today in cents per kWh",
"long_description": "Shows the average price per kWh (in cents) for the current day from your Tibber subscription",
"usage_tips": "Use this as a baseline for comparing current prices"
},
"price_rating": {
"name": "Current Price Rating",
"description": "How the current interval's price compares to historical data",
"long_description": "Shows how the current interval's price compares to historical price data as a percentage",
"usage_tips": "A positive percentage means the current price is above average, negative means below average",
"price_levels": {
"LOW": "Low",
"NORMAL": "Normal",
"HIGH": "High"
}
},
"daily_rating": {
"description": "Price comparison with historical data for the current day",
"long_description": "Shows how today's average price compares to historical data as a percentage difference",
"usage_tips": "Useful to understand if today is generally expensive or cheap"
"name": "Daily Price Rating",
"description": "How today's prices compare to historical data",
"long_description": "Shows how today's prices compare to historical price data as a percentage",
"usage_tips": "A positive percentage means today's prices are above average, negative means below average"
},
"monthly_rating": {
"description": "Price comparison with historical data for the current month",
"long_description": "Shows how this month's average price compares to historical data as a percentage difference",
"usage_tips": "Helpful for long-term energy budget planning"
"name": "Monthly Price Rating",
"description": "How this month's prices compare to historical data",
"long_description": "Shows how this month's prices compare to historical price data as a percentage",
"usage_tips": "A positive percentage means this month's prices are above average, negative means below average"
},
"data_timestamp": {
"description": "Timestamp of the most recent price data received from Tibber",
"long_description": "Shows when the price data was last updated from the Tibber API",
"usage_tips": "Monitor this to ensure your price data is current"
"name": "Latest Data Timestamp",
"description": "When the latest price data was received",
"long_description": "Shows the timestamp of the most recent price data update from Tibber",
"usage_tips": "Use this to check when price information was last updated"
},
"tomorrow_data_available": {
"description": "Indicates if price data for tomorrow is available",
"long_description": "Shows whether complete, partial, or no price data is available for tomorrow",
"usage_tips": "Use this to check if you can schedule appliances for tomorrow reliably"
"name": "Tomorrow's Data Status",
"description": "Whether price data for tomorrow is available",
"long_description": "Indicates if price data for the following day has been received from Tibber",
"usage_tips": "Use this to check if tomorrow's prices are available for planning"
},
"price_forecast": {
"name": "Price Forecast",
"description": "Forecast of upcoming electricity prices",
"long_description": "Shows upcoming electricity prices for future intervals in a format that's easy to use in dashboards",
"usage_tips": "Use this entity's attributes to display upcoming prices in charts or custom cards. Access either 'intervals' for all future intervals or 'hours' for hourly summaries."
}
},
"binary_sensor": {
"peak_interval": {
"name": "Peak Price Interval",
"description": "Whether the current interval is among the most expensive of the day",
"long_description": "Turns on when the current price is in the top 20% of today's prices",
"usage_tips": "Use this to avoid running high-consumption appliances during expensive intervals"
},
"best_price_interval": {
"name": "Best Price Interval",
"description": "Whether the current interval is among the cheapest of the day",
"long_description": "Turns on when the current price is in the bottom 20% of today's prices",
"usage_tips": "Use this to run high-consumption appliances during the cheapest intervals"
},
"peak_hour": {
"description": "Indicates whether the current hour has the highest price of the day",
"long_description": "Becomes active during hours that are among the most expensive of the day",
"usage_tips": "Use in automations to avoid running high-consumption devices during peak hours"
"name": "Peak Hour",
"description": "Whether the current hour is among the most expensive of the day",
"long_description": "Turns on when the current price is in the top 20% of today's prices",
"usage_tips": "Use this to avoid running high-consumption appliances during expensive hours"
},
"best_price_hour": {
"description": "Indicates whether the current hour has the lowest price of the day",
"long_description": "Becomes active during hours that are among the cheapest of the day",
"usage_tips": "Perfect for triggering energy-intensive appliances during optimal times"
"name": "Best Price Hour",
"description": "Whether the current hour is among the cheapest of the day",
"long_description": "Turns on when the current price is in the bottom 20% of today's prices",
"usage_tips": "Use this to run high-consumption appliances during the cheapest hours"
},
"connection": {
"description": "Shows connection status to the Tibber API",
"long_description": "Indicates whether the component is successfully connecting to the Tibber API",
"usage_tips": "Monitor this to ensure your price data is being updated correctly"
"name": "Tibber API Connection",
"description": "Whether the connection to the Tibber API is working",
"long_description": "Indicates if the integration can successfully connect to the Tibber API",
"usage_tips": "Use this to monitor the connection status to the Tibber API"
}
}
}

View file

@ -15,9 +15,7 @@ if TYPE_CHECKING:
TO_REDACT = {"access_token"}
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: ConfigEntry
) -> dict[str, Any]:
async def async_get_config_entry_diagnostics(hass: HomeAssistant, entry: ConfigEntry) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
coordinator = hass.data[DOMAIN][entry.entry_id].coordinator
@ -26,9 +24,7 @@ async def async_get_config_entry_diagnostics(
"coordinator_data": coordinator.data,
"last_update_success": coordinator.last_update_success,
"update_timestamps": {
"price": coordinator.last_price_update.isoformat()
if coordinator.last_price_update
else None,
"price": coordinator.last_price_update.isoformat() if coordinator.last_price_update else None,
"hourly_rating": coordinator.last_rating_update_hourly.isoformat()
if coordinator.last_rating_update_hourly
else None,

View file

@ -13,7 +13,7 @@ from homeassistant.components.sensor import (
SensorEntity,
SensorEntityDescription,
)
from homeassistant.const import CURRENCY_EURO, EntityCategory
from homeassistant.const import CURRENCY_CENT, CURRENCY_EURO, PERCENTAGE, EntityCategory, UnitOfPower, UnitOfTime
from homeassistant.util import dt as dt_util
from .const import (
@ -21,7 +21,7 @@ from .const import (
DEFAULT_EXTENDED_DESCRIPTIONS,
DOMAIN,
PRICE_LEVEL_MAPPING,
SENSOR_TYPE_PRICE_LEVEL,
PRICE_RATING_MAPPING,
async_get_entity_description,
get_entity_description,
)
@ -36,52 +36,56 @@ if TYPE_CHECKING:
from .coordinator import TibberPricesDataUpdateCoordinator
from .data import TibberPricesConfigEntry
PRICE_UNIT = "ct/kWh"
PRICE_UNIT_CENT = CURRENCY_CENT + "/" + UnitOfPower.KILO_WATT + UnitOfTime.HOURS
PRICE_UNIT_EURO = CURRENCY_EURO + "/" + UnitOfPower.KILO_WATT + UnitOfTime.HOURS
HOURS_IN_DAY = 24
LAST_HOUR_OF_DAY = 23
INTERVALS_PER_HOUR = 4 # 15-minute intervals
MINUTES_PER_INTERVAL = 15
MAX_FORECAST_INTERVALS = 8 # Show up to 8 future intervals (2 hours with 15-min intervals)
# Main price sensors that users will typically use in automations
PRICE_SENSORS = (
SensorEntityDescription(
key="current_price_eur",
translation_key="current_price",
name="Current Electricity Price",
icon="mdi:currency-eur",
device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement=CURRENCY_EURO,
entity_registry_enabled_default=False,
suggested_display_precision=2,
),
SensorEntityDescription(
key="current_price",
translation_key="current_price_cents",
name="Current Electricity Price",
icon="mdi:currency-eur",
device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement="ct/kWh",
suggested_display_precision=2,
native_unit_of_measurement=PRICE_UNIT_CENT,
suggested_display_precision=1,
),
SensorEntityDescription(
key="next_hour_price_eur",
translation_key="next_hour_price",
name="Next Hour Electricity Price",
icon="mdi:currency-eur-off",
key="current_price_eur",
translation_key="current_price",
name="Current Electricity Price",
icon="mdi:currency-eur",
device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement=CURRENCY_EURO,
native_unit_of_measurement=PRICE_UNIT_EURO,
entity_registry_enabled_default=False,
suggested_display_precision=2,
),
SensorEntityDescription(
key="next_hour_price",
translation_key="next_hour_price_cents",
name="Next Hour Electricity Price",
key="next_interval_price",
translation_key="next_interval_price_cents",
name="Next Interval Electricity Price",
icon="mdi:currency-eur-off",
device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement="ct/kWh",
native_unit_of_measurement=PRICE_UNIT_CENT,
suggested_display_precision=1,
),
SensorEntityDescription(
key="next_interval_price_eur",
translation_key="next_interval_price",
name="Next Interval Electricity Price",
icon="mdi:currency-eur-off",
device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement=PRICE_UNIT_EURO,
entity_registry_enabled_default=False,
suggested_display_precision=2,
),
SensorEntityDescription(
key=SENSOR_TYPE_PRICE_LEVEL,
key="price_level",
translation_key="price_level",
name="Current Price Level",
icon="mdi:meter-electric",
@ -90,32 +94,22 @@ PRICE_SENSORS = (
# Statistical price sensors
STATISTICS_SENSORS = (
SensorEntityDescription(
key="lowest_price_today_eur",
translation_key="lowest_price_today",
name="Today's Lowest Price",
icon="mdi:currency-eur",
device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement=CURRENCY_EURO,
entity_registry_enabled_default=False,
suggested_display_precision=2,
),
SensorEntityDescription(
key="lowest_price_today",
translation_key="lowest_price_today_cents",
name="Today's Lowest Price",
icon="mdi:currency-eur",
device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement="ct/kWh",
suggested_display_precision=2,
native_unit_of_measurement=PRICE_UNIT_CENT,
suggested_display_precision=1,
),
SensorEntityDescription(
key="highest_price_today_eur",
translation_key="highest_price_today",
name="Today's Highest Price",
key="lowest_price_today_eur",
translation_key="lowest_price_today",
name="Today's Lowest Price",
icon="mdi:currency-eur",
device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement=CURRENCY_EURO,
native_unit_of_measurement=PRICE_UNIT_EURO,
entity_registry_enabled_default=False,
suggested_display_precision=2,
),
@ -125,16 +119,16 @@ STATISTICS_SENSORS = (
name="Today's Highest Price",
icon="mdi:currency-eur",
device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement="ct/kWh",
suggested_display_precision=2,
native_unit_of_measurement=PRICE_UNIT_CENT,
suggested_display_precision=1,
),
SensorEntityDescription(
key="average_price_today_eur",
translation_key="average_price_today",
name="Today's Average Price",
key="highest_price_today_eur",
translation_key="highest_price_today",
name="Today's Highest Price",
icon="mdi:currency-eur",
device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement=CURRENCY_EURO,
native_unit_of_measurement=PRICE_UNIT_EURO,
entity_registry_enabled_default=False,
suggested_display_precision=2,
),
@ -144,7 +138,17 @@ STATISTICS_SENSORS = (
name="Today's Average Price",
icon="mdi:currency-eur",
device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement="ct/kWh",
native_unit_of_measurement=PRICE_UNIT_CENT,
suggested_display_precision=1,
),
SensorEntityDescription(
key="average_price_today_eur",
translation_key="average_price_today",
name="Today's Average Price",
icon="mdi:currency-eur",
device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement=PRICE_UNIT_EURO,
entity_registry_enabled_default=False,
suggested_display_precision=2,
),
)
@ -152,28 +156,22 @@ STATISTICS_SENSORS = (
# Rating sensors
RATING_SENSORS = (
SensorEntityDescription(
key="hourly_rating",
translation_key="hourly_rating",
name="Hourly Price Rating",
key="price_rating",
translation_key="price_rating",
name="Current Price Rating",
icon="mdi:clock-outline",
native_unit_of_measurement="%",
suggested_display_precision=1,
),
SensorEntityDescription(
key="daily_rating",
translation_key="daily_rating",
name="Daily Price Rating",
icon="mdi:calendar-today",
native_unit_of_measurement="%",
suggested_display_precision=1,
),
SensorEntityDescription(
key="monthly_rating",
translation_key="monthly_rating",
name="Monthly Price Rating",
icon="mdi:calendar-month",
native_unit_of_measurement="%",
suggested_display_precision=1,
),
)
@ -194,6 +192,13 @@ DIAGNOSTIC_SENSORS = (
icon="mdi:calendar-check",
entity_category=EntityCategory.DIAGNOSTIC,
),
SensorEntityDescription(
key="price_forecast",
translation_key="price_forecast",
name="Price Forecast",
icon="mdi:chart-line",
entity_category=EntityCategory.DIAGNOSTIC,
),
)
# Combine all sensors
@ -242,12 +247,12 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
# Map sensor keys to their handler methods
handlers = {
# Price level
SENSOR_TYPE_PRICE_LEVEL: self._get_price_level_value,
"price_level": self._get_price_level_value,
# Price sensors
"current_price": lambda: self._get_hourly_price_value(hour_offset=0, in_euro=False),
"current_price_eur": lambda: self._get_hourly_price_value(hour_offset=0, in_euro=True),
"next_hour_price": lambda: self._get_hourly_price_value(hour_offset=1, in_euro=False),
"next_hour_price_eur": lambda: self._get_hourly_price_value(hour_offset=1, in_euro=True),
"current_price": lambda: self._get_interval_price_value(interval_offset=0, in_euro=False),
"current_price_eur": lambda: self._get_interval_price_value(interval_offset=0, in_euro=True),
"next_interval_price": lambda: self._get_interval_price_value(interval_offset=1, in_euro=False),
"next_interval_price_eur": lambda: self._get_interval_price_value(interval_offset=1, in_euro=True),
# Statistics sensors
"lowest_price_today": lambda: self._get_statistics_value(stat_func=min, in_euro=False, decimals=2),
"lowest_price_today_eur": lambda: self._get_statistics_value(stat_func=min, in_euro=True, decimals=4),
@ -260,43 +265,59 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
stat_func=lambda prices: sum(prices) / len(prices), in_euro=True, decimals=4
),
# Rating sensors
"hourly_rating": lambda: self._get_rating_value(rating_type="hourly"),
"price_rating": lambda: self._get_rating_value(rating_type="hourly"),
"daily_rating": lambda: self._get_rating_value(rating_type="daily"),
"monthly_rating": lambda: self._get_rating_value(rating_type="monthly"),
# Diagnostic sensors
"data_timestamp": self._get_data_timestamp,
"tomorrow_data_available": self._get_tomorrow_data_status,
# Price forecast sensor
"price_forecast": self._get_price_forecast_value,
}
return handlers.get(key)
def _get_current_hour_data(self) -> dict | None:
"""Get the price data for the current hour."""
def _get_current_interval_data(self) -> dict | None:
"""Get the price data for the current interval using adaptive interval detection."""
if not self.coordinator.data:
return None
# Use HomeAssistant's dt_util to get the current time in the user's timezone
# Get the current time and price info
now = dt_util.now()
price_info = self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"]["priceInfo"]
for price_data in price_info.get("today", []):
# Parse the timestamp and convert to local time
starts_at = dt_util.parse_datetime(price_data["startsAt"])
if starts_at is None:
continue
# Make sure it's in the local timezone for proper comparison
starts_at = dt_util.as_local(starts_at)
if starts_at.hour == now.hour and starts_at.date() == now.date():
return price_data
return None
# Use our adaptive price data finder
return find_price_data_for_interval(price_info, now)
def _get_price_level_value(self) -> str | None:
"""Get the current price level value."""
current_hour_data = self._get_current_hour_data()
return current_hour_data["level"] if current_hour_data else None
"""
Get the current price level value as a translated string for the state.
The original (raw) value is stored for use as an attribute.
Returns:
The translated price level value for the state, or None if unavailable.
"""
current_interval_data = self._get_current_interval_data()
if not current_interval_data or "level" not in current_interval_data:
self._last_price_level = None
return None
level = current_interval_data["level"]
self._last_price_level = level
# Use the translation helper for price level, fallback to English if needed
if self.hass:
language = self.hass.config.language or "en"
from .const import get_price_level_translation
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 _get_price_value(self, price: float, *, in_euro: bool) -> float:
"""Convert price based on unit."""
@ -348,10 +369,51 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
return None
def _get_interval_price_value(self, *, interval_offset: int, in_euro: bool) -> float | None:
"""
Get price for the current interval or with offset, handling different interval granularities.
Args:
interval_offset: Number of intervals to offset from current time
in_euro: Whether to return value in EUR (True) or cents/kWh (False)
Returns:
Price value in the requested unit or None if not available
"""
if not self.coordinator.data:
return None
price_info = self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"]["priceInfo"]
# Determine data granularity
today_prices = price_info.get("today", [])
data_granularity = detect_interval_granularity(today_prices) if today_prices else MINUTES_PER_INTERVAL
# Use HomeAssistant's dt_util to get the current time in the user's timezone
now = dt_util.now()
# Calculate the target time based on detected granularity
target_datetime = now + timedelta(minutes=interval_offset * data_granularity)
# Find appropriate price data
price_data = find_price_data_for_interval(price_info, target_datetime, data_granularity)
if price_data:
return self._get_price_value(float(price_data["total"]), in_euro=in_euro)
return None
def _get_statistics_value(
self, *, stat_func: Callable[[list[float]], float], in_euro: bool, decimals: int | None = None
) -> float | None:
"""Handle statistics sensor values using the provided statistical function."""
"""
Handle statistics sensor values using the provided statistical function.
Returns:
The calculated value for the statistics sensor, or None if unavailable.
"""
if not self.coordinator.data:
return None
@ -371,37 +433,105 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
result = round(result, decimals)
return result
def _get_rating_value(self, *, rating_type: str) -> float | None:
"""Handle rating sensor values."""
if not self.coordinator.data:
return None
def _translate_rating_level(self, level: str) -> str:
"""Translate the rating level using custom translations, falling back to English or the raw value."""
if not self.hass or not level:
return level
language = self.hass.config.language or "en"
cache_key = f"{DOMAIN}_translations_{language}"
translations = self.hass.data.get(cache_key)
if (
translations
and "sensor" in translations
and "price_rating" in translations["sensor"]
and "price_levels" in translations["sensor"]["price_rating"]
and level in translations["sensor"]["price_rating"]["price_levels"]
):
return translations["sensor"]["price_rating"]["price_levels"][level]
# Fallback to English if not found
if language != "en":
en_cache_key = f"{DOMAIN}_translations_en"
en_translations = self.hass.data.get(en_cache_key)
if (
en_translations
and "sensor" in en_translations
and "price_rating" in en_translations["sensor"]
and "price_levels" in en_translations["sensor"]["price_rating"]
and level in en_translations["sensor"]["price_rating"]["price_levels"]
):
return en_translations["sensor"]["price_rating"]["price_levels"][level]
return level
def _find_rating_entry(
self, entries: list[dict], now: datetime, rating_type: str, subscription: dict
) -> dict | None:
"""Find the correct rating entry for the given type and time."""
if not entries:
return None
predicate = None
if rating_type == "hourly":
price_info = subscription.get("priceInfo", {})
today_prices = price_info.get("today", [])
data_granularity = detect_interval_granularity(today_prices) if today_prices else MINUTES_PER_INTERVAL
def interval_predicate(entry_time: datetime) -> bool:
interval_end = entry_time + timedelta(minutes=data_granularity)
return entry_time <= now < interval_end and entry_time.date() == now.date()
predicate = interval_predicate
elif rating_type == "daily":
def daily_predicate(entry_time: datetime) -> bool:
return dt_util.as_local(entry_time).date() == now.date()
predicate = daily_predicate
elif rating_type == "monthly":
def monthly_predicate(entry_time: datetime) -> bool:
local_time = dt_util.as_local(entry_time)
return local_time.month == now.month and local_time.year == now.year
predicate = monthly_predicate
if predicate:
for entry in entries:
entry_time = dt_util.parse_datetime(entry["time"])
if entry_time and predicate(entry_time):
return entry
# For hourly, fallback to hour match if not found
if rating_type == "hourly":
for entry in entries:
entry_time = dt_util.parse_datetime(entry["time"])
if entry_time:
entry_time = dt_util.as_local(entry_time)
if entry_time.hour == now.hour and entry_time.date() == now.date():
return entry
return None
def _get_rating_value(self, *, rating_type: str) -> str | None:
"""
Handle rating sensor values for hourly, daily, and monthly ratings.
Returns the translated rating level as the main status, and stores the original
level and percentage difference as attributes.
"""
if not self.coordinator.data:
self._last_rating_difference = None
self._last_rating_level = None
return None
subscription = self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"]
price_rating = subscription.get("priceRating", {}) or {}
now = dt_util.now()
rating_data = price_rating.get(rating_type, {})
entries = rating_data.get("entries", []) if rating_data else []
match_conditions = {
"hourly": lambda et: et.hour == now.hour and et.date() == now.date(),
"daily": lambda et: et.date() == now.date(),
"monthly": lambda et: et.month == now.month and et.year == now.year,
}
match_func = match_conditions.get(rating_type)
if not match_func:
return None
for entry in entries:
entry_time = dt_util.parse_datetime(entry["time"])
if entry_time is None:
continue
entry_time = dt_util.as_local(entry_time)
if match_func(entry_time):
return float(entry["difference"])
entry = self._find_rating_entry(entries, now, rating_type, dict(subscription))
if entry:
difference = entry.get("difference")
level = entry.get("level")
self._last_rating_difference = float(difference) if difference is not None else None
self._last_rating_level = level if level is not None else None
return self._translate_rating_level(level or "")
self._last_rating_difference = None
self._last_rating_level = None
return None
def _get_data_timestamp(self) -> datetime | None:
@ -432,12 +562,197 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
return "No"
return "Yes" if len(tomorrow_prices) == HOURS_IN_DAY else "Partial"
# Add method to get future price intervals
def _get_price_forecast_value(self) -> str | None:
"""Get the highest or lowest price status for the price forecast entity."""
future_prices = self._get_future_prices(max_intervals=MAX_FORECAST_INTERVALS)
if not future_prices:
return "No forecast data available"
# Return a simple status message indicating how much forecast data is available
return f"Forecast available for {len(future_prices)} intervals"
def _get_future_prices(self, max_intervals: int | None = None) -> list[dict] | None:
"""
Get future price data for multiple upcoming intervals.
Args:
max_intervals: Maximum number of future intervals to return
Returns:
List of upcoming price intervals with timestamps and prices
"""
if not self.coordinator.data:
return None
price_info = self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"]["priceInfo"]
price_rating = (
self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"].get("priceRating", {}) or {}
)
# Determine data granularity from the current price data
today_prices = price_info.get("today", [])
tomorrow_prices = price_info.get("tomorrow", [])
all_prices = today_prices + tomorrow_prices
if not all_prices:
return None
data_granularity = detect_interval_granularity(all_prices)
now = dt_util.now()
# Initialize the result list
future_prices = []
# Track the maximum intervals to return
intervals_to_return = MAX_FORECAST_INTERVALS if max_intervals is None else max_intervals
# Extract hourly rating data for enriching the forecast
rating_data = {}
hourly_rating = price_rating.get("hourly", {})
if hourly_rating and "entries" in hourly_rating:
for entry in hourly_rating.get("entries", []):
if entry.get("time"):
timestamp = dt_util.parse_datetime(entry["time"])
if timestamp:
timestamp = dt_util.as_local(timestamp)
# Store with ISO format key for easier lookup
time_key = timestamp.replace(second=0, microsecond=0).isoformat()
rating_data[time_key] = {
"difference": float(entry.get("difference", 0)),
"rating_level": entry.get("level"),
}
# Create a list of all future price data points
for day_key in ["today", "tomorrow"]:
for price_data in price_info.get(day_key, []):
starts_at = dt_util.parse_datetime(price_data["startsAt"])
if starts_at is None:
continue
starts_at = dt_util.as_local(starts_at)
interval_end = starts_at + timedelta(minutes=data_granularity)
# Only include future intervals
if starts_at > now:
# Format timestamp for rating lookup
starts_at_key = starts_at.replace(second=0, microsecond=0).isoformat()
# Try to find rating data for this interval
interval_rating = rating_data.get(starts_at_key) or {}
future_prices.append(
{
"interval_start": starts_at.isoformat(), # Renamed from starts_at to interval_start
"interval_end": interval_end.isoformat(),
"price": float(price_data["total"]),
"price_cents": round(float(price_data["total"]) * 100, 2),
"level": price_data.get("level", "NORMAL"), # Price level from priceInfo
"rating": interval_rating.get("difference", None), # Rating from priceRating
"rating_level": interval_rating.get("rating_level"), # Level from priceRating
"day": day_key,
}
)
# Sort by start time
future_prices.sort(key=lambda x: x["interval_start"]) # Updated sort key
# Limit to the requested number of intervals
return future_prices[:intervals_to_return] if future_prices else None
def _add_price_forecast_attributes(self, attributes: dict) -> None:
"""Add forecast attributes for the price forecast sensor."""
future_prices = self._get_future_prices(max_intervals=MAX_FORECAST_INTERVALS)
if not future_prices:
attributes["intervals"] = []
attributes["hours"] = []
attributes["data_available"] = False
return
attributes["intervals"] = future_prices
attributes["data_available"] = True
# Determine interval granularity for display purposes
min_intervals_for_granularity_detection = 2
if len(future_prices) >= min_intervals_for_granularity_detection:
start1 = datetime.fromisoformat(future_prices[0]["interval_start"])
start2 = datetime.fromisoformat(future_prices[1]["interval_start"])
minutes_diff = int((start2 - start1).total_seconds() / 60)
attributes["interval_minutes"] = minutes_diff
else:
attributes["interval_minutes"] = MINUTES_PER_INTERVAL
# Group by hour for easier consumption in dashboards
hours = {}
for interval in future_prices:
starts_at = datetime.fromisoformat(interval["interval_start"])
hour_key = starts_at.strftime("%Y-%m-%d %H")
if hour_key not in hours:
hours[hour_key] = {
"hour": starts_at.hour,
"day": interval["day"],
"date": starts_at.date().isoformat(),
"intervals": [],
"min_price": None,
"max_price": None,
"avg_price": 0,
"avg_rating": None, # Initialize rating tracking
"ratings_available": False, # Track if any ratings are available
}
# Create interval data with both price and rating info
interval_data = {
"minute": starts_at.minute,
"price": interval["price"],
"price_cents": interval["price_cents"],
"level": interval["level"], # Price level from priceInfo
"time": starts_at.strftime("%H:%M"),
}
# Add rating data if available
if interval["rating"] is not None:
interval_data["rating"] = interval["rating"]
interval_data["rating_level"] = interval["rating_level"]
hours[hour_key]["ratings_available"] = True
hours[hour_key]["intervals"].append(interval_data)
# Track min/max/avg for the hour
price = interval["price"]
if hours[hour_key]["min_price"] is None or price < hours[hour_key]["min_price"]:
hours[hour_key]["min_price"] = price
if hours[hour_key]["max_price"] is None or price > hours[hour_key]["max_price"]:
hours[hour_key]["max_price"] = price
# Calculate averages
for hour_data in hours.values():
prices = [interval["price"] for interval in hour_data["intervals"]]
if prices:
hour_data["avg_price"] = sum(prices) / len(prices)
hour_data["avg_price_cents"] = hour_data["avg_price"] * 100
hour_data["min_price_cents"] = hour_data["min_price"] * 100
hour_data["max_price_cents"] = hour_data["max_price"] * 100
# Calculate average rating if ratings are available
if hour_data["ratings_available"]:
ratings = [interval.get("rating") for interval in hour_data["intervals"] if "rating" in interval]
if ratings:
hour_data["avg_rating"] = sum(ratings) / len(ratings)
# Convert to list sorted by hour
attributes["hours"] = [hour_data for _, hour_data in sorted(hours.items())]
@property
def native_value(self) -> float | str | datetime | None:
"""Return the native value of the sensor."""
try:
if not self.coordinator.data or not self._value_getter:
return None
# For price_level, ensure we return the translated value as state
if self.entity_description.key == "price_level":
return self._get_price_level_value()
return self._value_getter()
except (KeyError, ValueError, TypeError) as ex:
self.coordinator.logger.exception(
@ -552,14 +867,17 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
attributes = {}
# Group sensors by type and delegate to specific handlers
if key in ["current_price", "current_price_eur", SENSOR_TYPE_PRICE_LEVEL]:
if key in ["current_price", "current_price_eur", "price_level"]:
self._add_current_price_attributes(attributes)
elif key in ["next_hour_price", "next_hour_price_eur"]:
self._add_next_hour_attributes(attributes)
elif any(
pattern in key for pattern in ["_price_today", "rating", "data_timestamp", "tomorrow_data_available"]
):
self._add_statistics_attributes(attributes)
elif key == "price_forecast":
self._add_price_forecast_attributes(attributes)
# For price_level, add the original level as attribute
if key == "price_level" and hasattr(self, "_last_price_level") and self._last_price_level is not None:
attributes["level_id"] = self._last_price_level
except (KeyError, ValueError, TypeError) as ex:
self.coordinator.logger.exception(
"Error getting sensor attributes",
@ -568,22 +886,28 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
"entity": self.entity_description.key,
},
)
return None
else:
return attributes if attributes else None
def _add_current_price_attributes(self, attributes: dict) -> None:
"""Add attributes for current price sensors."""
current_hour_data = self._get_current_hour_data()
attributes["timestamp"] = current_hour_data["startsAt"] if current_hour_data else None
current_interval_data = self._get_current_interval_data()
attributes["timestamp"] = current_interval_data["startsAt"] if current_interval_data else None
# Add price level info for the price level sensor
if (
self.entity_description.key == SENSOR_TYPE_PRICE_LEVEL
and current_hour_data
and "level" in current_hour_data
):
self._add_price_level_attributes(attributes, current_hour_data["level"])
if self.entity_description.key == "price_level" and current_interval_data and "level" in current_interval_data:
self._add_price_level_attributes(attributes, current_interval_data["level"])
# Add timestamp for next interval price sensors
if self.entity_description.key in ["next_interval_price", "next_interval_price_eur"]:
# Get the next interval's data
price_info = self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"]["priceInfo"]
today_prices = price_info.get("today", [])
data_granularity = detect_interval_granularity(today_prices) if today_prices else MINUTES_PER_INTERVAL
now = dt_util.now()
next_interval_time = now + timedelta(minutes=data_granularity)
next_interval_data = find_price_data_for_interval(price_info, next_interval_time, data_granularity)
attributes["timestamp"] = next_interval_data["startsAt"] if next_interval_data else None
def _add_price_level_attributes(self, attributes: dict, level: str) -> None:
"""
@ -594,75 +918,10 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
level: The price level value (e.g., VERY_CHEAP, NORMAL, etc.)
"""
if level not in PRICE_LEVEL_MAPPING:
return
# Add numeric value for sorting/comparison
attributes["level_value"] = PRICE_LEVEL_MAPPING[level]
# Add the original English level as a reliable identifier for automations
if level in PRICE_LEVEL_MAPPING:
attributes["level_value"] = PRICE_LEVEL_MAPPING[level]
attributes["level_id"] = level
# Default to the original level value if no translation is found
friendly_name = level
# Try to get localized friendly name from translations if Home Assistant is available
if self.hass:
# Get user's language preference (default to English)
language = self.hass.config.language or "en"
# Use direct dictionary lookup for better performance and reliability
# This matches how async_get_entity_description works
cache_key = f"{DOMAIN}_translations_{language}"
if cache_key in self.hass.data:
translations = self.hass.data[cache_key]
# Navigate through the translation dictionary
if (
"sensor" in translations
and "price_level" in translations["sensor"]
and "price_levels" in translations["sensor"]["price_level"]
and level in translations["sensor"]["price_level"]["price_levels"]
):
friendly_name = translations["sensor"]["price_level"]["price_levels"][level]
# If we didn't find a translation in the current language, try English
if friendly_name == level and language != "en":
en_cache_key = f"{DOMAIN}_translations_en"
if en_cache_key in self.hass.data:
en_translations = self.hass.data[en_cache_key]
# Try using English translation as fallback
if (
"sensor" in en_translations
and "price_level" in en_translations["sensor"]
and "price_levels" in en_translations["sensor"]["price_level"]
and level in en_translations["sensor"]["price_level"]["price_levels"]
):
friendly_name = en_translations["sensor"]["price_level"]["price_levels"][level]
# Add the friendly name to attributes
attributes["friendly_name"] = friendly_name
def _add_next_hour_attributes(self, attributes: dict) -> None:
"""Add attributes for next hour price sensors."""
price_info = self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"]["priceInfo"]
now = dt_util.now()
target_datetime = now.replace(microsecond=0) + timedelta(hours=1)
target_hour = target_datetime.hour
target_date = target_datetime.date()
# Determine which day's data we need
day_key = "tomorrow" if target_date > now.date() else "today"
# Try to find the timestamp in either day's data
self._find_price_timestamp(attributes, price_info, day_key, target_hour, target_date)
if "timestamp" not in attributes:
other_day_key = "today" if day_key == "tomorrow" else "tomorrow"
self._find_price_timestamp(attributes, price_info, other_day_key, target_hour, target_date)
def _find_price_timestamp(
self, attributes: dict, price_info: Any, day_key: str, target_hour: int, target_date: date
) -> None:
@ -679,6 +938,167 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
def _add_statistics_attributes(self, attributes: dict) -> None:
"""Add attributes for statistics, rating, and diagnostic sensors."""
key = self.entity_description.key
price_info = self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"]["priceInfo"]
first_timestamp = price_info.get("today", [{}])[0].get("startsAt")
attributes["timestamp"] = first_timestamp
now = dt_util.now()
if key == "price_rating":
today_prices = price_info.get("today", [])
data_granularity = detect_interval_granularity(today_prices) if today_prices else MINUTES_PER_INTERVAL
interval_data = find_price_data_for_interval(price_info, now, data_granularity)
attributes["timestamp"] = interval_data["startsAt"] if interval_data else None
if hasattr(self, "_last_rating_difference") and self._last_rating_difference is not None:
attributes["difference_" + PERCENTAGE] = self._last_rating_difference
if hasattr(self, "_last_rating_level") and self._last_rating_level is not None:
attributes["level_id"] = self._last_rating_level
attributes["level_value"] = PRICE_RATING_MAPPING.get(self._last_rating_level, self._last_rating_level)
elif key == "daily_rating":
attributes["timestamp"] = now.replace(hour=0, minute=0, second=0, microsecond=0).isoformat()
if hasattr(self, "_last_rating_difference") and self._last_rating_difference is not None:
attributes["difference_" + PERCENTAGE] = self._last_rating_difference
if hasattr(self, "_last_rating_level") and self._last_rating_level is not None:
attributes["level_id"] = self._last_rating_level
attributes["level_value"] = PRICE_RATING_MAPPING.get(self._last_rating_level, self._last_rating_level)
elif key == "monthly_rating":
first_of_month = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
attributes["timestamp"] = first_of_month.isoformat()
if hasattr(self, "_last_rating_difference") and self._last_rating_difference is not None:
attributes["difference_" + PERCENTAGE] = self._last_rating_difference
if hasattr(self, "_last_rating_level") and self._last_rating_level is not None:
attributes["level_id"] = self._last_rating_level
attributes["level_value"] = PRICE_RATING_MAPPING.get(self._last_rating_level, self._last_rating_level)
else:
# Fallback: use the first timestamp of today
first_timestamp = price_info.get("today", [{}])[0].get("startsAt")
attributes["timestamp"] = first_timestamp
def detect_interval_granularity(price_data: list[dict]) -> int:
"""
Detect the granularity of price intervals in minutes.
Args:
price_data: List of price data points with startsAt timestamps
Returns:
Minutes per interval (e.g., 60 for hourly, 15 for 15-minute intervals)
"""
min_datapoints_for_granularity = 2
if not price_data or len(price_data) < min_datapoints_for_granularity:
return MINUTES_PER_INTERVAL # Default to target value
# Sort data points by timestamp
sorted_data = sorted(price_data, key=lambda x: x["startsAt"])
# Calculate the time differences between consecutive timestamps
intervals = []
for i in range(1, min(10, len(sorted_data))): # Sample up to 10 intervals
start_time_1 = dt_util.parse_datetime(sorted_data[i - 1]["startsAt"])
start_time_2 = dt_util.parse_datetime(sorted_data[i]["startsAt"])
if start_time_1 and start_time_2:
diff_minutes = (start_time_2 - start_time_1).total_seconds() / 60
intervals.append(round(diff_minutes))
# If no valid intervals found, return default
if not intervals:
return MINUTES_PER_INTERVAL
# Return the most common interval (mode)
return max(set(intervals), key=intervals.count)
def get_interval_for_timestamp(timestamp: datetime, granularity: int) -> int:
"""
Calculate the interval index within an hour for a given timestamp.
Args:
timestamp: The timestamp to calculate interval for
granularity: Minutes per interval
Returns:
Interval index (0-based) within the hour
"""
# Calculate which interval this timestamp falls into
intervals_per_hour = 60 // granularity
return (timestamp.minute // granularity) % intervals_per_hour
def _match_hourly_price_data(day_prices: list, target_time: datetime) -> dict | None:
"""Match price data for hourly granularity."""
for price_data in day_prices:
starts_at = dt_util.parse_datetime(price_data["startsAt"])
if starts_at is None:
continue
starts_at = dt_util.as_local(starts_at)
if starts_at.hour == target_time.hour and starts_at.date() == target_time.date():
return price_data
return None
def _match_granular_price_data(day_prices: list, target_time: datetime, data_granularity: int) -> dict | None:
"""Match price data for sub-hourly granularity."""
for price_data in day_prices:
starts_at = dt_util.parse_datetime(price_data["startsAt"])
if starts_at is None:
continue
starts_at = dt_util.as_local(starts_at)
interval_end = starts_at + timedelta(minutes=data_granularity)
# Check if target time falls within this interval
if starts_at <= target_time < interval_end and starts_at.date() == target_time.date():
return price_data
return None
def find_price_data_for_interval(
price_info: Any, target_time: datetime, data_granularity: int | None = None
) -> dict | None:
"""
Find the price data for a specific timestamp, handling different interval granularities.
Args:
price_info: The price info dictionary from Tibber API
target_time: The target timestamp to find price data for
data_granularity: Override detected granularity with this value (minutes)
Returns:
Price data dict if found, None otherwise
"""
# Determine which day's data to search
day_key = "tomorrow" if target_time.date() > dt_util.now().date() else "today"
search_days = [day_key, "tomorrow" if day_key == "today" else "today"]
# Try to find price data in today or tomorrow
for search_day in search_days:
day_prices = price_info.get(search_day, [])
if not day_prices:
continue
# Detect the granularity if not provided
if data_granularity is None:
data_granularity = detect_interval_granularity(day_prices)
# Check for a match with appropriate granularity
if data_granularity >= MINUTES_PER_INTERVAL * 4: # 60 minutes = hourly
result = _match_hourly_price_data(day_prices, target_time)
else:
result = _match_granular_price_data(day_prices, target_time, data_granularity)
if result:
return result
# If not found and we have sub-hourly granularity, try to fall back to hourly data
if data_granularity is not None and data_granularity < MINUTES_PER_INTERVAL * 4:
hour_start = target_time.replace(minute=0, second=0, microsecond=0)
for search_day in search_days:
day_prices = price_info.get(search_day, [])
result = _match_hourly_price_data(day_prices, hour_start)
if result:
return result
return None

View file

@ -2,29 +2,34 @@
"config": {
"step": {
"user": {
"description": "Wenn du Hilfe bei der Konfiguration benötigst, schau hier nach: https://github.com/jpawlowski/hass.tibber_prices",
"description": "Richte Tibber Preisinformationen & Bewertungen ein. Um ein API-Zugriffstoken zu generieren, besuche developer.tibber.com.",
"data": {
"access_token": "Tibber Zugangstoken"
}
"access_token": "API-Zugriffstoken",
"extended_descriptions": "Erweiterte Beschreibungen in Entitätsattributen anzeigen"
},
"title": "Tibber Preisinformationen & Bewertungen"
}
},
"error": {
"auth": "Der Tibber Zugangstoken ist ungültig.",
"connection": "Verbindung zu Tibber nicht möglich. Bitte überprüfe deine Internetverbindung.",
"unknown": "Ein unerwarteter Fehler ist aufgetreten. Bitte überprüfe die Logs für Details."
"unknown": "Ein unerwarteter Fehler ist aufgetreten. Bitte überprüfe die Logs für Details.",
"cannot_connect": "Verbindung fehlgeschlagen",
"invalid_access_token": "Ungültiges Zugriffstoken"
},
"abort": {
"already_configured": "Dieser Eintrag ist bereits konfiguriert.",
"already_configured": "Integration ist bereits konfiguriert",
"entry_not_found": "Tibber Konfigurationseintrag nicht gefunden."
}
},
"options": {
"step": {
"init": {
"title": "Tibber Konfiguration aktualisieren",
"description": "Aktualisiere deinen Tibber API Zugangstoken. Wenn du einen neuen Token benötigst, kannst du einen unter https://developer.tibber.com/settings/access-token generieren",
"title": "Optionen für Tibber Preisinformationen & Bewertungen",
"description": "Konfiguriere Optionen für Tibber Preisinformationen & Bewertungen",
"data": {
"access_token": "Tibber Zugangstoken"
"access_token": "Tibber Zugangstoken",
"extended_descriptions": "Erweiterte Beschreibungen in Entitätsattributen anzeigen"
}
}
},
@ -41,13 +46,10 @@
"entity": {
"sensor": {
"current_price": {
"name": "Aktueller Preis"
},
"next_hour_price": {
"name": "Preis nächste Stunde"
"name": "Aktueller Strompreis"
},
"price_level": {
"name": "Preisniveau"
"name": "Aktuelles Preisniveau"
},
"lowest_price_today": {
"name": "Niedrigster Preis heute"
@ -58,8 +60,8 @@
"average_price_today": {
"name": "Durchschnittspreis heute"
},
"hourly_rating": {
"name": "Stündliche Preisbewertung"
"price_rating": {
"name": "Aktuelle Preisbewertung"
},
"daily_rating": {
"name": "Tägliche Preisbewertung"
@ -68,10 +70,16 @@
"name": "Monatliche Preisbewertung"
},
"data_timestamp": {
"name": "Preisprognose-Horizont"
"name": "Zeitstempel der neuesten Daten"
},
"tomorrow_data_available": {
"name": "Daten für morgen verfügbar"
"name": "Datenstatus für morgen"
},
"next_interval_price": {
"name": "Strompreis nächstes Intervall"
},
"price_forecast": {
"name": "Preisprognose"
}
},
"binary_sensor": {
@ -79,10 +87,10 @@
"name": "Spitzenstunde"
},
"best_price_hour": {
"name": "Beste Preisstunde"
"name": "Beste-Preis-Stunde"
},
"connection": {
"name": "Verbindungsstatus"
"name": "Tibber API-Verbindung"
}
}
}

View file

@ -2,29 +2,34 @@
"config": {
"step": {
"user": {
"description": "If you need help with the configuration have a look here: https://github.com/jpawlowski/hass.tibber_prices",
"description": "Set up Tibber Price Information & Ratings. To generate an API access token, visit developer.tibber.com.",
"data": {
"access_token": "Tibber Access Token"
}
"access_token": "API access token",
"extended_descriptions": "Show extended descriptions in entity attributes"
},
"title": "Tibber Price Information & Ratings"
}
},
"error": {
"auth": "The Tibber Access Token is invalid.",
"connection": "Unable to connect to Tibber. Please check your internet connection.",
"unknown": "An unexpected error occurred. Please check the logs for details."
"unknown": "Unexpected error",
"cannot_connect": "Failed to connect",
"invalid_access_token": "Invalid access token"
},
"abort": {
"already_configured": "This entry is already configured.",
"already_configured": "Integration is already configured",
"entry_not_found": "Tibber configuration entry not found."
}
},
"options": {
"step": {
"init": {
"title": "Update Tibber Configuration",
"description": "Update your Tibber API access token. If you need a new token, you can generate one at https://developer.tibber.com/settings/access-token",
"title": "Options for Tibber Price Information & Ratings",
"description": "Configure options for Tibber Price Information & Ratings",
"data": {
"access_token": "Tibber Access Token"
"access_token": "Tibber Access Token",
"extended_descriptions": "Show extended descriptions in entity attributes"
}
}
},
@ -41,25 +46,40 @@
"entity": {
"sensor": {
"current_price": {
"name": "Current Price"
"name": "Current Electricity Price"
},
"next_hour_price": {
"name": "Next Hour Price"
"current_price_cents": {
"name": "Current Electricity Price"
},
"next_interval_price": {
"name": "Next Interval Electricity Price"
},
"next_interval_price_cents": {
"name": "Next Interval Electricity Price"
},
"price_level": {
"name": "Price Level"
"name": "Current Price Level"
},
"lowest_price_today": {
"name": "Lowest Price Today"
"name": "Today's Lowest Price"
},
"lowest_price_today_cents": {
"name": "Today's Lowest Price"
},
"highest_price_today": {
"name": "Highest Price Today"
"name": "Today's Highest Price"
},
"highest_price_today_cents": {
"name": "Today's Highest Price"
},
"average_price_today": {
"name": "Average Price Today"
"name": "Today's Average Price"
},
"hourly_rating": {
"name": "Hourly Price Rating"
"average_price_today_cents": {
"name": "Today's Average Price"
},
"price_rating": {
"name": "Current Price Rating"
},
"daily_rating": {
"name": "Daily Price Rating"
@ -68,10 +88,13 @@
"name": "Monthly Price Rating"
},
"data_timestamp": {
"name": "Price Forecast Horizon"
"name": "Latest Data Available"
},
"tomorrow_data_available": {
"name": "Tomorrow's Data Available"
"name": "Tomorrow's Data Status"
},
"price_forecast": {
"name": "Price Forecast"
}
},
"binary_sensor": {
@ -82,7 +105,7 @@
"name": "Best Price Hour"
},
"connection": {
"name": "Connection Status"
"name": "Tibber API Connection"
}
}
}