Netgear add router switches (#72171)

This commit is contained in:
starkillerOG 2022-09-26 05:30:17 +02:00 committed by GitHub
parent b617d2bab0
commit d9b58e5ef1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 208 additions and 45 deletions

View File

@ -15,7 +15,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN, KEY_COORDINATOR, KEY_ROUTER
from .router import NetgearRouter, NetgearRouterEntity
from .router import NetgearRouter, NetgearRouterCoordinatorEntity
@dataclass
@ -55,7 +55,7 @@ async def async_setup_entry(
)
class NetgearRouterButtonEntity(NetgearRouterEntity, ButtonEntity):
class NetgearRouterButtonEntity(NetgearRouterCoordinatorEntity, ButtonEntity):
"""Netgear Router button entity."""
entity_description: NetgearButtonEntityDescription

View File

@ -20,7 +20,7 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, format_mac
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity import DeviceInfo, Entity
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
@ -87,14 +87,14 @@ class NetgearRouter:
)
self._consider_home = timedelta(seconds=consider_home_int)
self._api: Netgear = None
self._api_lock = asyncio.Lock()
self.api: Netgear = None
self.api_lock = asyncio.Lock()
self.devices: dict[str, Any] = {}
def _setup(self) -> bool:
"""Set up a Netgear router sync portion."""
self._api = get_api(
self.api = get_api(
self._password,
self._host,
self._username,
@ -102,7 +102,7 @@ class NetgearRouter:
self._ssl,
)
self._info = self._api.get_info()
self._info = self.api.get_info()
if self._info is None:
return False
@ -130,7 +130,7 @@ class NetgearRouter:
self.method_version = 2
if self.method_version == 2 and self.track_devices:
if not self._api.get_attached_devices_2():
if not self.api.get_attached_devices_2():
_LOGGER.error(
"Netgear Model '%s' in MODELS_V2 list, but failed to get attached devices using V2",
self.model,
@ -141,7 +141,7 @@ class NetgearRouter:
async def async_setup(self) -> bool:
"""Set up a Netgear router."""
async with self._api_lock:
async with self.api_lock:
if not await self.hass.async_add_executor_job(self._setup):
return False
@ -175,14 +175,14 @@ class NetgearRouter:
async def async_get_attached_devices(self) -> list:
"""Get the devices connected to the router."""
if self.method_version == 1:
async with self._api_lock:
async with self.api_lock:
return await self.hass.async_add_executor_job(
self._api.get_attached_devices
self.api.get_attached_devices
)
async with self._api_lock:
async with self.api_lock:
return await self.hass.async_add_executor_job(
self._api.get_attached_devices_2
self.api.get_attached_devices_2
)
async def async_update_device_trackers(self, now=None) -> bool:
@ -221,57 +221,57 @@ class NetgearRouter:
async def async_get_traffic_meter(self) -> dict[str, Any] | None:
"""Get the traffic meter data of the router."""
async with self._api_lock:
return await self.hass.async_add_executor_job(self._api.get_traffic_meter)
async with self.api_lock:
return await self.hass.async_add_executor_job(self.api.get_traffic_meter)
async def async_get_speed_test(self) -> dict[str, Any] | None:
"""Perform a speed test and get the results from the router."""
async with self._api_lock:
async with self.api_lock:
return await self.hass.async_add_executor_job(
self._api.get_new_speed_test_result
self.api.get_new_speed_test_result
)
async def async_get_link_status(self) -> dict[str, Any] | None:
"""Check the ethernet link status of the router."""
async with self._api_lock:
return await self.hass.async_add_executor_job(self._api.check_ethernet_link)
async with self.api_lock:
return await self.hass.async_add_executor_job(self.api.check_ethernet_link)
async def async_allow_block_device(self, mac: str, allow_block: str) -> None:
"""Allow or block a device connected to the router."""
async with self._api_lock:
async with self.api_lock:
await self.hass.async_add_executor_job(
self._api.allow_block_device, mac, allow_block
self.api.allow_block_device, mac, allow_block
)
async def async_get_utilization(self) -> dict[str, Any] | None:
"""Get the system information about utilization of the router."""
async with self._api_lock:
return await self.hass.async_add_executor_job(self._api.get_system_info)
async with self.api_lock:
return await self.hass.async_add_executor_job(self.api.get_system_info)
async def async_reboot(self) -> None:
"""Reboot the router."""
async with self._api_lock:
await self.hass.async_add_executor_job(self._api.reboot)
async with self.api_lock:
await self.hass.async_add_executor_job(self.api.reboot)
async def async_check_new_firmware(self) -> dict[str, Any] | None:
"""Check for new firmware of the router."""
async with self._api_lock:
return await self.hass.async_add_executor_job(self._api.check_new_firmware)
async with self.api_lock:
return await self.hass.async_add_executor_job(self.api.check_new_firmware)
async def async_update_new_firmware(self) -> None:
"""Update the router to the latest firmware."""
async with self._api_lock:
await self.hass.async_add_executor_job(self._api.update_new_firmware)
async with self.api_lock:
await self.hass.async_add_executor_job(self.api.update_new_firmware)
@property
def port(self) -> int:
"""Port used by the API."""
return self._api.port
return self.api.port
@property
def ssl(self) -> bool:
"""SSL used by the API."""
return self._api.ssl
return self.api.ssl
class NetgearBaseEntity(CoordinatorEntity):
@ -340,7 +340,7 @@ class NetgearDeviceEntity(NetgearBaseEntity):
)
class NetgearRouterEntity(CoordinatorEntity):
class NetgearRouterCoordinatorEntity(CoordinatorEntity):
"""Base class for a Netgear router entity."""
def __init__(
@ -379,3 +379,30 @@ class NetgearRouterEntity(CoordinatorEntity):
return DeviceInfo(
identifiers={(DOMAIN, self._router.unique_id)},
)
class NetgearRouterEntity(Entity):
"""Base class for a Netgear router entity without coordinator."""
def __init__(self, router: NetgearRouter) -> None:
"""Initialize a Netgear device."""
self._router = router
self._name = router.device_name
self._unique_id = router.serial_number
@property
def unique_id(self) -> str:
"""Return a unique ID."""
return self._unique_id
@property
def name(self) -> str:
"""Return the name."""
return self._name
@property
def device_info(self) -> DeviceInfo:
"""Return the device information."""
return DeviceInfo(
identifiers={(DOMAIN, self._router.unique_id)},
)

View File

@ -36,7 +36,7 @@ from .const import (
KEY_COORDINATOR_UTIL,
KEY_ROUTER,
)
from .router import NetgearDeviceEntity, NetgearRouter, NetgearRouterEntity
from .router import NetgearDeviceEntity, NetgearRouter, NetgearRouterCoordinatorEntity
_LOGGER = logging.getLogger(__name__)
@ -381,7 +381,7 @@ class NetgearSensorEntity(NetgearDeviceEntity, SensorEntity):
self._state = self._device[self._attribute]
class NetgearRouterSensorEntity(NetgearRouterEntity, RestoreSensor):
class NetgearRouterSensorEntity(NetgearRouterCoordinatorEntity, RestoreSensor):
"""Representation of a device connected to a Netgear router."""
_attr_entity_registry_enabled_default = False

View File

@ -1,4 +1,7 @@
"""Support for Netgear switches."""
from collections.abc import Callable
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import Any
@ -12,10 +15,11 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN, KEY_COORDINATOR, KEY_ROUTER
from .router import NetgearDeviceEntity, NetgearRouter
from .router import NetgearDeviceEntity, NetgearRouter, NetgearRouterEntity
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(seconds=300)
SWITCH_TYPES = [
SwitchEntityDescription(
@ -27,11 +31,96 @@ SWITCH_TYPES = [
]
@dataclass
class NetgearSwitchEntityDescriptionRequired:
"""Required attributes of NetgearSwitchEntityDescription."""
update: Callable[[NetgearRouter], bool]
action: Callable[[NetgearRouter], bool]
@dataclass
class NetgearSwitchEntityDescription(
SwitchEntityDescription, NetgearSwitchEntityDescriptionRequired
):
"""Class describing Netgear Switch entities."""
ROUTER_SWITCH_TYPES = [
NetgearSwitchEntityDescription(
key="access_control",
name="Access Control",
icon="mdi:block-helper",
entity_category=EntityCategory.CONFIG,
update=lambda router: router.api.get_block_device_enable_status,
action=lambda router: router.api.set_block_device_enable,
),
NetgearSwitchEntityDescription(
key="traffic_meter",
name="Traffic Meter",
icon="mdi:wifi-arrow-up-down",
entity_category=EntityCategory.CONFIG,
update=lambda router: router.api.get_traffic_meter_enabled,
action=lambda router: router.api.enable_traffic_meter,
),
NetgearSwitchEntityDescription(
key="parental_control",
name="Parental Control",
icon="mdi:account-child-outline",
entity_category=EntityCategory.CONFIG,
update=lambda router: router.api.get_parental_control_enable_status,
action=lambda router: router.api.enable_parental_control,
),
NetgearSwitchEntityDescription(
key="qos",
name="Quality of Service",
icon="mdi:wifi-star",
entity_category=EntityCategory.CONFIG,
update=lambda router: router.api.get_qos_enable_status,
action=lambda router: router.api.set_qos_enable_status,
),
NetgearSwitchEntityDescription(
key="2g_guest_wifi",
name="2.4G Guest Wifi",
icon="mdi:wifi",
entity_category=EntityCategory.CONFIG,
update=lambda router: router.api.get_2g_guest_access_enabled,
action=lambda router: router.api.set_2g_guest_access_enabled,
),
NetgearSwitchEntityDescription(
key="5g_guest_wifi",
name="5G Guest Wifi",
icon="mdi:wifi",
entity_category=EntityCategory.CONFIG,
update=lambda router: router.api.get_5g_guest_access_enabled,
action=lambda router: router.api.set_5g_guest_access_enabled,
),
NetgearSwitchEntityDescription(
key="smart_connect",
name="Smart Connect",
icon="mdi:wifi",
entity_category=EntityCategory.CONFIG,
update=lambda router: router.api.get_smart_connect_enabled,
action=lambda router: router.api.set_smart_connect_enabled,
),
]
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up switches for Netgear component."""
router = hass.data[DOMAIN][entry.entry_id][KEY_ROUTER]
# Router entities
router_entities = []
for description in ROUTER_SWITCH_TYPES:
router_entities.append(NetgearRouterSwitchEntity(router, description))
async_add_entities(router_entities)
# Entities per network device
coordinator = hass.data[DOMAIN][entry.entry_id][KEY_COORDINATOR]
tracked = set()
@ -80,14 +169,9 @@ class NetgearAllowBlock(NetgearDeviceEntity, SwitchEntity):
self.entity_description = entity_description
self._name = f"{self.get_device_name()} {self.entity_description.name}"
self._unique_id = f"{self._mac}-{self.entity_description.key}"
self._state = None
self._attr_is_on = None
self.async_update_device()
@property
def is_on(self):
"""Return true if switch is on."""
return self._state
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
await self._router.async_allow_block_device(self._mac, ALLOW)
@ -104,6 +188,58 @@ class NetgearAllowBlock(NetgearDeviceEntity, SwitchEntity):
self._device = self._router.devices[self._mac]
self._active = self._device["active"]
if self._device[self.entity_description.key] is None:
self._state = None
self._attr_is_on = None
else:
self._state = self._device[self.entity_description.key] == "Allow"
self._attr_is_on = self._device[self.entity_description.key] == "Allow"
class NetgearRouterSwitchEntity(NetgearRouterEntity, SwitchEntity):
"""Representation of a Netgear router switch."""
_attr_entity_registry_enabled_default = False
entity_description: NetgearSwitchEntityDescription
def __init__(
self,
router: NetgearRouter,
entity_description: NetgearSwitchEntityDescription,
) -> None:
"""Initialize a Netgear device."""
super().__init__(router)
self.entity_description = entity_description
self._name = f"{router.device_name} {entity_description.name}"
self._unique_id = f"{router.serial_number}-{entity_description.key}"
self._attr_is_on = None
self._attr_available = False
async def async_added_to_hass(self):
"""Fetch state when entity is added."""
await self.async_update()
await super().async_added_to_hass()
async def async_update(self):
"""Poll the state of the switch."""
async with self._router.api_lock:
response = await self.hass.async_add_executor_job(
self.entity_description.update(self._router)
)
if response is None:
self._attr_available = False
else:
self._attr_is_on = response
self._attr_available = True
async def async_turn_on(self, **kwargs):
"""Turn the switch on."""
async with self._router.api_lock:
await self.hass.async_add_executor_job(
self.entity_description.action(self._router), True
)
async def async_turn_off(self, **kwargs):
"""Turn the switch off."""
async with self._router.api_lock:
await self.hass.async_add_executor_job(
self.entity_description.action(self._router), False
)

View File

@ -15,7 +15,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN, KEY_COORDINATOR_FIRMWARE, KEY_ROUTER
from .router import NetgearRouter, NetgearRouterEntity
from .router import NetgearRouter, NetgearRouterCoordinatorEntity
LOGGER = logging.getLogger(__name__)
@ -31,7 +31,7 @@ async def async_setup_entry(
async_add_entities(entities)
class NetgearUpdateEntity(NetgearRouterEntity, UpdateEntity):
class NetgearUpdateEntity(NetgearRouterCoordinatorEntity, UpdateEntity):
"""Update entity for a Netgear device."""
_attr_device_class = UpdateDeviceClass.FIRMWARE