mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-05-28 18:43:40 +00:00
6e0613c055
New services for finding optimal electricity price windows: - find_cheapest_block: Cheapest contiguous time block (e.g., dishwasher) - find_cheapest_hours: Cheapest N hours, non-contiguous (e.g., EV charging) - find_cheapest_schedule: Multi-task scheduling with no-overlap (e.g., shared circuit) - find_most_expensive_block: Most expensive contiguous block (peak avoidance) - find_most_expensive_hours: Most expensive N hours (consumption shifting) Key features: - Flexible search range (today, tomorrow, today+tomorrow, rolling window) - Power profile support for variable consumption patterns - Price level filtering (e.g., only CHEAP/VERY_CHEAP intervals) - Comparison details showing savings vs. alternatives - Sliding window algorithm (O(n)) for block search, greedy scheduling for multi-task optimization Also includes: - Shared validation utilities (search range, price level, power profile) - entry_id now optional on all services (auto-selects single home) - Input validation for existing services (time range, filter conflicts) - Service icons for all new and existing services - Translations for all 5 languages (en, de, nb, nl, sv) - Removed 10 unused config.error translation keys (replaced by exceptions) - Tests for price window algorithms and search range resolution Impact: Users can find optimal time windows for appliances, EV charging, and multi-device scheduling via HA service calls. Existing services improved with optional entry_id and better input validation.
107 lines
2.9 KiB
Python
107 lines
2.9 KiB
Python
"""
|
|
User data refresh service handler.
|
|
|
|
This module implements the `refresh_user_data` service, which forces a refresh
|
|
of user profile and home information from the Tibber API.
|
|
|
|
Features:
|
|
- Force refresh of cached user data
|
|
- Bypass 24h cache TTL
|
|
- Return updated user profile and homes
|
|
- Error handling for API failures
|
|
|
|
Service: tibber_prices.refresh_user_data
|
|
Response: JSON with refresh status and updated data
|
|
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING, Any, Final
|
|
|
|
import voluptuous as vol
|
|
|
|
from custom_components.tibber_prices.api import (
|
|
TibberPricesApiClientAuthenticationError,
|
|
TibberPricesApiClientCommunicationError,
|
|
TibberPricesApiClientError,
|
|
)
|
|
from homeassistant.exceptions import ServiceValidationError
|
|
|
|
from .helpers import get_entry_and_data
|
|
|
|
if TYPE_CHECKING:
|
|
from homeassistant.core import ServiceCall
|
|
|
|
# Service constants
|
|
REFRESH_USER_DATA_SERVICE_NAME: Final = "refresh_user_data"
|
|
ATTR_ENTRY_ID: Final = "entry_id"
|
|
|
|
# Service schema
|
|
REFRESH_USER_DATA_SERVICE_SCHEMA: Final = vol.Schema(
|
|
{
|
|
vol.Optional(ATTR_ENTRY_ID, default=""): str,
|
|
}
|
|
)
|
|
|
|
|
|
async def handle_refresh_user_data(call: ServiceCall) -> dict[str, Any]:
|
|
"""
|
|
Refresh user data for a specific config entry.
|
|
|
|
Forces a refresh of user profile and home information from Tibber API,
|
|
bypassing the 24-hour cache TTL. Returns updated information or error details.
|
|
|
|
See services.yaml for detailed parameter documentation.
|
|
|
|
Args:
|
|
call: Service call with parameters
|
|
|
|
Returns:
|
|
Dictionary with refresh status and updated data
|
|
|
|
Raises:
|
|
ServiceValidationError: If entry_id is missing or invalid
|
|
|
|
"""
|
|
entry_id = call.data.get(ATTR_ENTRY_ID, "")
|
|
hass = call.hass
|
|
|
|
# Get the entry and coordinator
|
|
try:
|
|
_, coordinator, _ = get_entry_and_data(hass, entry_id)
|
|
except ServiceValidationError as ex:
|
|
return {
|
|
"success": False,
|
|
"message": f"Invalid entry ID: {ex}",
|
|
}
|
|
|
|
# Force refresh user data using the public method
|
|
try:
|
|
updated = await coordinator.refresh_user_data()
|
|
except (
|
|
TibberPricesApiClientAuthenticationError,
|
|
TibberPricesApiClientCommunicationError,
|
|
TibberPricesApiClientError,
|
|
) as ex:
|
|
return {
|
|
"success": False,
|
|
"message": f"API error refreshing user data: {ex!s}",
|
|
}
|
|
else:
|
|
if updated:
|
|
user_profile = coordinator.get_user_profile()
|
|
homes = coordinator.get_user_homes()
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "User data refreshed successfully",
|
|
"user_profile": user_profile,
|
|
"homes_count": len(homes),
|
|
"homes": homes,
|
|
"last_updated": user_profile.get("last_updated"),
|
|
}
|
|
return {
|
|
"success": False,
|
|
"message": "User data was already up to date",
|
|
}
|