Turn AVM FRITZ!Box Tools call deflection switches into coordinator entities

homeassistant/components/fritz/binary_sensor.py

@@ -80,7 +80,10 @@ class FritzBoxBinarySensor(FritzBoxBaseCoordinatorEntity, BinarySensorEntity):
     def is_on(self) -> bool | None:
         """Return true if the binary sensor is on."""
         if isinstance(
-            state := self.coordinator.data.get(self.entity_description.key), bool
+            state := self.coordinator.data["entity_states"].get(
+                self.entity_description.key
+            ),
+            bool,
         ):
             return state
         return None

homeassistant/components/fritz/common.py

@@ -19,6 +19,7 @@
 from fritzconnection.lib.fritzhosts import FritzHosts
 from fritzconnection.lib.fritzstatus import FritzStatus
 from fritzconnection.lib.fritzwlan import DEFAULT_PASSWORD_LENGTH, FritzGuestWLAN
+import xmltodict
 
 from homeassistant.components.device_tracker import (
     CONF_CONSIDER_HOME,
@@ -137,8 +138,15 @@ class HostInfo(TypedDict):
     status: bool
 
 
+class UpdateCoordinatorDataType(TypedDict):
+    """Update coordinator data type."""
+
+    call_deflections: dict[int, dict]
+    entity_states: dict[str, StateType | bool]
+
+
 class FritzBoxTools(
-    update_coordinator.DataUpdateCoordinator[dict[str, bool | StateType]]
+    update_coordinator.DataUpdateCoordinator[UpdateCoordinatorDataType]
 ):
     """FritzBoxTools class."""
 
@@ -173,6 +181,7 @@ def __init__(
         self.password = password
         self.port = port
         self.username = username
+        self.has_call_deflections: bool = False
         self._model: str | None = None
         self._current_firmware: str | None = None
         self._latest_firmware: str | None = None
@@ -243,6 +252,8 @@ def setup(self) -> None:
             )
             self.device_is_router = self.fritz_status.has_wan_enabled
 
+        self.has_call_deflections = "X_AVM-DE_OnTel1" in self.connection.services
+
     def register_entity_updates(
         self, key: str, update_fn: Callable[[FritzStatus, StateType], Any]
     ) -> Callable[[], None]:
@@ -259,20 +270,30 @@ def unregister_entity_updates() -> None:
             self._entity_update_functions[key] = update_fn
         return unregister_entity_updates
 
-    async def _async_update_data(self) -> dict[str, bool | StateType]:
+    async def _async_update_data(self) -> UpdateCoordinatorDataType:
         """Update FritzboxTools data."""
-        enity_data: dict[str, bool | StateType] = {}
+        entity_data: UpdateCoordinatorDataType = {
+            "call_deflections": {},
+            "entity_states": {},
+        }
         try:
             await self.async_scan_devices()
             for key, update_fn in self._entity_update_functions.items():
                 _LOGGER.debug("update entity %s", key)
-                enity_data[key] = await self.hass.async_add_executor_job(
+                entity_data["entity_states"][
+                    key
+                ] = await self.hass.async_add_executor_job(
                     update_fn, self.fritz_status, self.data.get(key)
                 )
+            if self.has_call_deflections:
+                entity_data[
+                    "call_deflections"
+                ] = await self.async_update_call_deflections()
         except FRITZ_EXCEPTIONS as ex:
             raise update_coordinator.UpdateFailed(ex) from ex
-        _LOGGER.debug("enity_data: %s", enity_data)
-        return enity_data
+
+        _LOGGER.debug("enity_data: %s", entity_data)
+        return entity_data
 
     @property
     def unique_id(self) -> str:
@@ -354,6 +375,22 @@ async def _async_update_device_info(self) -> tuple[bool, str | None, str | None]
         """Retrieve latest device information from the FRITZ!Box."""
         return await self.hass.async_add_executor_job(self._update_device_info)
 
+    async def async_update_call_deflections(
+        self,
+    ) -> dict[int, dict[str, Any]]:
+        """Call GetDeflections action from X_AVM-DE_OnTel service."""
+        raw_data = await self.hass.async_add_executor_job(
+            partial(self.connection.call_action, "X_AVM-DE_OnTel1", "GetDeflections")
+        )
+        if not raw_data:
+            return {}
+
+        items = xmltodict.parse(raw_data["NewDeflectionList"])["List"]["Item"]
+        if not isinstance(items, list):
+            items = [items]
+
+        return {int(item["DeflectionId"]): item for item in items}
+
     async def _async_get_wan_access(self, ip_address: str) -> bool | None:
         """Get WAN access rule for given IP address."""
         try:
@@ -772,18 +809,6 @@ async def async_get_wlan_configuration(self, index: int) -> dict[str, Any]:
             "WLANConfiguration", str(index), "GetInfo"
         )
 
-    async def async_get_ontel_num_deflections(self) -> dict[str, Any]:
-        """Call GetNumberOfDeflections action from X_AVM-DE_OnTel service."""
-
-        return await self._async_service_call(
-            "X_AVM-DE_OnTel", "1", "GetNumberOfDeflections"
-        )
-
-    async def async_get_ontel_deflections(self) -> dict[str, Any]:
-        """Call GetDeflections action from X_AVM-DE_OnTel service."""
-
-        return await self._async_service_call("X_AVM-DE_OnTel", "1", "GetDeflections")
-
     async def async_set_wlan_configuration(
         self, index: int, turn_on: bool
     ) -> dict[str, Any]:

homeassistant/components/fritz/sensor.py

@@ -309,4 +309,4 @@ class FritzBoxSensor(FritzBoxBaseCoordinatorEntity, SensorEntity):
     @property
     def native_value(self) -> StateType:
         """Return the value reported by the sensor."""
-        return self.coordinator.data.get(self.entity_description.key)
+        return self.coordinator.data["entity_states"].get(self.entity_description.key)

homeassistant/components/fritz/switch.py

@@ -4,17 +4,16 @@
 import logging
 from typing import Any
 
-import xmltodict
-
 from homeassistant.components.network import async_get_source_ip
-from homeassistant.components.switch import SwitchEntity
+from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import EntityCategory
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
 from homeassistant.helpers.dispatcher import async_dispatcher_connect
 from homeassistant.helpers.entity import DeviceInfo, Entity
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
 from homeassistant.util import slugify
 
 from .common import (
@@ -47,31 +46,15 @@ async def _async_deflection_entities_list(
 
     _LOGGER.debug("Setting up %s switches", SWITCH_TYPE_DEFLECTION)
 
-    deflections_response = await avm_wrapper.async_get_ontel_num_deflections()
-    if not deflections_response:
+    if (
+        call_deflections := avm_wrapper.data.get("call_deflections")
+    ) is None or not isinstance(call_deflections, dict):
         _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION)
         return []
 
-    _LOGGER.debug(
-        "Specific %s response: GetNumberOfDeflections=%s",
-        SWITCH_TYPE_DEFLECTION,
-        deflections_response,
-    )
-
-    if deflections_response["NewNumberOfDeflections"] == 0:
-        _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION)
-        return []
-
-    if not (deflection_list := await avm_wrapper.async_get_ontel_deflections()):
-        return []
-
-    items = xmltodict.parse(deflection_list["NewDeflectionList"])["List"]["Item"]
-    if not isinstance(items, list):
-        items = [items]
-
     return [
-        FritzBoxDeflectionSwitch(avm_wrapper, device_friendly_name, dict_of_deflection)
-        for dict_of_deflection in items
+        FritzBoxDeflectionSwitch(avm_wrapper, device_friendly_name, cd_id)
+        for cd_id in call_deflections
     ]
 
 
@@ -273,6 +256,61 @@ async def async_update_avm_device() -> None:
     )
 
 
