mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-05-28 18:43:40 +00:00
779e22a84e
Removed periods_found_total and replaced with period_count_today / period_count_tomorrow. The old attribute counted all periods including yesterday (coordinator scope), causing a discrepancy vs. the displayed list (sensor scope, today+tomorrow only). Renamed periods_total → period_count_total for naming consistency with the new per-day attributes. Recalculate period_position / period_count_total / periods_remaining after the today+tomorrow filter so all three navigation attributes reflect the filtered scope. period_count_tomorrow is always present (0 when no tomorrow data or no periods found), enabling automations without default(0) guards. Removed internal periods_found key from relaxation metadata — it was only consumed by add_calculation_summary_attributes which is now removed. BREAKING CHANGE: periods_found_total removed (replace with period_count_today + period_count_tomorrow). periods_total renamed to period_count_total. Impact: Period navigation attributes (position/total/remaining) now correctly reflect today+tomorrow scope. Per-day counts allow automations to distinguish "2 periods today, 0 tomorrow" from "1+1".
618 lines
23 KiB
Python
618 lines
23 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
|
|
|
|
# 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_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,
|
|
"periods_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_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 "periods_remaining" in current_period:
|
|
attributes["periods_remaining"] = current_period["periods_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. Detail information (period_interval_count, period_position, period_count_total, periods_remaining)
|
|
6. Relaxation information (relaxation_active, relaxation_level, relaxation_threshold_original_%,
|
|
relaxation_threshold_applied_%) - only if current period was relaxed
|
|
7. Calculation summary (min_periods_configured, flat_days_detected,
|
|
relaxation_incomplete) - diagnostic info about the overall calculation
|
|
8. 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. Detail information
|
|
add_detail_attributes(attributes, current_period)
|
|
|
|
# 5.5 Per-day period counts (how many cheap/peak periods per day)
|
|
add_period_count_attributes(attributes, period_summaries, time)
|
|
|
|
# 6. Relaxation information (only if current period was relaxed)
|
|
add_relaxation_attributes(attributes, current_period)
|
|
|
|
# 7. Calculation summary (diagnostic: min_periods_configured, flat_days_detected, etc.)
|
|
if period_metadata:
|
|
add_calculation_summary_attributes(attributes, period_metadata)
|
|
|
|
# 8. 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( # 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 or 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 or None
|