mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-03-30 05:13:40 +00:00
331 lines
12 KiB
Python
331 lines
12 KiB
Python
"""Binary sensor platform for tibber_prices."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import TYPE_CHECKING
|
|
|
|
from homeassistant.components.binary_sensor import (
|
|
BinarySensorDeviceClass,
|
|
BinarySensorEntity,
|
|
BinarySensorEntityDescription,
|
|
)
|
|
from homeassistant.const import EntityCategory
|
|
|
|
from .entity import TibberPricesEntity
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Callable
|
|
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
|
|
from .coordinator import TibberPricesDataUpdateCoordinator
|
|
from .data import TibberPricesConfigEntry
|
|
|
|
ENTITY_DESCRIPTIONS = (
|
|
BinarySensorEntityDescription(
|
|
key="peak_hour",
|
|
translation_key="peak_hour",
|
|
name="Peak Hour",
|
|
icon="mdi:clock-alert",
|
|
),
|
|
BinarySensorEntityDescription(
|
|
key="best_price_hour",
|
|
translation_key="best_price_hour",
|
|
name="Best Price Hour",
|
|
icon="mdi:clock-check",
|
|
),
|
|
BinarySensorEntityDescription(
|
|
key="connection",
|
|
translation_key="connection",
|
|
name="Tibber API Connection",
|
|
device_class=BinarySensorDeviceClass.CONNECTIVITY,
|
|
entity_category=EntityCategory.DIAGNOSTIC,
|
|
),
|
|
)
|
|
|
|
|
|
async def async_setup_entry(
|
|
_hass: HomeAssistant,
|
|
entry: TibberPricesConfigEntry,
|
|
async_add_entities: AddEntitiesCallback,
|
|
) -> None:
|
|
"""Set up the binary_sensor platform."""
|
|
async_add_entities(
|
|
TibberPricesBinarySensor(
|
|
coordinator=entry.runtime_data.coordinator,
|
|
entity_description=entity_description,
|
|
)
|
|
for entity_description in ENTITY_DESCRIPTIONS
|
|
)
|
|
|
|
|
|
class TibberPricesBinarySensor(TibberPricesEntity, BinarySensorEntity):
|
|
"""tibber_prices binary_sensor class."""
|
|
|
|
def __init__(
|
|
self,
|
|
coordinator: TibberPricesDataUpdateCoordinator,
|
|
entity_description: BinarySensorEntityDescription,
|
|
) -> None:
|
|
"""Initialize the binary_sensor class."""
|
|
super().__init__(coordinator)
|
|
self.entity_description = entity_description
|
|
self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{entity_description.key}"
|
|
self._state_getter: Callable | None = self._get_state_getter()
|
|
self._attribute_getter: Callable | None = self._get_attribute_getter()
|
|
|
|
def _get_state_getter(self) -> Callable | None:
|
|
"""Return the appropriate state getter method based on the sensor type."""
|
|
key = self.entity_description.key
|
|
|
|
if key == "peak_hour":
|
|
return lambda: self._get_price_threshold_state(threshold_percentage=0.8, high_is_active=True)
|
|
if key == "best_price_hour":
|
|
return lambda: self._get_price_threshold_state(threshold_percentage=0.2, high_is_active=False)
|
|
if key == "connection":
|
|
return lambda: True if self.coordinator.data else None
|
|
|
|
return None
|
|
|
|
def _get_attribute_getter(self) -> Callable | None:
|
|
"""Return the appropriate attribute getter method based on the sensor type."""
|
|
key = self.entity_description.key
|
|
|
|
if key == "peak_hour":
|
|
return lambda: self._get_price_hours_attributes(attribute_name="peak_hours", reverse_sort=True)
|
|
if key == "best_price_hour":
|
|
return lambda: self._get_price_hours_attributes(attribute_name="best_price_hours", reverse_sort=False)
|
|
|
|
return None
|
|
|
|
def _get_current_price_data(self) -> tuple[list[float], float] | None:
|
|
"""Get current price data if available."""
|
|
if not (
|
|
self.coordinator.data
|
|
and (
|
|
today_prices := self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"][
|
|
"priceInfo"
|
|
].get("today", [])
|
|
)
|
|
):
|
|
return None
|
|
|
|
now = datetime.now(tz=UTC).astimezone()
|
|
current_hour_data = next(
|
|
(
|
|
price_data
|
|
for price_data in today_prices
|
|
if datetime.fromisoformat(price_data["startsAt"]).hour == now.hour
|
|
),
|
|
None,
|
|
)
|
|
if not current_hour_data:
|
|
return None
|
|
|
|
prices = [float(price["total"]) for price in today_prices]
|
|
prices.sort()
|
|
return prices, float(current_hour_data["total"])
|
|
|
|
def _get_price_threshold_state(self, *, threshold_percentage: float, high_is_active: bool) -> bool | None:
|
|
"""
|
|
Determine if current price is above/below threshold.
|
|
|
|
Args:
|
|
threshold_percentage: The percentage point in the sorted list (0.0-1.0)
|
|
high_is_active: If True, value >= threshold is active, otherwise value <= threshold is active
|
|
|
|
"""
|
|
price_data = self._get_current_price_data()
|
|
if not price_data:
|
|
return None
|
|
|
|
prices, current_price = price_data
|
|
threshold_index = int(len(prices) * threshold_percentage)
|
|
|
|
if high_is_active:
|
|
return current_price >= prices[threshold_index]
|
|
|
|
return current_price <= prices[threshold_index]
|
|
|
|
def _get_price_hours_attributes(self, *, attribute_name: str, reverse_sort: bool) -> dict | None:
|
|
"""Get price hours attributes."""
|
|
if not self.coordinator.data:
|
|
return None
|
|
|
|
price_info = self.coordinator.data["data"]["viewer"]["homes"][0]["currentSubscription"]["priceInfo"]
|
|
|
|
today_prices = price_info.get("today", [])
|
|
if not today_prices:
|
|
return None
|
|
|
|
prices = [
|
|
(
|
|
datetime.fromisoformat(price["startsAt"]).hour,
|
|
float(price["total"]),
|
|
)
|
|
for price in today_prices
|
|
]
|
|
|
|
# Sort by price (high to low for peak, low to high for best)
|
|
sorted_hours = sorted(prices, key=lambda x: x[1], reverse=reverse_sort)[:5]
|
|
|
|
return {attribute_name: [{"hour": hour, "price": price} for hour, price in sorted_hours]}
|
|
|
|
@property
|
|
def is_on(self) -> bool | None:
|
|
"""Return true if the binary_sensor is on."""
|
|
try:
|
|
if not self.coordinator.data or not self._state_getter:
|
|
return None
|
|
|
|
return self._state_getter()
|
|
|
|
except (KeyError, ValueError, TypeError) as ex:
|
|
self.coordinator.logger.exception(
|
|
"Error getting binary sensor state",
|
|
extra={
|
|
"error": str(ex),
|
|
"entity": self.entity_description.key,
|
|
},
|
|
)
|
|
return None
|
|
|
|
@property
|
|
async def async_extra_state_attributes(self) -> dict | None:
|
|
"""Return additional state attributes asynchronously."""
|
|
try:
|
|
# Get the dynamic attributes if the getter is available
|
|
if not self.coordinator.data:
|
|
return None
|
|
|
|
attributes = {}
|
|
if self._attribute_getter:
|
|
dynamic_attrs = self._attribute_getter()
|
|
if dynamic_attrs:
|
|
attributes.update(dynamic_attrs)
|
|
|
|
# Add descriptions from the custom translations file
|
|
if self.entity_description.translation_key and self.hass is not None:
|
|
# Get user's language preference
|
|
language = self.hass.config.language if self.hass.config.language else "en"
|
|
|
|
# Import async function to get descriptions
|
|
from .const import (
|
|
CONF_EXTENDED_DESCRIPTIONS,
|
|
DEFAULT_EXTENDED_DESCRIPTIONS,
|
|
async_get_entity_description,
|
|
)
|
|
|
|
# Add basic description
|
|
description = await async_get_entity_description(
|
|
self.hass, "binary_sensor", self.entity_description.translation_key, language, "description"
|
|
)
|
|
if description:
|
|
attributes["description"] = description
|
|
|
|
# Check if extended descriptions are enabled in the config
|
|
extended_descriptions = self.coordinator.config_entry.options.get(
|
|
CONF_EXTENDED_DESCRIPTIONS,
|
|
self.coordinator.config_entry.data.get(CONF_EXTENDED_DESCRIPTIONS, DEFAULT_EXTENDED_DESCRIPTIONS),
|
|
)
|
|
|
|
# Add extended descriptions if enabled
|
|
if extended_descriptions:
|
|
# Add long description if available
|
|
long_desc = await async_get_entity_description(
|
|
self.hass,
|
|
"binary_sensor",
|
|
self.entity_description.translation_key,
|
|
language,
|
|
"long_description",
|
|
)
|
|
if long_desc:
|
|
attributes["long_description"] = long_desc
|
|
|
|
# Add usage tips if available
|
|
usage_tips = await async_get_entity_description(
|
|
self.hass, "binary_sensor", self.entity_description.translation_key, language, "usage_tips"
|
|
)
|
|
if usage_tips:
|
|
attributes["usage_tips"] = usage_tips
|
|
|
|
except (KeyError, ValueError, TypeError) as ex:
|
|
self.coordinator.logger.exception(
|
|
"Error getting binary sensor attributes",
|
|
extra={
|
|
"error": str(ex),
|
|
"entity": self.entity_description.key,
|
|
},
|
|
)
|
|
return None
|
|
else:
|
|
return attributes if attributes else None
|
|
|
|
@property
|
|
def extra_state_attributes(self) -> dict | None:
|
|
"""Return additional state attributes synchronously."""
|
|
try:
|
|
# Start with dynamic attributes if available
|
|
if not self.coordinator.data:
|
|
return None
|
|
|
|
attributes = {}
|
|
if self._attribute_getter:
|
|
dynamic_attrs = self._attribute_getter()
|
|
if dynamic_attrs:
|
|
attributes.update(dynamic_attrs)
|
|
|
|
# Add descriptions from the cache (non-blocking)
|
|
if self.entity_description.translation_key and self.hass is not None:
|
|
# Get user's language preference
|
|
language = self.hass.config.language if self.hass.config.language else "en"
|
|
|
|
# Import synchronous function to get cached descriptions
|
|
from .const import (
|
|
CONF_EXTENDED_DESCRIPTIONS,
|
|
DEFAULT_EXTENDED_DESCRIPTIONS,
|
|
get_entity_description,
|
|
)
|
|
|
|
# Add basic description from cache
|
|
description = get_entity_description(
|
|
"binary_sensor", self.entity_description.translation_key, language, "description"
|
|
)
|
|
if description:
|
|
attributes["description"] = description
|
|
|
|
# Check if extended descriptions are enabled in the config
|
|
extended_descriptions = self.coordinator.config_entry.options.get(
|
|
CONF_EXTENDED_DESCRIPTIONS,
|
|
self.coordinator.config_entry.data.get(CONF_EXTENDED_DESCRIPTIONS, DEFAULT_EXTENDED_DESCRIPTIONS),
|
|
)
|
|
|
|
# Add extended descriptions if enabled (from cache only)
|
|
if extended_descriptions:
|
|
# Add long description if available in cache
|
|
long_desc = get_entity_description(
|
|
"binary_sensor", self.entity_description.translation_key, language, "long_description"
|
|
)
|
|
if long_desc:
|
|
attributes["long_description"] = long_desc
|
|
|
|
# Add usage tips if available in cache
|
|
usage_tips = get_entity_description(
|
|
"binary_sensor", self.entity_description.translation_key, language, "usage_tips"
|
|
)
|
|
if usage_tips:
|
|
attributes["usage_tips"] = usage_tips
|
|
|
|
except (KeyError, ValueError, TypeError) as ex:
|
|
self.coordinator.logger.exception(
|
|
"Error getting binary sensor attributes",
|
|
extra={
|
|
"error": str(ex),
|
|
"entity": self.entity_description.key,
|
|
},
|
|
)
|
|
return None
|
|
else:
|
|
return attributes if attributes else None
|