Add home battery binary sensors to Zonneplan (#183905)

This commit is contained in:
Erwin Douna
2026-10-01 20:55:25 +02:00
committed by GitHub
parent 6edc3771fd
commit 9039d484a1
15 changed files with 1059 additions and 60 deletions
+27 -10
View File
@@ -1,30 +1,47 @@
"""The Zonneplan integration."""
from pyzonneplan import Token, Zonneplan
from pyzonneplan.const import ContractType
from homeassistant.const import CONF_EMAIL, CONF_TOKEN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .coordinator import ZonneplanConfigEntry, ZonneplanCoordinator
from .coordinator import (
ZonneplanBatteryCoordinator,
ZonneplanConfigEntry,
ZonneplanCoordinator,
ZonneplanRuntimeData,
)
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: ZonneplanConfigEntry) -> bool:
"""Set up Zonneplan from a config entry."""
coordinator = ZonneplanCoordinator(
hass,
entry,
Zonneplan(
email=entry.data[CONF_EMAIL],
session=async_get_clientsession(hass),
token=Token.from_dict(entry.data[CONF_TOKEN]),
),
zonneplan = Zonneplan(
email=entry.data[CONF_EMAIL],
session=async_get_clientsession(hass),
token=Token.from_dict(entry.data[CONF_TOKEN]),
)
coordinator = ZonneplanCoordinator(hass, entry, zonneplan)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
battery_coordinator = ZonneplanBatteryCoordinator(
hass,
entry,
zonneplan,
{
contract.uuid: connection.uuid
for connection in coordinator.data.account.connections
for contract in connection.contracts_of_type(ContractType.HOME_BATTERY)
},
)
await battery_coordinator.async_config_entry_first_refresh()
entry.runtime_data = ZonneplanRuntimeData(
coordinator=coordinator, battery_coordinator=battery_coordinator
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
@@ -5,6 +5,8 @@ from dataclasses import dataclass
from datetime import datetime
from typing import override
from pyzonneplan import Battery
from homeassistant.components.binary_sensor import (
BinarySensorEntity,
BinarySensorEntityDescription,
@@ -16,7 +18,7 @@ from homeassistant.util import dt as dt_util
from .const import ZONNEPLAN_TIMEZONE
from .coordinator import ZonneplanConfigEntry, ZonneplanCoordinator
from .entity import ZonneplanEntity
from .entity import ZonneplanBatteryEntity, ZonneplanEntity
PARALLEL_UPDATES = 0
@@ -48,17 +50,60 @@ BINARY_SENSORS: tuple[ZonneplanBinarySensorEntityDescription, ...] = (
)
@dataclass(frozen=True, kw_only=True)
class ZonneplanBatteryBinarySensorEntityDescription(BinarySensorEntityDescription):
"""Describes a Zonneplan home battery binary sensor."""
is_on_fn: Callable[[Battery], bool | None]
BATTERY_BINARY_SENSORS: tuple[ZonneplanBatteryBinarySensorEntityDescription, ...] = (
ZonneplanBatteryBinarySensorEntityDescription(
key="home_optimization_active",
translation_key="home_optimization_active",
is_on_fn=lambda battery: battery.home_optimization_active,
),
ZonneplanBatteryBinarySensorEntityDescription(
key="grid_congestion_active",
translation_key="grid_congestion_active",
is_on_fn=lambda battery: battery.grid_congestion_active,
),
ZonneplanBatteryBinarySensorEntityDescription(
key="dynamic_load_balancing_overload_active",
translation_key="dynamic_load_balancing_overload_active",
is_on_fn=lambda battery: battery.dynamic_load_balancing_overload_active,
),
ZonneplanBatteryBinarySensorEntityDescription(
key="backup_power_active",
translation_key="backup_power_active",
is_on_fn=lambda battery: battery.backup_power_active,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: ZonneplanConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Zonneplan binary sensor platform."""
coordinator = entry.runtime_data
coordinator = entry.runtime_data.coordinator
battery_coordinator = entry.runtime_data.battery_coordinator
async_add_entities(
ZonneplanBinarySensor(coordinator, description)
for description in BINARY_SENSORS
[
*(
ZonneplanBinarySensor(coordinator, description)
for description in BINARY_SENSORS
),
*(
ZonneplanBatteryBinarySensor(
battery_coordinator, contract_uuid, description
)
for contract_uuid in battery_coordinator.data
for description in BATTERY_BINARY_SENSORS
),
]
)
@@ -87,3 +132,15 @@ class ZonneplanBinarySensor(ZonneplanEntity, BinarySensorEntity):
def is_on(self) -> bool | None:
"""Return whether the current hour falls in the price block."""
return self.entity_description.is_on_fn(self.coordinator)
class ZonneplanBatteryBinarySensor(ZonneplanBatteryEntity, BinarySensorEntity):
"""Representation of a Zonneplan home battery binary sensor."""
entity_description: ZonneplanBatteryBinarySensorEntityDescription
@property
@override
def is_on(self) -> bool | None:
"""Return the state of the binary sensor."""
return self.entity_description.is_on_fn(self.battery)
+112 -41
View File
@@ -1,5 +1,6 @@
"""Coordinator for Zonneplan."""
from abc import abstractmethod
import asyncio
from dataclasses import dataclass
from datetime import date, timedelta
@@ -8,6 +9,7 @@ from typing import TYPE_CHECKING, override
from pyzonneplan import (
Account,
Battery,
Connection,
ConsumerPrices,
ElectricityChart,
@@ -33,8 +35,17 @@ from .const import DOMAIN
LOGGER = logging.getLogger(__name__)
UPDATE_INTERVAL = timedelta(minutes=15)
BATTERY_UPDATE_INTERVAL = timedelta(minutes=5)
type ZonneplanConfigEntry = ConfigEntry[ZonneplanCoordinator]
type ZonneplanConfigEntry = ConfigEntry[ZonneplanRuntimeData]
@dataclass(frozen=True, kw_only=True)
class ZonneplanRuntimeData:
"""Runtime data of a Zonneplan config entry."""
coordinator: ZonneplanCoordinator
battery_coordinator: ZonneplanBatteryCoordinator
@dataclass(frozen=True, kw_only=True)
@@ -61,13 +72,17 @@ def _connection(account: Account, market_segment: str) -> Connection | None:
)
class ZonneplanCoordinator(DataUpdateCoordinator[ZonneplanData]):
"""Coordinator to manage fetching Zonneplan account data."""
class ZonneplanBaseCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
"""Base coordinator for the Zonneplan API."""
config_entry: ZonneplanConfigEntry
def __init__(
self, hass: HomeAssistant, entry: ZonneplanConfigEntry, zonneplan: Zonneplan
self,
hass: HomeAssistant,
entry: ZonneplanConfigEntry,
zonneplan: Zonneplan,
update_interval: timedelta,
) -> None:
"""Initialize the coordinator."""
super().__init__(
@@ -75,50 +90,19 @@ class ZonneplanCoordinator(DataUpdateCoordinator[ZonneplanData]):
LOGGER,
config_entry=entry,
name=DOMAIN,
update_interval=UPDATE_INTERVAL,
update_interval=update_interval,
)
self.zonneplan = zonneplan
async def _async_fetch_electricity(
self, account: Account, today: date
) -> tuple[ConsumerPrices | None, ElectricityChart | None]:
"""Fetch electricity prices and usage, if the account has electricity."""
if (connection := _connection(account, "electricity")) is None:
return None, None
return await asyncio.gather(
self.zonneplan.async_get_consumer_prices(PriceChart.ELECTRICITY_HOURLY),
self.zonneplan.async_get_electricity_chart(
connection.uuid, today, ConsumptionChart.DAYS
),
)
async def _async_fetch_gas(
self, account: Account, today: date
) -> tuple[ConsumerPrices | None, GasChart | None]:
"""Fetch gas prices and usage, if the account has gas."""
if (connection := _connection(account, "gas")) is None:
return None, None
return await asyncio.gather(
self.zonneplan.async_get_consumer_prices(PriceChart.GAS_DAILY),
self.zonneplan.async_get_gas_chart(
connection.uuid, today, ConsumptionChart.DAYS
),
)
@abstractmethod
async def _async_fetch(self) -> _DataT:
"""Fetch the data from the Zonneplan API."""
@override
async def _async_update_data(self) -> ZonneplanData:
async def _async_update_data(self) -> _DataT:
"""Fetch data from the Zonneplan API."""
try:
account = await self.zonneplan.async_get_account()
# The consumption charts cover the month of the given local day.
today = dt_util.now().date()
(
(electricity_prices, electricity_usage),
(gas_prices, gas_usage),
) = await asyncio.gather(
self._async_fetch_electricity(account, today),
self._async_fetch_gas(account, today),
)
data = await self._async_fetch()
except ZonneplanAuthenticationError as err:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
@@ -156,6 +140,57 @@ class ZonneplanCoordinator(DataUpdateCoordinator[ZonneplanData]):
data={**self.config_entry.data, CONF_TOKEN: token},
)
return data
class ZonneplanCoordinator(ZonneplanBaseCoordinator[ZonneplanData]):
"""Coordinator to manage fetching Zonneplan account data."""
def __init__(
self, hass: HomeAssistant, entry: ZonneplanConfigEntry, zonneplan: Zonneplan
) -> None:
"""Initialize the coordinator."""
super().__init__(hass, entry, zonneplan, UPDATE_INTERVAL)
async def _async_fetch_electricity(
self, account: Account, today: date
) -> tuple[ConsumerPrices | None, ElectricityChart | None]:
"""Fetch electricity prices and usage, if the account has electricity."""
if (connection := _connection(account, "electricity")) is None:
return None, None
return await asyncio.gather(
self.zonneplan.async_get_consumer_prices(PriceChart.ELECTRICITY_HOURLY),
self.zonneplan.async_get_electricity_chart(
connection.uuid, today, ConsumptionChart.DAYS
),
)
async def _async_fetch_gas(
self, account: Account, today: date
) -> tuple[ConsumerPrices | None, GasChart | None]:
"""Fetch gas prices and usage, if the account has gas."""
if (connection := _connection(account, "gas")) is None:
return None, None
return await asyncio.gather(
self.zonneplan.async_get_consumer_prices(PriceChart.GAS_DAILY),
self.zonneplan.async_get_gas_chart(
connection.uuid, today, ConsumptionChart.DAYS
),
)
@override
async def _async_fetch(self) -> ZonneplanData:
"""Fetch the account, prices and usage."""
account = await self.zonneplan.async_get_account()
# The consumption charts cover the month of the given local day.
today = dt_util.now().date()
(
(electricity_prices, electricity_usage),
(gas_prices, gas_usage),
) = await asyncio.gather(
self._async_fetch_electricity(account, today),
self._async_fetch_gas(account, today),
)
return ZonneplanData(
account=account,
electricity_prices=electricity_prices,
@@ -163,3 +198,39 @@ class ZonneplanCoordinator(DataUpdateCoordinator[ZonneplanData]):
electricity_usage=electricity_usage,
gas_usage=gas_usage,
)
class ZonneplanBatteryCoordinator(ZonneplanBaseCoordinator[dict[str, Battery]]):
"""Coordinator to manage fetching the Zonneplan home batteries."""
def __init__(
self,
hass: HomeAssistant,
entry: ZonneplanConfigEntry,
zonneplan: Zonneplan,
connection_uuids: dict[str, str],
) -> None:
"""Initialize the coordinator with the connection UUID of each battery."""
super().__init__(hass, entry, zonneplan, BATTERY_UPDATE_INTERVAL)
self.connection_uuids = connection_uuids
@override
async def _async_fetch(self) -> dict[str, Battery]:
"""Fetch the state of every battery, keyed by contract UUID."""
installations = await asyncio.gather(
*(
self.zonneplan.async_get_battery(connection_uuid, contract_uuid)
for contract_uuid, connection_uuid in self.connection_uuids.items()
)
)
batteries: dict[str, Battery] = {}
for contract_uuid, installation in zip(
self.connection_uuids, installations, strict=True
):
if (battery := installation.battery) is None:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="battery_not_found",
)
batteries[contract_uuid] = battery
return batteries
@@ -26,6 +26,7 @@ TO_REDACT = {
"serial_number",
"uuid",
"external_contract_id",
"identifier",
}
@@ -33,7 +34,7 @@ async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: ZonneplanConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
data = entry.runtime_data.data
data = entry.runtime_data.coordinator.data
return async_redact_data(
{
@@ -52,6 +53,10 @@ async def async_get_config_entry_diagnostics(
data.electricity_usage.to_dict() if data.electricity_usage else None
),
"gas_usage": data.gas_usage.to_dict() if data.gas_usage else None,
"batteries": [
battery.contract.to_dict()
for battery in entry.runtime_data.battery_coordinator.data.values()
],
},
TO_REDACT,
)
+34 -1
View File
@@ -1,11 +1,13 @@
"""Base entity for Zonneplan."""
from pyzonneplan import Battery
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import ZonneplanCoordinator
from .coordinator import ZonneplanBatteryCoordinator, ZonneplanCoordinator
class ZonneplanEntity(CoordinatorEntity[ZonneplanCoordinator]):
@@ -27,3 +29,34 @@ class ZonneplanEntity(CoordinatorEntity[ZonneplanCoordinator]):
name="Zonneplan",
entry_type=DeviceEntryType.SERVICE,
)
class ZonneplanBatteryEntity(CoordinatorEntity[ZonneplanBatteryCoordinator]):
"""Base entity for a Zonneplan home battery."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: ZonneplanBatteryCoordinator,
contract_uuid: str,
entity_description: EntityDescription,
) -> None:
"""Initialize the entity."""
super().__init__(coordinator)
self.entity_description = entity_description
self._contract_uuid = contract_uuid
contract = self.battery.contract
self._attr_unique_id = f"{contract_uuid}_{entity_description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, contract_uuid)},
name=contract.label,
manufacturer="Zonneplan",
model=contract.model_name,
serial_number=contract.serial_number,
)
@property
def battery(self) -> Battery:
"""Return the battery of this entity."""
return self.coordinator.data[self._contract_uuid]
@@ -1,12 +1,33 @@
{
"entity": {
"binary_sensor": {
"backup_power_active": {
"default": "mdi:battery-alert-variant-outline",
"state": {
"on": "mdi:battery-alert-variant"
}
},
"dynamic_load_balancing_overload_active": {
"default": "mdi:scale-balance"
},
"electricity_price_low": {
"default": "mdi:cash-minus",
"state": {
"off": "mdi:cash-plus",
"on": "mdi:cash-minus"
}
},
"grid_congestion_active": {
"default": "mdi:transmission-tower",
"state": {
"on": "mdi:transmission-tower-off"
}
},
"home_optimization_active": {
"default": "mdi:home-lightning-bolt-outline",
"state": {
"on": "mdi:home-lightning-bolt"
}
}
},
"sensor": {
+1 -1
View File
@@ -390,7 +390,7 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Zonneplan sensor platform."""
coordinator = entry.runtime_data
coordinator = entry.runtime_data.coordinator
async_add_entities(
[
@@ -53,8 +53,20 @@
},
"entity": {
"binary_sensor": {
"backup_power_active": {
"name": "Backup power active"
},
"dynamic_load_balancing_overload_active": {
"name": "Load balancing overload"
},
"electricity_price_low": {
"name": "Electricity price low"
},
"grid_congestion_active": {
"name": "Grid congestion"
},
"home_optimization_active": {
"name": "Home optimization active"
}
},
"sensor": {
@@ -130,6 +142,9 @@
"api_error": {
"message": "Unexpected error from the Zonneplan API"
},
"battery_not_found": {
"message": "The Zonneplan API returned no data for the home battery"
},
"cannot_connect": {
"message": "Error communicating with the Zonneplan API"
},
+4
View File
@@ -7,6 +7,7 @@ from unittest.mock import AsyncMock, patch
import pytest
from pyzonneplan import (
Account,
BatteryInstallation,
ConsumerPrices,
ElectricityChart,
GasChart,
@@ -73,6 +74,9 @@ def mock_zonneplan_client() -> Generator[AsyncMock]:
client.async_get_gas_chart.return_value = GasChart.from_dict(
load_json_object_fixture("get_gas_chart_days.json", DOMAIN)
)
client.async_get_battery.return_value = BatteryInstallation.from_dict(
load_json_object_fixture("get_battery_installation.json", DOMAIN)
)
with (
patch("homeassistant.components.zonneplan.Zonneplan", return_value=client),
@@ -38,6 +38,16 @@
"deposit_type": "fixed",
"monthly_advanced_deposit_amount": 930000000
}
},
{
"uuid": "00000000-0000-4000-8000-000000000201",
"label": "Thuisbatterij",
"type": "home_battery_installation",
"start_date": "2026-04-09T22:00:00.000000Z",
"end_date": null,
"meta": {
"show_in_contract_screen": true
}
}
],
"features": [
@@ -0,0 +1,49 @@
{
"contracts": [
{
"uuid": "00000000-0000-4000-8000-000000000201",
"label": "Thuisbatterij",
"type": "home_battery_installation",
"start_date": "2026-04-09T22:00:00.000000Z",
"end_date": null,
"meta": {
"identifier": "SERIAL0000003",
"host_device_model_name": "Example Battery 5",
"first_measured_at": "2026-04-10T12:30:00.000000Z",
"last_measured_at": "2026-09-29T11:55:00.000000Z",
"state_of_charge": 655,
"power_ac": -800,
"inverter_state": "Discharging",
"battery_state": "Discharging",
"dynamic_charging_enabled": false,
"home_optimization_enabled": false,
"home_optimization_active": false,
"dynamic_load_balancing_enabled": true,
"dynamic_load_balancing_overload_active": false,
"grid_congestion_active": true,
"manual_control_enabled": false,
"manual_control_state": null,
"self_consumption_enabled": true,
"cycle_count": 100,
"show_in_contract_screen": true,
"reserve_discharge_cutoff_wh": 1000,
"backup_power_capable": true,
"backup_power_usable_capacity_wh": 4000,
"backup_power_active": false,
"total_earned": 1000000000,
"total_day": 5000000,
"delivery_day": 3000,
"production_day": 2500,
"average_day": 4000000,
"production_month": null
}
}
],
"measurement_groups": [
{
"date": "2026-09-28T22:00:00.000000Z",
"total": 5000000,
"measurements": []
}
]
}
@@ -1,4 +1,204 @@
# serializer version: 1
# name: test_binary_sensor[in_low_price_block][binary_sensor.thuisbatterij_backup_power_active-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.thuisbatterij_backup_power_active',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Backup power active',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Backup power active',
'platform': 'zonneplan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'backup_power_active',
'unique_id': '00000000-0000-4000-8000-000000000201_backup_power_active',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[in_low_price_block][binary_sensor.thuisbatterij_backup_power_active-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Thuisbatterij Backup power active',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.thuisbatterij_backup_power_active',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_binary_sensor[in_low_price_block][binary_sensor.thuisbatterij_grid_congestion-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.thuisbatterij_grid_congestion',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Grid congestion',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Grid congestion',
'platform': 'zonneplan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'grid_congestion_active',
'unique_id': '00000000-0000-4000-8000-000000000201_grid_congestion_active',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[in_low_price_block][binary_sensor.thuisbatterij_grid_congestion-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Thuisbatterij Grid congestion',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.thuisbatterij_grid_congestion',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_binary_sensor[in_low_price_block][binary_sensor.thuisbatterij_home_optimization_active-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.thuisbatterij_home_optimization_active',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Home optimization active',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Home optimization active',
'platform': 'zonneplan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'home_optimization_active',
'unique_id': '00000000-0000-4000-8000-000000000201_home_optimization_active',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[in_low_price_block][binary_sensor.thuisbatterij_home_optimization_active-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Thuisbatterij Home optimization active',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.thuisbatterij_home_optimization_active',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_binary_sensor[in_low_price_block][binary_sensor.thuisbatterij_load_balancing_overload-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.thuisbatterij_load_balancing_overload',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Load balancing overload',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Load balancing overload',
'platform': 'zonneplan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'dynamic_load_balancing_overload_active',
'unique_id': '00000000-0000-4000-8000-000000000201_dynamic_load_balancing_overload_active',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[in_low_price_block][binary_sensor.thuisbatterij_load_balancing_overload-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Thuisbatterij Load balancing overload',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.thuisbatterij_load_balancing_overload',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_binary_sensor[in_low_price_block][binary_sensor.zonneplan_electricity_price_low-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -49,6 +249,206 @@
'state': 'on',
})
# ---
# name: test_binary_sensor[no_prices_today][binary_sensor.thuisbatterij_backup_power_active-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.thuisbatterij_backup_power_active',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Backup power active',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Backup power active',
'platform': 'zonneplan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'backup_power_active',
'unique_id': '00000000-0000-4000-8000-000000000201_backup_power_active',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[no_prices_today][binary_sensor.thuisbatterij_backup_power_active-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Thuisbatterij Backup power active',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.thuisbatterij_backup_power_active',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_binary_sensor[no_prices_today][binary_sensor.thuisbatterij_grid_congestion-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.thuisbatterij_grid_congestion',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Grid congestion',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Grid congestion',
'platform': 'zonneplan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'grid_congestion_active',
'unique_id': '00000000-0000-4000-8000-000000000201_grid_congestion_active',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[no_prices_today][binary_sensor.thuisbatterij_grid_congestion-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Thuisbatterij Grid congestion',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.thuisbatterij_grid_congestion',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_binary_sensor[no_prices_today][binary_sensor.thuisbatterij_home_optimization_active-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.thuisbatterij_home_optimization_active',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Home optimization active',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Home optimization active',
'platform': 'zonneplan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'home_optimization_active',
'unique_id': '00000000-0000-4000-8000-000000000201_home_optimization_active',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[no_prices_today][binary_sensor.thuisbatterij_home_optimization_active-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Thuisbatterij Home optimization active',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.thuisbatterij_home_optimization_active',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_binary_sensor[no_prices_today][binary_sensor.thuisbatterij_load_balancing_overload-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.thuisbatterij_load_balancing_overload',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Load balancing overload',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Load balancing overload',
'platform': 'zonneplan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'dynamic_load_balancing_overload_active',
'unique_id': '00000000-0000-4000-8000-000000000201_dynamic_load_balancing_overload_active',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[no_prices_today][binary_sensor.thuisbatterij_load_balancing_overload-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Thuisbatterij Load balancing overload',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.thuisbatterij_load_balancing_overload',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_binary_sensor[no_prices_today][binary_sensor.zonneplan_electricity_price_low-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -99,6 +499,206 @@
'state': 'unknown',
})
# ---
# name: test_binary_sensor[outside_low_price_block][binary_sensor.thuisbatterij_backup_power_active-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.thuisbatterij_backup_power_active',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Backup power active',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Backup power active',
'platform': 'zonneplan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'backup_power_active',
'unique_id': '00000000-0000-4000-8000-000000000201_backup_power_active',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[outside_low_price_block][binary_sensor.thuisbatterij_backup_power_active-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Thuisbatterij Backup power active',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.thuisbatterij_backup_power_active',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_binary_sensor[outside_low_price_block][binary_sensor.thuisbatterij_grid_congestion-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.thuisbatterij_grid_congestion',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Grid congestion',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Grid congestion',
'platform': 'zonneplan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'grid_congestion_active',
'unique_id': '00000000-0000-4000-8000-000000000201_grid_congestion_active',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[outside_low_price_block][binary_sensor.thuisbatterij_grid_congestion-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Thuisbatterij Grid congestion',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.thuisbatterij_grid_congestion',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_binary_sensor[outside_low_price_block][binary_sensor.thuisbatterij_home_optimization_active-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.thuisbatterij_home_optimization_active',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Home optimization active',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Home optimization active',
'platform': 'zonneplan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'home_optimization_active',
'unique_id': '00000000-0000-4000-8000-000000000201_home_optimization_active',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[outside_low_price_block][binary_sensor.thuisbatterij_home_optimization_active-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Thuisbatterij Home optimization active',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.thuisbatterij_home_optimization_active',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_binary_sensor[outside_low_price_block][binary_sensor.thuisbatterij_load_balancing_overload-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.thuisbatterij_load_balancing_overload',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Load balancing overload',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Load balancing overload',
'platform': 'zonneplan',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'dynamic_load_balancing_overload_active',
'unique_id': '00000000-0000-4000-8000-000000000201_dynamic_load_balancing_overload_active',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[outside_low_price_block][binary_sensor.thuisbatterij_load_balancing_overload-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Thuisbatterij Load balancing overload',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.thuisbatterij_load_balancing_overload',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_binary_sensor[outside_low_price_block][binary_sensor.zonneplan_electricity_price_low-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -40,6 +40,16 @@
'type': 'electricity',
'uuid': '**REDACTED**',
}),
dict({
'end_date': None,
'label': 'Thuisbatterij',
'meta': dict({
'show_in_contract_screen': True,
}),
'start_date': '2026-04-09T22:00:00+00:00',
'type': 'home_battery_installation',
'uuid': '**REDACTED**',
}),
]),
'ean': '**REDACTED**',
'features': list([
@@ -99,6 +109,46 @@
'uuid': '**REDACTED**',
}),
}),
'batteries': list([
dict({
'end_date': None,
'label': 'Thuisbatterij',
'meta': dict({
'average_day': 4000000,
'backup_power_active': False,
'backup_power_capable': True,
'backup_power_usable_capacity_wh': 4000,
'battery_state': 'Discharging',
'cycle_count': 100,
'delivery_day': 3000,
'dynamic_charging_enabled': False,
'dynamic_load_balancing_enabled': True,
'dynamic_load_balancing_overload_active': False,
'first_measured_at': '2026-04-10T12:30:00.000000Z',
'grid_congestion_active': True,
'home_optimization_active': False,
'home_optimization_enabled': False,
'host_device_model_name': 'Example Battery 5',
'identifier': '**REDACTED**',
'inverter_state': 'Discharging',
'last_measured_at': '2026-09-29T11:55:00.000000Z',
'manual_control_enabled': False,
'manual_control_state': None,
'power_ac': -800,
'production_day': 2500,
'production_month': None,
'reserve_discharge_cutoff_wh': 1000,
'self_consumption_enabled': True,
'show_in_contract_screen': True,
'state_of_charge': 655,
'total_day': 5000000,
'total_earned': 1000000000,
}),
'start_date': '2026-04-09T22:00:00+00:00',
'type': 'home_battery_installation',
'uuid': '**REDACTED**',
}),
]),
'connections': list([
dict({
'contracts': list([
@@ -125,6 +175,16 @@
'type': 'electricity',
'uuid': '**REDACTED**',
}),
dict({
'end_date': None,
'label': 'Thuisbatterij',
'meta': dict({
'show_in_contract_screen': True,
}),
'start_date': '2026-04-09T22:00:00+00:00',
'type': 'home_battery_installation',
'uuid': '**REDACTED**',
}),
]),
'ean': '**REDACTED**',
'features': list([
@@ -1,18 +1,22 @@
"""Tests for the Zonneplan binary sensor platform."""
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from pyzonneplan import ZonneplanConnectionError
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.zonneplan import Platform
from homeassistant.components.zonneplan.coordinator import BATTERY_UPDATE_INTERVAL
from homeassistant.const import STATE_ON, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
LOW_PRICE_ENTITY_ID = "binary_sensor.zonneplan_electricity_price_low"
GRID_CONGESTION_ENTITY_ID = "binary_sensor.thuisbatterij_grid_congestion"
@pytest.mark.parametrize(
@@ -43,3 +47,32 @@ async def test_binary_sensor(
await hass.async_block_till_done()
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_battery_binary_sensor_unavailable_on_update_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_zonneplan_client: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the battery binary sensors become unavailable when an update fails."""
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.zonneplan.PLATFORMS",
[Platform.BINARY_SENSOR],
):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert (state := hass.states.get(GRID_CONGESTION_ENTITY_ID))
assert state.state == STATE_ON
mock_zonneplan_client.async_get_battery.side_effect = ZonneplanConnectionError(
"boom"
)
freezer.tick(BATTERY_UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert (state := hass.states.get(GRID_CONGESTION_ENTITY_ID))
assert state.state == STATE_UNAVAILABLE
+24
View File
@@ -6,6 +6,7 @@ from unittest.mock import AsyncMock
from freezegun.api import FrozenDateTimeFactory
import pytest
from pyzonneplan import (
BatteryInstallation,
Token,
ZonneplanAuthenticationError,
ZonneplanConnectionError,
@@ -70,6 +71,29 @@ async def test_setup_entry_update_failed(
assert mock_config_entry.state is expected_state
@pytest.mark.parametrize(
"side_effect",
[
pytest.param(ZonneplanConnectionError("boom"), id="connection_error"),
pytest.param([BatteryInstallation()], id="battery_not_found"),
],
)
async def test_setup_entry_battery_update_failed(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_zonneplan_client: AsyncMock,
side_effect: Exception | list[BatteryInstallation],
) -> None:
"""Test a failing home battery fetch marks the entry for retry."""
mock_zonneplan_client.async_get_battery.side_effect = side_effect
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_rate_limit_postpones_update(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,