1
0
mirror of https://github.com/home-assistant/core.git synced 2026-02-15 07:36:16 +00:00

div temporary work

Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
This commit is contained in:
Daniel Hjelseth Høyer
2026-02-12 06:52:20 +01:00
parent d7e0f4e5c3
commit 45d289565e
18 changed files with 1554 additions and 158 deletions

View File

@@ -10,7 +10,12 @@ from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .coordinator import HomevoltConfigEntry, HomevoltDataUpdateCoordinator
PLATFORMS: list[Platform] = [Platform.SENSOR]
PLATFORMS: list[Platform] = [
Platform.NUMBER,
Platform.SELECT,
Platform.SENSOR,
Platform.SWITCH,
]
async def async_setup_entry(hass: HomeAssistant, entry: HomevoltConfigEntry) -> bool:

View File

@@ -0,0 +1,64 @@
"""Shared entity helpers for Homevolt."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
from typing import Any, Concatenate
from homevolt import HomevoltAuthenticationError, HomevoltConnectionError, HomevoltError
from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, MANUFACTURER
from .coordinator import HomevoltDataUpdateCoordinator
class HomevoltEntity(CoordinatorEntity[HomevoltDataUpdateCoordinator]):
"""Base Homevolt entity."""
_attr_has_entity_name = True
def __init__(
self, coordinator: HomevoltDataUpdateCoordinator, device_identifier: str
) -> None:
"""Initialize the Homevolt entity."""
super().__init__(coordinator)
device_id = coordinator.data.unique_id
device_metadata = coordinator.data.device_metadata.get(device_identifier)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, f"{device_id}_{device_identifier}")},
configuration_url=coordinator.client.base_url,
manufacturer=MANUFACTURER,
model=device_metadata.model if device_metadata else None,
name=device_metadata.name if device_metadata else None,
)
def homevolt_exception_handler[_HomevoltEntityT: HomevoltEntity, **_P](
func: Callable[Concatenate[_HomevoltEntityT, _P], Coroutine[Any, Any, Any]],
) -> Callable[Concatenate[_HomevoltEntityT, _P], Coroutine[Any, Any, None]]:
"""Decorate Homevolt calls to handle exceptions."""
async def handler(
self: _HomevoltEntityT, *args: _P.args, **kwargs: _P.kwargs
) -> None:
try:
await func(self, *args, **kwargs)
except HomevoltAuthenticationError as error:
raise ConfigEntryAuthFailed("Authentication failed") from error
except HomevoltConnectionError as error:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="communication_error",
translation_placeholders={"error": str(error)},
) from error
except HomevoltError as error:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="unknown_error",
translation_placeholders={"error": str(error)},
) from error
return handler

View File

@@ -0,0 +1,160 @@
"""Support for Homevolt number entities."""
from __future__ import annotations
from dataclasses import dataclass
from homeassistant.components.number import (
NumberDeviceClass,
NumberEntity,
NumberEntityDescription,
)
from homeassistant.const import PERCENTAGE, EntityCategory, UnitOfPower
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import HomevoltConfigEntry, HomevoltDataUpdateCoordinator
from .entity import HomevoltEntity, homevolt_exception_handler
PARALLEL_UPDATES = 0 # Coordinator-based updates
@dataclass(frozen=True, kw_only=True)
class HomevoltNumberEntityDescription(NumberEntityDescription):
"""Describes a Homevolt number entity."""
available_modes: list[int] | None = None # None means available in all modes
def get_value(self, coordinator: HomevoltDataUpdateCoordinator) -> float | None:
"""Get the value from the coordinator based on the key."""
return coordinator.client.schedule.get(self.key)
NUMBER_DESCRIPTIONS: tuple[HomevoltNumberEntityDescription, ...] = (
HomevoltNumberEntityDescription(
key="setpoint",
translation_key="setpoint",
device_class=NumberDeviceClass.POWER,
entity_category=EntityCategory.CONFIG,
native_unit_of_measurement=UnitOfPower.WATT,
native_min_value=0,
native_max_value=7000,
native_step=1,
available_modes=[1, 2, 7, 8], # Inverter/solar charge/discharge modes
),
HomevoltNumberEntityDescription(
key="max_charge",
translation_key="max_charge",
device_class=NumberDeviceClass.POWER,
entity_category=EntityCategory.CONFIG,
native_unit_of_measurement=UnitOfPower.WATT,
native_min_value=0,
native_max_value=7000,
native_step=1,
),
HomevoltNumberEntityDescription(
key="max_discharge",
translation_key="max_discharge",
device_class=NumberDeviceClass.POWER,
entity_category=EntityCategory.CONFIG,
native_unit_of_measurement=UnitOfPower.WATT,
native_min_value=0,
native_max_value=7000,
native_step=1,
),
HomevoltNumberEntityDescription(
key="min_soc",
translation_key="min_soc",
device_class=NumberDeviceClass.BATTERY,
entity_category=EntityCategory.CONFIG,
native_unit_of_measurement=PERCENTAGE,
native_min_value=0,
native_max_value=100,
native_step=1,
),
HomevoltNumberEntityDescription(
key="max_soc",
translation_key="max_soc",
device_class=NumberDeviceClass.BATTERY,
entity_category=EntityCategory.CONFIG,
native_unit_of_measurement=PERCENTAGE,
native_min_value=0,
native_max_value=100,
native_step=1,
),
HomevoltNumberEntityDescription(
key="grid_import_limit",
translation_key="grid_import_limit",
device_class=NumberDeviceClass.POWER,
entity_category=EntityCategory.CONFIG,
native_unit_of_measurement=UnitOfPower.WATT,
native_min_value=0,
native_max_value=7000,
native_step=1,
available_modes=[3, 5], # Grid charge modes
),
HomevoltNumberEntityDescription(
key="grid_export_limit",
translation_key="grid_export_limit",
device_class=NumberDeviceClass.POWER,
entity_category=EntityCategory.CONFIG,
native_unit_of_measurement=UnitOfPower.WATT,
native_min_value=0,
native_max_value=7000,
native_step=1,
available_modes=[4, 5], # Grid discharge modes
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: HomevoltConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Homevolt number entities."""
coordinator = entry.runtime_data
async_add_entities(
HomevoltNumber(coordinator, description) for description in NUMBER_DESCRIPTIONS
)
class HomevoltNumber(HomevoltEntity, NumberEntity):
"""Representation of a Homevolt number entity."""
entity_description: HomevoltNumberEntityDescription
def __init__(
self,
coordinator: HomevoltDataUpdateCoordinator,
description: HomevoltNumberEntityDescription,
) -> None:
"""Initialize the number entity."""
self.entity_description = description
self._attr_unique_id = f"{coordinator.data.unique_id}_{description.key}"
device_id = coordinator.data.unique_id
super().__init__(coordinator, f"ems_{device_id}")
@property
def available(self) -> bool:
"""Return if entity is available based on current mode."""
if not super().available:
return False
if self.entity_description.available_modes is not None:
current_mode = self.coordinator.client.schedule_mode
if current_mode not in self.entity_description.available_modes:
return False
return True
@property
def native_value(self) -> float | None:
"""Return the current value."""
return self.entity_description.get_value(self.coordinator)
@homevolt_exception_handler
async def async_set_native_value(self, value: float) -> None:
"""Set the value."""
kwargs = {self.entity_description.key: int(value)}
await self.coordinator.client.set_battery_mode(**kwargs)
await self.coordinator.async_request_refresh()

