diff --git a/.coveragerc b/.coveragerc index 7fb301ec199..355645b9eb5 100644 --- a/.coveragerc +++ b/.coveragerc @@ -378,6 +378,7 @@ 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/sensor.py b/homeassistant/components/fritz/sensor.py index 781cde67957..d264d4b6603 100644 --- a/homeassistant/components/fritz/sensor.py +++ b/homeassistant/components/fritz/sensor.py @@ -7,13 +7,7 @@ from datetime import datetime, timedelta import logging from typing import Any, Literal -from fritzconnection.core.exceptions import ( - FritzActionError, - FritzActionFailedError, - FritzConnectionException, - FritzInternalError, - FritzServiceError, -) +from fritzconnection.core.exceptions import FritzConnectionException from fritzconnection.lib.fritzstatus import FritzStatus from homeassistant.components.sensor import ( @@ -36,6 +30,7 @@ from homeassistant.util.dt import utcnow from .common import FritzBoxBaseEntity, FritzBoxTools from .const import DOMAIN, DSL_CONNECTION, UPTIME_DEVIATION, MeshRoles +from .wrapper import AvmWrapper _LOGGER = logging.getLogger(__name__) @@ -282,22 +277,12 @@ async def async_setup_entry( """Set up entry.""" _LOGGER.debug("Setting up FRITZ!Box sensors") avm_device: FritzBoxTools = hass.data[DOMAIN][entry.entry_id] + avm_wrapper = AvmWrapper(avm_device) dsl: bool = False - try: - dslinterface = await hass.async_add_executor_job( - avm_device.connection.call_action, - "WANDSLInterfaceConfig:1", - "GetInfo", - ) + dslinterface = await avm_wrapper.get_wan_dsl_interface_config() + if dslinterface: dsl = dslinterface["NewEnable"] - except ( - FritzInternalError, - FritzActionError, - FritzActionFailedError, - FritzServiceError, - ): - pass entities = [ FritzBoxSensor(avm_device, entry.title, description) diff --git a/homeassistant/components/fritz/wrapper.py b/homeassistant/components/fritz/wrapper.py new file mode 100644 index 00000000000..805b431f44a --- /dev/null +++ b/homeassistant/components/fritz/wrapper.py @@ -0,0 +1,98 @@ +"""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", + )