+class FritzBoxBaseCoordinatorSwitch(CoordinatorEntity, SwitchEntity):
+    """Fritz switch coordinator base class."""
+
+    coordinator: AvmWrapper
+    entity_description: SwitchEntityDescription
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        avm_wrapper: AvmWrapper,
+        device_name: str,
+        description: SwitchEntityDescription,
+    ) -> None:
+        """Init device info class."""
+        super().__init__(avm_wrapper)
+        self.entity_description = description
+        self._device_name = device_name
+        self._attr_unique_id = f"{avm_wrapper.unique_id}-{description.key}"
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        """Return the device information."""
+        return DeviceInfo(
+            configuration_url=f"http://{self.coordinator.host}",
+            connections={(CONNECTION_NETWORK_MAC, self.coordinator.mac)},
+            identifiers={(DOMAIN, self.coordinator.unique_id)},
+            manufacturer="AVM",
+            model=self.coordinator.model,
+            name=self._device_name,
+            sw_version=self.coordinator.current_firmware,
+        )
+
+    @property
+    def data(self) -> dict[str, Any]:
+        """Return entity data from coordinator data."""
+        raise NotImplementedError()
+
+    @property
+    def available(self) -> bool:
+        """Return availability based on data availability."""
+        return super.available and bool(self.data)
+
+    async def _async_handle_turn_on_off(self, turn_on: bool) -> None:
+        """Handle switch state change request."""
+        raise NotImplementedError()
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Turn on switch."""
+        await self._async_handle_turn_on_off(turn_on=True)
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn off switch."""
+        await self._async_handle_turn_on_off(turn_on=False)
+
+
 class FritzBoxBaseSwitch(FritzBoxBaseEntity):
     """Fritz switch base class."""
 
