mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-03-30 21:33:39 +00:00
189d3ba84d
Add comprehensive data_lifecycle_status sensor showing real-time cache vs fresh API data status with 6 states and 13+ detailed attributes. Key features: - 6 lifecycle states: cached, fresh, refreshing, searching_tomorrow, turnover_pending, error - Push-update system for instant state changes (refreshing→fresh→error) - Quarter-hour polling for turnover_pending detection at 23:45 - Accurate next_api_poll prediction using Timer #1 offset tracking - Tomorrow prediction with actual timer schedule (not fixed 13:00) - 13+ formatted attributes: cache_age, data_completeness, api_calls_today, next_api_poll, etc. Implementation: - sensor/calculators/lifecycle.py: New calculator with state logic - sensor/attributes/lifecycle.py: Attribute builders with formatting - coordinator/core.py: Lifecycle tracking + callback system (+16 lines) - sensor/core.py: Push callback registration (+3 lines) - coordinator/constants.py: Added to TIME_SENSITIVE_ENTITY_KEYS - Translations: All 5 languages (de, en, nb, nl, sv) Timing optimization: - Extended turnover warning: 5min → 15min (catches 23:45 quarter boundary) - No minute-timer needed: quarter-hour updates + push = optimal - Push-updates: <1sec latency for refreshing/fresh/error states - Timer offset tracking: Accurate tomorrow predictions Removed obsolete sensors: - data_timestamp (replaced by lifecycle attributes) - price_forecast (never implemented, removed from definitions) Impact: Users can monitor data freshness, API call patterns, cache age, and understand integration behavior. Perfect for troubleshooting and visibility into when data updates occur.
93 lines
3.7 KiB
Python
93 lines
3.7 KiB
Python
"""Attribute builders for lifecycle diagnostic sensor."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
if TYPE_CHECKING:
|
|
from custom_components.tibber_prices.coordinator.core import (
|
|
TibberPricesDataUpdateCoordinator,
|
|
)
|
|
from custom_components.tibber_prices.sensor.calculators.lifecycle import (
|
|
TibberPricesLifecycleCalculator,
|
|
)
|
|
|
|
|
|
# Constants for cache age formatting
|
|
MINUTES_PER_HOUR = 60
|
|
MINUTES_PER_DAY = 1440 # 24 * 60
|
|
|
|
|
|
def build_lifecycle_attributes(
|
|
coordinator: TibberPricesDataUpdateCoordinator,
|
|
lifecycle_calculator: TibberPricesLifecycleCalculator,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Build attributes for data_lifecycle_status sensor.
|
|
|
|
Shows comprehensive cache status, data availability, and update timing.
|
|
|
|
Returns:
|
|
Dict with lifecycle attributes
|
|
|
|
"""
|
|
attributes: dict[str, Any] = {}
|
|
|
|
# Cache Status (formatted for readability)
|
|
cache_age = lifecycle_calculator.get_cache_age_minutes()
|
|
if cache_age is not None:
|
|
# Format cache age with units for better readability
|
|
if cache_age < MINUTES_PER_HOUR:
|
|
attributes["cache_age"] = f"{cache_age} min"
|
|
elif cache_age < MINUTES_PER_DAY: # Less than 24 hours
|
|
hours = cache_age // MINUTES_PER_HOUR
|
|
minutes = cache_age % MINUTES_PER_HOUR
|
|
attributes["cache_age"] = f"{hours}h {minutes}min" if minutes > 0 else f"{hours}h"
|
|
else: # 24+ hours
|
|
days = cache_age // MINUTES_PER_DAY
|
|
hours = (cache_age % MINUTES_PER_DAY) // MINUTES_PER_HOUR
|
|
attributes["cache_age"] = f"{days}d {hours}h" if hours > 0 else f"{days}d"
|
|
|
|
# Keep raw value for automations
|
|
attributes["cache_age_minutes"] = cache_age
|
|
|
|
cache_validity = lifecycle_calculator.get_cache_validity_status()
|
|
attributes["cache_validity"] = cache_validity
|
|
|
|
if coordinator._last_price_update: # noqa: SLF001 - Internal state access for diagnostic display
|
|
attributes["last_api_fetch"] = coordinator._last_price_update.isoformat() # noqa: SLF001
|
|
attributes["last_cache_update"] = coordinator._last_price_update.isoformat() # noqa: SLF001
|
|
|
|
# Data Availability & Completeness
|
|
data_completeness = lifecycle_calculator.get_data_completeness_status()
|
|
attributes["data_completeness"] = data_completeness
|
|
|
|
attributes["yesterday_available"] = lifecycle_calculator.is_data_available("yesterday")
|
|
attributes["today_available"] = lifecycle_calculator.is_data_available("today")
|
|
attributes["tomorrow_available"] = lifecycle_calculator.is_data_available("tomorrow")
|
|
attributes["tomorrow_expected_after"] = "13:00"
|
|
|
|
# Next Actions (only show if meaningful)
|
|
next_poll = lifecycle_calculator.get_next_api_poll_time()
|
|
if next_poll: # None means data is complete, no more polls needed
|
|
attributes["next_api_poll"] = next_poll.isoformat()
|
|
|
|
next_tomorrow_check = lifecycle_calculator.get_next_tomorrow_check_time()
|
|
if next_tomorrow_check:
|
|
attributes["next_tomorrow_check"] = next_tomorrow_check.isoformat()
|
|
|
|
next_midnight = lifecycle_calculator.get_next_midnight_turnover_time()
|
|
attributes["next_midnight_turnover"] = next_midnight.isoformat()
|
|
|
|
# Update Statistics
|
|
api_calls = lifecycle_calculator.get_api_calls_today()
|
|
attributes["updates_today"] = api_calls
|
|
|
|
if coordinator._last_actual_turnover: # noqa: SLF001 - Internal state access for diagnostic display
|
|
attributes["last_turnover"] = coordinator._last_actual_turnover.isoformat() # noqa: SLF001
|
|
|
|
# Last Error (if any)
|
|
if coordinator.last_exception:
|
|
attributes["last_error"] = str(coordinator.last_exception)
|
|
|
|
return attributes
|