mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-03-30 21:33:39 +00:00
2de793cfda
Changed from centralized main+subentry coordinator pattern to independent
coordinators per home. Each config entry now manages its own home data
with its own API client and access token.
Architecture changes:
- API Client: async_get_price_info() changed from home_ids: set[str] to home_id: str
* Removed GraphQL alias pattern (home0, home1, ...)
* Single-home query structure without aliasing
* Simplified response parsing (viewer.home instead of viewer.home0)
- Coordinator: Removed main/subentry distinction
* Deleted is_main_entry() and _has_existing_main_coordinator()
* Each coordinator fetches its own data independently
* Removed _find_main_coordinator() and _get_configured_home_ids()
* Simplified _async_update_data() - no subentry logic
* Added _home_id instance variable from config_entry.data
- __init__.py: New _get_access_token() helper
* Handles token retrieval for both parent and subentries
* Subentries find parent entry to get shared access token
* Creates single API client instance per coordinator
- Data structures: Flat single-home format
* Old: {"homes": {home_id: {"price_info": [...]}}}
* New: {"home_id": str, "price_info": [...], "currency": str}
* Attribute name: "periods" → "pricePeriods" (consistent with priceInfo)
- helpers.py: Removed get_configured_home_ids() (no longer needed)
* parse_all_timestamps() updated for single-home structure
Impact: Each home operates independently with its own lifecycle tracking,
caching, and period calculations. Simpler architecture, easier debugging,
better isolation between homes.
414 lines
15 KiB
Python
414 lines
15 KiB
Python
"""Attribute builders for binary sensors."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from custom_components.tibber_prices.coordinator.helpers import get_intervals_for_day_offsets
|
|
from custom_components.tibber_prices.entity_utils import add_icon_color_attribute
|
|
|
|
# Import TypedDict definitions for documentation (not used in signatures)
|
|
|
|
if TYPE_CHECKING:
|
|
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
|
|
|
|
if TYPE_CHECKING:
|
|
from datetime import datetime
|
|
|
|
from custom_components.tibber_prices.data import TibberPricesConfigEntry
|
|
from homeassistant.core import HomeAssistant
|
|
|
|
|
|
def get_tomorrow_data_available_attributes(
|
|
coordinator_data: dict,
|
|
*,
|
|
time: TibberPricesTimeService,
|
|
) -> dict | None:
|
|
"""
|
|
Build attributes for tomorrow_data_available sensor.
|
|
|
|
Returns TomorrowDataAvailableAttributes structure.
|
|
|
|
Args:
|
|
coordinator_data: Coordinator data dict
|
|
time: TibberPricesTimeService instance
|
|
|
|
Returns:
|
|
Attributes dict with intervals_available and data_status
|
|
|
|
"""
|
|
if not coordinator_data:
|
|
return None
|
|
|
|
# Use helper to get tomorrow's intervals
|
|
tomorrow_prices = get_intervals_for_day_offsets(coordinator_data, [1])
|
|
tomorrow_date = time.get_local_date(offset_days=1)
|
|
interval_count = len(tomorrow_prices)
|
|
|
|
# Get expected intervals for tomorrow (handles DST)
|
|
expected_intervals = time.get_expected_intervals_for_day(tomorrow_date)
|
|
|
|
if interval_count == 0:
|
|
status = "none"
|
|
elif interval_count == expected_intervals:
|
|
status = "full"
|
|
else:
|
|
status = "partial"
|
|
|
|
return {
|
|
"intervals_available": interval_count,
|
|
"data_status": status,
|
|
}
|
|
|
|
|
|
def get_price_intervals_attributes(
|
|
coordinator_data: dict,
|
|
*,
|
|
time: TibberPricesTimeService,
|
|
reverse_sort: bool,
|
|
) -> dict | None:
|
|
"""
|
|
Build attributes for period-based sensors (best/peak price).
|
|
|
|
Returns PeriodAttributes structure.
|
|
|
|
All data is already calculated in the coordinator - we just need to:
|
|
1. Get period summaries from coordinator (already filtered and fully calculated)
|
|
2. Add the current timestamp
|
|
3. Find current or next period based on time
|
|
|
|
Args:
|
|
coordinator_data: Coordinator data dict
|
|
time: TibberPricesTimeService instance (required)
|
|
reverse_sort: True for peak_price (highest first), False for best_price (lowest first)
|
|
|
|
Returns:
|
|
Attributes dict with current/next period and all periods list
|
|
|
|
"""
|
|
if not coordinator_data:
|
|
return build_no_periods_result(time=time)
|
|
|
|
# Get precomputed period summaries from coordinator
|
|
periods_data = coordinator_data.get("pricePeriods", {})
|
|
period_type = "peak_price" if reverse_sort else "best_price"
|
|
period_data = periods_data.get(period_type)
|
|
|
|
if not period_data:
|
|
return build_no_periods_result(time=time)
|
|
|
|
period_summaries = period_data.get("periods", [])
|
|
if not period_summaries:
|
|
return build_no_periods_result(time=time)
|
|
|
|
# Find current or next period based on current time
|
|
current_period = None
|
|
|
|
# First pass: find currently active period
|
|
for period in period_summaries:
|
|
start = period.get("start")
|
|
end = period.get("end")
|
|
if start and end and time.is_current_interval(start, end):
|
|
current_period = period
|
|
break
|
|
|
|
# Second pass: find next future period if none is active
|
|
if not current_period:
|
|
for period in period_summaries:
|
|
start = period.get("start")
|
|
if start and time.is_in_future(start):
|
|
current_period = period
|
|
break
|
|
|
|
# Build final attributes
|
|
return build_final_attributes_simple(current_period, period_summaries, time=time)
|
|
|
|
|
|
def build_no_periods_result(*, time: TibberPricesTimeService) -> dict:
|
|
"""
|
|
Build result when no periods exist (not filtered, just none available).
|
|
|
|
Returns:
|
|
A dict with empty periods and timestamp.
|
|
|
|
"""
|
|
# Calculate timestamp: current time rounded down to last quarter hour
|
|
now = time.now()
|
|
current_minute = (now.minute // 15) * 15
|
|
timestamp = now.replace(minute=current_minute, second=0, microsecond=0)
|
|
|
|
return {
|
|
"timestamp": timestamp,
|
|
"start": None,
|
|
"end": None,
|
|
"periods": [],
|
|
}
|
|
|
|
|
|
def add_time_attributes(attributes: dict, current_period: dict, timestamp: datetime) -> None:
|
|
"""Add time-related attributes (priority 1)."""
|
|
attributes["timestamp"] = timestamp
|
|
if "start" in current_period:
|
|
attributes["start"] = current_period["start"]
|
|
if "end" in current_period:
|
|
attributes["end"] = current_period["end"]
|
|
if "duration_minutes" in current_period:
|
|
attributes["duration_minutes"] = current_period["duration_minutes"]
|
|
|
|
|
|
def add_decision_attributes(attributes: dict, current_period: dict) -> None:
|
|
"""Add core decision attributes (priority 2)."""
|
|
if "level" in current_period:
|
|
attributes["level"] = current_period["level"]
|
|
if "rating_level" in current_period:
|
|
attributes["rating_level"] = current_period["rating_level"]
|
|
if "rating_difference_%" in current_period:
|
|
attributes["rating_difference_%"] = current_period["rating_difference_%"]
|
|
|
|
|
|
def add_price_attributes(attributes: dict, current_period: dict) -> None:
|
|
"""Add price statistics attributes (priority 3)."""
|
|
if "price_avg" in current_period:
|
|
attributes["price_avg"] = current_period["price_avg"]
|
|
if "price_min" in current_period:
|
|
attributes["price_min"] = current_period["price_min"]
|
|
if "price_max" in current_period:
|
|
attributes["price_max"] = current_period["price_max"]
|
|
if "price_spread" in current_period:
|
|
attributes["price_spread"] = current_period["price_spread"]
|
|
if "volatility" in current_period:
|
|
attributes["volatility"] = current_period["volatility"]
|
|
|
|
|
|
def add_comparison_attributes(attributes: dict, current_period: dict) -> None:
|
|
"""Add price comparison attributes (priority 4)."""
|
|
if "period_price_diff_from_daily_min" in current_period:
|
|
attributes["period_price_diff_from_daily_min"] = current_period["period_price_diff_from_daily_min"]
|
|
if "period_price_diff_from_daily_min_%" in current_period:
|
|
attributes["period_price_diff_from_daily_min_%"] = current_period["period_price_diff_from_daily_min_%"]
|
|
|
|
|
|
def add_detail_attributes(attributes: dict, current_period: dict) -> None:
|
|
"""Add detail information attributes (priority 5)."""
|
|
if "period_interval_count" in current_period:
|
|
attributes["period_interval_count"] = current_period["period_interval_count"]
|
|
if "period_position" in current_period:
|
|
attributes["period_position"] = current_period["period_position"]
|
|
if "periods_total" in current_period:
|
|
attributes["periods_total"] = current_period["periods_total"]
|
|
if "periods_remaining" in current_period:
|
|
attributes["periods_remaining"] = current_period["periods_remaining"]
|
|
|
|
|
|
def add_relaxation_attributes(attributes: dict, current_period: dict) -> None:
|
|
"""
|
|
Add relaxation information attributes (priority 6).
|
|
|
|
Only adds relaxation attributes if the period was actually relaxed.
|
|
If relaxation_active is False or missing, no attributes are added.
|
|
"""
|
|
if current_period.get("relaxation_active"):
|
|
attributes["relaxation_active"] = True
|
|
if "relaxation_level" in current_period:
|
|
attributes["relaxation_level"] = current_period["relaxation_level"]
|
|
if "relaxation_threshold_original_%" in current_period:
|
|
attributes["relaxation_threshold_original_%"] = current_period["relaxation_threshold_original_%"]
|
|
if "relaxation_threshold_applied_%" in current_period:
|
|
attributes["relaxation_threshold_applied_%"] = current_period["relaxation_threshold_applied_%"]
|
|
|
|
|
|
def build_final_attributes_simple(
|
|
current_period: dict | None,
|
|
period_summaries: list[dict],
|
|
*,
|
|
time: TibberPricesTimeService,
|
|
) -> dict:
|
|
"""
|
|
Build the final attributes dictionary from coordinator's period summaries.
|
|
|
|
All calculations are done in the coordinator - this just:
|
|
1. Adds the current timestamp (only thing calculated every 15min)
|
|
2. Uses the current/next period from summaries
|
|
3. Adds nested period summaries
|
|
|
|
Attributes are ordered following the documented priority:
|
|
1. Time information (timestamp, start, end, duration)
|
|
2. Core decision attributes (level, rating_level, rating_difference_%)
|
|
3. Price statistics (price_avg, price_min, price_max, price_spread, volatility)
|
|
4. Price differences (period_price_diff_from_daily_min, period_price_diff_from_daily_min_%)
|
|
5. Detail information (period_interval_count, period_position, periods_total, periods_remaining)
|
|
6. Relaxation information (relaxation_active, relaxation_level, relaxation_threshold_original_%,
|
|
relaxation_threshold_applied_%) - only if period was relaxed
|
|
7. Meta information (periods list)
|
|
|
|
Args:
|
|
current_period: The current or next period (already complete from coordinator)
|
|
period_summaries: All period summaries from coordinator
|
|
time: TibberPricesTimeService instance (required)
|
|
|
|
Returns:
|
|
Complete attributes dict with all fields
|
|
|
|
"""
|
|
now = time.now()
|
|
current_minute = (now.minute // 15) * 15
|
|
timestamp = now.replace(minute=current_minute, second=0, microsecond=0)
|
|
|
|
if current_period:
|
|
# Build attributes in priority order using helper methods
|
|
attributes = {}
|
|
|
|
# 1. Time information
|
|
add_time_attributes(attributes, current_period, timestamp)
|
|
|
|
# 2. Core decision attributes
|
|
add_decision_attributes(attributes, current_period)
|
|
|
|
# 3. Price statistics
|
|
add_price_attributes(attributes, current_period)
|
|
|
|
# 4. Price differences
|
|
add_comparison_attributes(attributes, current_period)
|
|
|
|
# 5. Detail information
|
|
add_detail_attributes(attributes, current_period)
|
|
|
|
# 6. Relaxation information (only if period was relaxed)
|
|
add_relaxation_attributes(attributes, current_period)
|
|
|
|
# 7. Meta information (periods array)
|
|
attributes["periods"] = period_summaries
|
|
|
|
return attributes
|
|
|
|
# No current/next period found - return all periods with timestamp
|
|
return {
|
|
"timestamp": timestamp,
|
|
"periods": period_summaries,
|
|
}
|
|
|
|
|
|
async def build_async_extra_state_attributes( # noqa: PLR0913
|
|
entity_key: str,
|
|
translation_key: str | None,
|
|
hass: HomeAssistant,
|
|
*,
|
|
time: TibberPricesTimeService,
|
|
config_entry: TibberPricesConfigEntry,
|
|
sensor_attrs: dict | None = None,
|
|
is_on: bool | None = None,
|
|
) -> dict | None:
|
|
"""
|
|
Build async extra state attributes for binary sensors.
|
|
|
|
Adds icon_color and translated descriptions.
|
|
|
|
Args:
|
|
entity_key: Entity key (e.g., "best_price_period")
|
|
translation_key: Translation key for entity
|
|
hass: Home Assistant instance
|
|
time: TibberPricesTimeService instance (required)
|
|
config_entry: Config entry with options (keyword-only)
|
|
sensor_attrs: Sensor-specific attributes (keyword-only)
|
|
is_on: Binary sensor state (keyword-only)
|
|
|
|
Returns:
|
|
Complete attributes dict with descriptions (synchronous)
|
|
|
|
"""
|
|
# Calculate default timestamp: current time rounded to nearest quarter hour
|
|
# This ensures all binary sensors have a consistent reference time for when calculations were made
|
|
# Individual sensors can override this via sensor_attrs if needed
|
|
now = time.now()
|
|
default_timestamp = time.round_to_nearest_quarter(now)
|
|
|
|
attributes = {
|
|
"timestamp": default_timestamp,
|
|
}
|
|
|
|
# Add sensor-specific attributes (may override timestamp)
|
|
if sensor_attrs:
|
|
# Copy and remove internal fields before exposing to user
|
|
clean_attrs = {k: v for k, v in sensor_attrs.items() if not k.startswith("_")}
|
|
# Merge sensor attributes (can override default timestamp)
|
|
attributes.update(clean_attrs)
|
|
|
|
# Add icon_color for best/peak price period sensors using shared utility
|
|
add_icon_color_attribute(attributes, entity_key, is_on=is_on)
|
|
|
|
# Add description attributes (always last, via central utility)
|
|
from ..entity_utils import async_add_description_attributes # noqa: PLC0415, TID252
|
|
|
|
await async_add_description_attributes(
|
|
attributes,
|
|
"binary_sensor",
|
|
translation_key,
|
|
hass,
|
|
config_entry,
|
|
position="end",
|
|
)
|
|
|
|
return attributes if attributes else None
|
|
|
|
|
|
def build_sync_extra_state_attributes( # noqa: PLR0913
|
|
entity_key: str,
|
|
translation_key: str | None,
|
|
hass: HomeAssistant,
|
|
*,
|
|
time: TibberPricesTimeService,
|
|
config_entry: TibberPricesConfigEntry,
|
|
sensor_attrs: dict | None = None,
|
|
is_on: bool | None = None,
|
|
) -> dict | None:
|
|
"""
|
|
Build synchronous extra state attributes for binary sensors.
|
|
|
|
Adds icon_color and cached translated descriptions.
|
|
|
|
Args:
|
|
entity_key: Entity key (e.g., "best_price_period")
|
|
translation_key: Translation key for entity
|
|
hass: Home Assistant instance
|
|
time: TibberPricesTimeService instance (required)
|
|
config_entry: Config entry with options (keyword-only)
|
|
sensor_attrs: Sensor-specific attributes (keyword-only)
|
|
is_on: Binary sensor state (keyword-only)
|
|
|
|
Returns:
|
|
Complete attributes dict with cached descriptions
|
|
|
|
"""
|
|
# Calculate default timestamp: current time rounded to nearest quarter hour
|
|
# This ensures all binary sensors have a consistent reference time for when calculations were made
|
|
# Individual sensors can override this via sensor_attrs if needed
|
|
now = time.now()
|
|
default_timestamp = time.round_to_nearest_quarter(now)
|
|
|
|
attributes = {
|
|
"timestamp": default_timestamp,
|
|
}
|
|
|
|
# Add sensor-specific attributes (may override timestamp)
|
|
if sensor_attrs:
|
|
# Copy and remove internal fields before exposing to user
|
|
clean_attrs = {k: v for k, v in sensor_attrs.items() if not k.startswith("_")}
|
|
# Merge sensor attributes (can override default timestamp)
|
|
attributes.update(clean_attrs)
|
|
|
|
# Add icon_color for best/peak price period sensors using shared utility
|
|
add_icon_color_attribute(attributes, entity_key, is_on=is_on)
|
|
|
|
# Add description attributes (always last, via central utility)
|
|
from ..entity_utils import add_description_attributes # noqa: PLC0415, TID252
|
|
|
|
add_description_attributes(
|
|
attributes,
|
|
"binary_sensor",
|
|
translation_key,
|
|
hass,
|
|
config_entry,
|
|
position="end",
|
|
)
|
|
|
|
return attributes if attributes else None
|