Add netgear speed test sensor (#72215)

* implement speed_test

* fix units

* restore last speedtest result

* fix import

* fix restore state is None

* fix styling

* fix mypy

* Use newer notation

* correct unit

* fix typing

* fix pylint

* fix issort

* use RestoreSensor

* fix import

* fix sensor restore

* do not extend SensorEntity

* fix mypy

* fix typing 2
This commit is contained in:
starkillerOG 2022-05-25 08:38:47 +02:00 committed by GitHub
parent 4c8a77fbd4
commit 9514f491f0
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 81 additions and 4 deletions

View file

@ -15,6 +15,7 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import (
DOMAIN,
KEY_COORDINATOR,
KEY_COORDINATOR_SPEED,
KEY_COORDINATOR_TRAFFIC,
KEY_ROUTER,
MODE_ROUTER,
@ -26,6 +27,7 @@ from .router import NetgearRouter
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(seconds=30)
SPEED_TEST_INTERVAL = timedelta(seconds=1800)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@ -78,6 +80,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Fetch data from the router."""
return await router.async_get_traffic_meter()
async def async_update_speed_test() -> dict[str, Any] | None:
"""Fetch data from the router."""
return await router.async_get_speed_test()
# Create update coordinators
coordinator = DataUpdateCoordinator(
hass,
@ -93,6 +99,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
update_method=async_update_traffic_meter,
update_interval=SCAN_INTERVAL,
)
coordinator_speed_test = DataUpdateCoordinator(
hass,
_LOGGER,
name=f"{router.device_name} Speed test",
update_method=async_update_speed_test,
update_interval=SPEED_TEST_INTERVAL,
)
if router.mode == MODE_ROUTER:
await coordinator.async_config_entry_first_refresh()
@ -102,6 +115,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
KEY_ROUTER: router,
KEY_COORDINATOR: coordinator,
KEY_COORDINATOR_TRAFFIC: coordinator_traffic_meter,
KEY_COORDINATOR_SPEED: coordinator_speed_test,
}
hass.config_entries.async_setup_platforms(entry, PLATFORMS)

View file

@ -12,6 +12,7 @@ CONF_CONSIDER_HOME = "consider_home"
KEY_ROUTER = "router"
KEY_COORDINATOR = "coordinator"
KEY_COORDINATOR_TRAFFIC = "coordinator_traffic"
KEY_COORDINATOR_SPEED = "coordinator_speed"
DEFAULT_CONSIDER_HOME = timedelta(seconds=180)
DEFAULT_NAME = "Netgear router"

View file

@ -208,6 +208,13 @@ class NetgearRouter:
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:
return await self.hass.async_add_executor_job(
self._api.get_new_speed_test_result
)
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:

View file

@ -1,20 +1,37 @@
"""Support for Netgear routers."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal
from homeassistant.components.sensor import (
RestoreSensor,
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import DATA_MEGABYTES, PERCENTAGE
from homeassistant.const import (
DATA_MEGABYTES,
DATA_RATE_MEGABITS_PER_SECOND,
PERCENTAGE,
TIME_MILLISECONDS,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN, KEY_COORDINATOR, KEY_COORDINATOR_TRAFFIC, KEY_ROUTER
from .const import (
DOMAIN,
KEY_COORDINATOR,
KEY_COORDINATOR_SPEED,
KEY_COORDINATOR_TRAFFIC,
KEY_ROUTER,
)
from .router import NetgearDeviceEntity, NetgearRouter, NetgearRouterEntity
SENSOR_TYPES = {
@ -200,6 +217,30 @@ SENSOR_TRAFFIC_TYPES = [
),
]
SENSOR_SPEED_TYPES = [
NetgearSensorEntityDescription(
key="NewOOKLAUplinkBandwidth",
name="Uplink Bandwidth",
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=DATA_RATE_MEGABITS_PER_SECOND,
icon="mdi:upload",
),
NetgearSensorEntityDescription(
key="NewOOKLADownlinkBandwidth",
name="Downlink Bandwidth",
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=DATA_RATE_MEGABITS_PER_SECOND,
icon="mdi:download",
),
NetgearSensorEntityDescription(
key="AveragePing",
name="Average Ping",
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=TIME_MILLISECONDS,
icon="mdi:wan",
),
]
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
@ -208,6 +249,7 @@ async def async_setup_entry(
router = hass.data[DOMAIN][entry.entry_id][KEY_ROUTER]
coordinator = hass.data[DOMAIN][entry.entry_id][KEY_COORDINATOR]
coordinator_traffic = hass.data[DOMAIN][entry.entry_id][KEY_COORDINATOR_TRAFFIC]
coordinator_speed = hass.data[DOMAIN][entry.entry_id][KEY_COORDINATOR_SPEED]
# Router entities
router_entities = []
@ -217,6 +259,11 @@ async def async_setup_entry(
NetgearRouterSensorEntity(coordinator_traffic, router, description)
)
for description in SENSOR_SPEED_TYPES:
router_entities.append(
NetgearRouterSensorEntity(coordinator_speed, router, description)
)
async_add_entities(router_entities)
# Entities per network device
@ -288,7 +335,7 @@ class NetgearSensorEntity(NetgearDeviceEntity, SensorEntity):
self._state = self._device[self._attribute]
class NetgearRouterSensorEntity(NetgearRouterEntity, SensorEntity):
class NetgearRouterSensorEntity(NetgearRouterEntity, RestoreSensor):
"""Representation of a device connected to a Netgear router."""
_attr_entity_registry_enabled_default = False
@ -306,7 +353,7 @@ class NetgearRouterSensorEntity(NetgearRouterEntity, SensorEntity):
self._name = f"{router.device_name} {entity_description.name}"
self._unique_id = f"{router.serial_number}-{entity_description.key}-{entity_description.index}"
self._value = None
self._value: StateType | date | datetime | Decimal = None
self.async_update_device()
@property
@ -314,6 +361,14 @@ class NetgearRouterSensorEntity(NetgearRouterEntity, SensorEntity):
"""Return the state of the sensor."""
return self._value
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
await super().async_added_to_hass()
if self.coordinator.data is None:
sensor_data = await self.async_get_last_sensor_data()
if sensor_data is not None:
self._value = sensor_data.native_value
@callback
def async_update_device(self) -> None:
"""Update the Netgear device."""