evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Turn AVM FRITZ!Box Tools call deflection switches into coordinator entities (#91913)

Browse source 
Co-authored-by: Franck Nijhof <frenck@frenck.nl>
This commit is contained in:
Michael 2023-04-29 18:51:38 +02:00 • committed by GitHub
parent d8bc37c695
commit ac4d9216d6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 140 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
homeassistant/components/fritz/binary_sensor.py
View file

@ -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

61
homeassistant/components/fritz/common.py
View file

@ -19,6 +19,7 @@ from fritzconnection.core.exceptions import (
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 @@ class FritzBoxTools(
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 @@ class FritzBoxTools(
)
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 @@ class FritzBoxTools(
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 @@ class FritzBoxTools(
"""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 @@ class AvmWrapper(FritzBoxTools):
"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]:

2
homeassistant/components/fritz/sensor.py
View file

@ -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)

164
homeassistant/components/fritz/switch.py
View file

@ -4,10 +4,8 @@ from __future__ import annotations
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.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_setup_entry(
)
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 @@ class FritzBoxPortSwitch(FritzBoxBaseSwitch, SwitchEntity):
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."""
@property
def data(self) -> dict[str, Any]:
"""Return call deflection data."""
return self.coordinator.data["call_deflections"].get(self.deflection_id, {})
resp = await self._avm_wrapper.async_get_ontel_deflections()
if not resp:
self._is_available = False
return
@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.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]
@property
def is_on(self) -> bool | None:
"""Switch status."""
return self.data.get("Enable") == "1"
_LOGGER.debug(
"Specific %s response: NewDeflectionList=%s",
SWITCH_TYPE_DEFLECTION,
self.dict_of_deflection,
)
self._attr_is_on = self.dict_of_deflection["Enable"] == "1"
self._is_available = True
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"]
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):