Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fritz cleanup: part1 (naming) (#63535)

pull/63542/head
Simone Chemelli 2022-01-06 12:15:40 +01:00 committed by GitHub
parent 53d86d4317
commit 60b6871b46
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 152 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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")
fritz_tools = FritzBoxTools(
avm_device = FritzBoxTools(
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 fritz_tools.async_setup(entry.options)
await avm_device.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] = fritz_tools
hass.data[DOMAIN][entry.entry_id] = avm_device
if DATA_FRITZ not in hass.data:
hass.data[DATA_FRITZ] = FritzData()
entry.async_on_unload(entry.add_update_listener(update_listener))
await fritz_tools.async_config_entry_first_refresh()
await avm_device.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."""
fritzbox: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
avm_device: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
fritz_data = hass.data[DATA_FRITZ]
fritz_data.tracked.pop(fritzbox.unique_id)
fritz_data.tracked.pop(avm_device.unique_id)
if not bool(fritz_data.tracked):
hass.data.pop(DATA_FRITZ)

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

@ -55,12 +55,12 @@ async def async_setup_entry(
) -> None:
"""Set up entry."""
_LOGGER.debug("Setting up FRITZ!Box binary sensors")
fritzbox_tools: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
avm_device: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
entities = [
FritzBoxBinarySensor(fritzbox_tools, entry.title, description)
FritzBoxBinarySensor(avm_device, entry.title, description)
for description in SENSOR_TYPES
if (description.exclude_mesh_role != fritzbox_tools.mesh_role)
if (description.exclude_mesh_role != avm_device.mesh_role)
]
async_add_entities(entities, True)
@ -71,27 +71,27 @@ class FritzBoxBinarySensor(FritzBoxBaseEntity, BinarySensorEntity):
def __init__(
self,
fritzbox_tools: FritzBoxTools,
avm_device: FritzBoxTools,
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"{fritzbox_tools.unique_id}-{description.key}"
super().__init__(fritzbox_tools, device_friendly_name)
self._attr_unique_id = f"{avm_device.unique_id}-{description.key}"
super().__init__(avm_device, 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._fritzbox_tools.update_available
self._attr_is_on = self._avm_device.update_available
self._attr_extra_state_attributes = {
"installed_version": self._fritzbox_tools.current_firmware,
"latest_available_version": self._fritzbox_tools.latest_firmware,
"installed_version": self._avm_device.current_firmware,
"latest_available_version": self._avm_device.latest_firmware,
}
if self.entity_description.key == "is_connected":
self._attr_is_on = bool(self._fritzbox_tools.fritz_status.is_connected)
self._attr_is_on = bool(self._avm_device.fritz_status.is_connected)
elif self.entity_description.key == "is_linked":
self._attr_is_on = bool(self._fritzbox_tools.fritz_status.is_linked)
self._attr_is_on = bool(self._avm_device.fritz_status.is_linked)

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

@ -42,21 +42,21 @@ BUTTONS: Final = [
name="Firmware Update",
device_class=ButtonDeviceClass.UPDATE,
entity_category=ENTITY_CATEGORY_CONFIG,
press_action=lambda router: router.async_trigger_firmware_update(),
press_action=lambda avm_device: avm_device.async_trigger_firmware_update(),
),
FritzButtonDescription(
key="reboot",
name="Reboot",
device_class=ButtonDeviceClass.RESTART,
entity_category=ENTITY_CATEGORY_CONFIG,
press_action=lambda router: router.async_trigger_reboot(),
press_action=lambda avm_device: avm_device.async_trigger_reboot(),
),
FritzButtonDescription(
key="reconnect",
name="Reconnect",
device_class=ButtonDeviceClass.RESTART,
entity_category=ENTITY_CATEGORY_CONFIG,
press_action=lambda router: router.async_trigger_reconnect(),
press_action=lambda avm_device: avm_device.async_trigger_reconnect(),
),
]
@ -68,9 +68,11 @@ async def async_setup_entry(
) -> None:
"""Set buttons for device."""
_LOGGER.debug("Setting up buttons")
router: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
avm_device: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
async_add_entities([FritzButton(router, entry.title, button) for button in BUTTONS])
async_add_entities(
[FritzButton(avm_device, entry.title, button) for button in BUTTONS]
)
class FritzButton(ButtonEntity):
@ -80,21 +82,21 @@ class FritzButton(ButtonEntity):
def __init__(
self,
router: FritzBoxTools,
avm_device: FritzBoxTools,
device_friendly_name: str,
description: FritzButtonDescription,
) -> None:
"""Initialize Fritz!Box button."""
self.entity_description = description
self.router = router
self.avm_device = avm_device
self._attr_name = f"{device_friendly_name} {description.name}"
self._attr_unique_id = f"{self.router.unique_id}-{description.key}"
self._attr_unique_id = f"{self.avm_device.unique_id}-{description.key}"
self._attr_device_info = DeviceInfo(
connections={(CONNECTION_NETWORK_MAC, router.mac)}
connections={(CONNECTION_NETWORK_MAC, avm_device.mac)}
)
async def async_press(self) -> None:
"""Triggers Fritz!Box service."""
await self.entity_description.press_action(self.router)
await self.entity_description.press_action(self.avm_device)

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

@ -287,7 +287,7 @@ class FritzBoxTools(update_coordinator.DataUpdateCoordinator):
def scan_devices(self, now: datetime | None = None) -> None:
"""Scan for new devices and return a list of found device ids."""
_LOGGER.debug("Checking host info for FRITZ!Box router %s", self.host)
_LOGGER.debug("Checking host info for FRITZ!Box device %s", self.host)
self._update_available, self._latest_firmware = self._update_device_info()
try:
@ -296,7 +296,7 @@ class FritzBoxTools(update_coordinator.DataUpdateCoordinator):
self.mesh_role = MeshRoles.SLAVE
return
_LOGGER.debug("Checking devices for FRITZ!Box router %s", self.host)
_LOGGER.debug("Checking devices for FRITZ!Box device %s", self.host)
_default_consider_home = DEFAULT_CONSIDER_HOME.total_seconds()
if self._options:
consider_home = self._options.get(
@ -400,7 +400,7 @@ class FritzBoxTools(update_coordinator.DataUpdateCoordinator):
self, service_call: ServiceCall, config_entry: ConfigEntry
) -> None:
"""Define FRITZ!Box services."""
_LOGGER.debug("FRITZ!Box router: %s", service_call.service)
_LOGGER.debug("FRITZ!Box service: %s", service_call.service)
if not self.connection:
raise HomeAssistantError("Unable to establish a connection")
@ -505,12 +505,12 @@ class FritzData:
class FritzDeviceBase(update_coordinator.CoordinatorEntity):
"""Entity base class for a device connected to a FRITZ!Box router."""
"""Entity base class for a device connected to a FRITZ!Box device."""
def __init__(self, router: FritzBoxTools, device: FritzDevice) -> None:
def __init__(self, avm_device: FritzBoxTools, device: FritzDevice) -> None:
"""Initialize a FRITZ!Box device."""
super().__init__(router)
self._router = router
super().__init__(avm_device)
self._avm_device = avm_device
self._mac: str = device.mac_address
self._name: str = device.hostname or DEFAULT_DEVICE_NAME
@ -523,7 +523,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._router.devices[self._mac].ip_address
return self._avm_device.devices[self._mac].ip_address
return None
@property
@ -535,7 +535,7 @@ class FritzDeviceBase(update_coordinator.CoordinatorEntity):
def hostname(self) -> str | None:
"""Return hostname of the device."""
if self._mac:
return self._router.devices[self._mac].hostname
return self._avm_device.devices[self._mac].hostname
return None
@property
@ -653,25 +653,25 @@ class SwitchInfo(TypedDict):
class FritzBoxBaseEntity:
"""Fritz host entity base class."""
def __init__(self, fritzbox_tools: FritzBoxTools, device_name: str) -> None:
def __init__(self, avm_device: FritzBoxTools, device_name: str) -> None:
"""Init device info class."""
self._fritzbox_tools = fritzbox_tools
self._avm_device = avm_device
self._device_name = device_name
@property
def mac_address(self) -> str:
"""Return the mac address of the main device."""
return self._fritzbox_tools.mac
return self._avm_device.mac
@property
def device_info(self) -> DeviceInfo:
"""Return the device information."""
return DeviceInfo(
configuration_url=f"http://{self._fritzbox_tools.host}",
configuration_url=f"http://{self._avm_device.host}",
connections={(CONNECTION_NETWORK_MAC, self.mac_address)},
identifiers={(DOMAIN, self._fritzbox_tools.unique_id)},
identifiers={(DOMAIN, self._avm_device.unique_id)},
manufacturer="AVM",
model=self._fritzbox_tools.model,
model=self._avm_device.model,
name=self._device_name,
sw_version=self._fritzbox_tools.current_firmware,
sw_version=self._avm_device.current_firmware,
)

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

@ -51,7 +51,7 @@ class FritzBoxToolsFlowHandler(ConfigFlow, domain=DOMAIN):
self._password: str
self._port: int | None = None
self._username: str
self.fritz_tools: FritzBoxTools
self.avm_device: FritzBoxTools
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.fritz_tools = FritzBoxTools(
self.avm_device = FritzBoxTools(
hass=self.hass,
host=self._host,
port=self._port,
@ -68,7 +68,7 @@ class FritzBoxToolsFlowHandler(ConfigFlow, domain=DOMAIN):
)
try:
await self.fritz_tools.async_setup()
await self.avm_device.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.fritz_tools.host,
CONF_PASSWORD: self.fritz_tools.password,
CONF_PORT: self.fritz_tools.port,
CONF_USERNAME: self.fritz_tools.username,
CONF_HOST: self.avm_device.host,
CONF_PASSWORD: self.avm_device.password,
CONF_PORT: self.avm_device.port,
CONF_USERNAME: self.avm_device.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.fritz_tools.model
self._name = self.avm_device.model
if await self.async_check_configured_entry():
error = "already_configured"

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

@ -1,4 +1,4 @@
"""Support for FRITZ!Box routers."""
"""Support for FRITZ!Box devices."""
from __future__ import annotations
import datetime
@ -74,56 +74,56 @@ async def async_setup_entry(
) -> None:
"""Set up device tracker for FRITZ!Box component."""
_LOGGER.debug("Starting FRITZ!Box device tracker")
router: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
avm_device: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
data_fritz: FritzData = hass.data[DATA_FRITZ]
@callback
def update_router() -> None:
"""Update the values of the router."""
_async_add_entities(router, async_add_entities, data_fritz)
def update_avm_device() -> None:
"""Update the values of AVM device."""
_async_add_entities(avm_device, async_add_entities, data_fritz)
entry.async_on_unload(
async_dispatcher_connect(hass, router.signal_device_new, update_router)
async_dispatcher_connect(hass, avm_device.signal_device_new, update_avm_device)
)
update_router()
update_avm_device()
@callback
def _async_add_entities(
router: FritzBoxTools,
avm_device: FritzBoxTools,
async_add_entities: AddEntitiesCallback,
data_fritz: FritzData,
) -> None:
"""Add new tracker entities from the router."""
"""Add new tracker entities from the AVM device."""
new_tracked = []
if router.unique_id not in data_fritz.tracked:
data_fritz.tracked[router.unique_id] = set()
if avm_device.unique_id not in data_fritz.tracked:
data_fritz.tracked[avm_device.unique_id] = set()
for mac, device in router.devices.items():
for mac, device in avm_device.devices.items():
if device_filter_out_from_trackers(mac, device, data_fritz.tracked.values()):
continue
new_tracked.append(FritzBoxTracker(router, device))
data_fritz.tracked[router.unique_id].add(mac)
new_tracked.append(FritzBoxTracker(avm_device, device))
data_fritz.tracked[avm_device.unique_id].add(mac)
if new_tracked:
async_add_entities(new_tracked)
class FritzBoxTracker(FritzDeviceBase, ScannerEntity):
"""This class queries a FRITZ!Box router."""
"""This class queries a FRITZ!Box device."""
def __init__(self, router: FritzBoxTools, device: FritzDevice) -> None:
def __init__(self, avm_device: FritzBoxTools, device: FritzDevice) -> None:
"""Initialize a FRITZ!Box device."""
super().__init__(router, device)
super().__init__(avm_device, device)
self._last_activity: datetime.datetime | None = device.last_activity
@property
def is_connected(self) -> bool:
"""Return device status."""
return self._router.devices[self._mac].is_connected
return self._avm_device.devices[self._mac].is_connected
@property
def unique_id(self) -> str:
@ -146,7 +146,7 @@ class FritzBoxTracker(FritzDeviceBase, ScannerEntity):
def extra_state_attributes(self) -> dict[str, str]:
"""Return the attributes."""
attrs: dict[str, str] = {}
device = self._router.devices[self._mac]
device = self._avm_device.devices[self._mac]
self._last_activity = device.last_activity
if self._last_activity is not None:
attrs["last_time_reachable"] = self._last_activity.isoformat(

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

@ -281,12 +281,12 @@ async def async_setup_entry(
) -> None:
"""Set up entry."""
_LOGGER.debug("Setting up FRITZ!Box sensors")
fritzbox_tools: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
avm_device: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
dsl: bool = False
try:
dslinterface = await hass.async_add_executor_job(
fritzbox_tools.connection.call_action,
avm_device.connection.call_action,
"WANDSLInterfaceConfig:1",
"GetInfo",
)
@ -300,10 +300,10 @@ async def async_setup_entry(
pass
entities = [
FritzBoxSensor(fritzbox_tools, entry.title, description)
FritzBoxSensor(avm_device, entry.title, description)
for description in SENSOR_TYPES
if (dsl or description.connection_type != DSL_CONNECTION)
and description.exclude_mesh_role != fritzbox_tools.mesh_role
and description.exclude_mesh_role != avm_device.mesh_role
]
async_add_entities(entities, True)
@ -316,7 +316,7 @@ class FritzBoxSensor(FritzBoxBaseEntity, SensorEntity):
def __init__(
self,
fritzbox_tools: FritzBoxTools,
avm_device: FritzBoxTools,
device_friendly_name: str,
description: FritzSensorEntityDescription,
) -> None:
@ -325,15 +325,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"{fritzbox_tools.unique_id}-{description.key}"
super().__init__(fritzbox_tools, device_friendly_name)
self._attr_unique_id = f"{avm_device.unique_id}-{description.key}"
super().__init__(avm_device, device_friendly_name)
def update(self) -> None:
"""Update data."""
_LOGGER.debug("Updating FRITZ!Box sensors")
try:
status: FritzStatus = self._fritzbox_tools.fritz_status
status: FritzStatus = self._avm_device.fritz_status
self._attr_available = True
except FritzConnectionException:
_LOGGER.error("Error getting the state from the FRITZ!Box", exc_info=True)

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

@ -31,7 +31,7 @@ async def async_setup_services(hass: HomeAssistant) -> None:
"""Call correct Fritz service."""
if not (
fritzbox_entry_ids := await _async_get_configured_fritz_tools(
fritzbox_entry_ids := await _async_get_configured_avm_device(
hass, service_call
)
):
@ -41,9 +41,9 @@ async def async_setup_services(hass: HomeAssistant) -> None:
for entry_id in fritzbox_entry_ids:
_LOGGER.debug("Executing service %s", service_call.service)
fritz_tools: FritzBoxTools = hass.data[DOMAIN][entry_id]
avm_device: FritzBoxTools = hass.data[DOMAIN][entry_id]
if config_entry := hass.config_entries.async_get_entry(entry_id):
await fritz_tools.service_fritzbox(service_call, config_entry)
await avm_device.service_fritzbox(service_call, config_entry)
else:
_LOGGER.error(
"Executing service %s failed, no config entry found",
@ -54,7 +54,7 @@ async def async_setup_services(hass: HomeAssistant) -> None:
hass.services.async_register(DOMAIN, service, async_call_fritz_service)
async def _async_get_configured_fritz_tools(
async def _async_get_configured_avm_device(
hass: HomeAssistant, service_call: ServiceCall
) -> list:
"""Get FritzBoxTools class from config entry."""

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

@ -48,17 +48,17 @@ _LOGGER = logging.getLogger(__name__)
async def async_service_call_action(
fritzbox_tools: FritzBoxTools,
avm_device: 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(
return await avm_device.hass.async_add_executor_job(
partial(
service_call_action,
fritzbox_tools,
avm_device,
service_name,
service_suffix,
action_name,
@ -68,7 +68,7 @@ async def async_service_call_action(
def service_call_action(
fritzbox_tools: FritzBoxTools,
avm_device: FritzBoxTools,
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 fritzbox_tools.connection.services:
if f"{service_name}{service_suffix}" not in avm_device.connection.services:
return None
try:
return fritzbox_tools.connection.call_action( # type: ignore[no-any-return]
return avm_device.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(
fritzbox_tools: FritzBoxTools, service_name: str
avm_device: FritzBoxTools, service_name: str
) -> list[OrderedDict[Any, Any]] | None:
"""Get deflection switch info."""
deflection_list = service_call_action(
fritzbox_tools,
avm_device,
service_name,
"1",
"GetDeflections",
@ -134,7 +134,7 @@ def get_deflections(
def deflection_entities_list(
fritzbox_tools: FritzBoxTools, device_friendly_name: str
avm_device: FritzBoxTools, 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(
fritzbox_tools, service_name, "1", "GetNumberOfDeflections"
avm_device, service_name, "1", "GetNumberOfDeflections"
)
if not deflections_response:
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION)
@ -158,20 +158,18 @@ def deflection_entities_list(
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION)
return []
deflection_list = get_deflections(fritzbox_tools, service_name)
deflection_list = get_deflections(avm_device, service_name)
if deflection_list is None:
return []
return [
FritzBoxDeflectionSwitch(
fritzbox_tools, device_friendly_name, dict_of_deflection
)
FritzBoxDeflectionSwitch(avm_device, device_friendly_name, dict_of_deflection)
for dict_of_deflection in deflection_list
]
def port_entities_list(
fritzbox_tools: FritzBoxTools, device_friendly_name: str, local_ip: str
avm_device: FritzBoxTools, device_friendly_name: str, local_ip: str
) -> list[FritzBoxPortSwitch]:
"""Get list of port forwarding entities."""
@ -179,7 +177,7 @@ def port_entities_list(
entities_list: list[FritzBoxPortSwitch] = []
service_name = "Layer3Forwarding"
connection_type = service_call_action(
fritzbox_tools, service_name, "1", "GetDefaultConnectionService"
avm_device, service_name, "1", "GetDefaultConnectionService"
)
if not connection_type:
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_PORTFORWARD)
@ -190,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(
fritzbox_tools, con_type, "1", "GetPortMappingNumberOfEntries"
avm_device, con_type, "1", "GetPortMappingNumberOfEntries"
)
if not resp:
_LOGGER.debug("The FRITZ!Box has no %s options", SWITCH_TYPE_DEFLECTION)
@ -204,12 +202,12 @@ def port_entities_list(
port_forwards_count,
)
_LOGGER.debug("IP source for %s is %s", fritzbox_tools.host, local_ip)
_LOGGER.debug("IP source for %s is %s", avm_device.host, local_ip)
for i in range(port_forwards_count):
portmap = service_call_action(
fritzbox_tools,
avm_device,
con_type,
"1",
"GetGenericPortMappingEntry",
@ -236,7 +234,7 @@ def port_entities_list(
port_name = f"{port_name} {portmap['NewExternalPort']}"
entities_list.append(
FritzBoxPortSwitch(
fritzbox_tools,
avm_device,
device_friendly_name,
portmap,
port_name,
@ -249,21 +247,21 @@ def port_entities_list(
def wifi_entities_list(
fritzbox_tools: FritzBoxTools, device_friendly_name: str
avm_device: FritzBoxTools, 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 fritzbox_tools.model == "FRITZ!Box 7390":
if avm_device.model == "FRITZ!Box 7390":
std_table = {"n": "5Ghz"}
networks: dict = {}
for i in range(4):
if not ("WLANConfiguration" + str(i)) in fritzbox_tools.connection.services:
if not ("WLANConfiguration" + str(i)) in avm_device.connection.services:
continue
network_info = service_call_action(
fritzbox_tools, "WLANConfiguration", str(i), "GetInfo"
avm_device, "WLANConfiguration", str(i), "GetInfo"
)
if network_info:
ssid = network_info["NewSSID"]
@ -281,53 +279,53 @@ def wifi_entities_list(
_LOGGER.debug("SSID normalized: <%s>", networks[i])
return [
FritzBoxWifiSwitch(fritzbox_tools, device_friendly_name, net, network_name)
FritzBoxWifiSwitch(avm_device, device_friendly_name, net, network_name)
for net, network_name in networks.items()
]
def profile_entities_list(
router: FritzBoxTools,
avm_device: FritzBoxTools,
data_fritz: FritzData,
) -> list[FritzBoxProfileSwitch]:
"""Add new tracker entities from the router."""
"""Add new tracker entities from the AVM device."""
new_profiles: list[FritzBoxProfileSwitch] = []
if "X_AVM-DE_HostFilter1" not in router.connection.services:
if "X_AVM-DE_HostFilter1" not in avm_device.connection.services:
return new_profiles
if router.unique_id not in data_fritz.profile_switches:
data_fritz.profile_switches[router.unique_id] = set()
if avm_device.unique_id not in data_fritz.profile_switches:
data_fritz.profile_switches[avm_device.unique_id] = set()
for mac, device in router.devices.items():
for mac, device in avm_device.devices.items():
if device_filter_out_from_trackers(
mac, device, data_fritz.profile_switches.values()
):
continue
new_profiles.append(FritzBoxProfileSwitch(router, device))
data_fritz.profile_switches[router.unique_id].add(mac)
new_profiles.append(FritzBoxProfileSwitch(avm_device, device))
data_fritz.profile_switches[avm_device.unique_id].add(mac)
return new_profiles
def all_entities_list(
fritzbox_tools: FritzBoxTools,
avm_device: FritzBoxTools,
device_friendly_name: str,
data_fritz: FritzData,
local_ip: str,
) -> list[Entity]:
"""Get a list of all entities."""
if fritzbox_tools.mesh_role == MeshRoles.SLAVE:
if avm_device.mesh_role == MeshRoles.SLAVE:
return []
return [
*deflection_entities_list(fritzbox_tools, device_friendly_name),
*port_entities_list(fritzbox_tools, device_friendly_name, local_ip),
*wifi_entities_list(fritzbox_tools, device_friendly_name),
*profile_entities_list(fritzbox_tools, data_fritz),
*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),
]
@ -336,18 +334,16 @@ async def async_setup_entry(
) -> None:
"""Set up entry."""
_LOGGER.debug("Setting up switches")
fritzbox_tools: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
avm_device: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
data_fritz: FritzData = hass.data[DATA_FRITZ]
_LOGGER.debug("Fritzbox services: %s", fritzbox_tools.connection.services)
_LOGGER.debug("Fritzbox services: %s", avm_device.connection.services)
local_ip = await async_get_source_ip(
fritzbox_tools.hass, target_ip=fritzbox_tools.host
)
local_ip = await async_get_source_ip(avm_device.hass, target_ip=avm_device.host)
entities_list = await hass.async_add_executor_job(
all_entities_list,
fritzbox_tools,
avm_device,
entry.title,
data_fritz,
local_ip,
@ -356,12 +352,12 @@ async def async_setup_entry(
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))
def update_avm_device() -> None:
"""Update the values of the AVM device."""
async_add_entities(profile_entities_list(avm_device, data_fritz))
entry.async_on_unload(
async_dispatcher_connect(hass, fritzbox_tools.signal_device_new, update_router)
async_dispatcher_connect(hass, avm_device.signal_device_new, update_avm_device)
)
@ -370,12 +366,12 @@ class FritzBoxBaseSwitch(FritzBoxBaseEntity):
def __init__(
self,
fritzbox_tools: FritzBoxTools,
avm_device: FritzBoxTools,
device_friendly_name: str,
switch_info: SwitchInfo,
) -> None:
"""Init Fritzbox port switch."""
super().__init__(fritzbox_tools, device_friendly_name)
super().__init__(avm_device, device_friendly_name)
self._description = switch_info["description"]
self._friendly_name = switch_info["friendly_name"]
@ -385,9 +381,7 @@ class FritzBoxBaseSwitch(FritzBoxBaseEntity):
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._unique_id = f"{self._avm_device.unique_id}-{slugify(self._description)}"
self._attributes: dict[str, str] = {}
self._is_available = True
@ -443,7 +437,7 @@ class FritzBoxPortSwitch(FritzBoxBaseSwitch, SwitchEntity):
def __init__(
self,
fritzbox_tools: FritzBoxTools,
avm_device: FritzBoxTools,
device_friendly_name: str,
port_mapping: dict[str, Any] | None,
port_name: str,
@ -451,7 +445,7 @@ class FritzBoxPortSwitch(FritzBoxBaseSwitch, SwitchEntity):
connection_type: str,
) -> None:
"""Init Fritzbox port switch."""
self._fritzbox_tools = fritzbox_tools
self._avm_device = avm_device
self._attributes = {}
self.connection_type = connection_type
@ -470,13 +464,13 @@ class FritzBoxPortSwitch(FritzBoxBaseSwitch, SwitchEntity):
callback_update=self._async_fetch_update,
callback_switch=self._async_handle_port_switch_on_off,
)
super().__init__(fritzbox_tools, device_friendly_name, switch_info)
super().__init__(avm_device, 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._avm_device,
self.connection_type,
"1",
"GetGenericPortMappingEntry",
@ -511,7 +505,7 @@ class FritzBoxPortSwitch(FritzBoxBaseSwitch, SwitchEntity):
self.port_mapping["NewEnabled"] = "1" if turn_on else "0"
resp = await async_service_call_action(
self._fritzbox_tools,
self._avm_device,
self.connection_type,
"1",
"AddPortMapping",
@ -526,12 +520,12 @@ class FritzBoxDeflectionSwitch(FritzBoxBaseSwitch, SwitchEntity):
def __init__(
self,
fritzbox_tools: FritzBoxTools,
avm_device: FritzBoxTools,
device_friendly_name: str,
dict_of_deflection: Any,
) -> None:
"""Init Fritxbox Deflection class."""
self._fritzbox_tools: FritzBoxTools = fritzbox_tools
self._avm_device = avm_device
self.dict_of_deflection = dict_of_deflection
self._attributes = {}
@ -546,13 +540,13 @@ class FritzBoxDeflectionSwitch(FritzBoxBaseSwitch, SwitchEntity):
callback_update=self._async_fetch_update,
callback_switch=self._async_switch_on_off_executor,
)
super().__init__(self._fritzbox_tools, device_friendly_name, switch_info)
super().__init__(self._avm_device, 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"
self._avm_device, "X_AVM-DE_OnTel", "1", "GetDeflections"
)
if not resp:
self._is_available = False
@ -586,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._fritzbox_tools,
self._avm_device,
"X_AVM-DE_OnTel",
"1",
"SetDeflectionEnable",
@ -600,9 +594,9 @@ class FritzBoxProfileSwitch(FritzDeviceBase, SwitchEntity):
_attr_icon = "mdi:router-wireless-settings"
def __init__(self, fritzbox_tools: FritzBoxTools, device: FritzDevice) -> None:
def __init__(self, avm_device: FritzBoxTools, device: FritzDevice) -> None:
"""Init Fritz profile."""
super().__init__(fritzbox_tools, device)
super().__init__(avm_device, device)
self._attr_is_on: bool = False
self._name = f"{device.hostname} Internet Access"
self._attr_unique_id = f"{self._mac}_internet_access"
@ -611,7 +605,7 @@ class FritzBoxProfileSwitch(FritzDeviceBase, SwitchEntity):
@property
def is_on(self) -> bool:
"""Switch status."""
return self._router.devices[self._mac].wan_access
return self._avm_device.devices[self._mac].wan_access
@property
def device_info(self) -> DeviceInfo:
@ -624,7 +618,7 @@ class FritzBoxProfileSwitch(FritzDeviceBase, SwitchEntity):
identifiers={(DOMAIN, self._mac)},
via_device=(
DOMAIN,
self._router.unique_id,
self._avm_device.unique_id,
),
)
@ -645,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._router,
self._avm_device,
"X_AVM-DE_HostFilter",
"1",
"DisallowWANAccessByIP",
@ -659,13 +653,13 @@ class FritzBoxWifiSwitch(FritzBoxBaseSwitch, SwitchEntity):
def __init__(
self,
fritzbox_tools: FritzBoxTools,
avm_device: FritzBoxTools,
device_friendly_name: str,
network_num: int,
network_name: str,
) -> None:
"""Init Fritz Wifi switch."""
self._fritzbox_tools = fritzbox_tools
self._avm_device = avm_device
self._attributes = {}
self._attr_entity_category = EntityCategory.CONFIG
@ -679,13 +673,13 @@ class FritzBoxWifiSwitch(FritzBoxBaseSwitch, SwitchEntity):
callback_update=self._async_fetch_update,
callback_switch=self._async_switch_on_off_executor,
)
super().__init__(self._fritzbox_tools, device_friendly_name, switch_info)
super().__init__(self._avm_device, device_friendly_name, switch_info)
async def _async_fetch_update(self) -> None:
"""Fetch updates."""
wifi_info = await async_service_call_action(
self._fritzbox_tools,
self._avm_device,
"WLANConfiguration",
str(self._network_num),
"GetInfo",
@ -711,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._fritzbox_tools,
self._avm_device,
"WLANConfiguration",
str(self._network_num),
"SetEnable",