jpawlowski/hass.tibber_prices
1
0
Fork 0
mirror of https://github.com/jpawlowski/hass.tibber_prices.git synced 2026-03-30 05:13:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

refactoring

Browse source
This commit is contained in:
Julian Pawlowski 2025-04-23 16:42:31 +00:00
parent 447987b628
commit dd94351278
14 changed files with 1317 additions and 793 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

140
.devcontainer/devcontainer.json
View file

@ -1,72 +1,74 @@
{ {
"name": "jpawlowski/hass.tibber_prices", "name": "jpawlowski/hass.tibber_prices",
"image": "mcr.microsoft.com/devcontainers/python:3.13", "image": "mcr.microsoft.com/devcontainers/python:3.13",
"postCreateCommand": "scripts/setup", "postCreateCommand": "scripts/setup",
"containerEnv": { "containerEnv": {
"PYTHONASYNCIODEBUG": "1" "PYTHONASYNCIODEBUG": "1"
}, },
"forwardPorts": [ "forwardPorts": [8123],
8123 "portsAttributes": {
], "8123": {
"portsAttributes": { "label": "Home Assistant",
"8123": { "onAutoForward": "notify"
"label": "Home Assistant",
"onAutoForward": "notify"
}
},
"customizations": {
"vscode": {
"extensions": [
"charliermarsh.ruff",
"github.vscode-pull-request-github",
"ms-python.python",
"ms-python.pylint",
"ms-python.vscode-pylance",
"ryanluker.vscode-coverage-gutters",
"visualstudioexptteam.vscodeintellicode",
"redhat.vscode-yaml",
"esbenp.prettier-vscode",
"github.copilot",
"ms-vscode-remote.remote-containers"
],
"settings": {
"editor.tabSize": 4,
"editor.formatOnPaste": false,
"editor.formatOnSave": true,
"editor.formatOnType": true,
"files.eol": "\n",
"files.trimTrailingWhitespace": true,
"python.analysis.typeCheckingMode": "basic",
"python.analysis.autoImportCompletions": true,
"python.defaultInterpreterPath": "/usr/local/bin/python",
"pylint.importStrategy": "fromEnvironment",
"python.terminal.activateEnvInCurrentTerminal": true,
"python.testing.pytestArgs": [
"--no-cov"
],
"yaml.customTags": [
"!input scalar",
"!secret scalar",
"!include_dir_named scalar",
"!include_dir_list scalar",
"!include_dir_merge_list scalar",
"!include_dir_merge_named scalar"
],
"[python]": {
"editor.defaultFormatter": "charliermarsh.ruff"
}
}
}
},
"remoteUser": "vscode",
"features": {
"ghcr.io/devcontainers/features/github-cli:1": {},
"ghcr.io/devcontainers-extra/features/apt-packages:1": {
"packages": [
"ffmpeg",
"libturbojpeg0",
"libpcap-dev"
]
}
} }
},
"customizations": {
"vscode": {
"extensions": [
"charliermarsh.ruff",
"github.vscode-pull-request-github",
"ms-python.python",
"ms-python.pylint",
"ms-python.vscode-pylance",
"ryanluker.vscode-coverage-gutters",
"visualstudioexptteam.vscodeintellicode",
"redhat.vscode-yaml",
"esbenp.prettier-vscode",
"github.copilot",
"ms-vscode-remote.remote-containers"
],
"settings": {
"editor.tabSize": 4,
"editor.formatOnPaste": false,
"editor.formatOnSave": false,
"editor.formatOnType": false,
"files.eol": "\n",
"files.trimTrailingWhitespace": true,
"python.analysis.typeCheckingMode": "basic",
"python.analysis.autoImportCompletions": true,
"python.defaultInterpreterPath": "/usr/local/bin/python",
"pylint.importStrategy": "fromEnvironment",
"python.terminal.activateEnvInCurrentTerminal": true,
"python.testing.pytestArgs": ["--no-cov"],
"yaml.customTags": [
"!input scalar",
"!secret scalar",
"!include_dir_named scalar",
"!include_dir_list scalar",
"!include_dir_merge_list scalar",
"!include_dir_merge_named scalar"
],
"[python]": {
"editor.defaultFormatter": "charliermarsh.ruff"
},
"json.schemas": [
{
"fileMatch": ["homeassistant/components/*/manifest.json"],
"url": "${containerWorkspaceFolder}/scripts/json_schemas/manifest_schema.json"
},
{
"fileMatch": ["homeassistant/components/*/translations/*.json"],
"url": "${containerWorkspaceFolder}/scripts/json_schemas/translation_schema.json"
}
]
}
}
},
"remoteUser": "vscode",
"features": {
"ghcr.io/devcontainers/features/github-cli:1": {},
"ghcr.io/devcontainers-extra/features/apt-packages:1": {
"packages": ["ffmpeg", "libturbojpeg0", "libpcap-dev"]
}
}
} }

13
custom_components/tibber_prices/api.py
View file

@ -108,7 +108,8 @@ async def _verify_graphql_response(response_json: dict) -> None:
def _is_data_empty(data: dict, query_type: str) -> bool: def _is_data_empty(data: dict, query_type: str) -> bool:
"""Check if the response data is empty or incomplete. """
Check if the response data is empty or incomplete.
For price info: For price info:
- Must have either range/edges or yesterday data - Must have either range/edges or yesterday data
@ -154,7 +155,8 @@ def _is_data_empty(data: dict, query_type: str) -> bool:
is_empty = not has_historical or not has_today is_empty = not has_historical or not has_today
_LOGGER.debug( _LOGGER.debug(
"Price info check - historical data (range: %s, yesterday: %s), today: %s, is_empty: %s", "Price info check - historical data "
"(range: %s, yesterday: %s), today: %s, is_empty: %s",
bool(has_range), bool(has_range),
bool(has_yesterday), bool(has_yesterday),
bool(has_today), bool(has_today),
@ -173,7 +175,9 @@ def _is_data_empty(data: dict, query_type: str) -> bool:
and "high" in rating["thresholdPercentages"] and "high" in rating["thresholdPercentages"]
) )
if not has_thresholds: if not has_thresholds:
_LOGGER.debug("Missing or invalid threshold percentages for %s rating", query_type) _LOGGER.debug(
"Missing or invalid threshold percentages for %s rating", query_type
)
return True return True
# Check rating entries # Check rating entries
@ -196,10 +200,11 @@ def _is_data_empty(data: dict, query_type: str) -> bool:
return is_empty return is_empty
_LOGGER.debug("Unknown query type %s, treating as non-empty", query_type) _LOGGER.debug("Unknown query type %s, treating as non-empty", query_type)
return False
except (KeyError, IndexError, TypeError) as error: except (KeyError, IndexError, TypeError) as error:
_LOGGER.debug("Error checking data emptiness: %s", error) _LOGGER.debug("Error checking data emptiness: %s", error)
return True return True
else:
return False
def _prepare_headers(access_token: str) -> dict[str, str]: def _prepare_headers(access_token: str) -> dict[str, str]:

120
custom_components/tibber_prices/binary_sensor.py
View file

