mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-05-28 18:43:40 +00:00
bbcfdd4443
Recompute merged relaxed periods from raw intervals, harden numeric period option normalization, update day-volatility handling for zero or negative averages, and expose day context on period binary sensors. Add focused regressions for overlap merges, cache invalidation, day statistics, and visible binary sensor attributes. Impact: Best and peak period entities stay consistent on negative-price days, refresh correctly when same-day prices change, and expose the documented day context attributes.
690 lines
26 KiB
Python
690 lines
26 KiB
Python
"""Attribute builders for binary sensors."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from custom_components.tibber_prices.const import get_display_unit_factor
|
|
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
|
|
from custom_components.tibber_prices.sensor.attributes.metadata import _find_current_segment_in_data
|
|
|
|
# Constants for price display conversion
|
|
_SUBUNIT_FACTOR = 100 # Conversion factor for subunit currency (ct/øre)
|
|
_SUBUNIT_PRECISION = 2 # Decimal places for subunit currency
|
|
_BASE_PRECISION = 4 # Decimal places for base currency
|
|
|
|
# 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_current_phase_type(coordinator_data: dict, *, time: TibberPricesTimeService) -> str | None:
|
|
"""
|
|
Return the type of the currently active intra-day price phase.
|
|
|
|
Delegates to the shared segment finder in sensor/attributes/metadata.py.
|
|
|
|
Args:
|
|
coordinator_data: The coordinator's data dict.
|
|
time: TibberPricesTimeService instance.
|
|
|
|
Returns:
|
|
Phase type string or None if no segment data is available.
|
|
|
|
"""
|
|
if not coordinator_data:
|
|
return None
|
|
current_index, segments = _find_current_segment_in_data(coordinator_data, time=time)
|
|
if current_index is None or segments is None:
|
|
return None
|
|
return segments[current_index].get("type")
|
|
|
|
|
|
def get_phase_attributes(coordinator_data: dict, *, time: TibberPricesTimeService) -> dict | None:
|
|
"""
|
|
Build start/end attributes for in_*_price_phase binary sensors.
|
|
|
|
Args:
|
|
coordinator_data: The coordinator's data dict.
|
|
time: TibberPricesTimeService instance.
|
|
|
|
Returns:
|
|
Dict with start and end timestamps, or None if unavailable.
|
|
|
|
"""
|
|
if not coordinator_data:
|
|
return None
|
|
current_index, segments = _find_current_segment_in_data(coordinator_data, time=time)
|
|
if current_index is None or segments is None:
|
|
return None
|
|
|
|
segment = segments[current_index]
|
|
attrs: dict = {}
|
|
|
|
if start := segment.get("start"):
|
|
attrs["start"] = start
|
|
if end := segment.get("end"):
|
|
attrs["end"] = end
|
|
|
|
return attrs or None
|
|
|
|
|
|
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,
|
|
config_entry: TibberPricesConfigEntry,
|
|
) -> 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
|
|
4. Convert prices to display units based on user configuration
|
|
|
|
Args:
|
|
coordinator_data: Coordinator data dict
|
|
time: TibberPricesTimeService instance (required)
|
|
reverse_sort: True for peak_price (highest first), False for best_price (lowest first)
|
|
config_entry: Config entry for display unit configuration
|
|
|
|
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)
|
|
|
|
# Filter periods for today+tomorrow (sensors don't show yesterday's periods)
|
|
# Coordinator cache contains yesterday/today/tomorrow, but sensors only need today+tomorrow
|
|
now = time.now()
|
|
today_start = time.start_of_local_day(now)
|
|
filtered_periods = [period for period in period_summaries if period.get("end") and period["end"] >= today_start]
|
|
|
|
if not filtered_periods:
|
|
return build_no_periods_result(time=time)
|
|
|
|
# Recalculate position metadata after filtering (coordinator stamped values include yesterday)
|
|
# Use shallow copies so coordinator dicts are not mutated
|
|
total_filtered = len(filtered_periods)
|
|
filtered_periods = [
|
|
period
|
|
| {
|
|
"period_position": i,
|
|
"period_count_total": total_filtered,
|
|
"period_count_remaining": total_filtered - i,
|
|
}
|
|
for i, period in enumerate(filtered_periods, 1)
|
|
]
|
|
|
|
# Find current or next period based on current time
|
|
current_period = None
|
|
|
|
# First pass: find currently active period
|
|
for period in filtered_periods:
|
|
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 filtered_periods:
|
|
start = period.get("start")
|
|
if start and time.is_in_future(start):
|
|
current_period = period
|
|
break
|
|
|
|
# Extract calculation metadata for diagnostic attributes
|
|
period_metadata = period_data.get("metadata", {})
|
|
|
|
# Build final attributes (use filtered_periods for display)
|
|
return build_final_attributes_simple(
|
|
current_period, filtered_periods, time=time, config_entry=config_entry, period_metadata=period_metadata
|
|
)
|
|
|
|
|
|
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, factor: int) -> None:
|
|
"""
|
|
Add price statistics attributes (priority 3).
|
|
|
|
Args:
|
|
attributes: Target dict to add attributes to
|
|
current_period: Period dict with price data (in base currency)
|
|
factor: Display unit conversion factor (100 for subunit, 1 for base)
|
|
|
|
"""
|
|
# Convert prices from base currency to display units
|
|
precision = _SUBUNIT_PRECISION if factor == _SUBUNIT_FACTOR else _BASE_PRECISION
|
|
|
|
if "price_mean" in current_period:
|
|
attributes["price_mean"] = round(current_period["price_mean"] * factor, precision)
|
|
if "price_median" in current_period:
|
|
attributes["price_median"] = round(current_period["price_median"] * factor, precision)
|
|
if "price_min" in current_period:
|
|
attributes["price_min"] = round(current_period["price_min"] * factor, precision)
|
|
if "price_max" in current_period:
|
|
attributes["price_max"] = round(current_period["price_max"] * factor, precision)
|
|
if "price_spread" in current_period:
|
|
attributes["price_spread"] = round(current_period["price_spread"] * factor, precision)
|
|
if "price_coefficient_variation_%" in current_period:
|
|
attributes["price_coefficient_variation_%"] = current_period["price_coefficient_variation_%"]
|
|
if "volatility" in current_period:
|
|
attributes["volatility"] = current_period["volatility"] # Volatility is not a price, keep as-is
|
|
|
|
|
|
def add_day_statistics_attributes(attributes: dict, current_period: dict) -> None:
|
|
"""Add per-day context attributes for the current/next period.
|
|
|
|
Day price range fields are already stored in minor currency units (ct/ore)
|
|
by the period summary builder and therefore must not be converted again here.
|
|
"""
|
|
if "day_volatility_%" in current_period:
|
|
attributes["day_volatility_%"] = current_period["day_volatility_%"]
|
|
if "day_price_min" in current_period:
|
|
attributes["day_price_min"] = current_period["day_price_min"]
|
|
if "day_price_max" in current_period:
|
|
attributes["day_price_max"] = current_period["day_price_max"]
|
|
if "day_price_span" in current_period:
|
|
attributes["day_price_span"] = current_period["day_price_span"]
|
|
|
|
|
|
def add_comparison_attributes(attributes: dict, current_period: dict, factor: int) -> None:
|
|
"""
|
|
Add price comparison attributes (priority 4).
|
|
|
|
Args:
|
|
attributes: Target dict to add attributes to
|
|
current_period: Period dict with price diff data (in base currency)
|
|
factor: Display unit conversion factor (100 for subunit, 1 for base)
|
|
|
|
"""
|
|
# Convert price differences from base currency to display units
|
|
precision = _SUBUNIT_PRECISION if factor == _SUBUNIT_FACTOR else _BASE_PRECISION
|
|
|
|
if "period_price_diff_from_daily_min" in current_period:
|
|
attributes["period_price_diff_from_daily_min"] = round(
|
|
current_period["period_price_diff_from_daily_min"] * factor, precision
|
|
)
|
|
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_max" in current_period:
|
|
attributes["period_price_diff_from_daily_max"] = round(
|
|
current_period["period_price_diff_from_daily_max"] * factor, precision
|
|
)
|
|
if "period_price_diff_from_daily_max_%" in current_period:
|
|
attributes["period_price_diff_from_daily_max_%"] = current_period["period_price_diff_from_daily_max_%"]
|
|
|
|
|
|
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 "period_count_total" in current_period:
|
|
attributes["period_count_total"] = current_period["period_count_total"]
|
|
if "period_count_remaining" in current_period:
|
|
attributes["period_count_remaining"] = current_period["period_count_remaining"]
|
|
|
|
|
|
def add_period_count_attributes(
|
|
attributes: dict,
|
|
period_summaries: list[dict],
|
|
time: TibberPricesTimeService,
|
|
) -> None:
|
|
"""
|
|
Add per-day period count attributes (priority 5.5).
|
|
|
|
Counts how many periods fall on today and tomorrow so automations can check
|
|
things like "only charge if there are at least 2 cheap periods today".
|
|
|
|
Args:
|
|
attributes: Target dict to add attributes to
|
|
period_summaries: All period summaries (already filtered to today+tomorrow)
|
|
time: TibberPricesTimeService instance for date comparison
|
|
|
|
"""
|
|
now = time.now()
|
|
today = time.get_local_date()
|
|
tomorrow = time.get_local_date(offset_days=1)
|
|
|
|
count_today = 0
|
|
count_tomorrow = 0
|
|
|
|
for period in period_summaries:
|
|
start = period.get("start")
|
|
if start is None:
|
|
continue
|
|
if hasattr(start, "date"):
|
|
period_date = start.date()
|
|
else:
|
|
from datetime import datetime # noqa: PLC0415
|
|
|
|
period_date = datetime.fromisoformat(str(start)).date()
|
|
|
|
if period_date == today:
|
|
count_today += 1
|
|
elif period_date == tomorrow:
|
|
count_tomorrow += 1
|
|
|
|
_ = now # used for clarity only
|
|
attributes["period_count_today"] = count_today
|
|
attributes["period_count_tomorrow"] = count_tomorrow
|
|
|
|
|
|
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 add_calculation_summary_attributes(attributes: dict, period_metadata: dict) -> None:
|
|
"""
|
|
Add calculation summary attributes (priority 7).
|
|
|
|
Provides diagnostic visibility into the period calculation: whether any flat days
|
|
triggered adaptive min_periods, and whether relaxation could not satisfy all days.
|
|
|
|
Only adds non-default/interesting values to avoid clutter:
|
|
- min_periods_configured: always added (useful reference for automations)
|
|
- flat_days_detected: only when > 0 (explains why fewer periods than configured)
|
|
- relaxation_incomplete: only when True (diagnostic flag for troubleshooting)
|
|
|
|
"""
|
|
relaxation_meta = period_metadata.get("relaxation", {})
|
|
if not relaxation_meta:
|
|
return
|
|
|
|
if "min_periods_requested" in relaxation_meta:
|
|
attributes["min_periods_configured"] = relaxation_meta["min_periods_requested"]
|
|
|
|
flat_days = relaxation_meta.get("flat_days_detected", 0)
|
|
if flat_days > 0:
|
|
attributes["flat_days_detected"] = flat_days
|
|
|
|
if relaxation_meta.get("relaxation_incomplete"):
|
|
attributes["relaxation_incomplete"] = True
|
|
|
|
|
|
def _convert_periods_to_display_units(period_summaries: list[dict], factor: int) -> list[dict]:
|
|
"""
|
|
Convert price values in periods array to display units.
|
|
|
|
Args:
|
|
period_summaries: List of period dicts with price data (in base currency)
|
|
factor: Display unit conversion factor (100 for subunit, 1 for base)
|
|
|
|
Returns:
|
|
New list with converted period dicts
|
|
|
|
"""
|
|
precision = _SUBUNIT_PRECISION if factor == _SUBUNIT_FACTOR else _BASE_PRECISION
|
|
converted_periods = []
|
|
|
|
for period in period_summaries:
|
|
converted = period.copy()
|
|
|
|
# Convert all price fields
|
|
price_fields = ["price_mean", "price_median", "price_min", "price_max", "price_spread"]
|
|
for field in price_fields:
|
|
if field in converted:
|
|
converted[field] = round(converted[field] * factor, precision)
|
|
|
|
# Convert price differences (not percentages)
|
|
if "period_price_diff_from_daily_min" in converted:
|
|
converted["period_price_diff_from_daily_min"] = round(
|
|
converted["period_price_diff_from_daily_min"] * factor, precision
|
|
)
|
|
if "period_price_diff_from_daily_max" in converted:
|
|
converted["period_price_diff_from_daily_max"] = round(
|
|
converted["period_price_diff_from_daily_max"] * factor, precision
|
|
)
|
|
|
|
converted_periods.append(converted)
|
|
|
|
return converted_periods
|
|
|
|
|
|
def build_final_attributes_simple(
|
|
current_period: dict | None,
|
|
period_summaries: list[dict],
|
|
*,
|
|
time: TibberPricesTimeService,
|
|
config_entry: TibberPricesConfigEntry,
|
|
period_metadata: dict | None = None,
|
|
) -> 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
|
|
4. Converts prices to display units based on user configuration
|
|
|
|
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_mean, price_median, price_min, price_max, price_spread, volatility)
|
|
4. Price differences (period_price_diff_from_daily_min, period_price_diff_from_daily_min_%)
|
|
5. Day context (day_volatility_%, day_price_min, day_price_max, day_price_span)
|
|
6. Detail information (period_interval_count, period_position, period_count_total, period_count_remaining)
|
|
7. Relaxation information (relaxation_active, relaxation_level, relaxation_threshold_original_%,
|
|
relaxation_threshold_applied_%) - only if current period was relaxed
|
|
8. Calculation summary (min_periods_configured, flat_days_detected,
|
|
relaxation_incomplete) - diagnostic info about the overall calculation
|
|
9. 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)
|
|
config_entry: Config entry for display unit configuration
|
|
period_metadata: Metadata from coordinator's period calculation (relaxation diagnostics)
|
|
|
|
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)
|
|
|
|
# Get display unit factor (100 for subunit, 1 for base currency)
|
|
factor = get_display_unit_factor(config_entry)
|
|
|
|
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 (converted to display units)
|
|
add_price_attributes(attributes, current_period, factor)
|
|
|
|
# 4. Price differences (converted to display units)
|
|
add_comparison_attributes(attributes, current_period, factor)
|
|
|
|
# 5. Day context attributes (already in minor units)
|
|
add_day_statistics_attributes(attributes, current_period)
|
|
|
|
# 6. Detail information
|
|
add_detail_attributes(attributes, current_period)
|
|
|
|
# 6.5 Per-day period counts (how many cheap/peak periods per day)
|
|
add_period_count_attributes(attributes, period_summaries, time)
|
|
|
|
# 7. Relaxation information (only if current period was relaxed)
|
|
add_relaxation_attributes(attributes, current_period)
|
|
|
|
# 8. Calculation summary (diagnostic: min_periods_configured, flat_days_detected, etc.)
|
|
if period_metadata:
|
|
add_calculation_summary_attributes(attributes, period_metadata)
|
|
|
|
# 9. Meta information (periods array - prices converted to display units)
|
|
attributes["periods"] = _convert_periods_to_display_units(period_summaries, factor)
|
|
|
|
return attributes
|
|
|
|
# No current/next period found - return all periods with timestamp (prices converted)
|
|
# Still add calculation summary so diagnostics are accessible even between periods
|
|
result: dict = {
|
|
"timestamp": timestamp,
|
|
}
|
|
add_period_count_attributes(result, period_summaries, time)
|
|
if period_metadata:
|
|
add_calculation_summary_attributes(result, period_metadata)
|
|
result["periods"] = _convert_periods_to_display_units(period_summaries, factor)
|
|
return result
|
|
|
|
|
|
async def build_async_extra_state_attributes(
|
|
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 or None
|
|
|
|
|
|
def build_sync_extra_state_attributes(
|
|
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 or None
|