From 09297520c0588c195bc191b7033375383942e566 Mon Sep 17 00:00:00 2001 From: Michael <35783820+mib1185@users.noreply.github.com> Date: Thu, 20 Jan 2022 12:43:32 +0100 Subject: [PATCH] Let the new wrapper just extend the `FritzBoxTools` class (#64133) * let wrapper just extend the FritzBoxTools class * keep avm_device in method names --- .coveragerc | 1 - homeassistant/components/fritz/__init__.py | 14 +- .../components/fritz/binary_sensor.py | 24 ++-- homeassistant/components/fritz/button.py | 24 ++-- homeassistant/components/fritz/common.py | 98 +++++++++++-- homeassistant/components/fritz/config_flow.py | 18 +-- .../components/fritz/device_tracker.py | 28 ++-- homeassistant/components/fritz/sensor.py | 18 ++- homeassistant/components/fritz/services.py | 6 +- homeassistant/components/fritz/switch.py | 132 +++++++++--------- homeassistant/components/fritz/wrapper.py | 98 ------------- 11 files changed, 217 insertions(+), 244 deletions(-) delete mode 100644 homeassistant/components/fritz/wrapper.py diff --git a/.coveragerc b/.coveragerc index ccb0135b4a3..2d2bede3b25 100644 --- a/.coveragerc +++ b/.coveragerc @@ -378,7 +378,6 @@ omit = homeassistant/components/fritz/sensor.py homeassistant/components/fritz/services.py homeassistant/components/fritz/switch.py - homeassistant/components/fritz/wrapper.py homeassistant/components/fritzbox_callmonitor/__init__.py homeassistant/components/fritzbox_callmonitor/const.py homeassistant/components/fritzbox_callmonitor/base.py diff --git a/homeassistant/components/fritz/__init__.py b/homeassistant/components/fritz/__init__.py index e2220103555..0db85b12077 100644 --- a/homeassistant/components/fritz/__init__.py +++ b/homeassistant/components/fritz/__init__.py @@ -12,7 +12,7 @@ from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNA from homeassistant.core import HomeAssistant from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady -from .common import FritzBoxTools, FritzData +from .common import AvmWrapper, FritzData from .const import DATA_FRITZ, DOMAIN, PLATFORMS from .services import async_setup_services, async_unload_services @@ -22,7 +22,7 @@ _LOGGER = logging.getLogger(__name__) async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Set up fritzboxtools from config entry.""" _LOGGER.debug("Setting up FRITZ!Box Tools component") - avm_device = FritzBoxTools( + avm_wrapper = AvmWrapper( hass=hass, host=entry.data[CONF_HOST], port=entry.data[CONF_PORT], @@ -31,21 +31,21 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: ) try: - await avm_device.async_setup(entry.options) + await avm_wrapper.async_setup(entry.options) except FritzSecurityError as ex: raise ConfigEntryAuthFailed from ex except (FritzConnectionException, FritzResourceError) as ex: raise ConfigEntryNotReady from ex hass.data.setdefault(DOMAIN, {}) - hass.data[DOMAIN][entry.entry_id] = avm_device + hass.data[DOMAIN][entry.entry_id] = avm_wrapper if DATA_FRITZ not in hass.data: hass.data[DATA_FRITZ] = FritzData() entry.async_on_unload(entry.add_update_listener(update_listener)) - await avm_device.async_config_entry_first_refresh() + await avm_wrapper.async_config_entry_first_refresh() # Load the other platforms like switch hass.config_entries.async_setup_platforms(entry, PLATFORMS) @@ -57,10 +57,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Unload FRITZ!Box Tools config entry.""" - avm_device: FritzBoxTools = hass.data[DOMAIN][entry.entry_id] + avm_wrapper: AvmWrapper = hass.data[DOMAIN][entry.entry_id] fritz_data = hass.data[DATA_FRITZ] - fritz_data.tracked.pop(avm_device.unique_id) + fritz_data.tracked.pop(avm_wrapper.unique_id) if not bool(fritz_data.tracked): hass.data.pop(DATA_FRITZ) diff --git a/homeassistant/components/fritz/binary_sensor.py b/homeassistant/components/fritz/binary_sensor.py index 51489376f7a..b416e0cfb11 100644 --- a/homeassistant/components/fritz/binary_sensor.py +++ b/homeassistant/components/fritz/binary_sensor.py @@ -14,7 +14,7 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers.entity import EntityCategory from homeassistant.helpers.entity_platform import AddEntitiesCallback -from .common import FritzBoxBaseEntity, FritzBoxTools +from .common import AvmWrapper, FritzBoxBaseEntity from .const import DOMAIN, MeshRoles _LOGGER = logging.getLogger(__name__) @@ -55,12 +55,12 @@ async def async_setup_entry( ) -> None: """Set up entry.""" _LOGGER.debug("Setting up FRITZ!Box binary sensors") - avm_device: FritzBoxTools = hass.data[DOMAIN][entry.entry_id] + avm_wrapper: AvmWrapper = hass.data[DOMAIN][entry.entry_id] entities = [ - FritzBoxBinarySensor(avm_device, entry.title, description) + FritzBoxBinarySensor(avm_wrapper, entry.title, description) for description in SENSOR_TYPES - if (description.exclude_mesh_role != avm_device.mesh_role) + if (description.exclude_mesh_role != avm_wrapper.mesh_role) ] async_add_entities(entities, True) @@ -71,27 +71,27 @@ class FritzBoxBinarySensor(FritzBoxBaseEntity, BinarySensorEntity): def __init__( self, - avm_device: FritzBoxTools, + avm_wrapper: AvmWrapper, device_friendly_name: str, description: BinarySensorEntityDescription, ) -> None: """Init FRITZ!Box connectivity class.""" self.entity_description = description self._attr_name = f"{device_friendly_name} {description.name}" - self._attr_unique_id = f"{avm_device.unique_id}-{description.key}" - super().__init__(avm_device, device_friendly_name) + self._attr_unique_id = f"{avm_wrapper.unique_id}-{description.key}" + super().__init__(avm_wrapper, device_friendly_name) def update(self) -> None: """Update data.""" _LOGGER.debug("Updating FRITZ!Box binary sensors") if self.entity_description.key == "firmware_update": - self._attr_is_on = self._avm_device.update_available + self._attr_is_on = self._avm_wrapper.update_available self._attr_extra_state_attributes = { - "installed_version": self._avm_device.current_firmware, - "latest_available_version": self._avm_device.latest_firmware, + "installed_version": self._avm_wrapper.current_firmware, + "latest_available_version": self._avm_wrapper.latest_firmware, } if self.entity_description.key == "is_connected": - self._attr_is_on = bool(self._avm_device.fritz_status.is_connected) + self._attr_is_on = bool(self._avm_wrapper.fritz_status.is_connected) elif self.entity_description.key == "is_linked": - self._attr_is_on = bool(self._avm_device.fritz_status.is_linked) + self._attr_is_on = bool(self._avm_wrapper.fritz_status.is_linked) diff --git a/homeassistant/components/fritz/button.py b/homeassistant/components/fritz/button.py index 3ca54de9577..d17d1f4d9ef 100644 --- a/homeassistant/components/fritz/button.py +++ b/homeassistant/components/fritz/button.py @@ -18,7 +18,7 @@ from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity_platform import AddEntitiesCallback -from .common import FritzBoxTools +from .common import AvmWrapper from .const import DOMAIN _LOGGER = logging.getLogger(__name__) @@ -42,28 +42,28 @@ BUTTONS: Final = [ name="Firmware Update", device_class=ButtonDeviceClass.UPDATE, entity_category=ENTITY_CATEGORY_CONFIG, - press_action=lambda avm_device: avm_device.async_trigger_firmware_update(), + press_action=lambda avm_wrapper: avm_wrapper.async_trigger_firmware_update(), ), FritzButtonDescription( key="reboot", name="Reboot", device_class=ButtonDeviceClass.RESTART, entity_category=ENTITY_CATEGORY_CONFIG, - press_action=lambda avm_device: avm_device.async_trigger_reboot(), + press_action=lambda avm_wrapper: avm_wrapper.async_trigger_reboot(), ), FritzButtonDescription( key="reconnect", name="Reconnect", device_class=ButtonDeviceClass.RESTART, entity_category=ENTITY_CATEGORY_CONFIG, - press_action=lambda avm_device: avm_device.async_trigger_reconnect(), + press_action=lambda avm_wrapper: avm_wrapper.async_trigger_reconnect(), ), FritzButtonDescription( key="cleanup", name="Cleanup", icon="mdi:broom", entity_category=ENTITY_CATEGORY_CONFIG, - press_action=lambda avm_device: avm_device.async_trigger_cleanup(), + press_action=lambda avm_wrapper: avm_wrapper.async_trigger_cleanup(), ), ] @@ -75,10 +75,10 @@ async def async_setup_entry( ) -> None: """Set buttons for device.""" _LOGGER.debug("Setting up buttons") - avm_device: FritzBoxTools = hass.data[DOMAIN][entry.entry_id] + avm_wrapper: AvmWrapper = hass.data[DOMAIN][entry.entry_id] async_add_entities( - [FritzButton(avm_device, entry.title, button) for button in BUTTONS] + [FritzButton(avm_wrapper, entry.title, button) for button in BUTTONS] ) @@ -89,21 +89,21 @@ class FritzButton(ButtonEntity): def __init__( self, - avm_device: FritzBoxTools, + avm_wrapper: AvmWrapper, device_friendly_name: str, description: FritzButtonDescription, ) -> None: """Initialize Fritz!Box button.""" self.entity_description = description - self.avm_device = avm_device + self.avm_wrapper = avm_wrapper self._attr_name = f"{device_friendly_name} {description.name}" - self._attr_unique_id = f"{self.avm_device.unique_id}-{description.key}" + self._attr_unique_id = f"{self.avm_wrapper.unique_id}-{description.key}" self._attr_device_info = DeviceInfo( - connections={(CONNECTION_NETWORK_MAC, avm_device.mac)} + connections={(CONNECTION_NETWORK_MAC, avm_wrapper.mac)} ) async def async_press(self) -> None: """Triggers Fritz!Box service.""" - await self.entity_description.press_action(self.avm_device) + await self.entity_description.press_action(self.avm_wrapper) diff --git a/homeassistant/components/fritz/common.py b/homeassistant/components/fritz/common.py index f6b6fed96c4..ae9845ad250 100644 --- a/homeassistant/components/fritz/common.py +++ b/homeassistant/components/fritz/common.py @@ -4,6 +4,7 @@ from __future__ import annotations from collections.abc import Callable, ValuesView from dataclasses import dataclass, field from datetime import datetime, timedelta +from functools import partial import logging from types import MappingProxyType from typing import Any, TypedDict, cast @@ -11,7 +12,9 @@ from typing import Any, TypedDict, cast from fritzconnection import FritzConnection from fritzconnection.core.exceptions import ( FritzActionError, + FritzActionFailedError, FritzConnectionException, + FritzLookUpError, FritzSecurityError, FritzServiceError, ) @@ -490,6 +493,77 @@ class FritzBoxTools(update_coordinator.DataUpdateCoordinator): raise HomeAssistantError("Service not supported") from ex +class AvmWrapper(FritzBoxTools): + """Setup AVM wrapper for API calls.""" + + def _service_call_action( + self, + service_name: str, + service_suffix: str, + action_name: str, + **kwargs: Any, + ) -> dict | None: + """Return service details.""" + + if f"{service_name}{service_suffix}" not in self.connection.services: + return None + + try: + result: dict = self.connection.call_action( + f"{service_name}:{service_suffix}", + action_name, + **kwargs, + ) + return result + except FritzSecurityError: + _LOGGER.error( + "Authorization Error: Please check the provided credentials and verify that you can log into the web interface", + exc_info=True, + ) + except ( + FritzActionError, + FritzActionFailedError, + FritzServiceError, + FritzLookUpError, + ): + _LOGGER.error( + "Service/Action Error: cannot execute service %s with action %s", + service_name, + action_name, + exc_info=True, + ) + except FritzConnectionException: + _LOGGER.error( + "Connection Error: Please check the device is properly configured for remote login", + exc_info=True, + ) + return None + + async def _async_service_call_action( + self, service_name: str, service_suffix: str, action_name: str, **kwargs: Any + ) -> dict[str, Any] | None: + """Make call_action async.""" + + return await self.hass.async_add_executor_job( + partial( + self._service_call_action, + service_name, + service_suffix, + action_name, + **kwargs, + ) + ) + + async def get_wan_dsl_interface_config(self) -> dict[str, Any] | None: + """Call WANDSLInterfaceConfig service.""" + + return await self._async_service_call_action( + "WANDSLInterfaceConfig", + "1", + "GetInfo", + ) + + @dataclass class FritzData: """Storage class for platform global data.""" @@ -501,10 +575,10 @@ class FritzData: class FritzDeviceBase(update_coordinator.CoordinatorEntity): """Entity base class for a device connected to a FRITZ!Box device.""" - def __init__(self, avm_device: FritzBoxTools, device: FritzDevice) -> None: + def __init__(self, avm_wrapper: AvmWrapper, device: FritzDevice) -> None: """Initialize a FRITZ!Box device.""" - super().__init__(avm_device) - self._avm_device = avm_device + super().__init__(avm_wrapper) + self._avm_wrapper = avm_wrapper self._mac: str = device.mac_address self._name: str = device.hostname or DEFAULT_DEVICE_NAME @@ -517,7 +591,7 @@ class FritzDeviceBase(update_coordinator.CoordinatorEntity): def ip_address(self) -> str | None: """Return the primary ip address of the device.""" if self._mac: - return self._avm_device.devices[self._mac].ip_address + return self._avm_wrapper.devices[self._mac].ip_address return None @property @@ -529,7 +603,7 @@ class FritzDeviceBase(update_coordinator.CoordinatorEntity): def hostname(self) -> str | None: """Return hostname of the device.""" if self._mac: - return self._avm_device.devices[self._mac].hostname + return self._avm_wrapper.devices[self._mac].hostname return None @property @@ -647,25 +721,25 @@ class SwitchInfo(TypedDict): class FritzBoxBaseEntity: """Fritz host entity base class.""" - def __init__(self, avm_device: FritzBoxTools, device_name: str) -> None: + def __init__(self, avm_wrapper: AvmWrapper, device_name: str) -> None: """Init device info class.""" - self._avm_device = avm_device + self._avm_wrapper = avm_wrapper self._device_name = device_name @property def mac_address(self) -> str: """Return the mac address of the main device.""" - return self._avm_device.mac + return self._avm_wrapper.mac @property def device_info(self) -> DeviceInfo: """Return the device information.""" return DeviceInfo( - configuration_url=f"http://{self._avm_device.host}", + configuration_url=f"http://{self._avm_wrapper.host}", connections={(dr.CONNECTION_NETWORK_MAC, self.mac_address)}, - identifiers={(DOMAIN, self._avm_device.unique_id)}, + identifiers={(DOMAIN, self._avm_wrapper.unique_id)}, manufacturer="AVM", - model=self._avm_device.model, + model=self._avm_wrapper.model, name=self._device_name, - sw_version=self._avm_device.current_firmware, + sw_version=self._avm_wrapper.current_firmware, ) diff --git a/homeassistant/components/fritz/config_flow.py b/homeassistant/components/fritz/config_flow.py index 6a5d563fdef..180a6239d9f 100644 --- a/homeassistant/components/fritz/config_flow.py +++ b/homeassistant/components/fritz/config_flow.py @@ -19,7 +19,7 @@ from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNA from homeassistant.core import callback from homeassistant.data_entry_flow import FlowResult -from .common import FritzBoxTools +from .common import AvmWrapper from .const import ( DEFAULT_HOST, DEFAULT_PORT, @@ -51,7 +51,7 @@ class FritzBoxToolsFlowHandler(ConfigFlow, domain=DOMAIN): self._password: str self._port: int | None = None self._username: str - self.avm_device: FritzBoxTools + self.avm_wrapper: AvmWrapper async def fritz_tools_init(self) -> str | None: """Initialize FRITZ!Box Tools class.""" @@ -59,7 +59,7 @@ class FritzBoxToolsFlowHandler(ConfigFlow, domain=DOMAIN): if not self._host or not self._port: return None - self.avm_device = FritzBoxTools( + self.avm_wrapper = AvmWrapper( hass=self.hass, host=self._host, port=self._port, @@ -68,7 +68,7 @@ class FritzBoxToolsFlowHandler(ConfigFlow, domain=DOMAIN): ) try: - await self.avm_device.async_setup() + await self.avm_wrapper.async_setup() except FritzSecurityError: return ERROR_AUTH_INVALID except FritzConnectionException: @@ -100,10 +100,10 @@ class FritzBoxToolsFlowHandler(ConfigFlow, domain=DOMAIN): return self.async_create_entry( title=self._name, data={ - CONF_HOST: self.avm_device.host, - CONF_PASSWORD: self.avm_device.password, - CONF_PORT: self.avm_device.port, - CONF_USERNAME: self.avm_device.username, + CONF_HOST: self.avm_wrapper.host, + CONF_PASSWORD: self.avm_wrapper.password, + CONF_PORT: self.avm_wrapper.port, + CONF_USERNAME: self.avm_wrapper.username, }, options={ CONF_CONSIDER_HOME: DEFAULT_CONSIDER_HOME.total_seconds(), @@ -204,7 +204,7 @@ class FritzBoxToolsFlowHandler(ConfigFlow, domain=DOMAIN): self._password = user_input[CONF_PASSWORD] if not (error := await self.fritz_tools_init()): - self._name = self.avm_device.model + self._name = self.avm_wrapper.model if await self.async_check_configured_entry(): error = "already_configured" diff --git a/homeassistant/components/fritz/device_tracker.py b/homeassistant/components/fritz/device_tracker.py index 9d444151391..34c3f64e1e5 100644 --- a/homeassistant/components/fritz/device_tracker.py +++ b/homeassistant/components/fritz/device_tracker.py @@ -12,7 +12,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity_platform import AddEntitiesCallback from .common import ( - FritzBoxTools, + AvmWrapper, FritzData, FritzDevice, FritzDeviceBase, @@ -31,16 +31,16 @@ async def async_setup_entry( ) -> None: """Set up device tracker for FRITZ!Box component.""" _LOGGER.debug("Starting FRITZ!Box device tracker") - avm_device: FritzBoxTools = hass.data[DOMAIN][entry.entry_id] + avm_wrapper: AvmWrapper = hass.data[DOMAIN][entry.entry_id] data_fritz: FritzData = hass.data[DATA_FRITZ] @callback def update_avm_device() -> None: """Update the values of AVM device.""" - _async_add_entities(avm_device, async_add_entities, data_fritz) + _async_add_entities(avm_wrapper, async_add_entities, data_fritz) entry.async_on_unload( - async_dispatcher_connect(hass, avm_device.signal_device_new, update_avm_device) + async_dispatcher_connect(hass, avm_wrapper.signal_device_new, update_avm_device) ) update_avm_device() @@ -48,22 +48,22 @@ async def async_setup_entry( @callback def _async_add_entities( - avm_device: FritzBoxTools, + avm_wrapper: AvmWrapper, async_add_entities: AddEntitiesCallback, data_fritz: FritzData, ) -> None: """Add new tracker entities from the AVM device.""" new_tracked = [] - if avm_device.unique_id not in data_fritz.tracked: - data_fritz.tracked[avm_device.unique_id] = set() + if avm_wrapper.unique_id not in data_fritz.tracked: + data_fritz.tracked[avm_wrapper.unique_id] = set() - for mac, device in avm_device.devices.items(): + for mac, device in avm_wrapper.devices.items(): if device_filter_out_from_trackers(mac, device, data_fritz.tracked.values()): continue - new_tracked.append(FritzBoxTracker(avm_device, device)) - data_fritz.tracked[avm_device.unique_id].add(mac) + new_tracked.append(FritzBoxTracker(avm_wrapper, device)) + data_fritz.tracked[avm_wrapper.unique_id].add(mac) if new_tracked: async_add_entities(new_tracked) @@ -72,15 +72,15 @@ def _async_add_entities( class FritzBoxTracker(FritzDeviceBase, ScannerEntity): """This class queries a FRITZ!Box device.""" - def __init__(self, avm_device: FritzBoxTools, device: FritzDevice) -> None: + def __init__(self, avm_wrapper: AvmWrapper, device: FritzDevice) -> None: """Initialize a FRITZ!Box device.""" - super().__init__(avm_device, device) + super().__init__(avm_wrapper, device) self._last_activity: datetime.datetime | None = device.last_activity @property def is_connected(self) -> bool: """Return device status.""" - return self._avm_device.devices[self._mac].is_connected + return self._avm_wrapper.devices[self._mac].is_connected @property def unique_id(self) -> str: @@ -103,7 +103,7 @@ class FritzBoxTracker(FritzDeviceBase, ScannerEntity): def extra_state_attributes(self) -> dict[str, str]: """Return the attributes.""" attrs: dict[str, str] = {} - device = self._avm_device.devices[self._mac] + device = self._avm_wrapper.devices[self._mac] self._last_activity = device.last_activity if self._last_activity is not None: attrs["last_time_reachable"] = self._last_activity.isoformat( diff --git a/homeassistant/components/fritz/sensor.py b/homeassistant/components/fritz/sensor.py index d264d4b6603..f3ebaf32ff9 100644 --- a/homeassistant/components/fritz/sensor.py +++ b/homeassistant/components/fritz/sensor.py @@ -28,9 +28,8 @@ from homeassistant.helpers.entity import EntityCategory from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.util.dt import utcnow -from .common import FritzBoxBaseEntity, FritzBoxTools +from .common import AvmWrapper, FritzBoxBaseEntity from .const import DOMAIN, DSL_CONNECTION, UPTIME_DEVIATION, MeshRoles -from .wrapper import AvmWrapper _LOGGER = logging.getLogger(__name__) @@ -276,8 +275,7 @@ async def async_setup_entry( ) -> None: """Set up entry.""" _LOGGER.debug("Setting up FRITZ!Box sensors") - avm_device: FritzBoxTools = hass.data[DOMAIN][entry.entry_id] - avm_wrapper = AvmWrapper(avm_device) + avm_wrapper: AvmWrapper = hass.data[DOMAIN][entry.entry_id] dsl: bool = False dslinterface = await avm_wrapper.get_wan_dsl_interface_config() @@ -285,10 +283,10 @@ async def async_setup_entry( dsl = dslinterface["NewEnable"] entities = [ - FritzBoxSensor(avm_device, entry.title, description) + FritzBoxSensor(avm_wrapper, entry.title, description) for description in SENSOR_TYPES if (dsl or description.connection_type != DSL_CONNECTION) - and description.exclude_mesh_role != avm_device.mesh_role + and description.exclude_mesh_role != avm_wrapper.mesh_role ] async_add_entities(entities, True) @@ -301,7 +299,7 @@ class FritzBoxSensor(FritzBoxBaseEntity, SensorEntity): def __init__( self, - avm_device: FritzBoxTools, + avm_wrapper: AvmWrapper, device_friendly_name: str, description: FritzSensorEntityDescription, ) -> None: @@ -310,15 +308,15 @@ class FritzBoxSensor(FritzBoxBaseEntity, SensorEntity): self._last_device_value: str | None = None self._attr_available = True self._attr_name = f"{device_friendly_name} {description.name}" - self._attr_unique_id = f"{avm_device.unique_id}-{description.key}" - super().__init__(avm_device, device_friendly_name) + self._attr_unique_id = f"{avm_wrapper.unique_id}-{description.key}" + super().__init__(avm_wrapper, device_friendly_name) def update(self) -> None: """Update data.""" _LOGGER.debug("Updating FRITZ!Box sensors") try: - status: FritzStatus = self._avm_device.fritz_status + status: FritzStatus = self._avm_wrapper.fritz_status self._attr_available = True except FritzConnectionException: _LOGGER.error("Error getting the state from the FRITZ!Box", exc_info=True) diff --git a/homeassistant/components/fritz/services.py b/homeassistant/components/fritz/services.py index 7fa458354c4..e32f4c7ffd7 100644 --- a/homeassistant/components/fritz/services.py +++ b/homeassistant/components/fritz/services.py @@ -6,7 +6,7 @@ from homeassistant.core import HomeAssistant, ServiceCall from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.service import async_extract_config_entry_ids -from .common import FritzBoxTools +from .common import AvmWrapper from .const import ( DOMAIN, FRITZ_SERVICES, @@ -42,9 +42,9 @@ async def async_setup_services(hass: HomeAssistant) -> None: for entry_id in fritzbox_entry_ids: _LOGGER.debug("Executing service %s", service_call.service) - avm_device: FritzBoxTools = hass.data[DOMAIN][entry_id] + avm_wrapper: AvmWrapper = hass.data[DOMAIN][entry_id] if config_entry := hass.config_entries.async_get_entry(entry_id): - await avm_device.service_fritzbox(service_call, config_entry) + await avm_wrapper.service_fritzbox(service_call, config_entry) else: _LOGGER.error( "Executing service %s failed, no config entry found", diff --git a/homeassistant/components/fritz/switch.py b/homeassistant/components/fritz/switch.py index a775f5d8e32..0b7a33692e6 100644 --- a/homeassistant/components/fritz/switch.py +++ b/homeassistant/components/fritz/switch.py @@ -27,8 +27,8 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.util import slugify from .common import ( + AvmWrapper, FritzBoxBaseEntity, - FritzBoxTools, FritzData, FritzDevice, FritzDeviceBase, @@ -48,17 +48,17 @@ _LOGGER = logging.getLogger(__name__) async def async_service_call_action( - avm_device: FritzBoxTools, + avm_wrapper: AvmWrapper, service_name: str, service_suffix: str | None, action_name: str, **kwargs: Any, ) -> None | dict: """Return service details.""" - return await avm_device.hass.async_add_executor_job( + return await avm_wrapper.hass.async_add_executor_job( partial( service_call_action, - avm_device, + avm_wrapper, service_name, service_suffix, action_name, @@ -68,7 +68,7 @@ async def async_service_call_action( def service_call_action( - avm_device: FritzBoxTools, + avm_wrapper: AvmWrapper, service_name: str, service_suffix: str | None, action_name: str, @@ -76,11 +76,11 @@ def service_call_action( ) -> dict | None: """Return service details.""" - if f"{service_name}{service_suffix}" not in avm_device.connection.services: + if f"{service_name}{service_suffix}" not in avm_wrapper.connection.services: return None try: - return avm_device.connection.call_action( # type: ignore[no-any-return] + return avm_wrapper.connection.call_action( # type: ignore[no-any-return] f"{service_name}:{service_suffix}", action_name, **kwargs, @@ -113,12 +113,12 @@ def service_call_action( def get_deflections( - avm_device: FritzBoxTools, service_name: str + avm_wrapper: AvmWrapper, service_name: str ) -> list[OrderedDict[Any, Any]] | None: """Get deflection switch info.""" deflection_list = service_call_action( - avm_device, + avm_wrapper, service_name, "1", "GetDeflections", @@ -134,7 +134,7 @@ def get_deflections( def deflection_entities_list( - avm_device: FritzBoxTools, device_friendly_name: str + avm_wrapper: AvmWrapper, device_friendly_name: str ) -> list[FritzBoxDeflectionSwitch]: """Get list of deflection entities.""" @@ -142,7 +142,7 @@ def deflection_entities_list( service_name = "X_AVM-DE_OnTel" deflections_response = service_call_action( - avm_device, service_name, "1", "GetNumberOfDeflections" + avm_wrapper, service_name, "1", "GetNumberOfDeflections" ) if not deflections_response: _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION) @@ -158,18 +158,18 @@ def deflection_entities_list( _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION) return [] - deflection_list = get_deflections(avm_device, service_name) + deflection_list = get_deflections(avm_wrapper, service_name) if deflection_list is None: return [] return [ - FritzBoxDeflectionSwitch(avm_device, device_friendly_name, dict_of_deflection) + FritzBoxDeflectionSwitch(avm_wrapper, device_friendly_name, dict_of_deflection) for dict_of_deflection in deflection_list ] def port_entities_list( - avm_device: FritzBoxTools, device_friendly_name: str, local_ip: str + avm_wrapper: AvmWrapper, device_friendly_name: str, local_ip: str ) -> list[FritzBoxPortSwitch]: """Get list of port forwarding entities.""" @@ -177,7 +177,7 @@ def port_entities_list( entities_list: list[FritzBoxPortSwitch] = [] service_name = "Layer3Forwarding" connection_type = service_call_action( - avm_device, service_name, "1", "GetDefaultConnectionService" + avm_wrapper, service_name, "1", "GetDefaultConnectionService" ) if not connection_type: _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_PORTFORWARD) @@ -188,7 +188,7 @@ def port_entities_list( # Query port forwardings and setup a switch for each forward for the current device resp = service_call_action( - avm_device, con_type, "1", "GetPortMappingNumberOfEntries" + avm_wrapper, con_type, "1", "GetPortMappingNumberOfEntries" ) if not resp: _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION) @@ -202,12 +202,12 @@ def port_entities_list( port_forwards_count, ) - _LOGGER.debug("IP source for %s is %s", avm_device.host, local_ip) + _LOGGER.debug("IP source for %s is %s", avm_wrapper.host, local_ip) for i in range(port_forwards_count): portmap = service_call_action( - avm_device, + avm_wrapper, con_type, "1", "GetGenericPortMappingEntry", @@ -234,7 +234,7 @@ def port_entities_list( port_name = f"{port_name} {portmap['NewExternalPort']}" entities_list.append( FritzBoxPortSwitch( - avm_device, + avm_wrapper, device_friendly_name, portmap, port_name, @@ -247,21 +247,21 @@ def port_entities_list( def wifi_entities_list( - avm_device: FritzBoxTools, device_friendly_name: str + avm_wrapper: AvmWrapper, device_friendly_name: str ) -> list[FritzBoxWifiSwitch]: """Get list of wifi entities.""" _LOGGER.debug("Setting up %s switches", SWITCH_TYPE_WIFINETWORK) std_table = {"ax": "Wifi6", "ac": "5Ghz", "n": "2.4Ghz"} - if avm_device.model == "FRITZ!Box 7390": + if avm_wrapper.model == "FRITZ!Box 7390": std_table = {"n": "5Ghz"} networks: dict = {} for i in range(4): - if not ("WLANConfiguration" + str(i)) in avm_device.connection.services: + if not ("WLANConfiguration" + str(i)) in avm_wrapper.connection.services: continue network_info = service_call_action( - avm_device, "WLANConfiguration", str(i), "GetInfo" + avm_wrapper, "WLANConfiguration", str(i), "GetInfo" ) if network_info: ssid = network_info["NewSSID"] @@ -279,53 +279,53 @@ def wifi_entities_list( _LOGGER.debug("SSID normalized: <%s>", networks[i]) return [ - FritzBoxWifiSwitch(avm_device, device_friendly_name, net, network_name) + FritzBoxWifiSwitch(avm_wrapper, device_friendly_name, net, network_name) for net, network_name in networks.items() ] def profile_entities_list( - avm_device: FritzBoxTools, + avm_wrapper: AvmWrapper, data_fritz: FritzData, ) -> list[FritzBoxProfileSwitch]: """Add new tracker entities from the AVM device.""" new_profiles: list[FritzBoxProfileSwitch] = [] - if "X_AVM-DE_HostFilter1" not in avm_device.connection.services: + if "X_AVM-DE_HostFilter1" not in avm_wrapper.connection.services: return new_profiles - if avm_device.unique_id not in data_fritz.profile_switches: - data_fritz.profile_switches[avm_device.unique_id] = set() + if avm_wrapper.unique_id not in data_fritz.profile_switches: + data_fritz.profile_switches[avm_wrapper.unique_id] = set() - for mac, device in avm_device.devices.items(): + for mac, device in avm_wrapper.devices.items(): if device_filter_out_from_trackers( mac, device, data_fritz.profile_switches.values() ): continue - new_profiles.append(FritzBoxProfileSwitch(avm_device, device)) - data_fritz.profile_switches[avm_device.unique_id].add(mac) + new_profiles.append(FritzBoxProfileSwitch(avm_wrapper, device)) + data_fritz.profile_switches[avm_wrapper.unique_id].add(mac) return new_profiles def all_entities_list( - avm_device: FritzBoxTools, + avm_wrapper: AvmWrapper, device_friendly_name: str, data_fritz: FritzData, local_ip: str, ) -> list[Entity]: """Get a list of all entities.""" - if avm_device.mesh_role == MeshRoles.SLAVE: + if avm_wrapper.mesh_role == MeshRoles.SLAVE: return [] return [ - *deflection_entities_list(avm_device, device_friendly_name), - *port_entities_list(avm_device, device_friendly_name, local_ip), - *wifi_entities_list(avm_device, device_friendly_name), - *profile_entities_list(avm_device, data_fritz), + *deflection_entities_list(avm_wrapper, device_friendly_name), + *port_entities_list(avm_wrapper, device_friendly_name, local_ip), + *wifi_entities_list(avm_wrapper, device_friendly_name), + *profile_entities_list(avm_wrapper, data_fritz), ] @@ -334,16 +334,16 @@ async def async_setup_entry( ) -> None: """Set up entry.""" _LOGGER.debug("Setting up switches") - avm_device: FritzBoxTools = hass.data[DOMAIN][entry.entry_id] + avm_wrapper: AvmWrapper = hass.data[DOMAIN][entry.entry_id] data_fritz: FritzData = hass.data[DATA_FRITZ] - _LOGGER.debug("Fritzbox services: %s", avm_device.connection.services) + _LOGGER.debug("Fritzbox services: %s", avm_wrapper.connection.services) - local_ip = await async_get_source_ip(avm_device.hass, target_ip=avm_device.host) + local_ip = await async_get_source_ip(avm_wrapper.hass, target_ip=avm_wrapper.host) entities_list = await hass.async_add_executor_job( all_entities_list, - avm_device, + avm_wrapper, entry.title, data_fritz, local_ip, @@ -354,10 +354,10 @@ async def async_setup_entry( @callback def update_avm_device() -> None: """Update the values of the AVM device.""" - async_add_entities(profile_entities_list(avm_device, data_fritz)) + async_add_entities(profile_entities_list(avm_wrapper, data_fritz)) entry.async_on_unload( - async_dispatcher_connect(hass, avm_device.signal_device_new, update_avm_device) + async_dispatcher_connect(hass, avm_wrapper.signal_device_new, update_avm_device) ) @@ -366,12 +366,12 @@ class FritzBoxBaseSwitch(FritzBoxBaseEntity): def __init__( self, - avm_device: FritzBoxTools, + avm_wrapper: AvmWrapper, device_friendly_name: str, switch_info: SwitchInfo, ) -> None: """Init Fritzbox port switch.""" - super().__init__(avm_device, device_friendly_name) + super().__init__(avm_wrapper, device_friendly_name) self._description = switch_info["description"] self._friendly_name = switch_info["friendly_name"] @@ -381,7 +381,7 @@ class FritzBoxBaseSwitch(FritzBoxBaseEntity): self._switch = switch_info["callback_switch"] self._name = f"{self._friendly_name} {self._description}" - self._unique_id = f"{self._avm_device.unique_id}-{slugify(self._description)}" + self._unique_id = f"{self._avm_wrapper.unique_id}-{slugify(self._description)}" self._attributes: dict[str, str] = {} self._is_available = True @@ -437,7 +437,7 @@ class FritzBoxPortSwitch(FritzBoxBaseSwitch, SwitchEntity): def __init__( self, - avm_device: FritzBoxTools, + avm_wrapper: AvmWrapper, device_friendly_name: str, port_mapping: dict[str, Any] | None, port_name: str, @@ -445,7 +445,7 @@ class FritzBoxPortSwitch(FritzBoxBaseSwitch, SwitchEntity): connection_type: str, ) -> None: """Init Fritzbox port switch.""" - self._avm_device = avm_device + self._avm_wrapper = avm_wrapper self._attributes = {} self.connection_type = connection_type @@ -464,13 +464,13 @@ class FritzBoxPortSwitch(FritzBoxBaseSwitch, SwitchEntity): callback_update=self._async_fetch_update, callback_switch=self._async_handle_port_switch_on_off, ) - super().__init__(avm_device, device_friendly_name, switch_info) + super().__init__(avm_wrapper, device_friendly_name, switch_info) async def _async_fetch_update(self) -> None: """Fetch updates.""" self.port_mapping = await async_service_call_action( - self._avm_device, + self._avm_wrapper, self.connection_type, "1", "GetGenericPortMappingEntry", @@ -505,7 +505,7 @@ class FritzBoxPortSwitch(FritzBoxBaseSwitch, SwitchEntity): self.port_mapping["NewEnabled"] = "1" if turn_on else "0" resp = await async_service_call_action( - self._avm_device, + self._avm_wrapper, self.connection_type, "1", "AddPortMapping", @@ -520,12 +520,12 @@ class FritzBoxDeflectionSwitch(FritzBoxBaseSwitch, SwitchEntity): def __init__( self, - avm_device: FritzBoxTools, + avm_wrapper: AvmWrapper, device_friendly_name: str, dict_of_deflection: Any, ) -> None: """Init Fritxbox Deflection class.""" - self._avm_device = avm_device + self._avm_wrapper = avm_wrapper self.dict_of_deflection = dict_of_deflection self._attributes = {} @@ -540,13 +540,13 @@ class FritzBoxDeflectionSwitch(FritzBoxBaseSwitch, SwitchEntity): callback_update=self._async_fetch_update, callback_switch=self._async_switch_on_off_executor, ) - super().__init__(self._avm_device, device_friendly_name, switch_info) + super().__init__(self._avm_wrapper, device_friendly_name, switch_info) async def _async_fetch_update(self) -> None: """Fetch updates.""" resp = await async_service_call_action( - self._avm_device, "X_AVM-DE_OnTel", "1", "GetDeflections" + self._avm_wrapper, "X_AVM-DE_OnTel", "1", "GetDeflections" ) if not resp: self._is_available = False @@ -580,7 +580,7 @@ class FritzBoxDeflectionSwitch(FritzBoxBaseSwitch, SwitchEntity): async def _async_switch_on_off_executor(self, turn_on: bool) -> None: """Handle deflection switch.""" await async_service_call_action( - self._avm_device, + self._avm_wrapper, "X_AVM-DE_OnTel", "1", "SetDeflectionEnable", @@ -594,9 +594,9 @@ class FritzBoxProfileSwitch(FritzDeviceBase, SwitchEntity): _attr_icon = "mdi:router-wireless-settings" - def __init__(self, avm_device: FritzBoxTools, device: FritzDevice) -> None: + def __init__(self, avm_wrapper: AvmWrapper, device: FritzDevice) -> None: """Init Fritz profile.""" - super().__init__(avm_device, device) + super().__init__(avm_wrapper, device) self._attr_is_on: bool = False self._name = f"{device.hostname} Internet Access" self._attr_unique_id = f"{self._mac}_internet_access" @@ -605,7 +605,7 @@ class FritzBoxProfileSwitch(FritzDeviceBase, SwitchEntity): @property def is_on(self) -> bool: """Switch status.""" - return self._avm_device.devices[self._mac].wan_access + return self._avm_wrapper.devices[self._mac].wan_access @property def device_info(self) -> DeviceInfo: @@ -618,7 +618,7 @@ class FritzBoxProfileSwitch(FritzDeviceBase, SwitchEntity): identifiers={(DOMAIN, self._mac)}, via_device=( DOMAIN, - self._avm_device.unique_id, + self._avm_wrapper.unique_id, ), ) @@ -639,7 +639,7 @@ class FritzBoxProfileSwitch(FritzDeviceBase, SwitchEntity): async def _async_switch_on_off(self, turn_on: bool) -> None: """Handle parental control switch.""" await async_service_call_action( - self._avm_device, + self._avm_wrapper, "X_AVM-DE_HostFilter", "1", "DisallowWANAccessByIP", @@ -653,13 +653,13 @@ class FritzBoxWifiSwitch(FritzBoxBaseSwitch, SwitchEntity): def __init__( self, - avm_device: FritzBoxTools, + avm_wrapper: AvmWrapper, device_friendly_name: str, network_num: int, network_name: str, ) -> None: """Init Fritz Wifi switch.""" - self._avm_device = avm_device + self._avm_wrapper = avm_wrapper self._attributes = {} self._attr_entity_category = EntityCategory.CONFIG @@ -673,13 +673,13 @@ class FritzBoxWifiSwitch(FritzBoxBaseSwitch, SwitchEntity): callback_update=self._async_fetch_update, callback_switch=self._async_switch_on_off_executor, ) - super().__init__(self._avm_device, device_friendly_name, switch_info) + super().__init__(self._avm_wrapper, device_friendly_name, switch_info) async def _async_fetch_update(self) -> None: """Fetch updates.""" wifi_info = await async_service_call_action( - self._avm_device, + self._avm_wrapper, "WLANConfiguration", str(self._network_num), "GetInfo", @@ -705,7 +705,7 @@ class FritzBoxWifiSwitch(FritzBoxBaseSwitch, SwitchEntity): async def _async_switch_on_off_executor(self, turn_on: bool) -> None: """Handle wifi switch.""" await async_service_call_action( - self._avm_device, + self._avm_wrapper, "WLANConfiguration", str(self._network_num), "SetEnable", diff --git a/homeassistant/components/fritz/wrapper.py b/homeassistant/components/fritz/wrapper.py deleted file mode 100644 index 805b431f44a..00000000000 --- a/homeassistant/components/fritz/wrapper.py +++ /dev/null @@ -1,98 +0,0 @@ -"""AVM FRITZ!Box API wrapper.""" -from __future__ import annotations - -from functools import partial -import logging -from typing import Any - -from fritzconnection.core.exceptions import ( - FritzActionError, - FritzActionFailedError, - FritzConnectionException, - FritzLookUpError, - FritzSecurityError, - FritzServiceError, -) - -from .common import FritzBoxTools - -_LOGGER = logging.getLogger(__name__) - - -class AvmWrapper: - """Setup AVM wrapper for API calls.""" - - def __init__(self, avm_device: FritzBoxTools) -> None: - """Init wrapper API class.""" - - self._avm_device = avm_device - - def _service_call_action( - self, - service_name: str, - service_suffix: str, - action_name: str, - **kwargs: Any, - ) -> dict | None: - """Return service details.""" - - if ( - f"{service_name}{service_suffix}" - not in self._avm_device.connection.services - ): - return None - - try: - result: dict = self._avm_device.connection.call_action( - f"{service_name}:{service_suffix}", - action_name, - **kwargs, - ) - return result - except FritzSecurityError: - _LOGGER.error( - "Authorization Error: Please check the provided credentials and verify that you can log into the web interface", - exc_info=True, - ) - except ( - FritzActionError, - FritzActionFailedError, - FritzServiceError, - FritzLookUpError, - ): - _LOGGER.error( - "Service/Action Error: cannot execute service %s with action %s", - service_name, - action_name, - exc_info=True, - ) - except FritzConnectionException: - _LOGGER.error( - "Connection Error: Please check the device is properly configured for remote login", - exc_info=True, - ) - return None - - async def _async_service_call_action( - self, service_name: str, service_suffix: str, action_name: str, **kwargs: Any - ) -> dict[str, Any] | None: - """Make call_action async.""" - - return await self._avm_device.hass.async_add_executor_job( - partial( - self._service_call_action, - service_name, - service_suffix, - action_name, - **kwargs, - ) - ) - - async def get_wan_dsl_interface_config(self) -> dict[str, Any] | None: - """Call WANDSLInterfaceConfig service.""" - - return await self._async_service_call_action( - "WANDSLInterfaceConfig", - "1", - "GetInfo", - )