Add BackupMode to Fronius integration.

homeassistant/components/fronius/__init__.py

@@ -37,7 +37,7 @@
 )
 
 _LOGGER: Final = logging.getLogger(__name__)
-PLATFORMS: Final = [Platform.SENSOR]
+PLATFORMS: Final = [Platform.BINARY_SENSOR, Platform.SENSOR]
 
 type FroniusConfigEntry = ConfigEntry[FroniusSolarNet]
 

homeassistant/components/fronius/binary_sensor.py

@@ -0,0 +1,95 @@
+"""Binary sensors for Fronius devices."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from homeassistant.components.binary_sensor import (
+    BinarySensorEntity,
+    BinarySensorEntityDescription,
+)
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .entity import FroniusEntity, FroniusEntityDescription
+
+if TYPE_CHECKING:
+    from . import FroniusConfigEntry
+    from .coordinator import FroniusPowerFlowUpdateCoordinator
+
+
+@dataclass(frozen=True)
+class FroniusBinarySensorEntityDescription(
+    FroniusEntityDescription, BinarySensorEntityDescription
+):
+    """Describes Fronius binary_sensor entity."""
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: FroniusConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up Fronius binary_sensor entities based on a config entry."""
+    solar_net = config_entry.runtime_data
+
+    if solar_net.power_flow_coordinator is not None:
+        solar_net.power_flow_coordinator.add_entities_for_seen_keys(
+            async_add_entities,
+            Platform.BINARY_SENSOR,
+            PowerFlowBinarySensor,
+        )
+
+
+POWER_FLOW_BINARY_ENTITY_DESCRIPTIONS: list[FroniusEntityDescription] = [
+    FroniusBinarySensorEntityDescription(
+        name="Backup mode",
+        key="backup_mode",
+    ),
+    FroniusBinarySensorEntityDescription(
+        name="Battery standby",
+        key="battery_standby",
+    ),
+]
+
+
+class _FroniusBinarySensorEntity(FroniusEntity, BinarySensorEntity):
+    """Defines a Fronius binary_sensor entity."""
+
+    entity_description: FroniusBinarySensorEntityDescription
+
+    def _get_entity_value(self) -> bool | None:
+        """Extract entity value from coordinator. Raises KeyError if not included in latest update."""
+        return bool(
+            self.coordinator.data[self.solar_net_id][self.response_key]["value"]
+        )
+
+    def _set_entity_value(self) -> None:
+        """binary_sensor requires a boolean value in _attr_is_on."""
+        self._attr_is_on = self._get_entity_value()
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        self._attr_is_on = self._get_entity_value()
+        self.async_write_ha_state()
+
+
+class PowerFlowBinarySensor(_FroniusBinarySensorEntity):
+    """Defines a Fronius power flow binary_sensor entity."""
+
+    def __init__(
+        self,
+        coordinator: FroniusPowerFlowUpdateCoordinator,
+        description: FroniusBinarySensorEntityDescription,
+        solar_net_id: str,
+    ) -> None:
+        """Set up an individual Fronius power flow binary_sensor."""
+        super().__init__(coordinator, description, solar_net_id)
+        # SolarNet device is already created in FroniusSolarNet._create_solar_net_device
+        self._attr_device_info = coordinator.solar_net.system_device_info
+        self._attr_unique_id = (
+            f"{coordinator.solar_net.solar_net_device_id}-power_flow-{description.key}"
+        )

homeassistant/components/fronius/coordinator.py

@@ -3,34 +3,37 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+from collections.abc import Mapping
 from datetime import timedelta
 from typing import TYPE_CHECKING, Any
 
 from pyfronius import BadStatusError, FroniusError
 
+from homeassistant.const import Platform
 from homeassistant.core import callback
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
+from .binary_sensor import POWER_FLOW_BINARY_ENTITY_DESCRIPTIONS
 from .const import (
     SOLAR_NET_ID_POWER_FLOW,
     SOLAR_NET_ID_SYSTEM,
     FroniusDeviceInfo,
     SolarNetId,
 )
