diff --git a/homeassistant/components/fritz/common.py b/homeassistant/components/fritz/common.py
index ae9845ad250..37dfdc0e554 100644
--- a/homeassistant/components/fritz/common.py
+++ b/homeassistant/components/fritz/common.py
@@ -502,11 +502,11 @@ class AvmWrapper(FritzBoxTools):
service_suffix: str,
action_name: str,
**kwargs: Any,
- ) -> dict | None:
+ ) -> dict:
"""Return service details."""
if f"{service_name}{service_suffix}" not in self.connection.services:
- return None
+ return {}
try:
result: dict = self.connection.call_action(
@@ -537,30 +537,154 @@ class AvmWrapper(FritzBoxTools):
"Connection Error: Please check the device is properly configured for remote login",
exc_info=True,
)
- return None
+ return {}
- 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."""
+ async def async_get_wan_dsl_interface_config(self) -> dict[str, Any]:
+ """Call WANDSLInterfaceConfig service."""
+
+ return await self.hass.async_add_executor_job(
+ partial(self.get_wan_dsl_interface_config)
+ )
+
+ async def async_get_port_mapping(self, con_type: str, index: int) -> dict[str, Any]:
+ """Call GetGenericPortMappingEntry action."""
+
+ return await self.hass.async_add_executor_job(
+ partial(self.get_port_mapping, con_type, index)
+ )
+
+ async def async_get_wlan_configuration(self, index: int) -> dict[str, Any]:
+ """Call WLANConfiguration service."""
+
+ return await self.hass.async_add_executor_job(
+ partial(self.get_wlan_configuration, index)
+ )
+
+ async def async_get_ontel_deflections(self) -> dict[str, Any]:
+ """Call GetDeflections action from X_AVM-DE_OnTel service."""
+
+ return await self.hass.async_add_executor_job(
+ partial(self.get_ontel_deflections)
+ )
+
+ async def async_set_wlan_configuration(
+ self, index: int, turn_on: bool
+ ) -> dict[str, Any]:
+ """Call SetEnable action from WLANConfiguration service."""
+
+ return await self.hass.async_add_executor_job(
+ partial(self.set_wlan_configuration, index, turn_on)
+ )
+
+ async def async_set_deflection_enable(
+ self, index: int, turn_on: bool
+ ) -> dict[str, Any]:
+ """Call SetDeflectionEnable service."""
+
+ return await self.hass.async_add_executor_job(
+ partial(self.set_deflection_enable, index, turn_on)
+ )
+
+ async def async_add_port_mapping(
+ self, con_type: str, port_mapping: Any
+ ) -> dict[str, Any]:
+ """Call AddPortMapping service."""
return await self.hass.async_add_executor_job(
partial(
- self._service_call_action,
- service_name,
- service_suffix,
- action_name,
- **kwargs,
+ self.add_port_mapping,
+ con_type,
+ port_mapping,
)
)
- async def get_wan_dsl_interface_config(self) -> dict[str, Any] | None:
+ async def async_set_allow_wan_access(
+ self, ip_address: str, turn_on: bool
+ ) -> dict[str, Any]:
+ """Call X_AVM-DE_HostFilter service."""
+
+ return await self.hass.async_add_executor_job(
+ partial(self.set_allow_wan_access, ip_address, turn_on)
+ )
+
+ def get_ontel_num_deflections(self) -> dict[str, Any]:
+ """Call GetNumberOfDeflections action from X_AVM-DE_OnTel service."""
+
+ return self._service_call_action(
+ "X_AVM-DE_OnTel", "1", "GetNumberOfDeflections"
+ )
+
+ def get_ontel_deflections(self) -> dict[str, Any]:
+ """Call GetDeflections action from X_AVM-DE_OnTel service."""
+
+ return self._service_call_action("X_AVM-DE_OnTel", "1", "GetDeflections")
+
+ def get_default_connection(self) -> dict[str, Any]:
+ """Call Layer3Forwarding service."""
+
+ return self._service_call_action(
+ "Layer3Forwarding", "1", "GetDefaultConnectionService"
+ )
+
+ def get_num_port_mapping(self, con_type: str) -> dict[str, Any]:
+ """Call GetPortMappingNumberOfEntries action."""
+
+ return self._service_call_action(con_type, "1", "GetPortMappingNumberOfEntries")
+
+ def get_port_mapping(self, con_type: str, index: int) -> dict[str, Any]:
+ """Call GetGenericPortMappingEntry action."""
+
+ return self._service_call_action(
+ con_type, "1", "GetGenericPortMappingEntry", NewPortMappingIndex=index
+ )
+
+ def get_wlan_configuration(self, index: int) -> dict[str, Any]:
+ """Call WLANConfiguration service."""
+
+ return self._service_call_action("WLANConfiguration", str(index), "GetInfo")
+
+ def get_wan_dsl_interface_config(self) -> dict[str, Any]:
"""Call WANDSLInterfaceConfig service."""
- return await self._async_service_call_action(
- "WANDSLInterfaceConfig",
+ return self._service_call_action("WANDSLInterfaceConfig", "1", "GetInfo")
+
+ def set_wlan_configuration(self, index: int, turn_on: bool) -> dict[str, Any]:
+ """Call SetEnable action from WLANConfiguration service."""
+
+ return self._service_call_action(
+ "WLANConfiguration",
+ str(index),
+ "SetEnable",
+ NewEnable="1" if turn_on else "0",
+ )
+
+ def set_deflection_enable(self, index: int, turn_on: bool) -> dict[str, Any]:
+ """Call SetDeflectionEnable service."""
+
+ return self._service_call_action(
+ "X_AVM-DE_OnTel",
"1",
- "GetInfo",
+ "SetDeflectionEnable",
+ NewDeflectionId=index,
+ NewEnable="1" if turn_on else "0",
+ )
+
+ def add_port_mapping(self, con_type: str, port_mapping: Any) -> dict[str, Any]:
+ """Call AddPortMapping service."""
+
+ return self._service_call_action(
+ con_type, "1", "AddPortMapping", **port_mapping
+ )
+
+ def set_allow_wan_access(self, ip_address: str, turn_on: bool) -> dict[str, Any]:
+ """Call X_AVM-DE_HostFilter service."""
+
+ return self._service_call_action(
+ "X_AVM-DE_HostFilter",
+ "1",
+ "DisallowWANAccessByIP",
+ NewIPv4Address=ip_address,
+ NewDisallow="0" if turn_on else "1",
)
diff --git a/homeassistant/components/fritz/sensor.py b/homeassistant/components/fritz/sensor.py
index f3ebaf32ff9..6155cdc5914 100644
--- a/homeassistant/components/fritz/sensor.py
+++ b/homeassistant/components/fritz/sensor.py
@@ -278,7 +278,7 @@ async def async_setup_entry(
avm_wrapper: AvmWrapper = hass.data[DOMAIN][entry.entry_id]
dsl: bool = False
- dslinterface = await avm_wrapper.get_wan_dsl_interface_config()
+ dslinterface = await avm_wrapper.async_get_wan_dsl_interface_config()
if dslinterface:
dsl = dslinterface["NewEnable"]
diff --git a/homeassistant/components/fritz/switch.py b/homeassistant/components/fritz/switch.py
index 2d6af382e44..0726741845a 100644
--- a/homeassistant/components/fritz/switch.py
+++ b/homeassistant/components/fritz/switch.py
@@ -1,19 +1,9 @@
"""Switches for AVM Fritz!Box functions."""
from __future__ import annotations
-from collections import OrderedDict
-from functools import partial
import logging
from typing import Any
-from fritzconnection.core.exceptions import (
- FritzActionError,
- FritzActionFailedError,
- FritzConnectionException,
- FritzLookUpError,
- FritzSecurityError,
- FritzServiceError,
-)
import xmltodict
from homeassistant.components.network import async_get_source_ip
@@ -47,92 +37,6 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
-async def async_service_call_action(
- avm_wrapper: AvmWrapper,
- service_name: str,
- service_suffix: str | None,
- action_name: str,
- **kwargs: Any,
-) -> None | dict:
- """Return service details."""
- return await avm_wrapper.hass.async_add_executor_job(
- partial(
- service_call_action,
- avm_wrapper,
- service_name,
- service_suffix,
- action_name,
- **kwargs,
- )
- )
-
-
-def service_call_action(
- avm_wrapper: AvmWrapper,
- service_name: str,
- service_suffix: str | None,
- action_name: str,
- **kwargs: Any,
-) -> dict | None:
- """Return service details."""
-
- if f"{service_name}{service_suffix}" not in avm_wrapper.connection.services:
- return None
-
- try:
- return avm_wrapper.connection.call_action( # type: ignore[no-any-return]
- f"{service_name}:{service_suffix}",
- action_name,
- **kwargs,
- )
- except FritzSecurityError:
- _LOGGER.error(
- "Authorization Error: Please check the provided credentials and verify that you can log into the web interface",
- exc_info=True,
- )
- return None
- except (
- FritzActionError,
- FritzActionFailedError,
- FritzServiceError,
- FritzLookUpError,
- ):
- _LOGGER.error(
- "Service/Action Error: cannot execute service %s with action %s",
- service_name,
- action_name,
- exc_info=True,
- )
- return None
- except FritzConnectionException:
- _LOGGER.error(
- "Connection Error: Please check the device is properly configured for remote login",
- exc_info=True,
- )
- return None
-
-
-def get_deflections(
- avm_wrapper: AvmWrapper, service_name: str
-) -> list[OrderedDict[Any, Any]] | None:
- """Get deflection switch info."""
-
- deflection_list = service_call_action(
- avm_wrapper,
- service_name,
- "1",
- "GetDeflections",
- )
-
- if not deflection_list:
- return []
-
- items = xmltodict.parse(deflection_list["NewDeflectionList"])["List"]["Item"]
- if not isinstance(items, list):
- return [items]
- return items
-
-
def deflection_entities_list(
avm_wrapper: AvmWrapper, device_friendly_name: str
) -> list[FritzBoxDeflectionSwitch]:
@@ -140,10 +44,7 @@ def deflection_entities_list(
_LOGGER.debug("Setting up %s switches", SWITCH_TYPE_DEFLECTION)
- service_name = "X_AVM-DE_OnTel"
- deflections_response = service_call_action(
- avm_wrapper, service_name, "1", "GetNumberOfDeflections"
- )
+ deflections_response = avm_wrapper.get_ontel_num_deflections()
if not deflections_response:
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION)
return []
@@ -158,13 +59,18 @@ def deflection_entities_list(
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION)
return []
- deflection_list = get_deflections(avm_wrapper, service_name)
- if deflection_list is None:
+ deflection_list = avm_wrapper.get_ontel_deflections()
+
+ if not deflection_list:
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 deflection_list
+ for dict_of_deflection in items
]
@@ -175,10 +81,7 @@ def port_entities_list(
_LOGGER.debug("Setting up %s switches", SWITCH_TYPE_PORTFORWARD)
entities_list: list[FritzBoxPortSwitch] = []
- service_name = "Layer3Forwarding"
- connection_type = service_call_action(
- avm_wrapper, service_name, "1", "GetDefaultConnectionService"
- )
+ connection_type = avm_wrapper.get_default_connection()
if not connection_type:
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_PORTFORWARD)
return []
@@ -187,9 +90,7 @@ def port_entities_list(
con_type: str = connection_type["NewDefaultConnectionService"][2:][:-2]
# Query port forwardings and setup a switch for each forward for the current device
- resp = service_call_action(
- avm_wrapper, con_type, "1", "GetPortMappingNumberOfEntries"
- )
+ resp = avm_wrapper.get_num_port_mapping(con_type)
if not resp:
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION)
return []
@@ -206,14 +107,7 @@ def port_entities_list(
for i in range(port_forwards_count):
- portmap = service_call_action(
- avm_wrapper,
- con_type,
- "1",
- "GetGenericPortMappingEntry",
- NewPortMappingIndex=i,
- )
-
+ portmap = avm_wrapper.get_port_mapping(con_type, i)
if not portmap:
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION)
continue
@@ -260,9 +154,7 @@ def wifi_entities_list(
if not ("WLANConfiguration" + str(i)) in avm_wrapper.connection.services:
continue
- network_info = service_call_action(
- avm_wrapper, "WLANConfiguration", str(i), "GetInfo"
- )
+ network_info = avm_wrapper.get_wlan_configuration(i)
if network_info:
ssid = network_info["NewSSID"]
_LOGGER.debug("SSID from device: <%s>", ssid)
@@ -462,24 +354,20 @@ class FritzBoxPortSwitch(FritzBoxBaseSwitch, SwitchEntity):
icon="mdi:check-network",
type=SWITCH_TYPE_PORTFORWARD,
callback_update=self._async_fetch_update,
- callback_switch=self._async_handle_port_switch_on_off,
+ callback_switch=self._async_switch_on_off_executor,
)
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_wrapper,
- self.connection_type,
- "1",
- "GetGenericPortMappingEntry",
- NewPortMappingIndex=self._idx,
+ self.port_mapping = await self._avm_wrapper.async_get_port_mapping(
+ self.connection_type, self._idx
)
_LOGGER.debug(
"Specific %s response: %s", SWITCH_TYPE_PORTFORWARD, self.port_mapping
)
- if self.port_mapping is None:
+ if not self.port_mapping:
self._is_available = False
return
@@ -497,21 +385,16 @@ class FritzBoxPortSwitch(FritzBoxBaseSwitch, SwitchEntity):
for key, attr in attributes_dict.items():
self._attributes[attr] = self.port_mapping[key]
- async def _async_handle_port_switch_on_off(self, turn_on: bool) -> bool:
+ async def _async_switch_on_off_executor(self, turn_on: bool) -> bool:
if self.port_mapping is None:
return False
self.port_mapping["NewEnabled"] = "1" if turn_on else "0"
- resp = await async_service_call_action(
- self._avm_wrapper,
- self.connection_type,
- "1",
- "AddPortMapping",
- **self.port_mapping,
+ resp = await self._avm_wrapper.async_add_port_mapping(
+ self.connection_type, self.port_mapping
)
-
return bool(resp is not None)
@@ -545,9 +428,7 @@ class FritzBoxDeflectionSwitch(FritzBoxBaseSwitch, SwitchEntity):
async def _async_fetch_update(self) -> None:
"""Fetch updates."""
- resp = await async_service_call_action(
- self._avm_wrapper, "X_AVM-DE_OnTel", "1", "GetDeflections"
- )
+ resp = await self._avm_wrapper.async_get_ontel_deflections()
if not resp:
self._is_available = False
return
@@ -579,14 +460,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_wrapper,
- "X_AVM-DE_OnTel",
- "1",
- "SetDeflectionEnable",
- NewDeflectionId=self.id,
- NewEnable="1" if turn_on else "0",
- )
+ await self._avm_wrapper.async_set_deflection_enable(self.id, turn_on)
class FritzBoxProfileSwitch(FritzDeviceBase, SwitchEntity):
@@ -632,21 +506,12 @@ class FritzBoxProfileSwitch(FritzDeviceBase, SwitchEntity):
async def _async_handle_turn_on_off(self, turn_on: bool) -> bool:
"""Handle switch state change request."""
- await self._async_switch_on_off(turn_on)
+ if not self.ip_address:
+ return False
+ await self._avm_wrapper.async_set_allow_wan_access(self.ip_address, turn_on)
self.async_write_ha_state()
return True
- async def _async_switch_on_off(self, turn_on: bool) -> None:
- """Handle parental control switch."""
- await async_service_call_action(
- self._avm_wrapper,
- "X_AVM-DE_HostFilter",
- "1",
- "DisallowWANAccessByIP",
- NewIPv4Address=self.ip_address,
- NewDisallow="0" if turn_on else "1",
- )
-
class FritzBoxWifiSwitch(FritzBoxBaseSwitch, SwitchEntity):
"""Defines a FRITZ!Box Tools Wifi switch."""
@@ -678,17 +543,14 @@ class FritzBoxWifiSwitch(FritzBoxBaseSwitch, SwitchEntity):
async def _async_fetch_update(self) -> None:
"""Fetch updates."""
- wifi_info = await async_service_call_action(
- self._avm_wrapper,
- "WLANConfiguration",
- str(self._network_num),
- "GetInfo",
+ wifi_info = await self._avm_wrapper.async_get_wlan_configuration(
+ self._network_num
)
_LOGGER.debug(
"Specific %s response: GetInfo=%s", SWITCH_TYPE_WIFINETWORK, wifi_info
)
- if wifi_info is None:
+ if not wifi_info:
self._is_available = False
return
@@ -704,10 +566,4 @@ 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_wrapper,
- "WLANConfiguration",
- str(self._network_num),
- "SetEnable",
- NewEnable="1" if turn_on else "0",
- )
+ await self._avm_wrapper.async_set_wlan_configuration(self._network_num, turn_on)