View File

@@ -0,0 +1,51 @@
"""Support for Homevolt select entities."""
from __future__ import annotations
from homevolt.const import SCHEDULE_TYPE
from homeassistant.components.select import SelectEntity
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import HomevoltConfigEntry, HomevoltDataUpdateCoordinator
from .entity import HomevoltEntity, homevolt_exception_handler
PARALLEL_UPDATES = 0 # Coordinator-based updates
async def async_setup_entry(
hass: HomeAssistant,
entry: HomevoltConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Homevolt select entities."""
coordinator = entry.runtime_data
async_add_entities([HomevoltModeSelect(coordinator)])
class HomevoltModeSelect(HomevoltEntity, SelectEntity):
"""Select entity for battery operational mode."""
_attr_entity_category = EntityCategory.CONFIG
_attr_translation_key = "battery_mode"
_attr_options = list(SCHEDULE_TYPE.values())
def __init__(self, coordinator: HomevoltDataUpdateCoordinator) -> None:
"""Initialize the select entity."""
self._attr_unique_id = f"{coordinator.data.unique_id}_battery_mode"
device_id = coordinator.data.unique_id
super().__init__(coordinator, f"ems_{device_id}")
@property
def current_option(self) -> str | None:
"""Return the current selected mode."""
mode_int = self.coordinator.client.schedule_mode
return SCHEDULE_TYPE.get(mode_int, "idle")
@homevolt_exception_handler
async def async_select_option(self, option: str) -> None:
"""Change the selected mode."""
await self.coordinator.client.set_battery_mode(mode=option)
await self.coordinator.async_request_refresh()

View File

@@ -30,6 +30,46 @@
}
},
"entity": {
"number": {
"grid_export_limit": {
"name": "Grid export limit"
},
"grid_import_limit": {
"name": "Grid import limit"
},
"max_charge": {
"name": "Max charge power"
},
"max_discharge": {
"name": "Max discharge power"
},
"max_soc": {
"name": "Maximum state of charge"
},
"min_soc": {
"name": "Minimum state of charge"
},
"setpoint": {
"name": "Power setpoint"
}
},
"select": {
"battery_mode": {
"name": "Battery mode",
"state": {
"frequency_reserve": "Frequency reserve",
"full_solar_export": "Full solar export",
"grid_charge": "Grid charge",
"grid_charge_discharge": "Grid charge/discharge",
"grid_discharge": "Grid discharge",
"idle": "Idle",
"inverter_charge": "Inverter charge",
"inverter_discharge": "Inverter discharge",
"solar_charge": "Solar charge",
"solar_charge_discharge": "Solar charge/discharge"
}
}
},
"sensor": {
"available_charging_energy": {
"name": "Available charging energy"
@@ -142,6 +182,19 @@
"tmin": {
"name": "Minimum temperature"
}
},
"switch": {
"local_mode": {
"name": "Local mode"
}
}
},
"exceptions": {
"communication_error": {
"message": "Failed to communicate with Homevolt: {error}"
},
"unknown_error": {
"message": "An unknown error occurred: {error}"
}
}
}

View File

@@ -0,0 +1,55 @@
"""Support for Homevolt switch entities."""
from __future__ import annotations
from typing import Any
from homeassistant.components.switch import SwitchEntity
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import HomevoltConfigEntry, HomevoltDataUpdateCoordinator
from .entity import HomevoltEntity, homevolt_exception_handler
PARALLEL_UPDATES = 0 # Coordinator-based updates
async def async_setup_entry(
hass: HomeAssistant,
entry: HomevoltConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Homevolt switch entities."""
coordinator = entry.runtime_data
async_add_entities([HomevoltLocalModeSwitch(coordinator)])
class HomevoltLocalModeSwitch(HomevoltEntity, SwitchEntity):
"""Switch entity for Homevolt local mode."""
_attr_entity_category = EntityCategory.CONFIG
_attr_translation_key = "local_mode"
def __init__(self, coordinator: HomevoltDataUpdateCoordinator) -> None:
"""Initialize the switch entity."""
self._attr_unique_id = f"{coordinator.data.unique_id}_local_mode"
device_id = coordinator.data.unique_id
super().__init__(coordinator, f"ems_{device_id}")
@property
def is_on(self) -> bool:
"""Return the local mode state."""
return self.coordinator.client.local_mode_enabled
@homevolt_exception_handler
async def async_turn_on(self, **kwargs: Any) -> None:
"""Enable local mode."""
await self.coordinator.client.enable_local_mode()
await self.coordinator.async_request_refresh()
@homevolt_exception_handler
async def async_turn_off(self, **kwargs: Any) -> None:
"""Disable local mode."""
await self.coordinator.client.disable_local_mode()
await self.coordinator.async_request_refresh()

View File

@@ -23,7 +23,7 @@ from homeassistant.helpers.typing import ConfigType
from homeassistant.util import dt as dt_util, ssl as ssl_util
from .const import AUTH_IMPLEMENTATION, DATA_HASS_CONFIG, DOMAIN, TibberConfigEntry
from .coordinator import TibberDataAPICoordinator
from .coordinator import TibberDataAPICoordinator, TibberDataCoordinator
from .services import async_setup_services
PLATFORMS = [Platform.BINARY_SENSOR, Platform.NOTIFY, Platform.SENSOR]
@@ -39,6 +39,7 @@ class TibberRuntimeData:
session: OAuth2Session
data_api_coordinator: TibberDataAPICoordinator | None = field(default=None)
data_coordinator: TibberDataCoordinator | None = field(default=None)
_client: tibber.Tibber | None = None
async def async_get_client(self, hass: HomeAssistant) -> tibber.Tibber:
@@ -124,9 +125,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: TibberConfigEntry) -> bo
except tibber.FatalHttpExceptionError as err:
raise ConfigEntryNotReady("Fatal HTTP error from Tibber API") from err
coordinator = TibberDataAPICoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data.data_api_coordinator = coordinator
data_api_coordinator = TibberDataAPICoordinator(hass, entry)
await data_api_coordinator.async_config_entry_first_refresh()
entry.runtime_data.data_api_coordinator = data_api_coordinator
data_coordinator = TibberDataCoordinator(hass, entry, entry.runtime_data)
await data_coordinator.async_config_entry_first_refresh()
entry.runtime_data.data_coordinator = data_coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True

