"""Switches for AVM Fritz!Box functions.""" import logging from typing import Any from homeassistant.components.network import async_get_source_ip from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription from homeassistant.const import EntityCategory from homeassistant.core import HomeAssistant from homeassistant.helpers import entity_registry as er from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity import Entity from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.util import slugify from .const import ( DOMAIN, SWITCH_TYPE_DEFLECTION, SWITCH_TYPE_PORTFORWARD, SWITCH_TYPE_PROFILE, SWITCH_TYPE_WIFINETWORK, MeshRoles, Platform, ) from .coordinator import FRITZ_DATA_KEY, AvmWrapper, FritzConfigEntry, FritzData from .entity import FritzBoxBaseEntity from .helpers import device_filter_out_from_trackers from .models import FritzDevice, SwitchInfo _LOGGER = logging.getLogger(__name__) # Set a sane value to avoid too many updates PARALLEL_UPDATES = 5 WIFI_STANDARD = {1: "2.4Ghz", 2: "5Ghz", 3: "5Ghz", 4: "Guest"} WIFI_BAND = { 0: {"band": "2.4Ghz"}, 1: {"band": "5Ghz"}, 3: {"band": "5Ghz High / 6Ghz"}, } def _wifi_naming( network_info: dict[str, Any], wifi_index: int, wifi_count: int ) -> str | None: """Return a friendly name for a Wi-Fi network.""" if wifi_index == 2 and wifi_count == 4: # In case of 4 Wi-Fi networks, the 2nd one is used # for internal communication between mesh devices and # should not be named like the others to avoid confusion return None if (wifi_index + 1) == wifi_count: # Last Wi-Fi network in the guest network, both bands available return "Guest" # Cast to correct type for type checker if (result := WIFI_BAND.get(wifi_index)) is not None: return f"Main {result['band']}" return None async def _get_wifi_networks_list(avm_wrapper: AvmWrapper) -> dict[int, dict[str, Any]]: """Get a list of wifi networks with friendly names.""" wifi_count = len( [ s for s in avm_wrapper.connection.services if s.startswith("WLANConfiguration") ] ) _LOGGER.debug("WiFi networks count: %s", wifi_count) networks: dict[int, dict[str, Any]] = {} for i in range(1, wifi_count + 1): network_info = await avm_wrapper.async_get_wlan_configuration(i) if (switch_name := _wifi_naming(network_info, i - 1, wifi_count)) is None: continue networks[i] = network_info networks[i]["switch_name"] = switch_name _LOGGER.debug("WiFi networks list: %s", networks) return networks async def _migrate_to_new_unique_id( hass: HomeAssistant, avm_wrapper: AvmWrapper ) -> None: """Migrate old unique ids to new unique ids.""" _LOGGER.debug("Migrating Wi-Fi switches") entity_registry = er.async_get(hass) networks = await _get_wifi_networks_list(avm_wrapper) for index, network in networks.items(): description = f"Wi-Fi {network['NewSSID']}" if ( len( [ j for j, n in networks.items() if slugify(n["NewSSID"]) == slugify(network["NewSSID"]) ] ) > 1 ): description += f" ({WIFI_STANDARD[index]})" old_unique_id = f"{avm_wrapper.unique_id}-{slugify(description)}" new_unique_id = ( f"{avm_wrapper.unique_id}-wi_fi_" f"{slugify(_wifi_naming(network, index - 1, len(networks)))}" ) entity_id = entity_registry.async_get_entity_id( Platform.SWITCH, DOMAIN, old_unique_id ) if entity_id is not None: entity_registry.async_update_entity( entity_id, new_unique_id=new_unique_id, ) _LOGGER.debug( "Migrating Wi-FI switch unique_id from [%s] to [%s]", old_unique_id, new_unique_id, ) _LOGGER.debug("Migration completed") async def _async_deflection_entities_list( avm_wrapper: AvmWrapper, device_friendly_name: str ) -> list[FritzBoxDeflectionSwitch]: """Get list of deflection entities.""" _LOGGER.debug("Setting up %s switches", SWITCH_TYPE_DEFLECTION) if not (call_deflections := avm_wrapper.data["call_deflections"]): _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION) return [] return [ FritzBoxDeflectionSwitch(avm_wrapper, device_friendly_name, cd_id) for cd_id in call_deflections ] async def _async_port_entities_list( avm_wrapper: AvmWrapper, device_friendly_name: str, local_ip: str ) -> list[FritzBoxPortSwitch]: """Get list of port forwarding entities.""" _LOGGER.debug("Setting up %s switches", SWITCH_TYPE_PORTFORWARD) entities_list: list[FritzBoxPortSwitch] = [] if not avm_wrapper.device_conn_type: _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_PORTFORWARD) return [] # Query port forwardings and setup a switch for each forward for the current device resp = await avm_wrapper.async_get_num_port_mapping(avm_wrapper.device_conn_type) if not resp: _LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_PORTFORWARD) return [] port_forwards_count: int = resp["NewPortMappingNumberOfEntries"] _LOGGER.debug( "Specific %s response: GetPortMappingNumberOfEntries=%s", SWITCH_TYPE_PORTFORWARD, port_forwards_count, ) _LOGGER.debug("IP source for %s is %s", avm_wrapper.host, local_ip) for i in range(port_forwards_count): portmap = await avm_wrapper.async_get_port_mapping( avm_wrapper.device_conn_type, 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: port_name = portmap["NewPortMappingDescription"] for entity in entities_list: if entity.port_mapping and ( port_name in entity.port_mapping["NewPortMappingDescription"] ): port_name = f"{port_name} {portmap['NewExternalPort']}" entities_list.append( FritzBoxPortSwitch( avm_wrapper, device_friendly_name, portmap, port_name, i, avm_wrapper.device_conn_type, ) ) return entities_list async def _async_wifi_entities_list( avm_wrapper: AvmWrapper, device_friendly_name: str ) -> list[FritzBoxWifiSwitch]: """Get list of wifi entities.""" _LOGGER.debug("Setting up %s switches", SWITCH_TYPE_WIFINETWORK) # # https://avm.de/fileadmin/user_upload/Global/Service/Schnittstellen/wlanconfigSCPD.pdf # networks = await _get_wifi_networks_list(avm_wrapper) return [ FritzBoxWifiSwitch(avm_wrapper, device_friendly_name, index, data) for index, data in networks.items() ] async def _async_profile_entities_list( avm_wrapper: AvmWrapper, data_fritz: FritzData, ) -> list[FritzBoxProfileSwitch]: """Add new tracker entities from the AVM device.""" _LOGGER.debug("Setting up %s switches", SWITCH_TYPE_PROFILE) new_profiles: list[FritzBoxProfileSwitch] = [] if "X_AVM-DE_HostFilter1" not in avm_wrapper.connection.services: return new_profiles for mac, device in avm_wrapper.devices.items(): if device_filter_out_from_trackers( mac, device, data_fritz.profile_switches.values() ): _LOGGER.debug( "Skipping profile switch creation for device %s", device.hostname ) continue new_profiles.append(FritzBoxProfileSwitch(avm_wrapper, device)) data_fritz.profile_switches[avm_wrapper.unique_id].add(mac) _LOGGER.debug("Creating %s profile switches", len(new_profiles)) return new_profiles async def async_all_entities_list( avm_wrapper: AvmWrapper, device_friendly_name: str, data_fritz: FritzData, local_ip: str, ) -> list[Entity]: """Get a list of all entities.""" if avm_wrapper.mesh_role == MeshRoles.SLAVE: if not avm_wrapper.mesh_wifi_uplink: return [*await _async_wifi_entities_list(avm_wrapper, device_friendly_name)] return [] return [ *await _async_deflection_entities_list(avm_wrapper, device_friendly_name), *await _async_port_entities_list(avm_wrapper, device_friendly_name, local_ip), *await _async_wifi_entities_list(avm_wrapper, device_friendly_name), *await _async_profile_entities_list(avm_wrapper, data_fritz), ] async def async_setup_entry( hass: HomeAssistant, entry: FritzConfigEntry, async_add_entities: AddConfigEntryEntitiesCallback, ) -> None: """Set up entry.""" _LOGGER.debug("Setting up switches") avm_wrapper = entry.runtime_data data_fritz = hass.data[FRITZ_DATA_KEY] _LOGGER.debug("Fritzbox services: %s", avm_wrapper.connection.services) local_ip = await async_get_source_ip(avm_wrapper.hass, target_ip=avm_wrapper.host) await _migrate_to_new_unique_id(hass, avm_wrapper) entities_list = await async_all_entities_list( avm_wrapper, entry.title, data_fritz, local_ip, ) async_add_entities(entities_list) async def async_update_avm_device() -> None: """Update the values of the AVM device.""" async_add_entities(await _async_profile_entities_list(avm_wrapper, data_fritz)) entry.async_on_unload( async_dispatcher_connect( hass, avm_wrapper.signal_device_new, async_update_avm_device ) ) class FritzBoxBaseCoordinatorSwitch(CoordinatorEntity[AvmWrapper], SwitchEntity): """Fritz switch coordinator base class.""" entity_description: SwitchEntityDescription _attr_has_entity_name = True def __init__( self, avm_wrapper: AvmWrapper, device_name: str, description: SwitchEntityDescription, ) -> None: """Init device info class.""" super().__init__(avm_wrapper) self.entity_description = description self._device_name = device_name self._attr_unique_id = f"{avm_wrapper.unique_id}-{description.key}" @property def device_info(self) -> DeviceInfo: """Return the device information.""" return DeviceInfo( configuration_url=f"http://{self.coordinator.host}", connections={(CONNECTION_NETWORK_MAC, self.coordinator.mac)}, identifiers={(DOMAIN, self.coordinator.unique_id)}, manufacturer="FRITZ!", model=self.coordinator.model, name=self._device_name, sw_version=self.coordinator.current_firmware, ) @property def data(self) -> dict[str, Any]: """Return entity data from coordinator data.""" raise NotImplementedError @property def available(self) -> bool: """Return availability based on data availability.""" return super().available and bool(self.data) async def _async_handle_turn_on_off(self, turn_on: bool) -> None: """Handle switch state change request.""" raise NotImplementedError 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) class FritzBoxBaseSwitch(FritzBoxBaseEntity, SwitchEntity): """Fritz switch base class.""" def __init__( self, avm_wrapper: AvmWrapper, device_friendly_name: str, switch_info: SwitchInfo, ) -> None: """Init Fritzbox base switch.""" super().__init__(avm_wrapper, device_friendly_name) description = switch_info["description"] self._type = switch_info["type"] self._update = switch_info["callback_update"] self._switch = switch_info["callback_switch"] self._attr_icon = switch_info["icon"] self._attr_is_on = switch_info["init_state"] self._attr_name = description self._attr_unique_id = f"{self._avm_wrapper.unique_id}-{slugify(description)}" self._attr_extra_state_attributes: dict[str, Any | None] = {} self._attr_available = True 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) -> None: """Handle switch state change request.""" await self._switch(turn_on) self._attr_is_on = turn_on class FritzBoxPortSwitch(FritzBoxBaseSwitch): """Defines a FRITZ!Box Tools PortForward switch.""" def __init__( self, avm_wrapper: AvmWrapper, device_friendly_name: str, port_mapping: dict[str, Any], port_name: str, idx: int, connection_type: str, ) -> None: """Init Fritzbox port switch.""" self.connection_type = connection_type # dict in the format as it comes from fritzconnection, # eg: {"NewRemoteHost": "0.0.0.0", "NewExternalPort": 22, ...} self.port_mapping = port_mapping self._idx = idx # needed for update routine self._attr_entity_category = EntityCategory.CONFIG switch_info = SwitchInfo( description=f"Port forward {port_name}", icon="mdi:check-network", type=SWITCH_TYPE_PORTFORWARD, callback_update=self._async_fetch_update, callback_switch=self._async_switch_on_off_executor, init_state=port_mapping["NewEnabled"], ) super().__init__(avm_wrapper, device_friendly_name, switch_info) async def _async_fetch_update(self) -> None: """Fetch updates.""" 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 not self.port_mapping: self._attr_available = False return self._attr_is_on = self.port_mapping["NewEnabled"] is True self._attr_available = True attributes_dict = { "NewInternalClient": "internal_ip", "NewInternalPort": "internal_port", "NewExternalPort": "external_port", "NewProtocol": "protocol", "NewPortMappingDescription": "description", } for key, attr in attributes_dict.items(): self._attr_extra_state_attributes[attr] = self.port_mapping[key] async def _async_switch_on_off_executor(self, turn_on: bool) -> None: self.port_mapping["NewEnabled"] = "1" if turn_on else "0" await self._avm_wrapper.async_add_port_mapping( self.connection_type, self.port_mapping ) class FritzBoxDeflectionSwitch(FritzBoxBaseCoordinatorSwitch): """Defines a FRITZ!Box Tools PortForward switch.""" _attr_entity_category = EntityCategory.CONFIG def __init__( self, avm_wrapper: AvmWrapper, device_friendly_name: str, deflection_id: int, ) -> None: """Init Fritxbox Deflection class.""" self.deflection_id = deflection_id description = SwitchEntityDescription( key=f"call_deflection_{self.deflection_id}", name=f"Call deflection {self.deflection_id}", icon="mdi:phone-forward", ) super().__init__(avm_wrapper, device_friendly_name, description) @property def data(self) -> dict[str, Any]: """Return call deflection data.""" return self.coordinator.data["call_deflections"].get(self.deflection_id, {}) @property def extra_state_attributes(self) -> dict[str, str]: """Return device attributes.""" return { "type": self.data["Type"], "number": self.data["Number"], "deflection_to_number": self.data["DeflectionToNumber"], "mode": self.data["Mode"][1:], "outgoing": self.data["Outgoing"], "phonebook_id": self.data["PhonebookID"], } @property def is_on(self) -> bool | None: """Switch status.""" return self.data.get("Enable") == "1" async def _async_handle_turn_on_off(self, turn_on: bool) -> None: """Handle deflection switch.""" await self.coordinator.async_set_deflection_enable(self.deflection_id, turn_on) deflection = self.coordinator.data["call_deflections"][self.deflection_id] deflection["Enable"] = "1" if turn_on else "0" self.async_write_ha_state() class FritzBoxProfileSwitch(FritzBoxBaseCoordinatorSwitch): """Defines a FRITZ!Box Tools DeviceProfile switch.""" _attr_translation_key = "internet_access" _attr_entity_category = EntityCategory.CONFIG def __init__(self, avm_wrapper: AvmWrapper, device: FritzDevice) -> None: """Init Fritz profile.""" self._mac = device.mac_address description = SwitchEntityDescription( key=f"{self._mac}_internet_access", ) super().__init__(avm_wrapper, device.hostname, description) self._attr_unique_id = f"{self._mac}_internet_access" @property def device_info(self) -> DeviceInfo: """Return the device information.""" return DeviceInfo( connections={(CONNECTION_NETWORK_MAC, self._mac)}, ) @property def _device(self) -> FritzDevice: """Return the device for this profile switch.""" return self.coordinator.devices[self._mac] @property def available(self) -> bool: """Return availability of the switch.""" if self._device.wan_access is None: return False return self.coordinator.last_update_success @property def is_on(self) -> bool | None: """Switch status.""" return self._device.wan_access async def _async_handle_turn_on_off(self, turn_on: bool) -> None: """Handle switch state change request.""" await self.coordinator.async_set_allow_wan_access( self._device.ip_address, turn_on ) self._device.wan_access = turn_on self.async_write_ha_state() class FritzBoxWifiSwitch(FritzBoxBaseSwitch): """Defines a FRITZ!Box Tools Wifi switch.""" def __init__( self, avm_wrapper: AvmWrapper, device_friendly_name: str, network_num: int, network_data: dict[str, Any], ) -> None: """Init Fritz Wifi switch.""" self._wifi_info = network_data self._attr_entity_category = EntityCategory.CONFIG self._attr_entity_registry_enabled_default = ( avm_wrapper.mesh_role is not MeshRoles.SLAVE ) self._network_num = network_num description = f"Wi-Fi {network_data['switch_name']}" self._attr_translation_key = slugify(description) switch_info = SwitchInfo( description=description, icon="mdi:wifi", type=SWITCH_TYPE_WIFINETWORK, callback_update=self._async_fetch_update, callback_switch=self._async_switch_on_off_executor, init_state=network_data["NewEnable"], ) super().__init__(avm_wrapper, device_friendly_name, switch_info) async def _async_fetch_update(self) -> None: """Fetch updates.""" 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 not wifi_info: self._attr_available = False return self._attr_is_on = wifi_info["NewEnable"] is True self._attr_available = True std = wifi_info["NewStandard"] self._attr_extra_state_attributes["standard"] = std or None self._attr_extra_state_attributes["bssid"] = wifi_info["NewBSSID"] self._attr_extra_state_attributes["mac_address_control"] = wifi_info[ "NewMACAddressControlEnabled" ] self._wifi_info = wifi_info async def _async_switch_on_off_executor(self, turn_on: bool) -> None: """Handle wifi switch.""" self._wifi_info["NewEnable"] = turn_on await self._avm_wrapper.async_set_wlan_configuration(self._network_num, turn_on)