@ -2,7 +2,7 @@
from __future__ import annotations from __future__ import annotations
from datetime import datetime from datetime import UTC, datetime
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from homeassistant.components.binary_sensor import ( from homeassistant.components.binary_sensor import (
@ -12,7 +12,6 @@ from homeassistant.components.binary_sensor import (
) )
from homeassistant.const import EntityCategory from homeassistant.const import EntityCategory
from .const import NAME, DOMAIN
from .entity import TibberPricesEntity from .entity import TibberPricesEntity
if TYPE_CHECKING: if TYPE_CHECKING:
@ -46,7 +45,7 @@ ENTITY_DESCRIPTIONS = (
async def async_setup_entry( async def async_setup_entry(
hass: HomeAssistant, _hass: HomeAssistant,
entry: TibberPricesConfigEntry, entry: TibberPricesConfigEntry,
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
@ -71,59 +70,61 @@ class TibberPricesBinarySensor(TibberPricesEntity, BinarySensorEntity):
"""Initialize the binary_sensor class.""" """Initialize the binary_sensor class."""
super().__init__(coordinator) super().__init__(coordinator)
self.entity_description = entity_description self.entity_description = entity_description
self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{entity_description.key}" self._attr_unique_id = (
f"{coordinator.config_entry.entry_id}_{entity_description.key}"
)
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", [])
)
):
return None
now = datetime.now(tz=UTC).astimezone()
current_hour_data = next(
(
price_data
for price_data in today_prices
if datetime.fromisoformat(price_data["startsAt"]).hour == now.hour
),
None,
)
if not current_hour_data:
return None
prices = [float(price["total"]) for price in today_prices]
prices.sort()
return prices, float(current_hour_data["total"])
@property @property
def is_on(self) -> bool | None: def is_on(self) -> bool | None:
"""Return true if the binary_sensor is on.""" """Return true if the binary_sensor is on."""
try: try:
if not self.coordinator.data: price_data = self._get_current_price_data()
if not price_data:
return None return None
subscription = self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"] prices, current_price = price_data
price_info = subscription["priceInfo"] match self.entity_description.key:
case "peak_hour":
now = datetime.now() threshold_index = int(len(prices) * 0.8)
current_hour_data = None return current_price >= prices[threshold_index]
today_prices = price_info.get("today", []) case "best_price_hour":
threshold_index = int(len(prices) * 0.2)
if not today_prices: return current_price <= prices[threshold_index]
return None case "connection":
return True
# Find current hour's data case _:
for price_data in today_prices: return None
starts_at = datetime.fromisoformat(price_data["startsAt"])
if starts_at.hour == now.hour:
current_hour_data = price_data
break
if not current_hour_data:
return None
if self.entity_description.key == "peak_hour":
# Consider it a peak hour if the price is in the top 20% of today's prices
prices = [float(price["total"]) for price in today_prices]
prices.sort()
threshold_index = int(len(prices) * 0.8)
peak_threshold = prices[threshold_index]
return float(current_hour_data["total"]) >= peak_threshold
elif self.entity_description.key == "best_price_hour":
# Consider it a best price hour if the price is in the bottom 20% of today's prices
prices = [float(price["total"]) for price in today_prices]
prices.sort()
threshold_index = int(len(prices) * 0.2)
best_threshold = prices[threshold_index]
return float(current_hour_data["total"]) <= best_threshold
elif self.entity_description.key == "connection":
# Check if we have valid current data
return bool(current_hour_data)
return None
except (KeyError, ValueError, TypeError) as ex: except (KeyError, ValueError, TypeError) as ex:
self.coordinator.logger.error( self.coordinator.logger.exception(
"Error getting binary sensor state", "Error getting binary sensor state",
extra={ extra={
"error": str(ex), "error": str(ex),
@ -139,20 +140,28 @@ class TibberPricesBinarySensor(TibberPricesEntity, BinarySensorEntity):
if not self.coordinator.data: if not self.coordinator.data:
return None return None
subscription = self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"] subscription = self.coordinator.data["data"]["viewer"]["homes"][0][
"currentSubscription"
]
price_info = subscription["priceInfo"] price_info = subscription["priceInfo"]
attributes = {} attributes = {}
if self.entity_description.key in ["peak_hour", "best_price_hour"]: if self.entity_description.key in ["peak_hour", "best_price_hour"]:
today_prices = price_info.get("today", []) today_prices = price_info.get("today", [])
if today_prices: if today_prices:
prices = [(datetime.fromisoformat(price["startsAt"]).hour, float(price["total"])) prices = [
for price in today_prices] (
datetime.fromisoformat(price["startsAt"]).hour,
float(price["total"]),
)
for price in today_prices
]
if self.entity_description.key == "peak_hour": if self.entity_description.key == "peak_hour":
# Get top 5 peak hours # Get top 5 peak hours
peak_hours = sorted(prices, key=lambda x: x[1], reverse=True)[:5] peak_hours = sorted(prices, key=lambda x: x[1], reverse=True)[
:5
]
attributes["peak_hours"] = [ attributes["peak_hours"] = [
{"hour": hour, "price": price} for hour, price in peak_hours {"hour": hour, "price": price} for hour, price in peak_hours
] ]
@ -162,11 +171,12 @@ class TibberPricesBinarySensor(TibberPricesEntity, BinarySensorEntity):
attributes["best_price_hours"] = [ attributes["best_price_hours"] = [
{"hour": hour, "price": price} for hour, price in best_hours {"hour": hour, "price": price} for hour, price in best_hours
] ]
return attributes
return attributes if attributes else None else:
return None
except (KeyError, ValueError, TypeError) as ex: except (KeyError, ValueError, TypeError) as ex:
self.coordinator.logger.error( self.coordinator.logger.exception(
"Error getting binary sensor attributes", "Error getting binary sensor attributes",
extra={ extra={
"error": str(ex), "error": str(ex),

4
custom_components/tibber_prices/config_flow.py
View file

@ -35,6 +35,10 @@ class TibberPricesFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Get the options flow for this handler.""" """Get the options flow for this handler."""
return TibberPricesOptionsFlowHandler(config_entry) return TibberPricesOptionsFlowHandler(config_entry)
def is_matching(self, other_flow: dict) -> bool:
"""Return True if match_dict matches this flow."""
return bool(other_flow.get("domain") == DOMAIN)
async def async_step_user( async def async_step_user(
self, self,
user_input: dict | None = None, user_input: dict | None = None,

648
custom_components/tibber_prices/coordinator.py
View file

@ -1,31 +1,28 @@
"""Coordinator for fetching Tibber price data.""" """Coordinator for fetching Tibber price data."""
from __future__ import annotations from __future__ import annotations
import asyncio
import random
from datetime import datetime, timedelta
import logging import logging
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Any, Final, TypedDict, cast from typing import TYPE_CHECKING, Any, Final, TypedDict, cast
import homeassistant.util.dt as dt_util
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.event import async_track_time_change from homeassistant.helpers.event import async_track_time_change
from homeassistant.helpers.storage import Store from homeassistant.helpers.storage import Store
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
import homeassistant.util.dt as dt_util
from .api import ( from .api import (
TibberPricesApiClient,
TibberPricesApiClientError,
TibberPricesApiClientAuthenticationError, TibberPricesApiClientAuthenticationError,
TibberPricesApiClientCommunicationError, TibberPricesApiClientCommunicationError,
TibberPricesApiClientError,
) )
from .const import ( from .const import DOMAIN, LOGGER
DOMAIN,
LOGGER,
)
if TYPE_CHECKING: if TYPE_CHECKING:
import asyncio
from .data import TibberPricesConfigEntry from .data import TibberPricesConfigEntry
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -36,6 +33,8 @@ RANDOM_DELAY_MAX_MINUTES: Final = 120 # Maximum random delay in minutes
NO_DATA_ERROR_MSG: Final = "No data available" NO_DATA_ERROR_MSG: Final = "No data available"
STORAGE_VERSION: Final = 1 STORAGE_VERSION: Final = 1
UPDATE_INTERVAL: Final = timedelta(days=1) # Both price and rating data update daily UPDATE_INTERVAL: Final = timedelta(days=1) # Both price and rating data update daily
UPDATE_FAILED_MSG: Final = "Update failed"
AUTH_FAILED_MSG: Final = "Authentication failed"
class TibberPricesPriceInfo(TypedDict): class TibberPricesPriceInfo(TypedDict):
@ -75,7 +74,9 @@ def _raise_no_data() -> None:
@callback @callback
def _get_latest_timestamp_from_prices(price_data: TibberPricesData | None) -> datetime | None: def _get_latest_timestamp_from_prices(
price_data: TibberPricesData | None,
) -> datetime | None:
"""Get the latest timestamp from price data.""" """Get the latest timestamp from price data."""
if not price_data or "data" not in price_data: if not price_data or "data" not in price_data:
return None return None
@ -90,7 +91,9 @@ def _get_latest_timestamp_from_prices(price_data: TibberPricesData | None) -> da
for price in today_prices: for price in today_prices:
if starts_at := price.get("startsAt"): if starts_at := price.get("startsAt"):
timestamp = dt_util.parse_datetime(starts_at) timestamp = dt_util.parse_datetime(starts_at)
if timestamp and (not latest_timestamp or timestamp > latest_timestamp): if timestamp and (
not latest_timestamp or timestamp > latest_timestamp
):
latest_timestamp = timestamp latest_timestamp = timestamp
# Check tomorrow's prices # Check tomorrow's prices
@ -98,17 +101,21 @@ def _get_latest_timestamp_from_prices(price_data: TibberPricesData | None) -> da
for price in tomorrow_prices: for price in tomorrow_prices:
if starts_at := price.get("startsAt"): if starts_at := price.get("startsAt"):
timestamp = dt_util.parse_datetime(starts_at) timestamp = dt_util.parse_datetime(starts_at)
if timestamp and (not latest_timestamp or timestamp > latest_timestamp): if timestamp and (
not latest_timestamp or timestamp > latest_timestamp
):
latest_timestamp = timestamp latest_timestamp = timestamp
return latest_timestamp
except (KeyError, IndexError, TypeError): except (KeyError, IndexError, TypeError):
return None return None
else:
return latest_timestamp
@callback @callback
def _get_latest_timestamp_from_rating(rating_data: TibberPricesData | None) -> datetime | None: def _get_latest_timestamp_from_rating(
rating_data: TibberPricesData | None,
) -> datetime | None:
"""Get the latest timestamp from rating data.""" """Get the latest timestamp from rating data."""
if not rating_data or "data" not in rating_data: if not rating_data or "data" not in rating_data:
return None return None
@ -124,13 +131,14 @@ def _get_latest_timestamp_from_rating(rating_data: TibberPricesData | None) -> d
for entry in rating_entries: for entry in rating_entries:
if time := entry.get("time"): if time := entry.get("time"):
timestamp = dt_util.parse_datetime(time) timestamp = dt_util.parse_datetime(time)
if timestamp and (not latest_timestamp or timestamp > latest_timestamp): if timestamp and (
not latest_timestamp or timestamp > latest_timestamp
):
latest_timestamp = timestamp latest_timestamp = timestamp
return latest_timestamp
except (KeyError, IndexError, TypeError): except (KeyError, IndexError, TypeError):
return None return None
else:
return latest_timestamp
# https://developers.home-assistant.io/docs/integration_fetching_data#coordinated-single-api-poll-for-data-for-all-entities # https://developers.home-assistant.io/docs/integration_fetching_data#coordinated-single-api-poll-for-data-for-all-entities
@ -181,14 +189,18 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
for listener in self._remove_update_listeners: for listener in self._remove_update_listeners:
listener() listener()
async def _async_handle_midnight_rotation(self, _now: datetime | None = None) -> None: async def _async_handle_midnight_rotation(
self, _now: datetime | None = None
) -> None:
"""Handle data rotation at midnight.""" """Handle data rotation at midnight."""
if not self._cached_price_data: if not self._cached_price_data:
return return
try: try:
LOGGER.debug("Starting midnight data rotation") LOGGER.debug("Starting midnight data rotation")
subscription = self._cached_price_data["data"]["viewer"]["homes"][0]["currentSubscription"] subscription = self._cached_price_data["data"]["viewer"]["homes"][0][
"currentSubscription"
]
price_info = subscription["priceInfo"] price_info = subscription["priceInfo"]
# Move today's data to yesterday # Move today's data to yesterday
@ -209,9 +221,39 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
# Trigger an update to refresh the entities # Trigger an update to refresh the entities
await self.async_request_refresh() await self.async_request_refresh()
except Exception as ex: except (KeyError, TypeError, ValueError) as ex:
LOGGER.error("Error during midnight data rotation: %s", ex) LOGGER.error("Error during midnight data rotation: %s", ex)
@callback
def _recover_timestamp(
self,
data: TibberPricesData | None,
stored_timestamp: str | None,
rating_type: str | None = None,
) -> datetime | None:
"""Recover timestamp from data or stored value."""
if stored_timestamp:
return dt_util.parse_datetime(stored_timestamp)
if not data:
return None
if rating_type:
timestamp = self._get_latest_timestamp_from_rating_type(data, rating_type)
else:
timestamp = _get_latest_timestamp_from_prices(data)
if timestamp:
LOGGER.debug(
"Recovered %s timestamp from data: %s",
rating_type or "price",
timestamp,
)
else:
return None
return timestamp
async def _async_initialize(self) -> None: async def _async_initialize(self) -> None:
"""Load stored data.""" """Load stored data."""
stored = await self._store.async_load() stored = await self._store.async_load()
@ -219,71 +261,44 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
if stored: if stored:
# Load cached data # Load cached data
self._cached_price_data = cast(TibberPricesData, stored.get("price_data")) self._cached_price_data = cast("TibberPricesData", stored.get("price_data"))
self._cached_rating_data_hourly = cast(TibberPricesData, stored.get("rating_data_hourly")) self._cached_rating_data_hourly = cast(
self._cached_rating_data_daily = cast(TibberPricesData, stored.get("rating_data_daily")) "TibberPricesData", stored.get("rating_data_hourly")
self._cached_rating_data_monthly = cast(TibberPricesData, stored.get("rating_data_monthly")) )
self._cached_rating_data_daily = cast(
"TibberPricesData", stored.get("rating_data_daily")
)
self._cached_rating_data_monthly = cast(
"TibberPricesData", stored.get("rating_data_monthly")
)
# Get timestamps from the actual data first # Recover timestamps
latest_price_timestamp = None self._last_price_update = self._recover_timestamp(
latest_hourly_timestamp = None self._cached_price_data, stored.get("last_price_update")
latest_daily_timestamp = None )
latest_monthly_timestamp = None self._last_rating_update_hourly = self._recover_timestamp(
self._cached_rating_data_hourly,
if self._cached_price_data: stored.get("last_rating_update_hourly"),
latest_price_timestamp = _get_latest_timestamp_from_prices(self._cached_price_data) "hourly",
if latest_price_timestamp and not stored.get("last_price_update"): )
self._last_price_update = latest_price_timestamp self._last_rating_update_daily = self._recover_timestamp(
LOGGER.debug("Recovered price update timestamp from data: %s", self._last_price_update) self._cached_rating_data_daily,
stored.get("last_rating_update_daily"),
if self._cached_rating_data_hourly: "daily",
latest_hourly_timestamp = self._get_latest_timestamp_from_rating_type( )
self._cached_rating_data_hourly, "hourly" self._last_rating_update_monthly = self._recover_timestamp(
) self._cached_rating_data_monthly,
if latest_hourly_timestamp and not stored.get("last_rating_update_hourly"): stored.get("last_rating_update_monthly"),
self._last_rating_update_hourly = latest_hourly_timestamp "monthly",
LOGGER.debug("Recovered hourly rating timestamp from data: %s", self._last_rating_update_hourly) )
if self._cached_rating_data_daily:
latest_daily_timestamp = self._get_latest_timestamp_from_rating_type(
self._cached_rating_data_daily, "daily"
)
if latest_daily_timestamp and not stored.get("last_rating_update_daily"):
self._last_rating_update_daily = latest_daily_timestamp
LOGGER.debug("Recovered daily rating timestamp from data: %s", self._last_rating_update_daily)
if self._cached_rating_data_monthly:
latest_monthly_timestamp = self._get_latest_timestamp_from_rating_type(
self._cached_rating_data_monthly, "monthly"
)
if latest_monthly_timestamp and not stored.get("last_rating_update_monthly"):
self._last_rating_update_monthly = latest_monthly_timestamp
LOGGER.debug("Recovered monthly rating timestamp from data: %s", self._last_rating_update_monthly)
# Then load stored timestamps if they exist
if last_price := stored.get("last_price_update"):
self._last_price_update = dt_util.parse_datetime(last_price)
if last_rating_hourly := stored.get("last_rating_update_hourly"):
self._last_rating_update_hourly = dt_util.parse_datetime(last_rating_hourly)
if last_rating_daily := stored.get("last_rating_update_daily"):
self._last_rating_update_daily = dt_util.parse_datetime(last_rating_daily)
if last_rating_monthly := stored.get("last_rating_update_monthly"):
self._last_rating_update_monthly = dt_util.parse_datetime(last_rating_monthly)
LOGGER.debug( LOGGER.debug(
"Loaded stored cache data - " "Loaded stored cache data - "
"Price update: %s (latest data: %s), " "Price update: %s, Rating hourly: %s, daily: %s, monthly: %s",
"Rating hourly: %s (latest data: %s), "
"daily: %s (latest data: %s), "
"monthly: %s (latest data: %s)",
self._last_price_update, self._last_price_update,
latest_price_timestamp,
self._last_rating_update_hourly, self._last_rating_update_hourly,
latest_hourly_timestamp,
self._last_rating_update_daily, self._last_rating_update_daily,
latest_daily_timestamp,
self._last_rating_update_monthly, self._last_rating_update_monthly,
latest_monthly_timestamp,
) )
async def _async_refresh_hourly(self, *_: Any) -> None: async def _async_refresh_hourly(self, *_: Any) -> None:
@ -291,7 +306,8 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
await self.async_refresh() await self.async_refresh()
async def _async_update_data(self) -> TibberPricesData: async def _async_update_data(self) -> TibberPricesData:
"""Fetch new state data for the coordinator. """
Fetch new state data for the coordinator.
This method will: This method will:
1. Initialize cached data if none exists 1. Initialize cached data if none exists
@ -304,6 +320,7 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
try: try:
current_time = dt_util.now() current_time = dt_util.now()
result = None
# If force update requested, fetch all data # If force update requested, fetch all data
if self._force_update: if self._force_update:
@ -317,81 +334,104 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
"hourly": self._last_rating_update_hourly, "hourly": self._last_rating_update_hourly,
"daily": self._last_rating_update_daily, "daily": self._last_rating_update_daily,
"monthly": self._last_rating_update_monthly, "monthly": self._last_rating_update_monthly,
} },
} },
) )
self._force_update = False # Reset force update flag self._force_update = False # Reset force update flag
return await self._fetch_all_data() result = await self._fetch_all_data()
else:
# Check if we need to update based on conditions result = await self._handle_conditional_update(current_time)
should_update_price = self._should_update_price_data(current_time)
should_update_hourly = self._should_update_rating_type(
current_time, self._cached_rating_data_hourly, self._last_rating_update_hourly, "hourly"
)
should_update_daily = self._should_update_rating_type(
current_time, self._cached_rating_data_daily, self._last_rating_update_daily, "daily"
)
should_update_monthly = self._should_update_rating_type(
current_time, self._cached_rating_data_monthly, self._last_rating_update_monthly, "monthly"
)
if any([should_update_price, should_update_hourly, should_update_daily, should_update_monthly]):
LOGGER.debug(
"Updating data based on conditions",
extra={
"update_price": should_update_price,
"update_hourly": should_update_hourly,
"update_daily": should_update_daily,
"update_monthly": should_update_monthly,
}
)
return await self._fetch_all_data()
# Use cached data if no update needed
if self._cached_price_data is not None:
LOGGER.debug("Using cached data")
return self._merge_all_cached_data()
# If we have no cached data and no updates needed, fetch new data
LOGGER.debug("No cached data available, fetching new data")
return await self._fetch_all_data()
except TibberPricesApiClientAuthenticationError as exception: except TibberPricesApiClientAuthenticationError as exception:
LOGGER.error( LOGGER.error(
"Authentication failed", "Authentication failed",
extra={"error": str(exception), "error_type": "auth_failed"} extra={"error": str(exception), "error_type": "auth_failed"},
)
raise ConfigEntryAuthFailed("Authentication failed while fetching data") from exception
except TibberPricesApiClientCommunicationError as exception:
LOGGER.error(
"API communication error",
extra={"error": str(exception), "error_type": "communication_error"}
) )
raise ConfigEntryAuthFailed(AUTH_FAILED_MSG) from exception
except (
TibberPricesApiClientCommunicationError,
TibberPricesApiClientError,
Exception,
) as exception:
if isinstance(exception, TibberPricesApiClientCommunicationError):
LOGGER.error(
"API communication error",
extra={
"error": str(exception),
"error_type": "communication_error",
},
)
elif isinstance(exception, TibberPricesApiClientError):
LOGGER.error(
"API client error",
extra={"error": str(exception), "error_type": "client_error"},
)
else:
LOGGER.exception(
"Unexpected error",
extra={"error": str(exception), "error_type": "unexpected"},
)
if self._cached_price_data is not None: if self._cached_price_data is not None:
LOGGER.info("Using cached data as fallback") LOGGER.info("Using cached data as fallback")
return self._merge_all_cached_data() return self._merge_all_cached_data()
raise UpdateFailed(f"Error communicating with API: {exception}") from exception raise UpdateFailed(UPDATE_FAILED_MSG) from exception
except TibberPricesApiClientError as exception: else:
LOGGER.error( return result
"API client error",
extra={"error": str(exception), "error_type": "client_error"} async def _handle_conditional_update(
self, current_time: datetime
) -> TibberPricesData:
"""Handle conditional update based on update conditions."""
should_update_price = self._should_update_price_data(current_time)
should_update_hourly = self._should_update_rating_type(
current_time,
self._cached_rating_data_hourly,
self._last_rating_update_hourly,
"hourly",
)
should_update_daily = self._should_update_rating_type(
current_time,
self._cached_rating_data_daily,
self._last_rating_update_daily,
"daily",
)
should_update_monthly = self._should_update_rating_type(
current_time,
self._cached_rating_data_monthly,
self._last_rating_update_monthly,
"monthly",
)
if any(
[
should_update_price,
should_update_hourly,
should_update_daily,
should_update_monthly,
]
):
LOGGER.debug(
"Updating data based on conditions",
extra={
"update_price": should_update_price,
"update_hourly": should_update_hourly,
"update_daily": should_update_daily,
"update_monthly": should_update_monthly,
},
) )
if self._cached_price_data is not None: return await self._fetch_all_data()
LOGGER.info("Using cached data as fallback")
return self._merge_all_cached_data() if self._cached_price_data is not None:
raise UpdateFailed(f"Error fetching data: {exception}") from exception LOGGER.debug("Using cached data")
except Exception as exception: return self._merge_all_cached_data()
LOGGER.exception(
"Unexpected error", LOGGER.debug("No cached data available, fetching new data")
extra={"error": str(exception), "error_type": "unexpected"} return await self._fetch_all_data()
)
if self._cached_price_data is not None:
LOGGER.info("Using cached data as fallback")
return self._merge_all_cached_data()
raise UpdateFailed(f"Unexpected error: {exception}") from exception
async def _fetch_all_data(self) -> TibberPricesData: async def _fetch_all_data(self) -> TibberPricesData:
"""Fetch all data from the API without checking update conditions. """
Fetch all data from the API without checking update conditions.
This method will: This method will:
1. Fetch all required data (price and rating data) 1. Fetch all required data (price and rating data)
@ -402,11 +442,7 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
current_time = dt_util.now() current_time = dt_util.now()
new_data = { new_data = {
"price_data": None, "price_data": None,
"rating_data": { "rating_data": {"hourly": None, "daily": None, "monthly": None},
"hourly": None,
"daily": None,
"monthly": None
}
} }
# First fetch all data without updating cache # First fetch all data without updating cache
@ -426,7 +462,9 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
except TibberPricesApiClientError as ex: except TibberPricesApiClientError as ex:
LOGGER.error("Failed to fetch price data: %s", ex) LOGGER.error("Failed to fetch price data: %s", ex)
if self._cached_price_data is not None: if self._cached_price_data is not None:
LOGGER.info("Using cached data as fallback after price data fetch failure") LOGGER.info(
"Using cached data as fallback after price data fetch failure"
)
return self._merge_all_cached_data() return self._merge_all_cached_data()
raise raise
@ -439,22 +477,30 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
_raise_no_data() _raise_no_data()
# Only update cache if we have valid data # Only update cache if we have valid data
self._cached_price_data = cast(TibberPricesData, new_data["price_data"]) self._cached_price_data = cast("TibberPricesData", new_data["price_data"])
self._last_price_update = current_time self._last_price_update = current_time
# Update rating data cache only for types that were successfully fetched # Update rating data cache only for types that were successfully fetched
for rating_type, rating_data in new_data["rating_data"].items(): for rating_type, rating_data in new_data["rating_data"].items():
if rating_data is not None: if rating_data is not None:
if rating_type == "hourly": if rating_type == "hourly":
self._cached_rating_data_hourly = cast(TibberPricesData, rating_data) self._cached_rating_data_hourly = cast(
"TibberPricesData", rating_data
)
self._last_rating_update_hourly = current_time self._last_rating_update_hourly = current_time
elif rating_type == "daily": elif rating_type == "daily":
self._cached_rating_data_daily = cast(TibberPricesData, rating_data) self._cached_rating_data_daily = cast(
"TibberPricesData", rating_data
)
self._last_rating_update_daily = current_time self._last_rating_update_daily = current_time
else: # monthly else: # monthly
self._cached_rating_data_monthly = cast(TibberPricesData, rating_data) self._cached_rating_data_monthly = cast(
"TibberPricesData", rating_data
)
self._last_rating_update_monthly = current_time self._last_rating_update_monthly = current_time
LOGGER.debug("Updated %s rating data cache at %s", rating_type, current_time) LOGGER.debug(
"Updated %s rating data cache at %s", rating_type, current_time
)
# Store the updated cache # Store the updated cache
await self._store_cache() await self._store_cache()
@ -467,20 +513,33 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
"""Store cache data.""" """Store cache data."""
# Recover any missing timestamps from the data # Recover any missing timestamps from the data
if self._cached_price_data and not self._last_price_update: if self._cached_price_data and not self._last_price_update:
latest_timestamp = _get_latest_timestamp_from_prices(self._cached_price_data) latest_timestamp = _get_latest_timestamp_from_prices(
self._cached_price_data
)
if latest_timestamp: if latest_timestamp:
self._last_price_update = latest_timestamp self._last_price_update = latest_timestamp
LOGGER.debug("Setting missing price update timestamp to: %s", self._last_price_update) LOGGER.debug(
"Setting missing price update timestamp to: %s",
self._last_price_update,
)
rating_types = { rating_types = {
"hourly": (self._cached_rating_data_hourly, self._last_rating_update_hourly), "hourly": (
self._cached_rating_data_hourly,
self._last_rating_update_hourly,
),
"daily": (self._cached_rating_data_daily, self._last_rating_update_daily), "daily": (self._cached_rating_data_daily, self._last_rating_update_daily),
"monthly": (self._cached_rating_data_monthly, self._last_rating_update_monthly), "monthly": (
self._cached_rating_data_monthly,
self._last_rating_update_monthly,
),
} }
for rating_type, (cached_data, last_update) in rating_types.items(): for rating_type, (cached_data, last_update) in rating_types.items():
if cached_data and not last_update: if cached_data and not last_update:
latest_timestamp = self._get_latest_timestamp_from_rating_type(cached_data, rating_type) latest_timestamp = self._get_latest_timestamp_from_rating_type(
cached_data, rating_type
)
if latest_timestamp: if latest_timestamp:
if rating_type == "hourly": if rating_type == "hourly":
self._last_rating_update_hourly = latest_timestamp self._last_rating_update_hourly = latest_timestamp
@ -488,21 +547,34 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
self._last_rating_update_daily = latest_timestamp self._last_rating_update_daily = latest_timestamp
else: # monthly else: # monthly
self._last_rating_update_monthly = latest_timestamp self._last_rating_update_monthly = latest_timestamp
LOGGER.debug("Setting missing %s rating timestamp to: %s", rating_type, latest_timestamp) LOGGER.debug(
"Setting missing %s rating timestamp to: %s",
rating_type,
latest_timestamp,
)
data = { data = {
"price_data": self._cached_price_data, "price_data": self._cached_price_data,
"rating_data_hourly": self._cached_rating_data_hourly, "rating_data_hourly": self._cached_rating_data_hourly,
"rating_data_daily": self._cached_rating_data_daily, "rating_data_daily": self._cached_rating_data_daily,
"rating_data_monthly": self._cached_rating_data_monthly, "rating_data_monthly": self._cached_rating_data_monthly,
"last_price_update": self._last_price_update.isoformat() if self._last_price_update else None, "last_price_update": self._last_price_update.isoformat()
"last_rating_update_hourly": self._last_rating_update_hourly.isoformat() if self._last_rating_update_hourly else None, if self._last_price_update
"last_rating_update_daily": self._last_rating_update_daily.isoformat() if self._last_rating_update_daily else None, else None,
"last_rating_update_monthly": self._last_rating_update_monthly.isoformat() if self._last_rating_update_monthly else None, "last_rating_update_hourly": self._last_rating_update_hourly.isoformat()
if self._last_rating_update_hourly
else None,
"last_rating_update_daily": self._last_rating_update_daily.isoformat()
if self._last_rating_update_daily
else None,
"last_rating_update_monthly": self._last_rating_update_monthly.isoformat()
if self._last_rating_update_monthly
else None,
} }
LOGGER.debug("Storing cache data with timestamps: %s", { LOGGER.debug(
k: v for k, v in data.items() if k.startswith("last_") "Storing cache data with timestamps: %s",
}) {k: v for k, v in data.items() if k.startswith("last_")},
)
await self._store.async_save(data) await self._store.async_save(data)
@callback @callback
@ -514,7 +586,9 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
return True return True
# Get the latest timestamp from our price data # Get the latest timestamp from our price data
latest_price_timestamp = _get_latest_timestamp_from_prices(self._cached_price_data) latest_price_timestamp = _get_latest_timestamp_from_prices(
self._cached_price_data
)
if not latest_price_timestamp: if not latest_price_timestamp:
LOGGER.debug("No valid timestamp found in price data, update needed") LOGGER.debug("No valid timestamp found in price data, update needed")
return True return True
@ -522,11 +596,16 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
# If we have price data but no last_update timestamp, set it # If we have price data but no last_update timestamp, set it
if not self._last_price_update: if not self._last_price_update:
self._last_price_update = latest_price_timestamp self._last_price_update = latest_price_timestamp
LOGGER.debug("Setting missing price update timestamp in check: %s", self._last_price_update) LOGGER.debug(
"Setting missing price update timestamp in check: %s",
self._last_price_update,
)
# Check if we're in the update window (13:00-15:00) # Check if we're in the update window (13:00-15:00)
current_hour = current_time.hour current_hour = current_time.hour
in_update_window = PRICE_UPDATE_RANDOM_MIN_HOUR <= current_hour <= PRICE_UPDATE_RANDOM_MAX_HOUR in_update_window = (
PRICE_UPDATE_RANDOM_MIN_HOUR <= current_hour <= PRICE_UPDATE_RANDOM_MAX_HOUR
)
# Get tomorrow's date at midnight # Get tomorrow's date at midnight
tomorrow = (current_time + timedelta(days=1)).replace( tomorrow = (current_time + timedelta(days=1)).replace(
@ -536,7 +615,8 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
# If we're in the update window and don't have tomorrow's complete data # If we're in the update window and don't have tomorrow's complete data
if in_update_window and latest_price_timestamp < tomorrow: if in_update_window and latest_price_timestamp < tomorrow:
LOGGER.debug( LOGGER.debug(
"In update window (%d:00) and latest price timestamp (%s) is before tomorrow, update needed", "In update window (%d:00) and latest price timestamp (%s) "
"is before tomorrow, update needed",
current_hour, current_hour,
latest_price_timestamp, latest_price_timestamp,
) )
@ -571,13 +651,19 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
"""Check if specific rating type should be updated.""" """Check if specific rating type should be updated."""
# If no cached data, we definitely need an update # If no cached data, we definitely need an update
if cached_data is None: if cached_data is None:
LOGGER.debug("No cached %s rating data available, update needed", rating_type) LOGGER.debug(
"No cached %s rating data available, update needed", rating_type
)
return True return True
# Get the latest timestamp from our rating data # Get the latest timestamp from our rating data
latest_timestamp = self._get_latest_timestamp_from_rating_type(cached_data, rating_type) latest_timestamp = self._get_latest_timestamp_from_rating_type(
cached_data, rating_type
)
if not latest_timestamp: if not latest_timestamp:
LOGGER.debug("No valid timestamp found in %s rating data, update needed", rating_type) LOGGER.debug(
"No valid timestamp found in %s rating data, update needed", rating_type
)
return True return True
# If we have rating data but no last_update timestamp, set it # If we have rating data but no last_update timestamp, set it
@ -588,64 +674,50 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
self._last_rating_update_daily = latest_timestamp self._last_rating_update_daily = latest_timestamp
else: # monthly else: # monthly
self._last_rating_update_monthly = latest_timestamp self._last_rating_update_monthly = latest_timestamp
LOGGER.debug("Setting missing %s rating timestamp in check: %s", rating_type, latest_timestamp) LOGGER.debug(
"Setting missing %s rating timestamp in check: %s",
rating_type,
latest_timestamp,
)
last_update = latest_timestamp last_update = latest_timestamp
current_hour = current_time.hour current_hour = current_time.hour
in_update_window = PRICE_UPDATE_RANDOM_MIN_HOUR <= current_hour <= PRICE_UPDATE_RANDOM_MAX_HOUR in_update_window = (
PRICE_UPDATE_RANDOM_MIN_HOUR <= current_hour <= PRICE_UPDATE_RANDOM_MAX_HOUR
)
should_update = False
if rating_type == "monthly": if rating_type == "monthly":
# For monthly ratings: current_month_start = current_time.replace(
# 1. Check if we have data for the current month day=1, hour=0, minute=0, second=0, microsecond=0
# 2. Update more frequently as the data changes throughout the month )
current_month_start = current_time.replace(day=1, hour=0, minute=0, second=0, microsecond=0) should_update = latest_timestamp < current_month_start or (
last_update and current_time - last_update >= timedelta(days=1)
if latest_timestamp < current_month_start: )
LOGGER.debug(
"Monthly rating data is from previous month (%s), update needed",
latest_timestamp,
)
return True
# Update monthly data daily to get updated calculations
if last_update and current_time - last_update >= timedelta(days=1):
LOGGER.debug(
"More than 24 hours since last monthly rating update (%s), update needed for latest calculations",
last_update,
)
return True
else: else:
# For hourly and daily ratings:
# Get tomorrow's date at midnight
tomorrow = (current_time + timedelta(days=1)).replace( tomorrow = (current_time + timedelta(days=1)).replace(
hour=0, minute=0, second=0, microsecond=0 hour=0, minute=0, second=0, microsecond=0
) )
should_update = (
in_update_window and latest_timestamp < tomorrow
) or current_time - last_update >= UPDATE_INTERVAL
# If we're in the update window and don't have tomorrow's data if should_update:
if in_update_window and latest_timestamp < tomorrow: LOGGER.debug(
LOGGER.debug( "Update needed for %s rating data - Last update: %s, Latest data: %s",
"In update window and %s rating data (%s) is before tomorrow, update needed", rating_type,
rating_type, last_update,
latest_timestamp, latest_timestamp,
) )
return True else:
LOGGER.debug(
"No %s rating update needed - Last update: %s, Latest data: %s",
rating_type,
last_update,
latest_timestamp,
)
# If it's been more than 24 hours since our last update return should_update
if current_time - last_update >= UPDATE_INTERVAL:
LOGGER.debug(
"More than 24 hours since last %s rating update (%s), update needed",
rating_type,
last_update,
)
return True
LOGGER.debug(
"No %s rating update needed - Last update: %s, Latest data: %s",
rating_type,
last_update,
latest_timestamp,
)
return False
@callback @callback
def _is_price_update_window(self, current_hour: int) -> bool: def _is_price_update_window(self, current_hour: int) -> bool:
@ -665,10 +737,14 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
try: try:
# Try to access data in the transformed structure first # Try to access data in the transformed structure first
try: try:
price_info = data["viewer"]["homes"][0]["currentSubscription"]["priceInfo"] price_info = data["viewer"]["homes"][0]["currentSubscription"][
"priceInfo"
]
except KeyError: except KeyError:
# If that fails, try the raw data structure # If that fails, try the raw data structure
price_info = data["data"]["viewer"]["homes"][0]["currentSubscription"]["priceInfo"] price_info = data["data"]["viewer"]["homes"][0]["currentSubscription"][
"priceInfo"
]
# Ensure we have all required fields # Ensure we have all required fields
extracted_price_info = { extracted_price_info = {
@ -676,35 +752,34 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
"tomorrow": price_info.get("tomorrow", []), "tomorrow": price_info.get("tomorrow", []),
"yesterday": price_info.get("yesterday", []), "yesterday": price_info.get("yesterday", []),
} }
return {
"data": {
"viewer": {
"homes": [{
"currentSubscription": {
"priceInfo": extracted_price_info
}
}]
}
}
}
except (KeyError, IndexError) as ex: except (KeyError, IndexError) as ex:
LOGGER.error("Error extracting price data: %s", ex) LOGGER.error("Error extracting price data: %s", ex)
return { return {
"data": { "data": {
"viewer": { "viewer": {
"homes": [{ "homes": [
"currentSubscription": { {
"priceInfo": { "currentSubscription": {
"today": [], "priceInfo": {
"tomorrow": [], "today": [],
"yesterday": [], "tomorrow": [],
"yesterday": [],
}
} }
} }
}] ]
} }
} }
} }
return {
"data": {
"viewer": {
"homes": [
{"currentSubscription": {"priceInfo": extracted_price_info}}
]
}
}
}
@callback @callback
def _get_latest_timestamp_from_rating_type( def _get_latest_timestamp_from_rating_type(
@ -715,21 +790,21 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
return None return None
try: try:
subscription = rating_data["data"]["viewer"]["homes"][0]["currentSubscription"] subscription = rating_data["data"]["viewer"]["homes"][0][
"currentSubscription"
]
price_rating = subscription["priceRating"] price_rating = subscription["priceRating"]
latest_timestamp = None result = None
if rating_entries := price_rating.get(rating_type, {}).get("entries", []): if rating_entries := price_rating.get(rating_type, {}).get("entries", []):
for entry in rating_entries: for entry in rating_entries:
if time := entry.get("time"): if time := entry.get("time"):
timestamp = dt_util.parse_datetime(time) timestamp = dt_util.parse_datetime(time)
if timestamp and (not latest_timestamp or timestamp > latest_timestamp): if timestamp and (not result or timestamp > result):
latest_timestamp = timestamp result = timestamp
return latest_timestamp
except (KeyError, IndexError, TypeError): except (KeyError, IndexError, TypeError):
return None return None
return result
async def _get_rating_data_for_type(self, rating_type: str) -> dict: async def _get_rating_data_for_type(self, rating_type: str) -> dict:
"""Get fresh rating data for a specific type.""" """Get fresh rating data for a specific type."""
@ -748,7 +823,9 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
except KeyError: except KeyError:
try: try:
# If that fails, try the raw data structure # If that fails, try the raw data structure
rating = data["data"]["viewer"]["homes"][0]["currentSubscription"]["priceRating"] rating = data["data"]["viewer"]["homes"][0]["currentSubscription"][
"priceRating"
]
except KeyError as ex: except KeyError as ex:
LOGGER.error("Failed to extract rating data: %s", ex) LOGGER.error("Failed to extract rating data: %s", ex)
raise TibberPricesApiClientError( raise TibberPricesApiClientError(
@ -756,29 +833,50 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
query_type=rating_type query_type=rating_type
) )
) from ex ) from ex
else:
return { return {
"data": { "data": {
"viewer": { "viewer": {
"homes": [ "homes": [
{ {
"currentSubscription": { "currentSubscription": {
"priceRating": { "priceRating": {
"thresholdPercentages": rating["thresholdPercentages"], "thresholdPercentages": rating[
rating_type: rating[rating_type], "thresholdPercentages"
],
rating_type: rating[rating_type],
}
}
}
]
}
}
}
else:
return {
"data": {
"viewer": {
"homes": [
{
"currentSubscription": {
"priceRating": {
"thresholdPercentages": rating[
"thresholdPercentages"
],
rating_type: rating[rating_type],
}
} }
} }
} ]
] }
} }
} }
}
@callback @callback
def _merge_all_cached_data(self) -> TibberPricesData: def _merge_all_cached_data(self) -> TibberPricesData:
"""Merge all cached data.""" """Merge all cached data."""
if not self._cached_price_data: if not self._cached_price_data:
return cast(TibberPricesData, {}) return cast("TibberPricesData", {})
# Start with price info # Start with price info
subscription = { subscription = {
@ -786,8 +884,8 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
"currentSubscription" "currentSubscription"
]["priceInfo"], ]["priceInfo"],
"priceRating": { "priceRating": {
"thresholdPercentages": None, # Will be set from any available rating data "thresholdPercentages": None,
} },
} }
# Add rating data if available # Add rating data if available
@ -799,19 +897,27 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
for rating_type, data in rating_data.items(): for rating_type, data in rating_data.items():
if data and "data" in data: if data and "data" in data:
rating = data["data"]["viewer"]["homes"][0]["currentSubscription"]["priceRating"] rating = data["data"]["viewer"]["homes"][0]["currentSubscription"][
"priceRating"
]
# Set thresholdPercentages from any available rating data # Set thresholdPercentages from any available rating data
if not subscription["priceRating"]["thresholdPercentages"]: if not subscription["priceRating"]["thresholdPercentages"]:
subscription["priceRating"]["thresholdPercentages"] = rating["thresholdPercentages"] subscription["priceRating"]["thresholdPercentages"] = rating[
"thresholdPercentages"
]
# Add the specific rating type data # Add the specific rating type data
subscription["priceRating"][rating_type] = rating[rating_type] subscription["priceRating"][rating_type] = rating[rating_type]
return cast(TibberPricesData, {"data": {"viewer": {"homes": [{"currentSubscription": subscription}]}}}) return cast(
"TibberPricesData",
{"data": {"viewer": {"homes": [{"currentSubscription": subscription}]}}},
)
async def async_request_refresh(self) -> None: async def async_request_refresh(self) -> None:
"""Request an immediate refresh of the data. """
Request an immediate refresh of the data.
This method will: This method will:
1. Set the force update flag to trigger a fresh data fetch 1. Set the force update flag to trigger a fresh data fetch
@ -824,4 +930,4 @@ class TibberPricesDataUpdateCoordinator(DataUpdateCoordinator[TibberPricesData])
def _transform_api_response(self, data: dict[str, Any]) -> TibberPricesData: def _transform_api_response(self, data: dict[str, Any]) -> TibberPricesData:
"""Transform API response to coordinator data format.""" """Transform API response to coordinator data format."""
return cast(TibberPricesData, data) return cast("TibberPricesData", data)

7
custom_components/tibber_prices/data.py
View file

@ -13,9 +13,6 @@ if TYPE_CHECKING:
from .coordinator import TibberPricesDataUpdateCoordinator from .coordinator import TibberPricesDataUpdateCoordinator
type TibberPricesConfigEntry = ConfigEntry[TibberPricesData]
@dataclass @dataclass
class TibberPricesData: class TibberPricesData:
"""Data for the tibber_prices integration.""" """Data for the tibber_prices integration."""
@ -23,3 +20,7 @@ class TibberPricesData:
client: TibberPricesApiClient client: TibberPricesApiClient
coordinator: TibberPricesDataUpdateCoordinator coordinator: TibberPricesDataUpdateCoordinator
integration: Integration integration: Integration
if TYPE_CHECKING:
type TibberPricesConfigEntry = ConfigEntry[TibberPricesData]

2
custom_components/tibber_prices/entity.py
View file

@ -5,7 +5,7 @@ from __future__ import annotations
from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import ATTRIBUTION, DOMAIN, NAME from .const import ATTRIBUTION, DOMAIN
from .coordinator import TibberPricesDataUpdateCoordinator from .coordinator import TibberPricesDataUpdateCoordinator

1
custom_components/tibber_prices/manifest.json
View file

@ -6,7 +6,6 @@
], ],
"config_flow": true, "config_flow": true,
"documentation": "https://github.com/jpawlowski/hass.tibber_prices", "documentation": "https://github.com/jpawlowski/hass.tibber_prices",
"icon": "mdi:chart-line",
"iot_class": "cloud_polling", "iot_class": "cloud_polling",
"issue_tracker": "https://github.com/jpawlowski/hass.tibber_prices/issues", "issue_tracker": "https://github.com/jpawlowski/hass.tibber_prices/issues",
"version": "0.1.0" "version": "0.1.0"

442
custom_components/tibber_prices/sensor.py
View file

@ -2,19 +2,17 @@
from __future__ import annotations from __future__ import annotations
from datetime import datetime from datetime import UTC, datetime
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from homeassistant.components.sensor import ( from homeassistant.components.sensor import (
SensorDeviceClass, SensorDeviceClass,
SensorEntity, SensorEntity,
SensorEntityDescription, SensorEntityDescription,
SensorStateClass,
) )
from homeassistant.const import EntityCategory from homeassistant.const import CURRENCY_EURO, EntityCategory
from homeassistant.util import dt as dt_util from homeassistant.util import dt as dt_util
from .const import DOMAIN
from .entity import TibberPricesEntity from .entity import TibberPricesEntity
if TYPE_CHECKING: if TYPE_CHECKING:
@ -25,7 +23,7 @@ if TYPE_CHECKING:
from .data import TibberPricesConfigEntry from .data import TibberPricesConfigEntry
PRICE_UNIT = "ct/kWh" PRICE_UNIT = "ct/kWh"
CURRENCY_EURO = "EUR/kWh" HOURS_IN_DAY = 24
# Main price sensors that users will typically use in automations # Main price sensors that users will typically use in automations
PRICE_SENSORS = ( PRICE_SENSORS = (
@ -36,7 +34,7 @@ PRICE_SENSORS = (
icon="mdi:currency-eur", icon="mdi:currency-eur",
device_class=SensorDeviceClass.MONETARY, device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement=CURRENCY_EURO, native_unit_of_measurement=CURRENCY_EURO,
entity_registry_enabled_default=False, # Hidden by default as it's mainly for the Energy Dashboard entity_registry_enabled_default=False,
), ),
SensorEntityDescription( SensorEntityDescription(
key="current_price", key="current_price",
@ -53,7 +51,7 @@ PRICE_SENSORS = (
icon="mdi:currency-eur-off", icon="mdi:currency-eur-off",
device_class=SensorDeviceClass.MONETARY, device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement=CURRENCY_EURO, native_unit_of_measurement=CURRENCY_EURO,
entity_registry_enabled_default=False, # Hidden by default as it's mainly for the Energy Dashboard entity_registry_enabled_default=False,
), ),
SensorEntityDescription( SensorEntityDescription(
key="next_hour_price", key="next_hour_price",
@ -80,7 +78,7 @@ STATISTICS_SENSORS = (
icon="mdi:currency-eur", icon="mdi:currency-eur",
device_class=SensorDeviceClass.MONETARY, device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement=CURRENCY_EURO, native_unit_of_measurement=CURRENCY_EURO,
entity_registry_enabled_default=False, # Hidden by default as it's mainly for the Energy Dashboard entity_registry_enabled_default=False,
), ),
SensorEntityDescription( SensorEntityDescription(
key="lowest_price_today", key="lowest_price_today",
@ -97,7 +95,7 @@ STATISTICS_SENSORS = (
icon="mdi:currency-eur", icon="mdi:currency-eur",
device_class=SensorDeviceClass.MONETARY, device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement=CURRENCY_EURO, native_unit_of_measurement=CURRENCY_EURO,
entity_registry_enabled_default=False, # Hidden by default as it's mainly for the Energy Dashboard entity_registry_enabled_default=False,
), ),
SensorEntityDescription( SensorEntityDescription(
key="highest_price_today", key="highest_price_today",
@ -114,7 +112,7 @@ STATISTICS_SENSORS = (
icon="mdi:currency-eur", icon="mdi:currency-eur",
device_class=SensorDeviceClass.MONETARY, device_class=SensorDeviceClass.MONETARY,
native_unit_of_measurement=CURRENCY_EURO, native_unit_of_measurement=CURRENCY_EURO,
entity_registry_enabled_default=False, # Hidden by default as it's mainly for the Energy Dashboard entity_registry_enabled_default=False,
), ),
SensorEntityDescription( SensorEntityDescription(
key="average_price_today", key="average_price_today",
@ -180,7 +178,7 @@ ENTITY_DESCRIPTIONS = (
async def async_setup_entry( async def async_setup_entry(
hass: HomeAssistant, _hass: HomeAssistant,
entry: TibberPricesConfigEntry, entry: TibberPricesConfigEntry,
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
@ -205,224 +203,298 @@ class TibberPricesSensor(TibberPricesEntity, SensorEntity):
"""Initialize the sensor class.""" """Initialize the sensor class."""
super().__init__(coordinator) super().__init__(coordinator)
self.entity_description = entity_description self.entity_description = entity_description
self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{entity_description.key}" self._attr_unique_id = (
f"{coordinator.config_entry.entry_id}_{entity_description.key}"
)
self._attr_has_entity_name = True self._attr_has_entity_name = True
def _get_current_hour_data(self) -> dict | None:
"""Get the price data for the current hour."""
if not self.coordinator.data:
return None
now = datetime.now(tz=UTC).astimezone()
price_info = self.coordinator.data["data"]["viewer"]["homes"][0][
"currentSubscription"
]["priceInfo"]
for price_data in price_info.get("today", []):
starts_at = datetime.fromisoformat(price_data["startsAt"])
if starts_at.hour == now.hour:
return price_data
return None
def _get_price_value(self, price: float) -> float:
"""Convert price based on unit."""
return (
price * 100
if self.entity_description.native_unit_of_measurement == "ct/kWh"
else price
)
def _get_price_sensor_value(self) -> float | None:
"""Handle price sensor values."""
if not self.coordinator.data:
return None
subscription = self.coordinator.data["data"]["viewer"]["homes"][0][
"currentSubscription"
]
price_info = subscription["priceInfo"]
now = datetime.now(tz=UTC).astimezone()
current_hour_data = self._get_current_hour_data()
key = self.entity_description.key
if key in ["current_price", "current_price_eur"]:
if not current_hour_data:
return None
return (
self._get_price_value(float(current_hour_data["total"]))
if key == "current_price"
else float(current_hour_data["total"])
)
if key in ["next_hour_price", "next_hour_price_eur"]:
next_hour = (now.hour + 1) % 24
for price_data in price_info.get("today", []):
starts_at = datetime.fromisoformat(price_data["startsAt"])
if starts_at.hour == next_hour:
return (
self._get_price_value(float(price_data["total"]))
if key == "next_hour_price"
else float(price_data["total"])
)
return None
return None
def _get_statistics_value(self) -> float | None:
"""Handle statistics sensor values."""
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
key = self.entity_description.key
prices = [float(price["total"]) for price in today_prices]
if key in ["lowest_price_today", "lowest_price_today_eur"]:
value = min(prices)
elif key in ["highest_price_today", "highest_price_today_eur"]:
value = max(prices)
elif key in ["average_price_today", "average_price_today_eur"]:
value = sum(prices) / len(prices)
else:
return None
return self._get_price_value(value) if key.endswith("today") else value
def _get_rating_value(self) -> float | None:
"""Handle rating sensor values."""
if not self.coordinator.data:
return None
def check_hourly(entry: dict) -> bool:
return datetime.fromisoformat(entry["time"]).hour == now.hour
def check_daily(entry: dict) -> bool:
return datetime.fromisoformat(entry["time"]).date() == now.date()
def check_monthly(entry: dict) -> bool:
dt = datetime.fromisoformat(entry["time"])
return dt.month == now.month and dt.year == now.year
subscription = self.coordinator.data["data"]["viewer"]["homes"][0][
"currentSubscription"
]
price_rating = subscription.get("priceRating", {}) or {}
now = datetime.now(tz=UTC).astimezone()
key = self.entity_description.key
if key == "hourly_rating":
rating_data = price_rating.get("hourly", {})
entries = rating_data.get("entries", []) if rating_data else []
time_match = check_hourly
elif key == "daily_rating":
rating_data = price_rating.get("daily", {})
entries = rating_data.get("entries", []) if rating_data else []
time_match = check_daily
elif key == "monthly_rating":
rating_data = price_rating.get("monthly", {})
entries = rating_data.get("entries", []) if rating_data else []
time_match = check_monthly
else:
return None
for entry in entries:
if time_match(entry):
return round(float(entry["difference"]) * 100, 1)
return None
def _get_diagnostic_value(self) -> datetime | str | None:
"""Handle diagnostic sensor values."""
if not self.coordinator.data:
return None
price_info = self.coordinator.data["data"]["viewer"]["homes"][0][
"currentSubscription"
]["priceInfo"]
key = self.entity_description.key
if key == "data_timestamp":
latest_timestamp = None
for day in ["today", "tomorrow"]:
for price_data in price_info.get(day, []):
timestamp = datetime.fromisoformat(price_data["startsAt"])
if not latest_timestamp or timestamp > latest_timestamp:
latest_timestamp = timestamp
return dt_util.as_utc(latest_timestamp) if latest_timestamp else None
if key == "tomorrow_data_available":
tomorrow_prices = price_info.get("tomorrow", [])
if not tomorrow_prices:
return "No"
return "Yes" if len(tomorrow_prices) == HOURS_IN_DAY else "Partial"
return None
@property @property
def native_value(self) -> float | str | datetime | None: def native_value(self) -> float | str | datetime | None:
"""Return the native value of the sensor.""" """Return the native value of the sensor."""
result = None
try: try:
if not self.coordinator.data: if self.coordinator.data:
return None key = self.entity_description.key
current_hour_data = self._get_current_hour_data()
subscription = self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"]
price_info = subscription["priceInfo"]
price_rating = subscription.get("priceRating") or {}
# Get current hour's data
now = datetime.now()
current_hour_data = None
for price_data in price_info.get("today", []):
starts_at = datetime.fromisoformat(price_data["startsAt"])
if starts_at.hour == now.hour:
current_hour_data = price_data
break
# Helper function to convert price based on unit
def get_price_value(price: float) -> float:
if self.entity_description.native_unit_of_measurement == "ct/kWh":
return price * 100
return price
if self.entity_description.key == "current_price":
return get_price_value(float(current_hour_data["total"])) if current_hour_data else None
elif self.entity_description.key == "current_price_eur":
return float(current_hour_data["total"]) if current_hour_data else None
elif self.entity_description.key == "next_hour_price":
next_hour = (now.hour + 1) % 24
for price_data in price_info.get("today", []):
starts_at = datetime.fromisoformat(price_data["startsAt"])
if starts_at.hour == next_hour:
return get_price_value(float(price_data["total"]))
return None
elif self.entity_description.key == "next_hour_price_eur":
next_hour = (now.hour + 1) % 24
for price_data in price_info.get("today", []):
starts_at = datetime.fromisoformat(price_data["startsAt"])
if starts_at.hour == next_hour:
return float(price_data["total"])
return None
elif self.entity_description.key == "lowest_price_today":
today_prices = price_info.get("today", [])
if not today_prices:
return None
return get_price_value(min(float(price["total"]) for price in today_prices))
elif self.entity_description.key == "lowest_price_today_eur":
today_prices = price_info.get("today", [])
if not today_prices:
return None
return min(float(price["total"]) for price in today_prices)
elif self.entity_description.key == "highest_price_today":
today_prices = price_info.get("today", [])
if not today_prices:
return None
return get_price_value(max(float(price["total"]) for price in today_prices))
elif self.entity_description.key == "highest_price_today_eur":
today_prices = price_info.get("today", [])
if not today_prices:
return None
return max(float(price["total"]) for price in today_prices)
elif self.entity_description.key == "average_price_today":
today_prices = price_info.get("today", [])
if not today_prices:
return None
avg = sum(float(price["total"]) for price in today_prices) / len(today_prices)
return get_price_value(avg)
elif self.entity_description.key == "average_price_today_eur":
today_prices = price_info.get("today", [])
if not today_prices:
return None
return sum(float(price["total"]) for price in today_prices) / len(today_prices)
elif self.entity_description.key == "price_level":
return current_hour_data["level"] if current_hour_data else None
elif self.entity_description.key == "hourly_rating":
hourly = price_rating.get("hourly", {})
entries = hourly.get("entries", []) if hourly else []
if not entries:
return None
for entry in entries:
starts_at = datetime.fromisoformat(entry["time"])
if starts_at.hour == now.hour:
return round(float(entry["difference"]) * 100, 1)
return None
elif self.entity_description.key == "daily_rating":
daily = price_rating.get("daily", {})
entries = daily.get("entries", []) if daily else []
if not entries:
return None
for entry in entries:
starts_at = datetime.fromisoformat(entry["time"])
if starts_at.date() == now.date():
return round(float(entry["difference"]) * 100, 1)
return None
elif self.entity_description.key == "monthly_rating":
monthly = price_rating.get("monthly", {})
entries = monthly.get("entries", []) if monthly else []
if not entries:
return None
for entry in entries:
starts_at = datetime.fromisoformat(entry["time"])
if starts_at.month == now.month and starts_at.year == now.year:
return round(float(entry["difference"]) * 100, 1)
return None
elif self.entity_description.key == "data_timestamp":
# Return the latest timestamp from any data we have
latest_timestamp = None
# Check today's data
for price_data in price_info.get("today", []):
timestamp = datetime.fromisoformat(price_data["startsAt"])
if not latest_timestamp or timestamp > latest_timestamp:
latest_timestamp = timestamp
# Check tomorrow's data
for price_data in price_info.get("tomorrow", []):
timestamp = datetime.fromisoformat(price_data["startsAt"])
if not latest_timestamp or timestamp > latest_timestamp:
latest_timestamp = timestamp
return dt_util.as_utc(latest_timestamp) if latest_timestamp else None
elif self.entity_description.key == "tomorrow_data_available":
tomorrow_prices = price_info.get("tomorrow", [])
if not tomorrow_prices:
return "No"
# Check if we have a full day of data (24 hours)
return "Yes" if len(tomorrow_prices) == 24 else "Partial"
return None
if key == "price_level":
result = current_hour_data["level"] if current_hour_data else None
elif key in [
"current_price",
"current_price_eur",
"next_hour_price",
"next_hour_price_eur",
]:
result = self._get_price_sensor_value()
elif "price_today" in key:
result = self._get_statistics_value()
elif "rating" in key:
result = self._get_rating_value()
elif key in ["data_timestamp", "tomorrow_data_available"]:
result = self._get_diagnostic_value()
else:
result = None
else:
result = None
except (KeyError, ValueError, TypeError) as ex: except (KeyError, ValueError, TypeError) as ex:
self.coordinator.logger.error( self.coordinator.logger.exception(
"Error getting sensor value", "Error getting sensor value",
extra={ extra={
"error": str(ex), "error": str(ex),
"entity": self.entity_description.key, "entity": self.entity_description.key,
}, },
) )
return None result = None
return result
@property @property
def extra_state_attributes(self) -> dict | None: def extra_state_attributes(self) -> dict | None: # noqa: PLR0912
"""Return additional state attributes.""" """Return additional state attributes."""
try: try:
if not self.coordinator.data: if not self.coordinator.data:
return None return None
subscription = self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"] subscription = self.coordinator.data["data"]["viewer"]["homes"][0][
"currentSubscription"
]
price_info = subscription["priceInfo"] price_info = subscription["priceInfo"]
attributes = {} attributes = {}
# Get current hour's data for timestamp # Get current hour's data for timestamp
now = datetime.now() now = datetime.now(tz=UTC).astimezone()
current_hour_data = None current_hour_data = self._get_current_hour_data()
for price_data in price_info.get("today", []):
starts_at = datetime.fromisoformat(price_data["startsAt"])
if starts_at.hour == now.hour:
current_hour_data = price_data
break
if self.entity_description.key in ["current_price", "current_price_eur"]: if self.entity_description.key in ["current_price", "current_price_eur"]:
attributes["timestamp"] = current_hour_data["startsAt"] if current_hour_data else None attributes["timestamp"] = (
elif self.entity_description.key in ["next_hour_price", "next_hour_price_eur"]: current_hour_data["startsAt"] if current_hour_data else None
)
if self.entity_description.key in [
"next_hour_price",
"next_hour_price_eur",
]:
next_hour = (now.hour + 1) % 24 next_hour = (now.hour + 1) % 24
for price_data in price_info.get("today", []): for price_data in price_info.get("today", []):
starts_at = datetime.fromisoformat(price_data["startsAt"]) starts_at = datetime.fromisoformat(price_data["startsAt"])
if starts_at.hour == next_hour: if starts_at.hour == next_hour:
attributes["timestamp"] = price_data["startsAt"] attributes["timestamp"] = price_data["startsAt"]
break break
elif self.entity_description.key == "price_level":
attributes["timestamp"] = current_hour_data["startsAt"] if current_hour_data else None if self.entity_description.key == "price_level":
elif self.entity_description.key == "lowest_price_today": attributes["timestamp"] = (
attributes["timestamp"] = price_info.get("today", [{}])[0].get("startsAt") current_hour_data["startsAt"] if current_hour_data else None
elif self.entity_description.key == "highest_price_today": )
attributes["timestamp"] = price_info.get("today", [{}])[0].get("startsAt")
elif self.entity_description.key == "average_price_today": if self.entity_description.key == "lowest_price_today":
attributes["timestamp"] = price_info.get("today", [{}])[0].get("startsAt") attributes["timestamp"] = price_info.get("today", [{}])[0].get(
elif self.entity_description.key == "hourly_rating": "startsAt"
attributes["timestamp"] = current_hour_data["startsAt"] if current_hour_data else None )
elif self.entity_description.key == "daily_rating":
attributes["timestamp"] = price_info.get("today", [{}])[0].get("startsAt") if self.entity_description.key == "highest_price_today":
elif self.entity_description.key == "monthly_rating": attributes["timestamp"] = price_info.get("today", [{}])[0].get(
attributes["timestamp"] = price_info.get("today", [{}])[0].get("startsAt") "startsAt"
elif self.entity_description.key == "data_timestamp": )
attributes["timestamp"] = price_info.get("today", [{}])[0].get("startsAt")
elif self.entity_description.key == "tomorrow_data_available": if self.entity_description.key == "average_price_today":
attributes["timestamp"] = price_info.get("today", [{}])[0].get("startsAt") attributes["timestamp"] = price_info.get("today", [{}])[0].get(
"startsAt"
)
if self.entity_description.key == "hourly_rating":
attributes["timestamp"] = (
current_hour_data["startsAt"] if current_hour_data else None
)
if self.entity_description.key == "daily_rating":
attributes["timestamp"] = price_info.get("today", [{}])[0].get(
"startsAt"
)
if self.entity_description.key == "monthly_rating":
attributes["timestamp"] = price_info.get("today", [{}])[0].get(
"startsAt"
)
if self.entity_description.key == "data_timestamp":
attributes["timestamp"] = price_info.get("today", [{}])[0].get(
"startsAt"
)
if self.entity_description.key == "tomorrow_data_available":
attributes["timestamp"] = price_info.get("today", [{}])[0].get(
"startsAt"
)
# Add translated description # Add translated description
if self.hass is not None: if self.hass is not None:
key = f"entity.sensor.{self.entity_description.translation_key}.description" base_key = "entity.sensor"
language_config = getattr(self.hass.config, 'language', None) key = (
f"{base_key}.{self.entity_description.translation_key}.description"
)
language_config = getattr(self.hass.config, "language", None)
if isinstance(language_config, dict): if isinstance(language_config, dict):
description = language_config.get(key) description = language_config.get(key)
if description is not None: if description is not None:
attributes["description"] = description attributes["description"] = description
return attributes if attributes else None return attributes if attributes else None # noqa: TRY300
except (KeyError, ValueError, TypeError) as ex: except (KeyError, ValueError, TypeError) as ex:
self.coordinator.logger.error( self.coordinator.logger.exception(
"Error getting sensor attributes", "Error getting sensor attributes",
extra={ extra={
"error": str(ex), "error": str(ex),

103
custom_components/tibber_prices/translations/de.json
View file

@ -1,103 +0,0 @@
{
"config": {
"step": {
"user": {
"description": "Wenn Sie Hilfe bei der Konfiguration benötigen, schauen Sie hier: https://github.com/jpawlowski/hass.tibber_prices",
"data": {
"access_token": "Tibber Zugangstoken"
}
}
},
"error": {
"auth": "Der Tibber Zugangstoken ist ungültig.",
"connection": "Keine Verbindung zu Tibber möglich. Bitte überprüfen Sie Ihre Internetverbindung.",
"unknown": "Ein unerwarteter Fehler ist aufgetreten. Bitte prüfen Sie die Logs für Details."
},
"abort": {
"already_configured": "Dieser Eintrag ist bereits konfiguriert.",
"entry_not_found": "Tibber-Konfigurationseintrag nicht gefunden."
}
},
"options": {
"step": {
"init": {
"title": "Tibber-Konfiguration aktualisieren",
"description": "Aktualisieren Sie Ihren Tibber-API-Zugangstoken. Wenn Sie einen neuen Token benötigen, können Sie diesen unter https://developer.tibber.com/settings/access-token generieren",
"data": {
"access_token": "Tibber Zugangstoken"
}
}
},
"error": {
"auth": "Der Tibber Zugangstoken ist ungültig.",
"connection": "Keine Verbindung zu Tibber möglich. Bitte überprüfen Sie Ihre Internetverbindung.",
"unknown": "Ein unerwarteter Fehler ist aufgetreten. Bitte prüfen Sie die Logs für Details.",
"different_account": "Der neue Zugangstoken gehört zu einem anderen Tibber-Konto. Bitte verwenden Sie einen Token desselben Kontos oder erstellen Sie eine neue Konfiguration für das andere Konto."
},
"abort": {
"entry_not_found": "Tibber-Konfigurationseintrag nicht gefunden."
}
},
"entity": {
"sensor": {
"current_price": {
"name": "Aktueller Preis",
"description": "Der aktuelle Strompreis dieser Stunde inklusive Steuern"
},
"next_hour_price": {
"name": "Preis nächste Stunde",
"description": "Der Strompreis der nächsten Stunde inklusive Steuern"
},
"price_level": {
"name": "Preisniveau",
"description": "Aktueller Preisindikator (SEHR_GÜNSTIG, GÜNSTIG, NORMAL, TEUER, SEHR_TEUER)"
},
"lowest_price_today": {
"name": "Niedrigster Preis heute",
"description": "Der niedrigste Strompreis des aktuellen Tages"
},
"highest_price_today": {
"name": "Höchster Preis heute",
"description": "Der höchste Strompreis des aktuellen Tages"
},
"average_price_today": {
"name": "Durchschnittspreis heute",
"description": "Der durchschnittliche Strompreis des aktuellen Tages"
},
"hourly_rating": {
"name": "Stündliche Preisbewertung",
"description": "Preisvergleich mit historischen Daten für die aktuelle Stunde (prozentuale Abweichung)"
},
"daily_rating": {
"name": "Tägliche Preisbewertung",
"description": "Preisvergleich mit historischen Daten für den aktuellen Tag (prozentuale Abweichung)"
},
"monthly_rating": {
"name": "Monatliche Preisbewertung",
"description": "Preisvergleich mit historischen Daten für den aktuellen Monat (prozentuale Abweichung)"
},
"data_timestamp": {
"name": "Letzte Datenaktualisierung",
"description": "Zeitstempel der zuletzt von Tibber empfangenen Preisdaten"
},
"tomorrow_data_available": {
"name": "Daten für morgen verfügbar",
"description": "Zeigt an, ob Preisdaten für morgen verfügbar sind (Ja/Nein/Teilweise)"
}
},
"binary_sensor": {
"peak_hour": {
"name": "Spitzenstunde",
"description": "Zeigt an, ob die aktuelle Stunde zu den 20% teuersten Stunden des Tages gehört"
},
"best_price_hour": {
"name": "Beste Preisstunde",
"description": "Zeigt an, ob die aktuelle Stunde zu den 20% günstigsten Stunden des Tages gehört"
},
"connection": {
"name": "Verbindungsstatus",
"description": "Zeigt an, ob aktuelle gültige Preisdaten von Tibber vorliegen"
}
}
}
}

180
custom_components/tibber_prices/translations/en.json
View file

@ -1,103 +1,89 @@
{ {
"config": { "config": {
"step": { "step": {
"user": { "user": {
"description": "If you need help with the configuration have a look here: https://github.com/jpawlowski/hass.tibber_prices", "description": "If you need help with the configuration have a look here: https://github.com/jpawlowski/hass.tibber_prices",
"data": { "data": {
"access_token": "Tibber Access Token" "access_token": "Tibber Access Token"
}
}
},
"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."
},
"abort": {
"already_configured": "This entry is already configured.",
"entry_not_found": "Tibber configuration entry not found."
} }
}
}, },
"options": { "error": {
"step": { "auth": "The Tibber Access Token is invalid.",
"init": { "connection": "Unable to connect to Tibber. Please check your internet connection.",
"title": "Update Tibber Configuration", "unknown": "An unexpected error occurred. Please check the logs for details."
"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",
"data": {
"access_token": "Tibber Access Token"
}
}
},
"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.",
"different_account": "The new access token belongs to a different Tibber account. Please use a token from the same account or create a new configuration for the other account."
},
"abort": {
"entry_not_found": "Tibber configuration entry not found."
}
}, },
"entity": { "abort": {
"sensor": { "already_configured": "This entry is already configured.",
"current_price": { "entry_not_found": "Tibber configuration entry not found."
"name": "Current Price",
"description": "The current hour's electricity price including taxes"
},
"next_hour_price": {
"name": "Next Hour Price",
"description": "The next hour's electricity price including taxes"
},
"price_level": {
"name": "Price Level",
"description": "Current price level indicator (VERY_CHEAP, CHEAP, NORMAL, EXPENSIVE, VERY_EXPENSIVE)"
},
"lowest_price_today": {
"name": "Lowest Price Today",
"description": "The lowest electricity price for the current day"
},
"highest_price_today": {
"name": "Highest Price Today",
"description": "The highest electricity price for the current day"
},
"average_price_today": {
"name": "Average Price Today",
"description": "The average electricity price for the current day"
},
"hourly_rating": {
"name": "Hourly Price Rating",
"description": "Price comparison with historical data for the current hour (percentage difference)"
},
"daily_rating": {
"name": "Daily Price Rating",
"description": "Price comparison with historical data for the current day (percentage difference)"
},
"monthly_rating": {
"name": "Monthly Price Rating",
"description": "Price comparison with historical data for the current month (percentage difference)"
},
"data_timestamp": {
"name": "Last Data Update",
"description": "Timestamp of the most recent price data received from Tibber"
},
"tomorrow_data_available": {
"name": "Tomorrow's Data Available",
"description": "Indicates if price data for tomorrow is available (Yes/No/Partial)"
}
},
"binary_sensor": {
"peak_hour": {
"name": "Peak Hour",
"description": "Indicates if the current hour is in the top 20% most expensive hours of the day"
},
"best_price_hour": {
"name": "Best Price Hour",
"description": "Indicates if the current hour is in the bottom 20% cheapest hours of the day"
},
"connection": {
"name": "Connection Status",
"description": "Indicates if we have valid current price data from Tibber"
}
}
} }
},
"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",
"data": {
"access_token": "Tibber Access Token"
}
}
},
"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.",
"different_account": "The new access token belongs to a different Tibber account. Please use a token from the same account or create a new configuration for the other account."
},
"abort": {
"entry_not_found": "Tibber configuration entry not found."
}
},
"entity": {
"sensor": {
"current_price": {
"name": "Current Price"
},
"next_hour_price": {
"name": "Next Hour Price"
},
"price_level": {
"name": "Price Level"
},
"lowest_price_today": {
"name": "Lowest Price Today"
},
"highest_price_today": {
"name": "Highest Price Today"
},
"average_price_today": {
"name": "Average Price Today"
},
"hourly_rating": {
"name": "Hourly Price Rating"
},
"daily_rating": {
"name": "Daily Price Rating"
},
"monthly_rating": {
"name": "Monthly Price Rating"
},
"data_timestamp": {
"name": "Last Data Update"
},
"tomorrow_data_available": {
"name": "Tomorrow's Data Available"
}
},
"binary_sensor": {
"peak_hour": {
"name": "Peak Hour"
},
"best_price_hour": {
"name": "Best Price Hour"
},
"connection": {
"name": "Connection Status"
}
}
}
} }

6
hacs.json
View file

@ -1,5 +1,5 @@
{ {
"name": "Tibber Price Information & Ratings", "name": "Tibber Price Information & Ratings",
"homeassistant": "2025.4.2", "homeassistant": "2025.4.2",
"hacs": "2.0.1" "hacs": "2.0.1"
} }

391
scripts/json_schemas/manifest_schema.json Normal file
View file

@ -0,0 +1,391 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "Home Assistant integration manifest",
"description": "The manifest for a Home Assistant integration",
"type": "object",
"if": {
"properties": { "integration_type": { "const": "virtual" } },
"required": ["integration_type"]
},
"then": {
"oneOf": [
{
"properties": {
"domain": {
"description": "The domain identifier of the integration.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#domain",
"examples": ["mobile_app"],
"type": "string",
"pattern": "[0-9a-z_]+"
},
"name": {
"description": "The friendly name of the integration.",
"type": "string"
},
"integration_type": {
"description": "The integration type.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#integration-type",
"const": "virtual"
},
"iot_standards": {
"description": "The IoT standards which supports devices or services of this virtual integration.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#iot-standards",
"type": "array",
"minItems": 1,
"items": {
"type": "string",
"enum": ["homekit", "zigbee", "zwave"]
}
}
},
"additionalProperties": false,
"required": ["domain", "name", "integration_type", "iot_standards"]
},
{
"properties": {
"domain": {
"description": "The domain identifier of the integration.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#domain",
"examples": ["mobile_app"],
"type": "string",
"pattern": "[0-9a-z_]+"
},
"name": {
"description": "The friendly name of the integration.",
"type": "string"
},
"integration_type": {
"description": "The integration type.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#integration-type",
"const": "virtual"
},
"supported_by": {
"description": "The integration which supports devices or services of this virtual integration.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#supported-by",
"type": "string"
}
},
"additionalProperties": false,
"required": ["domain", "name", "integration_type", "supported_by"]
}
]
},
"else": {
"properties": {
"domain": {
"description": "The domain identifier of the integration.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#domain",
"examples": ["mobile_app"],
"type": "string",
"pattern": "[0-9a-z_]+"
},
"name": {
"description": "The friendly name of the integration.",
"type": "string"
},
"integration_type": {
"description": "The integration type.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#integration-type",
"type": "string",
"default": "hub",
"enum": [
"device",
"entity",
"hardware",
"helper",
"hub",
"service",
"system"
]
},
"config_flow": {
"description": "Whether the integration is configurable from the UI.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#config-flow",
"type": "boolean"
},
"mqtt": {
"description": "A list of topics to subscribe for the discovery of devices via MQTT.\nThis requires to specify \"mqtt\" in either the \"dependencies\" or \"after_dependencies\".\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#mqtt",
"type": "array",
"items": {
"type": "string"
},
"uniqueItems": true
},
"zeroconf": {
"description": "A list containing service domains to search for devices to discover via Zeroconf. Items can either be strings, which discovers all devices in the specific service domain, and/or objects which include filters. (useful for generic service domains like _http._tcp.local.)\nA device is discovered if it matches one of the items, but inside the individual item all properties have to be matched.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#zeroconf",
"type": "array",
"minItems": 1,
"items": {
"anyOf": [
{
"type": "string",
"pattern": "^.*\\.local\\.$",
"description": "Service domain to search for devices."
},
{
"type": "object",
"properties": {
"type": {
"description": "The service domain to search for devices.",
"examples": ["_http._tcp.local."],
"type": "string",
"pattern": "^.*\\.local\\.$"
},
"name": {
"description": "The name or name pattern of the devices to filter.",
"type": "string"
},
"properties": {
"description": "The properties of the Zeroconf advertisement to filter.",
"type": "object",
"additionalProperties": { "type": "string" }
}
},
"required": ["type"],
"additionalProperties": false
}
]
},
"uniqueItems": true
},
"ssdp": {
"description": "A list of matchers to find devices discoverable via SSDP/UPnP. In order to be discovered, the device has to match all properties of any of the matchers.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#ssdp",
"type": "array",
"minItems": 1,
"items": {
"description": "A matcher for the SSDP discovery.",
"type": "object",
"properties": {
"st": {
"type": "string"
},
"deviceType": {
"type": "string"
},
"manufacturer": {
"type": "string"
},
"modelDescription": {
"type": "string"
}
},
"additionalProperties": { "type": "string" }
}
},
"bluetooth": {
"description": "A list of matchers to find devices discoverable via Bluetooth. In order to be discovered, the device has to match all properties of any of the matchers.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#bluetooth",
"type": "array",
"minItems": 1,
"items": {
"description": "A matcher for the bluetooth discovery",
"type": "object",
"properties": {
"connectable": {
"description": "Whether the device needs to be connected to or it works with just advertisement data.",
"type": "boolean"
},
"local_name": {
"description": "The name or a name pattern of the device to match.",
"type": "string",
"pattern": "^([^*]+|[^*]{3,}[*].*)$"
},
"service_uuid": {
"description": "The 128-bit service data UUID to match.",
"type": "string",
"pattern": "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
},
"service_data_uuid": {
"description": "The 16-bit service data UUID to match, converted into the corresponding 128-bit UUID by replacing the 3rd and 4th byte of `00000000-0000-1000-8000-00805f9b34fb` with the 16-bit UUID.",
"examples": ["0000fd3d-0000-1000-8000-00805f9b34fb"],
"type": "string",
"pattern": "0000[0-9a-f]{4}-0000-1000-8000-00805f9b34fb"
},
"manufacturer_id": {
"description": "The Manufacturer ID to match.",
"type": "integer"
},
"manufacturer_data_start": {
"description": "The start bytes of the manufacturer data to match.",
"type": "array",
"minItems": 1,
"items": {
"type": "integer",
"minimum": 0,
"maximum": 255
}
}
},
"additionalProperties": false
},
"uniqueItems": true
},
"homekit": {
"description": "A list of model names to find devices which are discoverable via HomeKit. A device is discovered if the model name of the device starts with any of the specified model names.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#homekit",
"type": "object",
"properties": {
"models": {
"description": "The model names to search for.",
"type": "array",
"items": {
"type": "string"
},
"uniqueItems": true
}
},
"required": ["models"],
"additionalProperties": false
},
"dhcp": {
"description": "A list of matchers to find devices discoverable via DHCP. In order to be discovered, the device has to match all properties of any of the matchers.\nYou can specify an item with \"registered_devices\" set to true to check for devices with MAC addresses specified in the device registry.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#dhcp",
"type": "array",
"items": {
"anyOf": [
{
"type": "object",
"properties": {
"registered_devices": {
"description": "Whether the MAC addresses of devices in the device registry should be used for discovery, useful if the discovery is used to update the IP address of already registered devices.",
"const": true
}
},
"additionalProperties": false
},
{
"type": "object",
"properties": {
"hostname": {
"description": "The hostname or hostname pattern to match.",
"type": "string"
},
"macaddress": {
"description": "The MAC address or MAC address pattern to match.",
"type": "string",
"maxLength": 12
}
},
"additionalProperties": false
}
]
},
"uniqueItems": true
},
"usb": {
"description": "A list of matchers to find devices discoverable via USB. In order to be discovered, the device has to match all properties of any of the matchers.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#usb",
"type": "array",
"uniqueItems": true,
"items": {
"type": "object",
"additionalProperties": false,
"properties": {
"vid": {
"description": "The vendor ID to match.",
"type": "string",
"pattern": "[0-9A-F]{4}"
},
"pid": {
"description": "The product ID to match.",
"type": "string",
"pattern": "[0-9A-F]{4}"
},
"description": {
"description": "The USB device description to match.",
"type": "string"
},
"manufacturer": {
"description": "The manufacturer to match.",
"type": "string"
},
"serial_number": {
"description": "The serial number to match.",
"type": "string"
},
"known_devices": {
"type": "array",
"items": {
"type": "string"
}
}
}
}
},
"documentation": {
"description": "The website containing the documentation for the integration. It has to be in the format \"https://www.home-assistant.io/integrations/[domain]\"\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#documentation",
"type": "string",
"pattern": "^https://www.home-assistant.io/integrations/[0-9a-z_]+$",
"format": "uri"
},
"quality_scale": {
"description": "The quality scale of the integration.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#integration-quality-scale",
"type": "string",
"enum": ["bronze", "silver", "gold", "platinum", "internal", "legacy"]
},
"requirements": {
"description": "The PyPI package requirements for the integration. The package has to be pinned to a specific version.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#requirements",
"type": "array",
"items": {
"type": "string",
"pattern": ".+==.+"
},
"uniqueItems": true
},
"dependencies": {
"description": "A list of integrations which need to be loaded before this integration can be set up.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#dependencies",
"type": "array",
"items": {
"type": "string"
},
"minItems": 1,
"uniqueItems": true
},
"after_dependencies": {
"description": "A list of integrations which need to be loaded before this integration is set up when it is configured. The integration will still be set up when the \"after_dependencies\" are not configured.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#after-dependencies",
"type": "array",
"items": {
"type": "string"
},
"minItems": 1,
"uniqueItems": true
},
"codeowners": {
"description": "A list of GitHub usernames or GitHub team names of the integration owners.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#code-owners",
"type": "array",
"minItems": 0,
"items": {
"type": "string",
"pattern": "^@.+$"
},
"uniqueItems": true
},
"loggers": {
"description": "A list of logger names used by the requirements.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#loggers",
"type": "array",
"minItems": 1,
"items": {
"type": "string"
},
"uniqueItems": true
},
"disabled": {
"description": "The reason for the integration being disabled.",
"type": "string"
},
"iot_class": {
"description": "The IoT class of the integration, describing how the integration connects to the device or service.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#iot-class",
"type": "string",
"enum": [
"assumed_state",
"cloud_polling",
"cloud_push",
"local_polling",
"local_push",
"calculated"
]
},
"single_config_entry": {
"description": "Whether the integration only supports a single config entry.\nhttps://developers.home-assistant.io/docs/creating_integration_manifest/#single-config-entry-only",
"const": true
}
},
"additionalProperties": false,
"required": ["domain", "name", "codeowners", "documentation"],
"dependencies": {
"mqtt": {
"anyOf": [
{ "required": ["dependencies"] },
{ "required": ["after_dependencies"] }
]
}
}
}
}

51
scripts/json_schemas/translation_schema.json Normal file
View file

@ -0,0 +1,51 @@
{
"title": "Component Title",
"config": {
"step": {
"user": {
"title": "Step Title",
"description": "Step Description",
"data": {
"field_name": "Field Label"
}
}
},
"error": {
"error_key": "Error Message"
},
"abort": {
"abort_key": "Abort Message"
}
},
"entity": {
"sensor": {
"sensor_key": {
"name": "Sensor Name",
"state": {
"state_key": "State Translation"
}
}
},
"binary_sensor": {
"sensor_key": {
"name": "Binary Sensor Name",
"state": {
"on": "On State Translation",
"off": "Off State Translation"
}
}
}
},
"services": {
"service_name": {
"name": "Service Name",
"description": "Service Description",
"fields": {
"field_name": {
"name": "Field Name",
"description": "Field Description"
}
}
}
}
}