View File

@@ -2,7 +2,8 @@
from __future__ import annotations
from datetime import timedelta
from dataclasses import dataclass
from datetime import datetime, timedelta
import logging
from typing import TYPE_CHECKING, cast
@@ -31,6 +32,9 @@ from homeassistant.util.unit_conversion import EnergyConverter
from .const import DOMAIN
if TYPE_CHECKING:
from tibber import TibberHome
from . import TibberRuntimeData
from .const import TibberConfigEntry
FIVE_YEARS = 5 * 365 * 24
@@ -38,8 +42,52 @@ FIVE_YEARS = 5 * 365 * 24
_LOGGER = logging.getLogger(__name__)
class TibberDataCoordinator(DataUpdateCoordinator[None]):
"""Handle Tibber data and insert statistics."""
@dataclass
class TibberHomeData:
"""Structured data per Tibber home from GraphQL and price API."""
currency: str
price_unit: str
current_price: float | None
current_price_time: datetime | None
intraday_price_ranking: float | None
max_price: float
avg_price: float
min_price: float
off_peak_1: float
peak: float
off_peak_2: float
month_cost: float | None
peak_hour: float | None
peak_hour_time: datetime | None
month_cons: float | None
def _build_home_data(home: TibberHome) -> TibberHomeData:
"""Build TibberHomeData from a TibberHome after price info has been fetched."""
price_value, price_time, price_rank = home.current_price_data()
attrs = home.current_attributes()
return TibberHomeData(
currency=home.currency,
price_unit=home.price_unit,
current_price=price_value,
current_price_time=price_time,
intraday_price_ranking=price_rank,
max_price=attrs.get("max_price", 0.0),
avg_price=attrs.get("avg_price", 0.0),
min_price=attrs.get("min_price", 0.0),
off_peak_1=attrs.get("off_peak_1", 0.0),
peak=attrs.get("peak", 0.0),
off_peak_2=attrs.get("off_peak_2", 0.0),
month_cost=getattr(home, "month_cost", None),
peak_hour=getattr(home, "peak_hour", None),
peak_hour_time=getattr(home, "peak_hour_time", None),
month_cons=getattr(home, "month_cons", None),
)
class TibberDataCoordinator(DataUpdateCoordinator[dict[str, TibberHomeData]]):
"""Handle Tibber data, insert statistics, and expose per-home data for sensors."""
config_entry: TibberConfigEntry
@@ -47,24 +95,39 @@ class TibberDataCoordinator(DataUpdateCoordinator[None]):
self,
hass: HomeAssistant,
config_entry: TibberConfigEntry,
tibber_connection: tibber.Tibber,
runtime_data: TibberRuntimeData,
) -> None:
"""Initialize the data handler."""
super().__init__(
hass,
_LOGGER,
config_entry=config_entry,
name=f"Tibber {tibber_connection.name}",
name="Tibber",
update_interval=timedelta(minutes=20),
)
self._tibber_connection = tibber_connection
self._runtime_data = runtime_data
async def _async_update_data(self) -> None:
"""Update data via API."""
async def _async_update_data(self) -> dict[str, TibberHomeData]:
"""Update data via API and return per-home data for sensors."""
tibber_connection = await self._runtime_data.async_get_client(self.hass)
try:
await self._tibber_connection.fetch_consumption_data_active_homes()
await self._tibber_connection.fetch_production_data_active_homes()
await self._insert_statistics()
await tibber_connection.fetch_consumption_data_active_homes()
await tibber_connection.fetch_production_data_active_homes()
now = dt_util.now()
for home in tibber_connection.get_homes(only_active=True):
update_needed = False
last_data_timestamp = home.last_data_timestamp
if last_data_timestamp is None:
update_needed = True
else:
remaining_seconds = (last_data_timestamp - now).total_seconds()
if remaining_seconds < 11 * 3600:
update_needed = True
if update_needed:
await home.update_info_and_price_info()
await self._insert_statistics(tibber_connection)
except tibber.RetryableHttpExceptionError as err:
raise UpdateFailed(f"Error communicating with API ({err.status})") from err
except tibber.FatalHttpExceptionError:
@@ -72,10 +135,16 @@ class TibberDataCoordinator(DataUpdateCoordinator[None]):
self.hass.async_create_task(
self.hass.config_entries.async_reload(self.config_entry.entry_id)
)
return self.data if self.data is not None else {}
async def _insert_statistics(self) -> None:
result: dict[str, TibberHomeData] = {}
for home in tibber_connection.get_homes(only_active=True):
result[home.home_id] = _build_home_data(home)
return result
async def _insert_statistics(self, tibber_connection: tibber.Tibber) -> None:
"""Insert Tibber statistics."""
for home in self._tibber_connection.get_homes():
for home in tibber_connection.get_homes():
sensors: list[tuple[str, bool, str | None, str]] = []
if home.hourly_consumption_data:
sensors.append(

View File

@@ -3,11 +3,9 @@
from __future__ import annotations
from collections.abc import Callable
import datetime
from datetime import timedelta
import logging
from random import randrange
from typing import Any
from typing import Any, cast
import aiohttp
from tibber import FatalHttpExceptionError, RetryableHttpExceptionError, TibberHome
@@ -42,18 +40,16 @@ from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from homeassistant.util import Throttle, dt as dt_util
from homeassistant.util import dt as dt_util
from .const import DOMAIN, MANUFACTURER, TibberConfigEntry
from .const import DOMAIN, TibberConfigEntry
from .coordinator import TibberDataAPICoordinator, TibberDataCoordinator
from .entity import TibberDataCoordinatorEntity, TibberSensor
_LOGGER = logging.getLogger(__name__)
ICON = "mdi:currency-usd"
SCAN_INTERVAL = timedelta(minutes=1)
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=5)
PARALLEL_UPDATES = 0
TWENTY_MINUTES = 20 * 60
RT_SENSORS_UNIQUE_ID_MIGRATION = {
"accumulated_consumption_last_hour": "accumulated consumption current hour",
@@ -262,6 +258,48 @@ SENSORS: tuple[SensorEntityDescription, ...] = (
),
)
PRICE_SENSORS: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key="current_price",
translation_key="electricity_price",
state_class=SensorStateClass.MEASUREMENT,
),
SensorEntityDescription(
key="max_price",
translation_key="max_price",
device_class=SensorDeviceClass.MONETARY,
),
SensorEntityDescription(
key="avg_price",
translation_key="avg_price",
device_class=SensorDeviceClass.MONETARY,
),
SensorEntityDescription(
key="min_price",
translation_key="min_price",
device_class=SensorDeviceClass.MONETARY,
),
SensorEntityDescription(
key="off_peak_1",
translation_key="off_peak_1",
device_class=SensorDeviceClass.MONETARY,
),
SensorEntityDescription(
key="peak",
translation_key="peak",
device_class=SensorDeviceClass.MONETARY,
),
SensorEntityDescription(
key="off_peak_2",
translation_key="off_peak_2",
device_class=SensorDeviceClass.MONETARY,
),
SensorEntityDescription(
key="intraday_price_ranking",
translation_key="intraday_price_ranking",
state_class=SensorStateClass.MEASUREMENT,
),
)
DATA_API_SENSORS: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
@@ -603,14 +641,13 @@ async def _async_setup_graphql_sensors(
entry: TibberConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Tibber sensor."""
"""Set up the Tibber GraphQL-based sensors."""
tibber_connection = await entry.runtime_data.async_get_client(hass)
entity_registry = er.async_get(hass)
coordinator: TibberDataCoordinator | None = None
entities: list[TibberSensor] = []
active_homes: list[TibberHome] = []
for home in tibber_connection.get_homes(only_active=False):
try:
await home.update_info()
@@ -626,13 +663,7 @@ async def _async_setup_graphql_sensors(
raise PlatformNotReady from err
if home.has_active_subscription:
entities.append(TibberSensorElPrice(home))
if coordinator is None:
coordinator = TibberDataCoordinator(hass, entry, tibber_connection)
entities.extend(
TibberDataSensor(home, coordinator, entity_description)
for entity_description in SENSORS
)
active_homes.append(home)
if home.has_real_time_consumption:
entity_creator = TibberRtEntityCreator(
@@ -647,6 +678,18 @@ async def _async_setup_graphql_sensors(
).async_set_updated_data
)
entities: list[TibberSensor] = []
coordinator = entry.runtime_data.data_coordinator
if coordinator is not None and active_homes:
for home in active_homes:
entities.extend(
TibberDataSensor(home, coordinator, desc, model="Price Sensor")
for desc in PRICE_SENSORS
)
entities.extend(
TibberDataSensor(home, coordinator, desc) for desc in SENSORS
)
async_add_entities(entities)
@@ -707,139 +750,69 @@ class TibberDataAPISensor(CoordinatorEntity[TibberDataAPICoordinator], SensorEnt
return sensor.value if sensor else None
class TibberSensor(SensorEntity):
"""Representation of a generic Tibber sensor."""
_attr_has_entity_name = True
def __init__(self, *args: Any, tibber_home: TibberHome, **kwargs: Any) -> None:
"""Initialize the sensor."""
super().__init__(*args, **kwargs)
self._tibber_home = tibber_home
self._home_name = tibber_home.info["viewer"]["home"]["appNickname"]
if self._home_name is None:
self._home_name = tibber_home.info["viewer"]["home"]["address"].get(
"address1", ""
)
self._device_name: str | None = None
self._model: str | None = None
@property
def device_info(self) -> DeviceInfo:
"""Return the device_info of the device."""
device_info = DeviceInfo(
identifiers={(DOMAIN, self._tibber_home.home_id)},
name=self._device_name,
manufacturer=MANUFACTURER,
)
if self._model is not None:
device_info["model"] = self._model
return device_info
class TibberSensorElPrice(TibberSensor):
"""Representation of a Tibber sensor for el price."""
_attr_state_class = SensorStateClass.MEASUREMENT
_attr_translation_key = "electricity_price"
def __init__(self, tibber_home: TibberHome) -> None:
"""Initialize the sensor."""
super().__init__(tibber_home=tibber_home)
self._last_updated: datetime.datetime | None = None
self._spread_load_constant = randrange(TWENTY_MINUTES)
self._attr_available = False
self._attr_extra_state_attributes = {
"app_nickname": None,
"grid_company": None,
"estimated_annual_consumption": None,
"max_price": None,
"avg_price": None,
"min_price": None,
"off_peak_1": None,
"peak": None,
"off_peak_2": None,
"intraday_price_ranking": None,
}
self._attr_icon = ICON
self._attr_unique_id = self._tibber_home.home_id
self._model = "Price Sensor"
self._device_name = self._home_name
async def async_update(self) -> None:
"""Get the latest data and updates the states."""
now = dt_util.now()
if (
not self._tibber_home.last_data_timestamp
or (self._tibber_home.last_data_timestamp - now).total_seconds()
< 10 * 3600 - self._spread_load_constant
or not self.available
):
_LOGGER.debug("Asking for new data")
await self._fetch_data()
elif (
self._tibber_home.price_total
and self._last_updated
and self._last_updated.hour == now.hour
and now - self._last_updated < timedelta(minutes=15)
and self._tibber_home.last_data_timestamp
):
return
res = self._tibber_home.current_price_data()
self._attr_native_value, self._last_updated, price_rank = res
self._attr_extra_state_attributes["intraday_price_ranking"] = price_rank
attrs = self._tibber_home.current_attributes()
self._attr_extra_state_attributes.update(attrs)
self._attr_available = self._attr_native_value is not None
self._attr_native_unit_of_measurement = self._tibber_home.price_unit
@Throttle(MIN_TIME_BETWEEN_UPDATES)
async def _fetch_data(self) -> None:
_LOGGER.debug("Fetching data")
try:
await self._tibber_home.update_info_and_price_info()
except TimeoutError, aiohttp.ClientError:
return
data = self._tibber_home.info["viewer"]["home"]
self._attr_extra_state_attributes["app_nickname"] = data["appNickname"]
self._attr_extra_state_attributes["grid_company"] = data["meteringPointData"][
"gridCompany"
]
self._attr_extra_state_attributes["estimated_annual_consumption"] = data[
"meteringPointData"
]["estimatedAnnualConsumption"]
class TibberDataSensor(TibberSensor, CoordinatorEntity[TibberDataCoordinator]):
"""Representation of a Tibber sensor."""
class TibberDataSensor(TibberDataCoordinatorEntity):
"""Representation of a Tibber sensor reading from coordinator data."""
def __init__(
self,
tibber_home: TibberHome,
coordinator: TibberDataCoordinator,
entity_description: SensorEntityDescription,
*,
model: str | None = None,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator=coordinator, tibber_home=tibber_home)
self.entity_description = entity_description
self._attr_unique_id = (
f"{self._tibber_home.home_id}_{self.entity_description.key}"
)
if entity_description.key == "month_cost":
self._attr_native_unit_of_measurement = self._tibber_home.currency
if self.entity_description.key == "current_price":
# Preserve the existing unique ID for the electricity price
# entity to avoid breaking user setups.
self._attr_unique_id = self._tibber_home.home_id
else:
self._attr_unique_id = (
f"{self._tibber_home.home_id}_{self.entity_description.key}"
)
self._device_name = self._home_name
if model is not None:
self._model = model
@property
def native_value(self) -> StateType:
"""Return the value of the sensor."""
return getattr(self._tibber_home, self.entity_description.key) # type: ignore[no-any-return]
"""Return the value of the sensor from coordinator data."""
home_data = self._get_home_data()
if home_data is None:
return None
return cast(
StateType,
getattr(home_data, self.entity_description.key, None),
)
@property
def native_unit_of_measurement(self) -> str | None:
"""Return the unit from coordinator data for monetary sensors."""
if self.entity_description.key == "current_price":
home_data = self._get_home_data()
if home_data is None:
return None
return home_data.price_unit
if self.entity_description.device_class == SensorDeviceClass.MONETARY:
home_data = self._get_home_data()
if home_data is None:
return None
if self.entity_description.key in {
"max_price",
"avg_price",
"min_price",
"off_peak_1",
"peak",
"off_peak_2",
}:
return home_data.price_unit
return home_data.currency
return self.entity_description.native_unit_of_measurement
class TibberSensorRT(TibberSensor, CoordinatorEntity["TibberRtDataCoordinator"]):
@@ -987,7 +960,7 @@ class TibberRtEntityCreator:
self._async_add_entities(new_entities)
class TibberRtDataCoordinator(DataUpdateCoordinator): # pylint: disable=hass-enforce-class-module
class TibberRtDataCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""Handle Tibber realtime data."""
def __init__(

View File

@@ -43,6 +43,9 @@
"average_power": {
"name": "Average power"
},
"avg_price": {
"name": "Average price today"
},
"cellular_rssi": {
"name": "Cellular signal strength"
},
@@ -136,6 +139,9 @@
"grid_phase_count": {
"name": "Number of grid phases"
},
"intraday_price_ranking": {
"name": "Intraday price ranking"
},
"last_meter_consumption": {
"name": "Last meter consumption"
},
@@ -145,15 +151,30 @@
"max_power": {
"name": "Max power"
},
"max_price": {
"name": "Max price today"
},
"min_power": {
"name": "Min power"
},
"min_price": {
"name": "Min price today"
},
"month_cons": {
"name": "Monthly net consumption"
},
"month_cost": {
"name": "Monthly cost"
},
"off_peak_1": {
"name": "Off-peak 1 average"
},
"off_peak_2": {
"name": "Off-peak 2 average"
},
"peak": {
"name": "Peak average"
},
"peak_hour": {
"name": "Monthly peak hour consumption"
},

View File

@@ -83,6 +83,34 @@ def mock_homevolt_client() -> Generator[MagicMock]:
# Load schedule data from fixture
client.current_schedule = json.loads(load_fixture("schedule.json", DOMAIN))
# Add convenience properties for new client interface
schedule_data = client.current_schedule
schedule = (
schedule_data.get("schedule", [{}])[0]
if schedule_data.get("schedule")
else {}
)
params = schedule.get("params", {})
client.schedule_mode = schedule.get("type", 0)
client.local_mode_enabled = schedule_data.get("local_mode", False)
client.schedule_setpoint = params.get("setpoint")
client.schedule_max_charge = schedule.get("max_charge")
client.schedule_max_discharge = schedule.get("max_discharge")
client.schedule_min_soc = params.get("min_soc") or params.get("min")
client.schedule_max_soc = params.get("max_soc") or params.get("max")
client.schedule_grid_import_limit = params.get("grid_import_limit")
client.schedule_grid_export_limit = params.get("grid_export_limit")
client.schedule_threshold_high = params.get("threshold_high")
client.schedule_threshold_low = params.get("threshold_low")
client.schedule_freq_reg_droop_up = params.get("freq_reg_droop_up")
client.schedule_freq_reg_droop_down = params.get("freq_reg_droop_down")
# Add methods
client.set_battery_mode = AsyncMock()
client.enable_local_mode = AsyncMock()
client.disable_local_mode = AsyncMock()
yield client

View File

@@ -3,10 +3,10 @@
"schedule": [
{
"type": 1,
"max_charge": 6028,
"max_discharge": 6028,
"params": {
"setpoint": 0,
"max_charge": 6028,
"max_discharge": 6028,
"min_soc": 10,
"max_soc": 95
}

View File

@@ -0,0 +1,661 @@
# serializer version: 1
# name: test_entities[number.homevolt_ems_battery-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'max': 100,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.homevolt_ems_battery',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Battery',
'options': dict({
}),
'original_device_class': <NumberDeviceClass.BATTERY: 'battery'>,
'original_icon': None,
'original_name': 'Battery',
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_min_soc',
'unique_id': '40580137858664_battery_min_soc',
'unit_of_measurement': '%',
})
# ---
# name: test_entities[number.homevolt_ems_battery-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'battery',
'friendly_name': 'Homevolt EMS Battery',
'max': 100,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'number.homevolt_ems_battery',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '10',
})
# ---
# name: test_entities[number.homevolt_ems_battery_2-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'max': 100,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.homevolt_ems_battery_2',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Battery',
'options': dict({
}),
'original_device_class': <NumberDeviceClass.BATTERY: 'battery'>,
'original_icon': None,
'original_name': 'Battery',
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_max_soc',
'unique_id': '40580137858664_battery_max_soc',
'unit_of_measurement': '%',
})
# ---
# name: test_entities[number.homevolt_ems_battery_2-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'battery',
'friendly_name': 'Homevolt EMS Battery',
'max': 100,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'number.homevolt_ems_battery_2',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '95',
})
# ---
# name: test_entities[number.homevolt_ems_battery_3-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'max': 100,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.homevolt_ems_battery_3',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Battery',
'options': dict({
}),
'original_device_class': <NumberDeviceClass.BATTERY: 'battery'>,
'original_icon': None,
'original_name': 'Battery',
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_threshold_high',
'unique_id': '40580137858664_battery_threshold_high',
'unit_of_measurement': '%',
})
# ---
# name: test_entities[number.homevolt_ems_battery_3-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'battery',
'friendly_name': 'Homevolt EMS Battery',
'max': 100,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'number.homevolt_ems_battery_3',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unavailable',
})
# ---
# name: test_entities[number.homevolt_ems_battery_4-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'max': 100,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.homevolt_ems_battery_4',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Battery',
'options': dict({
}),
'original_device_class': <NumberDeviceClass.BATTERY: 'battery'>,
'original_icon': None,
'original_name': 'Battery',
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_threshold_low',
'unique_id': '40580137858664_battery_threshold_low',
'unit_of_measurement': '%',
})
# ---
# name: test_entities[number.homevolt_ems_battery_4-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'battery',
'friendly_name': 'Homevolt EMS Battery',
'max': 100,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'number.homevolt_ems_battery_4',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unavailable',
})
# ---
# name: test_entities[number.homevolt_ems_power-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'max': 7000,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.homevolt_ems_power',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Power',
'options': dict({
}),
'original_device_class': <NumberDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Power',
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_power_setpoint',
'unique_id': '40580137858664_battery_power_setpoint',
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
})
# ---
# name: test_entities[number.homevolt_ems_power-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Homevolt EMS Power',
'max': 7000,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
}),
'context': <ANY>,
'entity_id': 'number.homevolt_ems_power',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_entities[number.homevolt_ems_power_2-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'max': 7000,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.homevolt_ems_power_2',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Power',
'options': dict({
}),
'original_device_class': <NumberDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Power',
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_max_charge_power',
'unique_id': '40580137858664_battery_max_charge_power',
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
})
# ---
# name: test_entities[number.homevolt_ems_power_2-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Homevolt EMS Power',
'max': 7000,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
}),
'context': <ANY>,
'entity_id': 'number.homevolt_ems_power_2',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '6028',
})
# ---
# name: test_entities[number.homevolt_ems_power_3-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'max': 7000,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.homevolt_ems_power_3',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Power',
'options': dict({
}),
'original_device_class': <NumberDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Power',
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_max_discharge_power',
'unique_id': '40580137858664_battery_max_discharge_power',
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
})
# ---
# name: test_entities[number.homevolt_ems_power_3-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Homevolt EMS Power',
'max': 7000,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
}),
'context': <ANY>,
'entity_id': 'number.homevolt_ems_power_3',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '6028',
})
# ---
# name: test_entities[number.homevolt_ems_power_4-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'max': 7000,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.homevolt_ems_power_4',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Power',
'options': dict({
}),
'original_device_class': <NumberDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Power',
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_grid_import_limit',
'unique_id': '40580137858664_battery_grid_import_limit',
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
})
# ---
# name: test_entities[number.homevolt_ems_power_4-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Homevolt EMS Power',
'max': 7000,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
}),
'context': <ANY>,
'entity_id': 'number.homevolt_ems_power_4',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unavailable',
})
# ---
# name: test_entities[number.homevolt_ems_power_5-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'max': 7000,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.homevolt_ems_power_5',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Power',
'options': dict({
}),
'original_device_class': <NumberDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Power',
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_grid_export_limit',
'unique_id': '40580137858664_battery_grid_export_limit',
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
})
# ---
# name: test_entities[number.homevolt_ems_power_5-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Homevolt EMS Power',
'max': 7000,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
}),
'context': <ANY>,
'entity_id': 'number.homevolt_ems_power_5',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unavailable',
})
# ---
# name: test_entities[number.homevolt_ems_power_6-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'max': 7000,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.homevolt_ems_power_6',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Power',
'options': dict({
}),
'original_device_class': <NumberDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Power',
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_freq_reg_droop_up',
'unique_id': '40580137858664_battery_freq_reg_droop_up',
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
})
# ---
# name: test_entities[number.homevolt_ems_power_6-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Homevolt EMS Power',
'max': 7000,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
}),
'context': <ANY>,
'entity_id': 'number.homevolt_ems_power_6',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unavailable',
})
# ---
# name: test_entities[number.homevolt_ems_power_7-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'max': 7000,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'number',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'number.homevolt_ems_power_7',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Power',
'options': dict({
}),
'original_device_class': <NumberDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Power',
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_freq_reg_droop_down',
'unique_id': '40580137858664_battery_freq_reg_droop_down',
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
})
# ---
# name: test_entities[number.homevolt_ems_power_7-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Homevolt EMS Power',
'max': 7000,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
}),
'context': <ANY>,
'entity_id': 'number.homevolt_ems_power_7',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unavailable',
})
# ---

