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-05-18 14:11:27 +00:00
parent 1dc2457564
commit 71b81a9812
5 changed files with 217 additions and 121 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
custom_components/tibber_prices/api.py
View file

@ -38,7 +38,7 @@ class QueryType(Enum):
DAILY_RATING = "daily"
HOURLY_RATING = "hourly"
MONTHLY_RATING = "monthly"
TEST = "test"
VIEWER = "viewer"
class TibberPricesApiClientError(Exception):
@ -215,7 +215,7 @@ def _transform_data(data: dict, query_type: QueryType) -> dict:
QueryType.MONTHLY_RATING,
):
return data
if query_type == QueryType.TEST:
if query_type == QueryType.VIEWER:
return data
_LOGGER.warning("Unknown query type %s, returning raw data", query_type)
@ -286,19 +286,32 @@ class TibberPricesApiClient:
self._max_retries = 3
self._retry_delay = 2
async def async_test_connection(self) -> Any:
async def async_get_viewer_details(self) -> Any:
"""Test connection to the API."""
return await self._api_wrapper(
data={
"query": """
query {
{
viewer {
userId
name
login
homes {
id
type
appNickname
address {
address1
postalCode
city
country
}
}
}
}
"""
},
query_type=QueryType.TEST,
query_type=QueryType.VIEWER,
)
async def async_get_price_info(self) -> Any:
@ -306,7 +319,7 @@ class TibberPricesApiClient:
return await self._api_wrapper(
data={
"query": """
{viewer{homes{currentSubscription{priceInfo{
{viewer{homes{id,currentSubscription{priceInfo{
range(resolution:HOURLY,last:48){edges{node{
startsAt total energy tax level
}}}
@ -322,7 +335,7 @@ class TibberPricesApiClient:
return await self._api_wrapper(
data={
"query": """
{viewer{homes{currentSubscription{priceRating{
{viewer{homes{id,currentSubscription{priceRating{
thresholdPercentages{low high}
daily{entries{time total energy tax difference level}}
}}}}}"""
@ -335,7 +348,7 @@ class TibberPricesApiClient:
return await self._api_wrapper(
data={
"query": """
{viewer{homes{currentSubscription{priceRating{
{viewer{homes{id,currentSubscription{priceRating{
thresholdPercentages{low high}
hourly{entries{time total energy tax difference level}}
}}}}}"""
@ -348,7 +361,7 @@ class TibberPricesApiClient:
return await self._api_wrapper(
data={
"query": """
{viewer{homes{currentSubscription{priceRating{
{viewer{homes{id,currentSubscription{priceRating{
thresholdPercentages{low high}
monthly{
currency
@ -455,7 +468,7 @@ class TibberPricesApiClient:
query_type,
)
if query_type != QueryType.TEST and _is_data_empty(response_data, query_type.value):
if query_type != QueryType.VIEWER and _is_data_empty(response_data, query_type.value):
_LOGGER.debug("Empty data detected for query_type: %s", query_type)
raise TibberPricesApiClientError(
TibberPricesApiClientError.EMPTY_DATA_ERROR.format(query_type=query_type.value)
@ -467,7 +480,7 @@ class TibberPricesApiClient:
self,
data: dict | None = None,
headers: dict | None = None,
query_type: QueryType = QueryType.TEST,
query_type: QueryType = QueryType.VIEWER,
) -> Any:
"""Get information from the API with rate limiting and retry logic."""
headers = headers or _prepare_headers(self._access_token)

183
custom_components/tibber_prices/config_flow.py
View file

@ -3,7 +3,6 @@
from __future__ import annotations
import voluptuous as vol
from slugify import slugify
from homeassistant import config_entries
from homeassistant.const import CONF_ACCESS_TOKEN
@ -37,6 +36,7 @@ class TibberPricesFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Initialize the config flow."""
super().__init__()
self._reauth_entry: config_entries.ConfigEntry | None = None
self._pending_user_input: dict | None = None
@staticmethod
def async_get_options_flow(
@ -53,11 +53,11 @@ class TibberPricesFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
self,
user_input: dict | None = None,
) -> config_entries.ConfigFlowResult:
"""Handle a flow initialized by the user."""
"""Handle a flow initialized by the user. Only ask for access token."""
_errors = {}
if user_input is not None:
try:
name = await self._test_credentials(access_token=user_input[CONF_ACCESS_TOKEN])
viewer = await self._get_viewer_details(access_token=user_input[CONF_ACCESS_TOKEN])
except TibberPricesApiClientAuthenticationError as exception:
LOGGER.warning(exception)
_errors["base"] = "auth"
@ -68,12 +68,12 @@ class TibberPricesFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
LOGGER.exception(exception)
_errors["base"] = "unknown"
else:
await self.async_set_unique_id(unique_id=slugify(name))
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=name,
data=user_input,
)
# Store viewer for use in finish step
self._pending_user_input = {
"access_token": user_input[CONF_ACCESS_TOKEN],
"viewer": viewer,
}
return await self.async_step_finish()
return self.async_show_form(
step_id="user",
@ -87,89 +87,84 @@ class TibberPricesFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
type=selector.TextSelectorType.TEXT,
),
),
vol.Optional(
CONF_EXTENDED_DESCRIPTIONS,
default=(user_input or {}).get(CONF_EXTENDED_DESCRIPTIONS, DEFAULT_EXTENDED_DESCRIPTIONS),
): selector.BooleanSelector(),
vol.Optional(
CONF_BEST_PRICE_FLEX,
default=(user_input or {}).get(CONF_BEST_PRICE_FLEX, DEFAULT_BEST_PRICE_FLEX),
): selector.NumberSelector(
selector.NumberSelectorConfig(
min=0,
max=20,
step=1,
mode=selector.NumberSelectorMode.SLIDER,
),
),
vol.Optional(
CONF_PEAK_PRICE_FLEX,
default=(user_input or {}).get(CONF_PEAK_PRICE_FLEX, DEFAULT_PEAK_PRICE_FLEX),
): selector.NumberSelector(
selector.NumberSelectorConfig(
min=0,
max=20,
step=1,
mode=selector.NumberSelectorMode.SLIDER,
),
),
},
),
errors=_errors,
)
async def _test_credentials(self, access_token: str) -> str:
"""Validate credentials and return the user's name."""
async def async_step_finish(self, user_input: dict | None = None) -> config_entries.ConfigFlowResult:
"""Show a finish screen after successful setup, then create entry on submit."""
if self._pending_user_input is not None and user_input is None:
# First visit: show home selection
viewer = self._pending_user_input["viewer"]
homes = viewer.get("homes", [])
# Build choices: label = address or nickname, value = id
home_choices = {}
for home in homes:
label = home.get("appNickname") or home.get("address", {}).get("address1") or home["id"]
if home.get("address", {}).get("city"):
label += f", {home['address']['city']}"
home_choices[home["id"]] = label
schema = vol.Schema({vol.Required("home_id"): vol.In(home_choices)})
return self.async_show_form(
step_id="finish",
data_schema=schema,
description_placeholders={},
errors={},
last_step=True,
)
if self._pending_user_input is not None and user_input is not None:
# User selected home, create entry
home_id = user_input["home_id"]
viewer = self._pending_user_input["viewer"]
# Use the same label as shown to the user for the config entry title
home_label = None
for home in viewer.get("homes", []):
if home["id"] == home_id:
home_label = home.get("appNickname") or home.get("address", {}).get("address1") or home_id
if home.get("address", {}).get("city"):
home_label += f", {home['address']['city']}"
break
if not home_label:
home_label = viewer.get("name", "Tibber")
data = {
CONF_ACCESS_TOKEN: self._pending_user_input["access_token"],
CONF_EXTENDED_DESCRIPTIONS: DEFAULT_EXTENDED_DESCRIPTIONS,
CONF_BEST_PRICE_FLEX: DEFAULT_BEST_PRICE_FLEX,
CONF_PEAK_PRICE_FLEX: DEFAULT_PEAK_PRICE_FLEX,
}
self._pending_user_input = None
# Set unique_id to home_id
await self.async_set_unique_id(unique_id=home_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=home_label,
data=data,
)
return self.async_abort(reason="setup_complete")
async def _get_viewer_details(self, access_token: str) -> dict:
"""Validate credentials and return information about the account (viewer object)."""
client = TibberPricesApiClient(
access_token=access_token,
session=async_create_clientsession(self.hass),
)
result = await client.async_test_connection()
return result["viewer"]["name"]
result = await client.async_get_viewer_details()
return result["viewer"]
class TibberPricesOptionsFlowHandler(config_entries.OptionsFlow):
"""Tibber Prices config flow options handler."""
def __init__(self, _: config_entries.ConfigEntry) -> None:
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
super().__init__()
self.config_entry = config_entry
async def async_step_init(self, user_input: dict | None = None) -> config_entries.ConfigFlowResult:
"""Manage the options."""
errors: dict[str, str] = {}
if user_input is not None:
try:
# Test the new access token and get account name
client = TibberPricesApiClient(
access_token=user_input[CONF_ACCESS_TOKEN],
session=async_create_clientsession(self.hass),
)
result = await client.async_test_connection()
new_account_name = result["viewer"]["name"]
# Check if this token is for the same account
current_unique_id = self.config_entry.unique_id
new_unique_id = slugify(new_account_name)
if current_unique_id != new_unique_id:
# Token is for a different account
errors["base"] = "different_account"
else:
# Update the config entry with the new access token and options
return self.async_create_entry(title="", data=user_input)
except TibberPricesApiClientAuthenticationError as exception:
LOGGER.warning(exception)
errors["base"] = "auth"
except TibberPricesApiClientCommunicationError as exception:
LOGGER.error(exception)
errors["base"] = "connection"
except TibberPricesApiClientError as exception:
LOGGER.exception(exception)
errors["base"] = "unknown"
# Build options schema
options = {
vol.Required(
@ -221,8 +216,54 @@ class TibberPricesOptionsFlowHandler(config_entries.OptionsFlow):
),
}
if user_input is not None:
# Validate new access token if changed
new_token = user_input.get(CONF_ACCESS_TOKEN, self.config_entry.data.get(CONF_ACCESS_TOKEN, "")) or ""
current_home_id = self.config_entry.data.get("home_id", "")
errors = {}
if new_token != self.config_entry.data.get(CONF_ACCESS_TOKEN, ""):
try:
client = TibberPricesApiClient(
access_token=new_token,
session=async_create_clientsession(self.hass),
)
result = await client.async_get_viewer_details()
homes = result["viewer"].get("homes", [])
if not any(home["id"] == current_home_id for home in homes):
errors[CONF_ACCESS_TOKEN] = "different_home"
except TibberPricesApiClientAuthenticationError as exception:
LOGGER.warning(exception)
errors[CONF_ACCESS_TOKEN] = "auth"
except TibberPricesApiClientCommunicationError as exception:
LOGGER.error(exception)
errors[CONF_ACCESS_TOKEN] = "connection"
except TibberPricesApiClientError as exception:
LOGGER.exception(exception)
errors[CONF_ACCESS_TOKEN] = "unknown"
if errors:
# Show form again with errors
description_placeholders = {
"access_token": new_token,
"home_id": current_home_id,
}
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(options),
errors=errors,
description_placeholders=description_placeholders,
)
# Only update options and access token if valid
return self.async_create_entry(title="", data=user_input)
# Prepare read-only info for description placeholders
description_placeholders = {
"access_token": self.config_entry.data.get(CONF_ACCESS_TOKEN, ""),
"unique_id": self.config_entry.unique_id or "",
}
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(options),
errors=errors,
description_placeholders=description_placeholders,
)

40
custom_components/tibber_prices/entity.py
View file

@ -2,7 +2,7 @@
from __future__ import annotations
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import ATTRIBUTION, DOMAIN
@ -19,21 +19,43 @@ class TibberPricesEntity(CoordinatorEntity[TibberPricesDataUpdateCoordinator]):
"""Initialize."""
super().__init__(coordinator)
# Get home name from Tibber API if available
home_name = None
# enum of home types
home_types = {
"APARTMENT": "Apartment",
"ROWHOUSE": "Rowhouse",
"HOUSE": "House",
"COTTAGE": "Cottage",
}
# Get home info from Tibber API if available
home_name = "Tibber Home"
home_id = self.coordinator.config_entry.unique_id
home_type = None
city = None
app_nickname = None
address1 = None
if coordinator.data:
try:
home = coordinator.data["data"]["viewer"]["homes"][0]
home_name = home.get("address", {}).get("address1", "Tibber Home")
except (KeyError, IndexError):
home_id = self.unique_id
address1 = str(coordinator.data.get("address", {}).get("address1", ""))
city = str(coordinator.data.get("address", {}).get("city", ""))
app_nickname = str(coordinator.data.get("appNickname", ""))
home_type = str(coordinator.data.get("type", ""))
# Compose a nice name
home_name = "Tibber " + (app_nickname or address1 or "Home")
if city:
home_name = f"{home_name}, {city}"
except (KeyError, IndexError, TypeError):
home_name = "Tibber Home"
else:
home_name = "Tibber Home"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, coordinator.config_entry.entry_id)},
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN, coordinator.config_entry.unique_id or coordinator.config_entry.entry_id)},
name=home_name,
manufacturer="Tibber",
model="Price API",
sw_version=str(coordinator.config_entry.version),
model=home_types.get(home_type, "Unknown") if home_type else "Unknown",
model_id=home_type if home_type else None,
serial_number=home_id if home_id else None,
)

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

@ -2,14 +2,20 @@
"config": {
"step": {
"user": {
"description": "Richte Tibber Preisinformationen & Bewertungen ein. Um ein API-Zugriffstoken zu generieren, besuche developer.tibber.com.",
"description": "Richte Tibber Preisinformationen & Bewertungen ein.\n\nUm einen API-Zugriffstoken zu generieren, besuche https://developer.tibber.com.",
"data": {
"access_token": "API-Zugriffstoken",
"extended_descriptions": "Erweiterte Beschreibungen in Entitätsattributen anzeigen",
"best_price_flex": "Flexibilität für Bestpreis (%)",
"peak_price_flex": "Flexibilität für Spitzenpreis (%)"
"access_token": "API-Zugriffstoken"
},
"title": "Tibber Preisinformationen & Bewertungen"
"title": "Tibber Preisinformationen & Bewertungen",
"submit": "Token validieren"
},
"finish": {
"description": "Wähle ein Zuhause, um Preisinformationen und Bewertungen abzurufen.",
"data": {
"home_id": "Home ID"
},
"title": "Wähle ein Zuhause",
"submit": "Zuhause auswählen"
}
},
"error": {
@ -21,31 +27,37 @@
},
"abort": {
"already_configured": "Integration ist bereits konfiguriert",
"entry_not_found": "Tibber Konfigurationseintrag nicht gefunden."
"entry_not_found": "Tibber Konfigurationseintrag nicht gefunden.",
"setup_complete": "Einrichtung abgeschlossen! Du kannst zusätzliche Optionen für Tibber Preise in den Integrationsoptionen ändern, nachdem du diesen Dialog geschlossen hast."
}
},
"options": {
"step": {
"init": {
"title": "Optionen für Tibber Preisinformationen & Bewertungen",
"description": "Konfiguriere Optionen für Tibber Preisinformationen & Bewertungen",
"description": "Home ID: {unique_id}",
"data": {
"access_token": "Tibber Zugangstoken",
"access_token": "API-Zugriffstoken",
"extended_descriptions": "Erweiterte Beschreibungen in Entitätsattributen anzeigen",
"best_price_flex": "Flexibilität für Bestpreis (%)",
"peak_price_flex": "Flexibilität für Spitzenpreis (%)"
}
},
"title": "Optionen für Tibber Preisinformationen & Bewertungen",
"submit": "Optionen speichern"
}
},
"error": {
"auth": "Der Tibber Zugangstoken ist ungültig.",
"connection": "Verbindung zu Tibber nicht möglich. Bitte überprüfe deine Internetverbindung.",
"unknown": "Ein unerwarteter Fehler ist aufgetreten. Bitte überprüfe die Logs für Details.",
"different_account": "Der neue Zugangstoken gehört zu einem anderen Tibber-Konto. Bitte verwende einen Token vom selben Konto oder erstelle eine neue Konfiguration für das andere Konto."
"cannot_connect": "Verbindung fehlgeschlagen",
"invalid_access_token": "Ungültiges Zugriffstoken",
"different_home": "Der Zugriffstoken ist nicht gültig für die Home ID, für die diese Integration konfiguriert ist."
},
"abort": {
"entry_not_found": "Tibber Konfigurationseintrag nicht gefunden."
}
},
"best_price_flex": "Bestpreis Flexibilität (%)",
"peak_price_flex": "Spitzenpreis Flexibilität (%)"
},
"entity": {
"sensor": {
@ -98,4 +110,4 @@
}
}
}
}
}

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

@ -2,14 +2,20 @@
"config": {
"step": {
"user": {
"description": "Set up Tibber Price Information & Ratings. To generate an API access token, visit developer.tibber.com.",
"description": "Set up Tibber Price Information & Ratings.\n\nTo generate an API access token, visit https://developer.tibber.com.",
"data": {
"access_token": "API access token",
"extended_descriptions": "Show extended descriptions in entity attributes",
"best_price_flex": "Best Price Flexibility (%)",
"peak_price_flex": "Peak Price Flexibility (%)"
"access_token": "API access token"
},
"title": "Tibber Price Information & Ratings"
"title": "Tibber Price Information & Ratings",
"submit": "Validate Token"
},
"finish": {
"description": "Select a home to fetch price information and ratings.",
"data": {
"home_id": "Home ID"
},
"title": "Pick a home",
"submit": "Select Home"
}
},
"error": {
@ -21,29 +27,31 @@
},
"abort": {
"already_configured": "Integration is already configured",
"entry_not_found": "Tibber configuration entry not found."
},
"best_price_flex": "Best Price Flexibility (%)",
"peak_price_flex": "Peak Price Flexibility (%)"
"entry_not_found": "Tibber configuration entry not found.",
"setup_complete": "Setup complete! You can change additional options for Tibber Prices in the integration's options after closing this dialog."
}
},
"options": {
"step": {
"init": {
"title": "Options for Tibber Price Information & Ratings",
"description": "Configure options for Tibber Price Information & Ratings",
"description": "Home ID: {unique_id}",
"data": {
"access_token": "Tibber Access Token",
"access_token": "API access token",
"extended_descriptions": "Show extended descriptions in entity attributes",
"best_price_flex": "Best Price Flexibility (%)",
"peak_price_flex": "Peak Price Flexibility (%)"
}
},
"title": "Options for Tibber Price Information & Ratings",
"submit": "Save Options"
}
},
"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."
"cannot_connect": "Failed to connect",
"invalid_access_token": "Invalid access token",
"different_home": "The access token is not valid for the home ID this integration is configured for."
},
"abort": {
"entry_not_found": "Tibber configuration entry not found."
@ -117,4 +125,4 @@
}
}
}
}
}