@@ -417,69 +455,51 @@ async def _async_switch_on_off_executor(self, turn_on: bool) -> bool:
         return bool(resp is not None)
 
 
-class FritzBoxDeflectionSwitch(FritzBoxBaseSwitch, SwitchEntity):
+class FritzBoxDeflectionSwitch(FritzBoxBaseCoordinatorSwitch):
     """Defines a FRITZ!Box Tools PortForward switch."""
 
+    _attr_entity_category = EntityCategory.CONFIG
+
     def __init__(
         self,
         avm_wrapper: AvmWrapper,
         device_friendly_name: str,
-        dict_of_deflection: Any,
+        deflection_id: int,
     ) -> None:
         """Init Fritxbox Deflection class."""
-        self._avm_wrapper = avm_wrapper
-
-        self.dict_of_deflection = dict_of_deflection
-        self._attributes = {}
-        self.id = int(self.dict_of_deflection["DeflectionId"])
-        self._attr_entity_category = EntityCategory.CONFIG
-
-        switch_info = SwitchInfo(
-            description=f"Call deflection {self.id}",
-            friendly_name=device_friendly_name,
+        self.deflection_id = deflection_id
+        description = SwitchEntityDescription(
+            key=f"call_deflection_{self.deflection_id}",
+            name=f"Call deflection {self.deflection_id}",
             icon="mdi:phone-forward",
-            type=SWITCH_TYPE_DEFLECTION,
-            callback_update=self._async_fetch_update,
-            callback_switch=self._async_switch_on_off_executor,
         )
-        super().__init__(self._avm_wrapper, device_friendly_name, switch_info)
+        super().__init__(avm_wrapper, device_friendly_name, description)
 
-    async def _async_fetch_update(self) -> None:
-        """Fetch updates."""
-
-        resp = await self._avm_wrapper.async_get_ontel_deflections()
-        if not resp:
-            self._is_available = False
-            return
-
-        self.dict_of_deflection = xmltodict.parse(resp["NewDeflectionList"])["List"][
-            "Item"
-        ]
-        if isinstance(self.dict_of_deflection, list):
-            self.dict_of_deflection = self.dict_of_deflection[self.id]
-
-        _LOGGER.debug(
-            "Specific %s response: NewDeflectionList=%s",
-            SWITCH_TYPE_DEFLECTION,
-            self.dict_of_deflection,
-        )
+    @property
+    def data(self) -> dict[str, Any]:
+        """Return call deflection data."""
+        return self.coordinator.data["call_deflections"].get(self.deflection_id, {})
 
-        self._attr_is_on = self.dict_of_deflection["Enable"] == "1"
-        self._is_available = True
+    @property
+    def extra_state_attributes(self) -> dict[str, str]:
+        """Return device attributes."""
+        return {
+            "type": self.data["Type"],
+            "number": self.data["Number"],
+            "deflection_to_number": self.data["DeflectionToNumber"],
+            "mode": self.data["Mode"][1:],
+            "outgoing": self.data["Outgoing"],
+            "phonebook_id": self.data["PhonebookID"],
+        }
 
-        self._attributes["type"] = self.dict_of_deflection["Type"]
-        self._attributes["number"] = self.dict_of_deflection["Number"]
-        self._attributes["deflection_to_number"] = self.dict_of_deflection[
-            "DeflectionToNumber"
-        ]
-        # Return mode sample: "eImmediately"
-        self._attributes["mode"] = self.dict_of_deflection["Mode"][1:]
-        self._attributes["outgoing"] = self.dict_of_deflection["Outgoing"]
-        self._attributes["phonebook_id"] = self.dict_of_deflection["PhonebookID"]
+    @property
+    def is_on(self) -> bool | None:
+        """Switch status."""
+        return self.data.get("Enable") == "1"
 
-    async def _async_switch_on_off_executor(self, turn_on: bool) -> None:
+    async def _async_handle_turn_on_off(self, turn_on: bool) -> None:
         """Handle deflection switch."""
-        await self._avm_wrapper.async_set_deflection_enable(self.id, turn_on)
+        await self.coordinator.async_set_deflection_enable(self.deflection_id, turn_on)
 
 
 class FritzBoxProfileSwitch(FritzDeviceBase, SwitchEntity):