View File

@@ -0,0 +1,75 @@
# serializer version: 1
# name: test_entities[select.homevolt_ems-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'idle',
'inverter_charge',
'inverter_discharge',
'grid_charge',
'grid_discharge',
'grid_charge_discharge',
'frequency_reserve',
'solar_charge',
'solar_charge_discharge',
'full_solar_export',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'select',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'select.homevolt_ems',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_mode',
'unique_id': '40580137858664_battery_mode',
'unit_of_measurement': None,
})
# ---
# name: test_entities[select.homevolt_ems-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Homevolt EMS',
'options': list([
'idle',
'inverter_charge',
'inverter_discharge',
'grid_charge',
'grid_discharge',
'grid_charge_discharge',
'frequency_reserve',
'solar_charge',
'solar_charge_discharge',
'full_solar_export',
]),
}),
'context': <ANY>,
'entity_id': 'select.homevolt_ems',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'inverter_charge',
})
# ---

View File

@@ -0,0 +1,50 @@
# serializer version: 1
# name: test_entities[switch.homevolt_ems-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'switch.homevolt_ems',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'local_mode',
'unique_id': '40580137858664_local_mode',
'unit_of_measurement': None,
})
# ---
# name: test_entities[switch.homevolt_ems-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Homevolt EMS',
}),
'context': <ANY>,
'entity_id': 'switch.homevolt_ems',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---

