From 7959225fef303245fe1460c8f4f2fb46c904b6ad Mon Sep 17 00:00:00 2001 From: Simone Chemelli Date: Tue, 29 Jun 2021 17:57:34 +0200 Subject: [PATCH] Add switch platform to Fritz (#51610) * Add switch platform to Fritz * Fix tests * Pylint * Small fix * Bump fritzprofiles to fix log level and identifier * Fix different WiFi networks with same name * Changed exposed attributes * Moved to extra_state * Remove redundant lambda * Add missing wait * Removed identifiers * Update homeassistant/components/fritz/switch.py Co-authored-by: J. Nick Koston * Update homeassistant/components/fritz/switch.py Co-authored-by: J. Nick Koston * Add mapping dict * Device Profile disabled by default * Heavy cleanup * Tweak * Bug fix * Update homeassistant/components/fritz/switch.py Co-authored-by: Aaron David Schneider * Fix port forward switching + small log improvement * Cleanup from old approach * Handle port mapping hot removal (from device) * Minor fixes * Typying * Removed lambda call * Last missing strict typing * Split get entities * Func rename * Move FritzBoxBaseSwitch to switch.py * Removed lambda * Update homeassistant/components/fritz/common.py Co-authored-by: J. Nick Koston * Update homeassistant/components/fritz/common.py Co-authored-by: J. Nick Koston * Update homeassistant/components/fritz/switch.py Co-authored-by: J. Nick Koston * Update homeassistant/components/fritz/switch.py Co-authored-by: J. Nick Koston * Update homeassistant/components/fritz/switch.py Co-authored-by: J. Nick Koston * Update homeassistant/components/fritz/switch.py Co-authored-by: J. Nick Koston * Fixes after applying comments * Remvoed redundant try block * Removed broad-except * Optimized async/sync switch * Update homeassistant/components/fritz/switch.py Co-authored-by: J. Nick Koston * Update homeassistant/components/fritz/switch.py Co-authored-by: J. Nick Koston * Update homeassistant/components/fritz/switch.py Co-authored-by: J. Nick Koston * Address remaining comments * Optimize return list * More optimization for return lists * Some missing strict typing * Redundant typing * Update homeassistant/components/fritz/switch.py Co-authored-by: J. Nick Koston * Wrong if * Introduce const for profile status * Update homeassistant/components/fritz/switch.py Co-authored-by: J. Nick Koston * Update homeassistant/components/fritz/switch.py Co-authored-by: J. Nick Koston * Fix mypy * Switch back to get_local_ip() * Address latest comments Co-authored-by: J. Nick Koston Co-authored-by: Aaron David Schneider --- .coveragerc | 1 + homeassistant/components/fritz/common.py | 24 +- homeassistant/components/fritz/const.py | 10 +- homeassistant/components/fritz/manifest.json | 5 +- homeassistant/components/fritz/switch.py | 639 +++++++++++++++++++ requirements_all.txt | 4 + requirements_test_all.txt | 4 + tests/components/fritz/test_config_flow.py | 57 +- 8 files changed, 735 insertions(+), 9 deletions(-) create mode 100644 homeassistant/components/fritz/switch.py diff --git a/.coveragerc b/.coveragerc index 6fb95b595cc..eb0fbdc1fcf 100644 --- a/.coveragerc +++ b/.coveragerc @@ -343,6 +343,7 @@ omit = homeassistant/components/fritz/device_tracker.py homeassistant/components/fritz/sensor.py homeassistant/components/fritz/services.py + homeassistant/components/fritz/switch.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/common.py b/homeassistant/components/fritz/common.py index a255ef6439f..776c7a7a22e 100644 --- a/homeassistant/components/fritz/common.py +++ b/homeassistant/components/fritz/common.py @@ -5,7 +5,7 @@ from dataclasses import dataclass, field from datetime import datetime, timedelta import logging from types import MappingProxyType -from typing import Any, TypedDict +from typing import Any, Callable, TypedDict from fritzconnection import FritzConnection from fritzconnection.core.exceptions import ( @@ -15,6 +15,7 @@ from fritzconnection.core.exceptions import ( ) from fritzconnection.lib.fritzhosts import FritzHosts from fritzconnection.lib.fritzstatus import FritzStatus +from fritzprofiles import FritzProfileSwitch, get_all_profiles from homeassistant.components.device_tracker.const import ( CONF_CONSIDER_HOME, @@ -44,7 +45,7 @@ _LOGGER = logging.getLogger(__name__) class ClassSetupMissing(Exception): """Raised when a Class func is called before setup.""" - def __init__(self): + def __init__(self) -> None: """Init custom exception.""" super().__init__("Function called before Class setup") @@ -85,6 +86,7 @@ class FritzBoxTools: self._unique_id: str | None = None self.connection: FritzConnection = None self.fritz_hosts: FritzHosts = None + self.fritz_profiles: dict[str, FritzProfileSwitch] = {} self.fritz_status: FritzStatus = None self.hass = hass self.host = host @@ -117,6 +119,13 @@ class FritzBoxTools: self._model = info.get("NewModelName") self._sw_version = info.get("NewSoftwareVersion") + self.fritz_profiles = { + profile: FritzProfileSwitch( + "http://" + self.host, self.username, self.password, profile + ) + for profile in get_all_profiles(self.host, self.username, self.password) + } + async def async_start(self, options: MappingProxyType[str, Any]) -> None: """Start FritzHosts connection.""" self.fritz_hosts = FritzHosts(fc=self.connection) @@ -306,6 +315,17 @@ class FritzDevice: return self._last_activity +class SwitchInfo(TypedDict): + """FRITZ!Box switch info class.""" + + description: str + friendly_name: str + icon: str + type: str + callback_update: Callable + callback_switch: Callable + + class FritzBoxBaseEntity: """Fritz host entity base class.""" diff --git a/homeassistant/components/fritz/const.py b/homeassistant/components/fritz/const.py index 266e24b6be3..776c7a7dafa 100644 --- a/homeassistant/components/fritz/const.py +++ b/homeassistant/components/fritz/const.py @@ -2,7 +2,7 @@ DOMAIN = "fritz" -PLATFORMS = ["binary_sensor", "device_tracker", "sensor"] +PLATFORMS = ["binary_sensor", "device_tracker", "sensor", "switch"] DATA_FRITZ = "fritz_data" @@ -19,6 +19,14 @@ FRITZ_SERVICES = "fritz_services" SERVICE_REBOOT = "reboot" SERVICE_RECONNECT = "reconnect" +SWITCH_PROFILE_STATUS_OFF = "never" +SWITCH_PROFILE_STATUS_ON = "unlimited" + +SWITCH_TYPE_DEFLECTION = "CallDeflection" +SWITCH_TYPE_DEVICEPROFILE = "DeviceProfile" +SWITCH_TYPE_PORTFORWARD = "PortForward" +SWITCH_TYPE_WIFINETWORK = "WiFiNetwork" + TRACKER_SCAN_INTERVAL = 30 UPTIME_DEVIATION = 5 diff --git a/homeassistant/components/fritz/manifest.json b/homeassistant/components/fritz/manifest.json index 1158ea2e797..d1c096a2ef5 100644 --- a/homeassistant/components/fritz/manifest.json +++ b/homeassistant/components/fritz/manifest.json @@ -3,8 +3,11 @@ "name": "AVM FRITZ!Box Tools", "documentation": "https://www.home-assistant.io/integrations/fritz", "requirements": [ - "fritzconnection==1.4.2" + "fritzconnection==1.4.2", + "fritzprofiles==0.6.1", + "xmltodict==0.12.0" ], + "dependencies": ["network"], "codeowners": [ "@mammuth", "@AaronDavidSchneider", diff --git a/homeassistant/components/fritz/switch.py b/homeassistant/components/fritz/switch.py new file mode 100644 index 00000000000..09bae36a29c --- /dev/null +++ b/homeassistant/components/fritz/switch.py @@ -0,0 +1,639 @@ +"""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, + FritzSecurityError, + FritzServiceError, +) +import xmltodict + +from homeassistant.components.switch import SwitchEntity +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity import Entity +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.util import get_local_ip, slugify + +from .common import FritzBoxBaseEntity, FritzBoxTools, SwitchInfo +from .const import ( + DOMAIN, + SWITCH_PROFILE_STATUS_OFF, + SWITCH_PROFILE_STATUS_ON, + SWITCH_TYPE_DEFLECTION, + SWITCH_TYPE_DEVICEPROFILE, + SWITCH_TYPE_PORTFORWARD, + SWITCH_TYPE_WIFINETWORK, +) + +_LOGGER = logging.getLogger(__name__) + + +async def async_service_call_action( + fritzbox_tools: FritzBoxTools, + service_name: str, + service_suffix: str | None, + action_name: str, + **kwargs: Any, +) -> None | dict: + """Return service details.""" + return await fritzbox_tools.hass.async_add_executor_job( + partial( + service_call_action, + fritzbox_tools, + service_name, + service_suffix, + action_name, + **kwargs, + ) + ) + + +def service_call_action( + fritzbox_tools: FritzBoxTools, + 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 fritzbox_tools.connection.services: + return None + + try: + return fritzbox_tools.connection.call_action( + 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): + _LOGGER.error( + "Service/Action Error: cannot execute service %s", + service_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( + fritzbox_tools: FritzBoxTools, service_name: str +) -> list[OrderedDict[Any, Any]] | None: + """Get deflection switch info.""" + + deflection_list = service_call_action( + fritzbox_tools, + service_name, + "1", + "GetDeflections", + ) + + if not deflection_list: + return [] + + return [xmltodict.parse(deflection_list["NewDeflectionList"])["List"]["Item"]] + + +def deflection_entities_list( + fritzbox_tools: FritzBoxTools, device_friendly_name: str +) -> list[FritzBoxDeflectionSwitch]: + """Get list of deflection entities.""" + + _LOGGER.debug("Setting up %s switches", SWITCH_TYPE_DEFLECTION) + + service_name = "X_AVM-DE_OnTel" + deflections_response = service_call_action( + fritzbox_tools, service_name, "1", "GetNumberOfDeflections" + ) + if not deflections_response: + _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION) + return [] + + _LOGGER.debug( + "Specific %s response: GetNumberOfDeflections=%s", + SWITCH_TYPE_DEFLECTION, + deflections_response, + ) + + if deflections_response["NewNumberOfDeflections"] == 0: + _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION) + return [] + + deflection_list = get_deflections(fritzbox_tools, service_name) + if deflection_list is None: + return [] + + return [ + FritzBoxDeflectionSwitch( + fritzbox_tools, device_friendly_name, dict_of_deflection + ) + for dict_of_deflection in deflection_list + ] + + +def port_entities_list( + fritzbox_tools: FritzBoxTools, device_friendly_name: str +) -> list[FritzBoxPortSwitch]: + """Get list of port forwarding entities.""" + + _LOGGER.debug("Setting up %s switches", SWITCH_TYPE_PORTFORWARD) + entities_list: list = [] + service_name = "Layer3Forwarding" + connection_type = service_call_action( + fritzbox_tools, service_name, "1", "GetDefaultConnectionService" + ) + if not connection_type: + _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_PORTFORWARD) + return [] + + # Return NewDefaultConnectionService sample: "1.WANPPPConnection.1" + 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( + fritzbox_tools, con_type, "1", "GetPortMappingNumberOfEntries" + ) + if not resp: + _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION) + return [] + + port_forwards_count: int = resp["NewPortMappingNumberOfEntries"] + + _LOGGER.debug( + "Specific %s response: GetPortMappingNumberOfEntries=%s", + SWITCH_TYPE_PORTFORWARD, + port_forwards_count, + ) + + local_ip = get_local_ip() + _LOGGER.debug("IP source for %s is %s", fritzbox_tools.host, local_ip) + + for i in range(port_forwards_count): + + portmap = service_call_action( + fritzbox_tools, + con_type, + "1", + "GetGenericPortMappingEntry", + NewPortMappingIndex=i, + ) + + if not portmap: + _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION) + continue + + _LOGGER.debug( + "Specific %s response: GetGenericPortMappingEntry=%s", + SWITCH_TYPE_PORTFORWARD, + portmap, + ) + + # We can only handle port forwards of the given device + if portmap["NewInternalClient"] == local_ip: + entities_list.append( + FritzBoxPortSwitch( + fritzbox_tools, + device_friendly_name, + portmap, + i, + con_type, + ) + ) + + return entities_list + + +def profile_entities_list( + fritzbox_tools: FritzBoxTools, device_friendly_name: str +) -> list[FritzBoxProfileSwitch]: + """Get list of profile entities.""" + _LOGGER.debug("Setting up %s switches", SWITCH_TYPE_DEVICEPROFILE) + if len(fritzbox_tools.fritz_profiles) <= 0: + _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEVICEPROFILE) + return [] + + return [ + FritzBoxProfileSwitch(fritzbox_tools, device_friendly_name, profile) + for profile in fritzbox_tools.fritz_profiles.keys() + ] + + +def wifi_entities_list( + fritzbox_tools: FritzBoxTools, device_friendly_name: str +) -> list[FritzBoxWifiSwitch]: + """Get list of wifi entities.""" + _LOGGER.debug("Setting up %s switches", SWITCH_TYPE_WIFINETWORK) + std_table = {"ac": "5Ghz", "n": "2.4Ghz"} + networks: dict = {} + for i in range(4): + if not ("WLANConfiguration" + str(i)) in fritzbox_tools.connection.services: + continue + + network_info = service_call_action( + fritzbox_tools, "WLANConfiguration", str(i), "GetInfo" + ) + if network_info: + ssid = network_info["NewSSID"] + if ssid in networks.values(): + networks[i] = f'{ssid} {std_table[network_info["NewStandard"]]}' + else: + networks[i] = ssid + + return [ + FritzBoxWifiSwitch(fritzbox_tools, device_friendly_name, net, networks[net]) + for net in networks + ] + + +def all_entities_list( + fritzbox_tools: FritzBoxTools, device_friendly_name: str +) -> list[Entity]: + """Get a list of all entities.""" + return [ + *deflection_entities_list(fritzbox_tools, device_friendly_name), + *port_entities_list(fritzbox_tools, device_friendly_name), + *profile_entities_list(fritzbox_tools, device_friendly_name), + *wifi_entities_list(fritzbox_tools, device_friendly_name), + ] + + +async def async_setup_entry( + hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback +) -> None: + """Set up entry.""" + _LOGGER.debug("Setting up switches") + fritzbox_tools: FritzBoxTools = hass.data[DOMAIN][entry.entry_id] + + _LOGGER.debug("Fritzbox services: %s", fritzbox_tools.connection.services) + + entities_list = await hass.async_add_executor_job( + all_entities_list, fritzbox_tools, entry.title + ) + async_add_entities(entities_list) + + +class FritzBoxBaseSwitch(FritzBoxBaseEntity): + """Fritz switch base class.""" + + def __init__( + self, + fritzbox_tools: FritzBoxTools, + device_friendly_name: str, + switch_info: SwitchInfo, + ) -> None: + """Init Fritzbox port switch.""" + super().__init__(fritzbox_tools, device_friendly_name) + + self._description = switch_info["description"] + self._friendly_name = switch_info["friendly_name"] + self._icon = switch_info["icon"] + self._type = switch_info["type"] + self._update = switch_info["callback_update"] + self._switch = switch_info["callback_switch"] + + self._name = f"{self._friendly_name} {self._description}" + self._unique_id = ( + f"{self._fritzbox_tools.unique_id}-{slugify(self._description)}" + ) + + self._attributes: dict[str, str] = {} + self._is_available = True + + self._attr_is_on = False + + @property + def name(self) -> str: + """Return name.""" + return self._name + + @property + def icon(self) -> str: + """Return name.""" + return self._icon + + @property + def unique_id(self) -> str: + """Return unique id.""" + return self._unique_id + + @property + def available(self) -> bool: + """Return availability.""" + return self._is_available + + @property + def extra_state_attributes(self) -> dict[str, str]: + """Return device attributes.""" + return self._attributes + + async def async_update(self) -> None: + """Update data.""" + _LOGGER.debug("Updating '%s' (%s) switch state", self.name, self._type) + await self._update() + + async def async_turn_on(self, **kwargs: Any) -> None: + """Turn on switch.""" + await self._async_handle_turn_on_off(turn_on=True) + + async def async_turn_off(self, **kwargs: Any) -> None: + """Turn off switch.""" + await self._async_handle_turn_on_off(turn_on=False) + + async def _async_handle_turn_on_off(self, turn_on: bool) -> bool: + """Handle switch state change request.""" + await self._switch(turn_on) + self._attr_is_on = turn_on + return True + + +class FritzBoxPortSwitch(FritzBoxBaseSwitch, SwitchEntity): + """Defines a FRITZ!Box Tools PortForward switch.""" + + def __init__( + self, + fritzbox_tools: FritzBoxTools, + device_friendly_name: str, + port_mapping: dict[str, Any] | None, + idx: int, + connection_type: str, + ) -> None: + """Init Fritzbox port switch.""" + self._fritzbox_tools = fritzbox_tools + + self._attributes = {} + self.connection_type = connection_type + self.port_mapping = port_mapping # dict in the format as it comes from fritzconnection. eg: {'NewRemoteHost': '0.0.0.0', 'NewExternalPort': 22, 'NewProtocol': 'TCP', 'NewInternalPort': 22, 'NewInternalClient': '192.168.178.31', 'NewEnabled': True, 'NewPortMappingDescription': 'Beast SSH ', 'NewLeaseDuration': 0} + self._idx = idx # needed for update routine + + if port_mapping is None: + return + + switch_info = SwitchInfo( + description=f'Port forward {port_mapping["NewPortMappingDescription"]}', + friendly_name=device_friendly_name, + icon="mdi:check-network", + type=SWITCH_TYPE_PORTFORWARD, + callback_update=self._async_fetch_update, + callback_switch=self._async_handle_port_switch_on_off, + ) + super().__init__(fritzbox_tools, device_friendly_name, switch_info) + + async def _async_fetch_update(self) -> None: + """Fetch updates.""" + + self.port_mapping = await async_service_call_action( + self._fritzbox_tools, + self.connection_type, + "1", + "GetGenericPortMappingEntry", + NewPortMappingIndex=self._idx, + ) + _LOGGER.debug( + "Specific %s response: %s", SWITCH_TYPE_PORTFORWARD, self.port_mapping + ) + if self.port_mapping is None: + self._is_available = False + return + + self._attr_is_on = self.port_mapping["NewEnabled"] is True + self._is_available = True + + attributes_dict = { + "NewInternalClient": "internalIP", + "NewInternalPort": "internalPort", + "NewExternalPort": "externalPort", + "NewProtocol": "protocol", + "NewPortMappingDescription": "description", + } + + for key in attributes_dict: + self._attributes[attributes_dict[key]] = self.port_mapping[key] + + async def _async_handle_port_switch_on_off(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._fritzbox_tools, + self.connection_type, + "1", + "AddPortMapping", + **self.port_mapping, + ) + + return bool(resp is not None) + + +class FritzBoxDeflectionSwitch(FritzBoxBaseSwitch, SwitchEntity): + """Defines a FRITZ!Box Tools PortForward switch.""" + + def __init__( + self, + fritzbox_tools: FritzBoxTools, + device_friendly_name: str, + dict_of_deflection: Any, + ) -> None: + """Init Fritxbox Deflection class.""" + self._fritzbox_tools: FritzBoxTools = fritzbox_tools + + self.dict_of_deflection = dict_of_deflection + self._attributes = {} + self.id = int(self.dict_of_deflection["DeflectionId"]) + + switch_info = SwitchInfo( + description=f"Call deflection {self.id}", + friendly_name=device_friendly_name, + icon="mdi:phone-forward", + type=SWITCH_TYPE_DEFLECTION, + callback_update=self._async_fetch_update, + callback_switch=self._async_switch_on_off_executor, + ) + super().__init__(self._fritzbox_tools, device_friendly_name, switch_info) + + async def _async_fetch_update(self) -> None: + """Fetch updates.""" + + resp = await async_service_call_action( + self._fritzbox_tools, "X_AVM-DE_OnTel", "1", "GetDeflections" + ) + if not resp: + self._is_available = False + return + + self.dict_of_deflection = xmltodict.parse(resp["NewDeflectionList"])["List"][ + "Item" + ] + if isinstance(self.dict_of_deflection, list): + self.dict_of_deflection = self.dict_of_deflection[self.id] + + _LOGGER.debug( + "Specific %s response: NewDeflectionList=%s", + SWITCH_TYPE_DEFLECTION, + self.dict_of_deflection, + ) + + self._attr_is_on = self.dict_of_deflection["Enable"] == "1" + self._is_available = True + + self._attributes["Type"] = self.dict_of_deflection["Type"] + self._attributes["Number"] = self.dict_of_deflection["Number"] + self._attributes["DeflectionToNumber"] = self.dict_of_deflection[ + "DeflectionToNumber" + ] + # Return mode sample: "eImmediately" + self._attributes["Mode"] = self.dict_of_deflection["Mode"][1:] + self._attributes["Outgoing"] = self.dict_of_deflection["Outgoing"] + self._attributes["PhonebookID"] = self.dict_of_deflection["PhonebookID"] + + async def _async_switch_on_off_executor(self, turn_on: bool) -> None: + """Handle deflection switch.""" + await async_service_call_action( + self._fritzbox_tools, + "X_AVM-DE_OnTel", + "1", + "SetDeflectionEnable", + NewDeflectionId=self.id, + NewEnable="1" if turn_on else "0", + ) + + +class FritzBoxProfileSwitch(FritzBoxBaseSwitch, SwitchEntity): + """Defines a FRITZ!Box Tools DeviceProfile switch.""" + + def __init__( + self, fritzbox_tools: FritzBoxTools, device_friendly_name: str, profile: str + ) -> None: + """Init Fritz profile.""" + self._fritzbox_tools: FritzBoxTools = fritzbox_tools + self.profile = profile + + switch_info = SwitchInfo( + description=f"Profile {profile}", + friendly_name=device_friendly_name, + icon="mdi:router-wireless-settings", + type=SWITCH_TYPE_DEVICEPROFILE, + callback_update=self._async_fetch_update, + callback_switch=self._async_switch_on_off_executor, + ) + super().__init__(self._fritzbox_tools, device_friendly_name, switch_info) + + async def _async_fetch_update(self) -> None: + """Update data.""" + try: + status = await self.hass.async_add_executor_job( + self._fritzbox_tools.fritz_profiles[self.profile].get_state + ) + _LOGGER.debug( + "Specific %s response: get_State()=%s", + SWITCH_TYPE_DEVICEPROFILE, + status, + ) + if status == SWITCH_PROFILE_STATUS_OFF: + self._attr_is_on = False + self._is_available = True + elif status == SWITCH_PROFILE_STATUS_ON: + self._attr_is_on = True + self._is_available = True + else: + self._is_available = False + except Exception: # pylint: disable=broad-except + _LOGGER.error("Could not get %s state", self.name, exc_info=True) + self._is_available = False + + async def _async_switch_on_off_executor(self, turn_on: bool) -> None: + """Handle profile switch.""" + state = SWITCH_PROFILE_STATUS_ON if turn_on else SWITCH_PROFILE_STATUS_OFF + await self.hass.async_add_executor_job( + self._fritzbox_tools.fritz_profiles[self.profile].set_state, state + ) + + @property + def entity_registry_enabled_default(self) -> bool: + """Return if the entity should be enabled when first added to the entity registry.""" + return False + + +class FritzBoxWifiSwitch(FritzBoxBaseSwitch, SwitchEntity): + """Defines a FRITZ!Box Tools Wifi switch.""" + + def __init__( + self, + fritzbox_tools: FritzBoxTools, + device_friendly_name: str, + network_num: int, + network_name: str, + ) -> None: + """Init Fritz Wifi switch.""" + self._fritzbox_tools = fritzbox_tools + + self._attributes = {} + self._network_num = network_num + + switch_info = SwitchInfo( + description=f"Wi-Fi {network_name}", + friendly_name=device_friendly_name, + icon="mdi:wifi", + type=SWITCH_TYPE_WIFINETWORK, + callback_update=self._async_fetch_update, + callback_switch=self._async_switch_on_off_executor, + ) + super().__init__(self._fritzbox_tools, device_friendly_name, switch_info) + + async def _async_fetch_update(self) -> None: + """Fetch updates.""" + + wifi_info = await async_service_call_action( + self._fritzbox_tools, + "WLANConfiguration", + str(self._network_num), + "GetInfo", + ) + _LOGGER.debug( + "Specific %s response: GetInfo=%s", SWITCH_TYPE_WIFINETWORK, wifi_info + ) + + if wifi_info is None: + self._is_available = False + return + + self._attr_is_on = wifi_info["NewEnable"] is True + self._is_available = True + + std = wifi_info["NewStandard"] + self._attributes["standard"] = std if std else None + self._attributes["BSSID"] = wifi_info["NewBSSID"] + self._attributes["mac_address_control"] = wifi_info[ + "NewMACAddressControlEnabled" + ] + + async def _async_switch_on_off_executor(self, turn_on: bool) -> None: + """Handle wifi switch.""" + await async_service_call_action( + self._fritzbox_tools, + "WLANConfiguration", + str(self._network_num), + "SetEnable", + NewEnable="1" if turn_on else "0", + ) diff --git a/requirements_all.txt b/requirements_all.txt index b95c9a26ae6..27d81dd8eec 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -640,6 +640,9 @@ freesms==0.2.0 # homeassistant.components.fritzbox_netmonitor fritzconnection==1.4.2 +# homeassistant.components.fritz +fritzprofiles==0.6.1 + # homeassistant.components.google_translate gTTS==2.2.3 @@ -2390,6 +2393,7 @@ xboxapi==2.0.1 xknx==0.18.7 # homeassistant.components.bluesound +# homeassistant.components.fritz # homeassistant.components.rest # homeassistant.components.startca # homeassistant.components.ted5000 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 933122e1c49..ff553882fe2 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -349,6 +349,9 @@ freebox-api==0.0.10 # homeassistant.components.fritzbox_netmonitor fritzconnection==1.4.2 +# homeassistant.components.fritz +fritzprofiles==0.6.1 + # homeassistant.components.google_translate gTTS==2.2.3 @@ -1308,6 +1311,7 @@ xbox-webapi==2.0.11 xknx==0.18.7 # homeassistant.components.bluesound +# homeassistant.components.fritz # homeassistant.components.rest # homeassistant.components.startca # homeassistant.components.ted5000 diff --git a/tests/components/fritz/test_config_flow.py b/tests/components/fritz/test_config_flow.py index a68aee2edff..1551a508277 100644 --- a/tests/components/fritz/test_config_flow.py +++ b/tests/components/fritz/test_config_flow.py @@ -56,6 +56,8 @@ MOCK_SSDP_DATA = { ATTR_UPNP_UDN: "uuid:only-a-test", } +MOCK_REQUEST = b'xxxxxxxxxxxxxxxxxxxxxxxx0Dial2App2HomeAuto2BoxAdmin2Phone2NAS2FakeFritzUser\n' + @pytest.fixture() def fc_class_mock(): @@ -72,7 +74,16 @@ async def test_user(hass: HomeAssistant, fc_class_mock): side_effect=fc_class_mock, ), patch("homeassistant.components.fritz.common.FritzStatus"), patch( "homeassistant.components.fritz.async_setup_entry" - ) as mock_setup_entry: + ) as mock_setup_entry, patch( + "requests.get" + ) as mock_request_get, patch( + "requests.post" + ) as mock_request_post: + + mock_request_get.return_value.status_code = 200 + mock_request_get.return_value.content = MOCK_REQUEST + mock_request_post.return_value.status_code = 200 + mock_request_post.return_value.text = MOCK_REQUEST result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": SOURCE_USER} @@ -106,7 +117,16 @@ async def test_user_already_configured(hass: HomeAssistant, fc_class_mock): with patch( "homeassistant.components.fritz.common.FritzConnection", side_effect=fc_class_mock, - ), patch("homeassistant.components.fritz.common.FritzStatus"): + ), patch("homeassistant.components.fritz.common.FritzStatus"), patch( + "requests.get" + ) as mock_request_get, patch( + "requests.post" + ) as mock_request_post: + + mock_request_get.return_value.status_code = 200 + mock_request_get.return_value.content = MOCK_REQUEST + mock_request_post.return_value.status_code = 200 + mock_request_post.return_value.text = MOCK_REQUEST result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": SOURCE_USER} @@ -202,7 +222,16 @@ async def test_reauth_successful(hass: HomeAssistant, fc_class_mock): side_effect=fc_class_mock, ), patch("homeassistant.components.fritz.common.FritzStatus"), patch( "homeassistant.components.fritz.async_setup_entry" - ) as mock_setup_entry: + ) as mock_setup_entry, patch( + "requests.get" + ) as mock_request_get, patch( + "requests.post" + ) as mock_request_post: + + mock_request_get.return_value.status_code = 200 + mock_request_get.return_value.content = MOCK_REQUEST + mock_request_post.return_value.status_code = 200 + mock_request_post.return_value.text = MOCK_REQUEST result = await hass.config_entries.flow.async_init( DOMAIN, @@ -355,7 +384,16 @@ async def test_ssdp(hass: HomeAssistant, fc_class_mock): side_effect=fc_class_mock, ), patch("homeassistant.components.fritz.common.FritzStatus"), patch( "homeassistant.components.fritz.async_setup_entry" - ) as mock_setup_entry: + ) as mock_setup_entry, patch( + "requests.get" + ) as mock_request_get, patch( + "requests.post" + ) as mock_request_post: + + mock_request_get.return_value.status_code = 200 + mock_request_get.return_value.content = MOCK_REQUEST + mock_request_post.return_value.status_code = 200 + mock_request_post.return_value.text = MOCK_REQUEST result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": SOURCE_SSDP}, data=MOCK_SSDP_DATA @@ -411,7 +449,16 @@ async def test_import(hass: HomeAssistant, fc_class_mock): side_effect=fc_class_mock, ), patch("homeassistant.components.fritz.common.FritzStatus"), patch( "homeassistant.components.fritz.async_setup_entry" - ) as mock_setup_entry: + ) as mock_setup_entry, patch( + "requests.get" + ) as mock_request_get, patch( + "requests.post" + ) as mock_request_post: + + mock_request_get.return_value.status_code = 200 + mock_request_get.return_value.content = MOCK_REQUEST + mock_request_post.return_value.status_code = 200 + mock_request_post.return_value.text = MOCK_REQUEST result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": SOURCE_IMPORT}, data=MOCK_IMPORT_CONFIG