+from .entity import FroniusEntity
 from .sensor import (
     INVERTER_ENTITY_DESCRIPTIONS,
     LOGGER_ENTITY_DESCRIPTIONS,
     METER_ENTITY_DESCRIPTIONS,
     OHMPILOT_ENTITY_DESCRIPTIONS,
     POWER_FLOW_ENTITY_DESCRIPTIONS,
     STORAGE_ENTITY_DESCRIPTIONS,
-    FroniusSensorEntityDescription,
 )
 
 if TYPE_CHECKING:
     from . import FroniusSolarNet
-    from .sensor import _FroniusSensorEntity
+    from .entity import FroniusEntityDescription
 
 
 class FroniusCoordinatorBase(
@@ -40,7 +43,7 @@ class FroniusCoordinatorBase(
 
     default_interval: timedelta
     error_interval: timedelta
-    valid_descriptions: list[FroniusSensorEntityDescription]
+    valid_descriptions: Mapping[Platform, list[FroniusEntityDescription]]
 
     MAX_FAILED_UPDATES = 3
 
@@ -50,7 +53,7 @@ def __init__(self, *args: Any, solar_net: FroniusSolarNet, **kwargs: Any) -> Non
         self.solar_net = solar_net
         # unregistered_descriptors are used to create entities in platform module
         self.unregistered_descriptors: dict[
-            SolarNetId, list[FroniusSensorEntityDescription]
+            SolarNetId, dict[Platform, list[FroniusEntityDescription]]
         ] = {}
         super().__init__(*args, update_interval=self.default_interval, **kwargs)
 
@@ -76,29 +79,35 @@ async def _async_update_data(self) -> dict[SolarNetId, Any]:
             for solar_net_id in data:
                 if solar_net_id not in self.unregistered_descriptors:
                     # id seen for the first time
-                    self.unregistered_descriptors[solar_net_id] = (
-                        self.valid_descriptions.copy()
-                    )
+                    self.unregistered_descriptors[solar_net_id] = {
+                        platform: descriptors.copy()
+                        for platform, descriptors in self.valid_descriptions.items()
+                    }
             return data
 
     @callback
-    def add_entities_for_seen_keys[_FroniusEntityT: _FroniusSensorEntity](
+    def add_entities_for_seen_keys[FroniusEntityT: FroniusEntity](
         self,
         async_add_entities: AddEntitiesCallback,
-        entity_constructor: type[_FroniusEntityT],
+        platform: Platform,
+        entity_constructor: type[FroniusEntityT],
     ) -> None:
         """Add entities for received keys and registers listener for future seen keys.
 
         Called from a platforms `async_setup_entry`.
+
+        Only those descriptions matching the supplied platform will be added.
         """
 
         @callback
         def _add_entities_for_unregistered_descriptors() -> None:
             """Add entities for keys seen for the first time."""
-            new_entities: list[_FroniusEntityT] = []
+            new_entities: list[FroniusEntityT] = []
             for solar_net_id, device_data in self.data.items():
                 remaining_unregistered_descriptors = []
-                for description in self.unregistered_descriptors[solar_net_id]:
+                for description in self.unregistered_descriptors[solar_net_id][
+                    platform
+                ]:
                     key = description.response_key or description.key
                     if key not in device_data:
                         remaining_unregistered_descriptors.append(description)
@@ -113,7 +122,7 @@ def _add_entities_for_unregistered_descriptors() -> None:
                             solar_net_id=solar_net_id,
                         )
                     )
-                self.unregistered_descriptors[solar_net_id] = (
+                self.unregistered_descriptors[solar_net_id][platform] = (
                     remaining_unregistered_descriptors
                 )
             async_add_entities(new_entities)
@@ -129,7 +138,7 @@ class FroniusInverterUpdateCoordinator(FroniusCoordinatorBase):
 
     default_interval = timedelta(minutes=1)
     error_interval = timedelta(minutes=10)
-    valid_descriptions = INVERTER_ENTITY_DESCRIPTIONS
+    valid_descriptions = {Platform.SENSOR: INVERTER_ENTITY_DESCRIPTIONS}
 
     SILENT_RETRIES = 3
 
@@ -165,7 +174,7 @@ class FroniusLoggerUpdateCoordinator(FroniusCoordinatorBase):
 
     default_interval = timedelta(hours=1)
     error_interval = timedelta(hours=1)
-    valid_descriptions = LOGGER_ENTITY_DESCRIPTIONS
+    valid_descriptions = {Platform.SENSOR: LOGGER_ENTITY_DESCRIPTIONS}
 
     async def _update_method(self) -> dict[SolarNetId, Any]:
         """Return data per solar net id from pyfronius."""
@@ -178,7 +187,7 @@ class FroniusMeterUpdateCoordinator(FroniusCoordinatorBase):
 
     default_interval = timedelta(minutes=1)
     error_interval = timedelta(minutes=10)
-    valid_descriptions = METER_ENTITY_DESCRIPTIONS
+    valid_descriptions = {Platform.SENSOR: METER_ENTITY_DESCRIPTIONS}
 
     async def _update_method(self) -> dict[SolarNetId, Any]:
         """Return data per solar net id from pyfronius."""
@@ -191,7 +200,7 @@ class FroniusOhmpilotUpdateCoordinator(FroniusCoordinatorBase):
 
     default_interval = timedelta(minutes=1)
     error_interval = timedelta(minutes=10)
-    valid_descriptions = OHMPILOT_ENTITY_DESCRIPTIONS
+    valid_descriptions = {Platform.SENSOR: OHMPILOT_ENTITY_DESCRIPTIONS}
 
     async def _update_method(self) -> dict[SolarNetId, Any]:
         """Return data per solar net id from pyfronius."""
@@ -204,7 +213,10 @@ class FroniusPowerFlowUpdateCoordinator(FroniusCoordinatorBase):
 
     default_interval = timedelta(seconds=10)
     error_interval = timedelta(minutes=3)
-    valid_descriptions = POWER_FLOW_ENTITY_DESCRIPTIONS
+    valid_descriptions = {
+        Platform.SENSOR: POWER_FLOW_ENTITY_DESCRIPTIONS,
+        Platform.BINARY_SENSOR: POWER_FLOW_BINARY_ENTITY_DESCRIPTIONS,
+    }
 
     async def _update_method(self) -> dict[SolarNetId, Any]:
         """Return data per solar net id from pyfronius."""
@@ -217,7 +229,7 @@ class FroniusStorageUpdateCoordinator(FroniusCoordinatorBase):
 
     default_interval = timedelta(minutes=1)
     error_interval = timedelta(minutes=10)
-    valid_descriptions = STORAGE_ENTITY_DESCRIPTIONS
+    valid_descriptions = {Platform.SENSOR: STORAGE_ENTITY_DESCRIPTIONS}
 
     async def _update_method(self) -> dict[SolarNetId, Any]:
         """Return data per solar net id from pyfronius."""

homeassistant/components/fronius/entity.py

@@ -0,0 +1,60 @@
+"""Support for Fronius devices."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, Any
+
+from homeassistant.helpers.entity import Entity, EntityDescription
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+
+class FroniusEntityDescription(EntityDescription):
+    """Base class for Fronius entity descriptions."""
+
+    response_key: str | None = None
+
+
+if TYPE_CHECKING:
+    from .coordinator import FroniusCoordinatorBase
+
+
+class FroniusEntity(ABC, CoordinatorEntity["FroniusCoordinatorBase"], Entity):
+    """Defines a Fronius coordinator entity."""
+
+    entity_description: FroniusEntityDescription
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        coordinator: FroniusCoordinatorBase,
+        description: FroniusEntityDescription,
+        solar_net_id: str,
+    ) -> None:
+        """Set up an individual Fronius meter sensor."""
+        super().__init__(coordinator)
+        self.entity_description = description
+        self.response_key = description.response_key or description.key
+        self.solar_net_id = solar_net_id
+        self._set_entity_value()
+        self._attr_translation_key = description.key
+
+    def _device_data(self) -> dict[str, Any]:
+        """Extract information for SolarNet device from coordinator data."""
+        return self.coordinator.data[self.solar_net_id]
+
+    @abstractmethod
+    def _get_entity_value(self) -> Any:
+        """Extract entity value from coordinator.
+
+        Raises KeyError if not included in latest update.
+        """
+
+    @abstractmethod
+    def _set_entity_value(self) -> None:
+        """Set the entity value correctly based on the platform."""
+
+    @abstractmethod
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""