View File

@@ -0,0 +1,42 @@
"""Tests for the Homevolt number platform."""
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.homevolt.const import DOMAIN
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
pytestmark = pytest.mark.usefixtures(
"entity_registry_enabled_by_default", "init_integration"
)
@pytest.fixture
def platforms() -> list[Platform]:
"""Fixture to specify platforms to test."""
return [Platform.NUMBER]
async def test_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the number entities."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, "40580137858664_ems_40580137858664")}
)
assert device_entry
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
for entity_entry in entity_entries:
assert entity_entry.device_id == device_entry.id

View File

@@ -0,0 +1,42 @@
"""Tests for the Homevolt select platform."""
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.homevolt.const import DOMAIN
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
pytestmark = pytest.mark.usefixtures(
"entity_registry_enabled_by_default", "init_integration"
)
@pytest.fixture
def platforms() -> list[Platform]:
"""Fixture to specify platforms to test."""
return [Platform.SELECT]
async def test_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the select entities."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, "40580137858664_ems_40580137858664")}
)
assert device_entry
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
for entity_entry in entity_entries:
assert entity_entry.device_id == device_entry.id

View File

@@ -0,0 +1,42 @@
"""Tests for the Homevolt switch platform."""
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.homevolt.const import DOMAIN
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
pytestmark = pytest.mark.usefixtures(
"entity_registry_enabled_by_default", "init_integration"
)
@pytest.fixture
def platforms() -> list[Platform]:
"""Fixture to specify platforms to test."""
return [Platform.SWITCH]
async def test_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the switch entities."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, "40580137858664_ems_40580137858664")}
)
assert device_entry
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
for entity_entry in entity_entries:
assert entity_entry.device_id == device_entry.id