jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-24 05:37:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Let the new wrapper just extend the FritzBoxTools class (#64133)

Browse Source 
* let wrapper just extend the FritzBoxTools class

* keep avm_device in method names
This commit is contained in:
Michael 2022-01-20 12:43:32 +01:00 committed by GitHub
parent 76229bc188
commit 09297520c0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 217 additions and 244 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

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

14
homeassistant/components/fritz/__init__.py
View File

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

24
homeassistant/components/fritz/binary_sensor.py
View File

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

24
homeassistant/components/fritz/button.py
View File

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

98
homeassistant/components/fritz/common.py
View File

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

18
homeassistant/components/fritz/config_flow.py
View File

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

28
homeassistant/components/fritz/device_tracker.py
View File

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

18
homeassistant/components/fritz/sensor.py
View File

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

6
homeassistant/components/fritz/services.py
View File

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

132
homeassistant/components/fritz/switch.py
View File

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

98
homeassistant/components/fritz/wrapper.py
View File

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