diff --git a/homeassistant/components/fritz/binary_sensor.py b/homeassistant/components/fritz/binary_sensor.py index d2edb99e02650..6d371a82c95aa 100644 --- a/homeassistant/components/fritz/binary_sensor.py +++ b/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 diff --git a/homeassistant/components/fritz/common.py b/homeassistant/components/fritz/common.py index 89a51581bf704..fa35b240d9859 100644 --- a/homeassistant/components/fritz/common.py +++ b/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]: diff --git a/homeassistant/components/fritz/sensor.py b/homeassistant/components/fritz/sensor.py index 2b156046098d2..d6b78c1cfc090 100644 --- a/homeassistant/components/fritz/sensor.py +++ b/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) diff --git a/homeassistant/components/fritz/switch.py b/homeassistant/components/fritz/switch.py index a26a0b2313fc0..c8a7952ae2b5a 100644 --- a/homeassistant/components/fritz/switch.py +++ b/homeassistant/components/fritz/switch.py @@ -4,10 +4,8 @@ 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 @@ -15,6 +13,7 @@ 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):