evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Replace fritz profile switches by per device parental control switches (#52721)

Browse source 
* removes old profile switches and add new switches based on new method

* use Ellipsis instead of pass

* refactor async_add_profile_switches

* - add forgotten update_ha_state
- add notimplemtederror for devicebase

* Update homeassistant/components/fritz/common.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Update homeassistant/components/fritz/common.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* comments

* fix for devices that were not connected

* Update homeassistant/components/fritz/common.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Update homeassistant/components/fritz/switch.py

Co-authored-by: J. Nick Koston <nick@koston.org>

Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Aaron David Schneider 2021-07-16 13:38:37 +02:00 • committed by GitHub
parent 7d52c30e36
commit 3d3db4b044
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 212 additions and 181 deletions
Download patch file Download diff file Expand all files Collapse all files

102
homeassistant/components/fritz/common.py
View file

@ -15,7 +15,6 @@ 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,
@ -24,12 +23,13 @@ from homeassistant.components.device_tracker.const import (
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.dispatcher import dispatcher_send
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect, dispatcher_send
from homeassistant.helpers.entity import DeviceInfo, Entity
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.util import dt as dt_util
from .const import (
DEFAULT_DEVICE_NAME,
DEFAULT_HOST,
DEFAULT_PORT,
DEFAULT_USERNAME,
@ -81,12 +81,11 @@ class FritzBoxTools:
) -> None:
"""Initialize FritzboxTools class."""
self._cancel_scan: CALLBACK_TYPE | None = None
self._devices: dict[str, Any] = {}
self._devices: dict[str, FritzDevice] = {}
self._options: MappingProxyType[str, Any] | None = None
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
@ -123,13 +122,6 @@ 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)
@ -260,10 +252,92 @@ class FritzData:
"""Storage class for platform global data."""
tracked: dict = field(default_factory=dict)
profile_switches: dict = field(default_factory=dict)
class FritzDeviceBase(Entity):
"""Entity base class for a device connected to a FRITZ!Box router."""
def __init__(self, router: FritzBoxTools, device: FritzDevice) -> None:
"""Initialize a FRITZ!Box device."""
self._router = router
self._mac: str = device.mac_address
self._name: str = device.hostname or DEFAULT_DEVICE_NAME
@property
def name(self) -> str:
"""Return device name."""
return self._name
@property
def ip_address(self) -> str | None:
"""Return the primary ip address of the device."""
if self._mac:
device: FritzDevice = self._router.devices[self._mac]
return device.ip_address
return None
@property
def mac_address(self) -> str:
"""Return the mac address of the device."""
return self._mac
@property
def hostname(self) -> str | None:
"""Return hostname of the device."""
if self._mac:
device: FritzDevice = self._router.devices[self._mac]
return device.hostname
return None
@property
def device_info(self) -> DeviceInfo:
"""Return the device information."""
return {
"connections": {(CONNECTION_NETWORK_MAC, self._mac)},
"identifiers": {(DOMAIN, self._mac)},
"default_name": self.name,
"default_manufacturer": "AVM",
"default_model": "FRITZ!Box Tracked device",
"via_device": (
DOMAIN,
self._router.unique_id,
),
}
@property
def should_poll(self) -> bool:
"""No polling needed."""
return False
@property
def entity_registry_enabled_default(self) -> bool:
"""Return if the entity should be enabled when first added to the entity registry."""
return False
async def async_process_update(self) -> None:
"""Update device."""
raise NotImplementedError()
async def async_on_demand_update(self) -> None:
"""Update state."""
await self.async_process_update()
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Register state update callback."""
await self.async_process_update()
self.async_on_remove(
async_dispatcher_connect(
self.hass,
self._router.signal_device_update,
self.async_on_demand_update,
)
)
class FritzDevice:
"""FritzScanner device."""
"""Representation of a device connected to the FRITZ!Box."""
def __init__(self, mac: str, name: str) -> None:
"""Initialize device info."""
@ -292,7 +366,7 @@ class FritzDevice:
if dev_home:
self._last_activity = utc_point_in_time
self._ip_address = dev_info.ip_address if self._connected else None
self._ip_address = dev_info.ip_address
@property
def is_connected(self) -> bool:

4
homeassistant/components/fritz/const.py
View file

@ -19,11 +19,7 @@ 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"

97
homeassistant/components/fritz/device_tracker.py
View file

@ -16,14 +16,12 @@ from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType
from .common import Device, FritzBoxTools, FritzData, FritzDevice
from .const import DATA_FRITZ, DEFAULT_DEVICE_NAME, DOMAIN
from .common import FritzBoxTools, FritzData, FritzDevice, FritzDeviceBase
from .const import DATA_FRITZ, DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -93,7 +91,7 @@ def _async_add_entities(
) -> None:
"""Add new tracker entities from the router."""
def _is_tracked(mac: str, device: Device) -> bool:
def _is_tracked(mac: str) -> bool:
for tracked in data_fritz.tracked.values():
if mac in tracked:
return True
@ -105,7 +103,7 @@ def _async_add_entities(
data_fritz.tracked[router.unique_id] = set()
for mac, device in router.devices.items():
if device.ip_address == "" or _is_tracked(mac, device):
if device.ip_address == "" or _is_tracked(mac):
continue
new_tracked.append(FritzBoxTracker(router, device))
@ -115,14 +113,12 @@ def _async_add_entities(
async_add_entities(new_tracked)
class FritzBoxTracker(ScannerEntity):
class FritzBoxTracker(FritzDeviceBase, ScannerEntity):
"""This class queries a FRITZ!Box router."""
def __init__(self, router: FritzBoxTools, device: FritzDevice) -> None:
"""Initialize a FRITZ!Box device."""
self._router = router
self._mac: str = device.mac_address
self._name: str = device.hostname or DEFAULT_DEVICE_NAME
super().__init__(router, device)
self._last_activity: datetime.datetime | None = device.last_activity
self._active = False
@ -131,59 +127,10 @@ class FritzBoxTracker(ScannerEntity):
"""Return device status."""
return self._active
@property
def name(self) -> str:
"""Return device name."""
return self._name
@property
def unique_id(self) -> str:
"""Return device unique id."""
return self._mac
@property
def ip_address(self) -> str | None:
"""Return the primary ip address of the device."""
if self._mac:
return self._router.devices[self._mac].ip_address # type: ignore[no-any-return]
return None
@property
def mac_address(self) -> str:
"""Return the mac address of the device."""
return self._mac
@property
def hostname(self) -> str | None:
"""Return hostname of the device."""
if self._mac:
return self._router.devices[self._mac].hostname # type: ignore[no-any-return]
return None
@property
def source_type(self) -> str:
"""Return tracker source type."""
return SOURCE_TYPE_ROUTER
@property
def device_info(self) -> DeviceInfo:
"""Return the device information."""
return {
"connections": {(CONNECTION_NETWORK_MAC, self._mac)},
"identifiers": {(DOMAIN, self.unique_id)},
"default_name": self.name,
"default_manufacturer": "AVM",
"default_model": "FRITZ!Box Tracked device",
"via_device": (
DOMAIN,
self._router.unique_id,
),
}
@property
def should_poll(self) -> bool:
"""No polling needed."""
return False
return f"{self._mac}_tracker"
@property
def icon(self) -> str:
@ -192,11 +139,6 @@ class FritzBoxTracker(ScannerEntity):
return "mdi:lan-connect"
return "mdi:lan-disconnect"
@property
def entity_registry_enabled_default(self) -> bool:
"""Return if the entity should be enabled when first added to the entity registry."""
return False
@property
def extra_state_attributes(self) -> dict[str, str]:
"""Return the attributes."""
@ -207,8 +149,12 @@ class FritzBoxTracker(ScannerEntity):
)
return attrs
@callback
def async_process_update(self) -> None:
@property
def source_type(self) -> str:
"""Return tracker source type."""
return SOURCE_TYPE_ROUTER
async def async_process_update(self) -> None:
"""Update device."""
if not self._mac:
return
@ -216,20 +162,3 @@ class FritzBoxTracker(ScannerEntity):
device = self._router.devices[self._mac]
self._active = device.is_connected
self._last_activity = device.last_activity
@callback
def async_on_demand_update(self) -> None:
"""Update state."""
self.async_process_update()
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Register state update callback."""
self.async_process_update()
self.async_on_remove(
async_dispatcher_connect(
self.hass,
self._router.signal_device_update,
self.async_on_demand_update,
)
)

1
homeassistant/components/fritz/manifest.json
View file

@ -4,7 +4,6 @@
"documentation": "https://www.home-assistant.io/integrations/fritz",
"requirements": [
"fritzconnection==1.4.2",
"fritzprofiles==0.6.1",
"xmltodict==0.12.0"
],
"dependencies": ["network"],

183
homeassistant/components/fritz/switch.py
View file

@ -17,18 +17,24 @@ import xmltodict
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
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 .common import (
FritzBoxBaseEntity,
FritzBoxTools,
FritzData,
FritzDevice,
FritzDeviceBase,
SwitchInfo,
)
from .const import (
DATA_FRITZ,
DOMAIN,
SWITCH_PROFILE_STATUS_OFF,
SWITCH_PROFILE_STATUS_ON,
SWITCH_TYPE_DEFLECTION,
SWITCH_TYPE_DEVICEPROFILE,
SWITCH_TYPE_PORTFORWARD,
SWITCH_TYPE_WIFINETWORK,
)
@ -225,21 +231,6 @@ def port_entities_list(
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]:
@ -267,15 +258,45 @@ def wifi_entities_list(
]
def profile_entities_list(
router: FritzBoxTools, data_fritz: FritzData
) -> list[FritzBoxProfileSwitch]:
"""Add new tracker entities from the router."""
def _is_tracked(mac: str) -> bool:
for tracked in data_fritz.profile_switches.values():
if mac in tracked:
return True
return False
new_profiles: list[FritzBoxProfileSwitch] = []
if "X_AVM-DE_HostFilter1" not in router.connection.services:
return new_profiles
if router.unique_id not in data_fritz.profile_switches:
data_fritz.profile_switches[router.unique_id] = set()
for mac, device in router.devices.items():
if device.ip_address == "" or _is_tracked(mac):
continue
new_profiles.append(FritzBoxProfileSwitch(router, device))
data_fritz.profile_switches[router.unique_id].add(mac)
return new_profiles
def all_entities_list(
fritzbox_tools: FritzBoxTools, device_friendly_name: str
fritzbox_tools: FritzBoxTools, device_friendly_name: str, data_fritz: FritzData
) -> 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),
*profile_entities_list(fritzbox_tools, data_fritz),
]
@ -285,14 +306,25 @@ async def async_setup_entry(
"""Set up entry."""
_LOGGER.debug("Setting up switches")
fritzbox_tools: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
data_fritz: FritzData = hass.data[DATA_FRITZ]
_LOGGER.debug("Fritzbox services: %s", fritzbox_tools.connection.services)
entities_list = await hass.async_add_executor_job(
all_entities_list, fritzbox_tools, entry.title
all_entities_list, fritzbox_tools, entry.title, data_fritz
)
async_add_entities(entities_list)
@callback
def update_router() -> None:
"""Update the values of the router."""
async_add_entities(profile_entities_list(fritzbox_tools, data_fritz))
entry.async_on_unload(
async_dispatcher_connect(hass, fritzbox_tools.signal_device_new, update_router)
)
class FritzBoxBaseSwitch(FritzBoxBaseEntity):
"""Fritz switch base class."""
@ -522,60 +554,67 @@ class FritzBoxDeflectionSwitch(FritzBoxBaseSwitch, SwitchEntity):
)
class FritzBoxProfileSwitch(FritzBoxBaseSwitch, SwitchEntity):
class FritzBoxProfileSwitch(FritzDeviceBase, SwitchEntity):
"""Defines a FRITZ!Box Tools DeviceProfile switch."""
def __init__(
self, fritzbox_tools: FritzBoxTools, device_friendly_name: str, profile: str
) -> None:
def __init__(self, fritzbox_tools: FritzBoxTools, device: FritzDevice) -> 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
)
super().__init__(fritzbox_tools, device)
self._attr_is_on: bool = False
@property
def entity_registry_enabled_default(self) -> bool:
"""Return if the entity should be enabled when first added to the entity registry."""
return False
def unique_id(self) -> str:
"""Return device unique id."""
return f"{self._mac}_switch"
@property
def icon(self) -> str:
"""Return device icon."""
return "mdi:router-wireless-settings"
async def async_process_update(self) -> None:
"""Update device."""
if not self._mac or not self.ip_address:
return
wan_disable_info = await async_service_call_action(
self._router,
"X_AVM-DE_HostFilter",
"1",
"GetWANAccessByIP",
NewIPv4Address=self.ip_address,
)
if wan_disable_info is None:
return
self._attr_is_on = not wan_disable_info["NewDisallow"]
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._async_switch_on_off(turn_on)
self._attr_is_on = 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._router,
"X_AVM-DE_HostFilter",
"1",
"DisallowWANAccessByIP",
NewIPv4Address=self.ip_address,
NewDisallow="0" if turn_on else "1",
)
class FritzBoxWifiSwitch(FritzBoxBaseSwitch, SwitchEntity):

3
requirements_all.txt
View file

@ -639,9 +639,6 @@ freesms==0.2.0
# homeassistant.components.fritzbox_callmonitor
fritzconnection==1.4.2
# homeassistant.components.fritz
fritzprofiles==0.6.1
# homeassistant.components.google_translate
gTTS==2.2.3

3
requirements_test_all.txt
View file

@ -348,9 +348,6 @@ freebox-api==0.0.10
# homeassistant.components.fritzbox_callmonitor
fritzconnection==1.4.2
# homeassistant.components.fritz
fritzprofiles==0.6.1
# homeassistant.components.google_translate
gTTS==2.2.3