From 698c5eca00fdeace3916d1c2f0bd167fcced0358 Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Thu, 5 Mar 2026 11:49:28 +0100 Subject: [PATCH] Migrate remaining netgear coordinators to separate module (#164826) --- homeassistant/components/netgear/__init__.py | 78 ++----------- homeassistant/components/netgear/button.py | 4 +- .../components/netgear/coordinator.py | 107 +++++++++++++++--- .../components/netgear/device_tracker.py | 4 +- homeassistant/components/netgear/entity.py | 6 +- homeassistant/components/netgear/sensor.py | 8 +- homeassistant/components/netgear/switch.py | 4 +- 7 files changed, 119 insertions(+), 92 deletions(-) diff --git a/homeassistant/components/netgear/__init__.py b/homeassistant/components/netgear/__init__.py index 13565061593a..cbde5ccccadc 100644 --- a/homeassistant/components/netgear/__init__.py +++ b/homeassistant/components/netgear/__init__.py @@ -2,9 +2,7 @@ from __future__ import annotations -from datetime import timedelta import logging -from typing import Any from homeassistant.const import CONF_PORT, CONF_SSL from homeassistant.core import HomeAssistant @@ -14,18 +12,19 @@ from homeassistant.helpers import device_registry as dr, entity_registry as er from .const import PLATFORMS from .coordinator import ( NetgearConfigEntry, - NetgearDataCoordinator, NetgearFirmwareCoordinator, + NetgearLinkCoordinator, NetgearRuntimeData, + NetgearSpeedTestCoordinator, + NetgearTrackerCoordinator, + NetgearTrafficMeterCoordinator, + NetgearUtilizationCoordinator, ) from .errors import CannotLoginException from .router import NetgearRouter _LOGGER = logging.getLogger(__name__) -SCAN_INTERVAL = timedelta(seconds=30) -SPEED_TEST_INTERVAL = timedelta(hours=2) - async def async_setup_entry(hass: HomeAssistant, entry: NetgearConfigEntry) -> bool: """Set up Netgear component.""" @@ -52,70 +51,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: NetgearConfigEntry) -> b router.ssl, ) - async def async_update_devices() -> bool: - """Fetch data from the router.""" - if router.track_devices: - return await router.async_update_device_trackers() - return False - - async def async_update_traffic_meter() -> dict[str, Any] | None: - """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() - - async def async_update_utilization() -> dict[str, Any] | None: - """Fetch data from the router.""" - return await router.async_get_utilization() - - async def async_check_link_status() -> dict[str, Any] | None: - """Fetch data from the router.""" - return await router.async_get_link_status() - # Create update coordinators - coordinator_tracker = NetgearDataCoordinator( - hass, - router, - entry, - name="Devices", - update_method=async_update_devices, - update_interval=SCAN_INTERVAL, - ) - coordinator_traffic_meter = NetgearDataCoordinator( - hass, - router, - entry, - name="Traffic meter", - update_method=async_update_traffic_meter, - update_interval=SCAN_INTERVAL, - ) - coordinator_speed_test = NetgearDataCoordinator( - hass, - router, - entry, - name="Speed test", - update_method=async_update_speed_test, - update_interval=SPEED_TEST_INTERVAL, - ) + coordinator_tracker = NetgearTrackerCoordinator(hass, router, entry) + coordinator_traffic_meter = NetgearTrafficMeterCoordinator(hass, router, entry) + coordinator_speed_test = NetgearSpeedTestCoordinator(hass, router, entry) coordinator_firmware = NetgearFirmwareCoordinator(hass, router, entry) - coordinator_utilization = NetgearDataCoordinator( - hass, - router, - entry, - name="Utilization", - update_method=async_update_utilization, - update_interval=SCAN_INTERVAL, - ) - coordinator_link = NetgearDataCoordinator( - hass, - router, - entry, - name="Ethernet Link Status", - update_method=async_check_link_status, - update_interval=SCAN_INTERVAL, - ) + coordinator_utilization = NetgearUtilizationCoordinator(hass, router, entry) + coordinator_link = NetgearLinkCoordinator(hass, router, entry) if router.track_devices: await coordinator_tracker.async_config_entry_first_refresh() diff --git a/homeassistant/components/netgear/button.py b/homeassistant/components/netgear/button.py index 63308ca91b29..7ddd11bceafc 100644 --- a/homeassistant/components/netgear/button.py +++ b/homeassistant/components/netgear/button.py @@ -13,7 +13,7 @@ from homeassistant.const import EntityCategory from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback -from .coordinator import NetgearConfigEntry, NetgearDataCoordinator +from .coordinator import NetgearConfigEntry, NetgearTrackerCoordinator from .entity import NetgearRouterCoordinatorEntity from .router import NetgearRouter @@ -56,7 +56,7 @@ class NetgearRouterButtonEntity(NetgearRouterCoordinatorEntity, ButtonEntity): def __init__( self, - coordinator: NetgearDataCoordinator[bool], + coordinator: NetgearTrackerCoordinator, router: NetgearRouter, entity_description: NetgearButtonEntityDescription, ) -> None: diff --git a/homeassistant/components/netgear/coordinator.py b/homeassistant/components/netgear/coordinator.py index bc30e918c97b..9ee6b7b7342c 100644 --- a/homeassistant/components/netgear/coordinator.py +++ b/homeassistant/components/netgear/coordinator.py @@ -2,7 +2,6 @@ from __future__ import annotations -from collections.abc import Callable, Coroutine from dataclasses import dataclass from datetime import timedelta import logging @@ -16,7 +15,9 @@ from .router import NetgearRouter _LOGGER = logging.getLogger(__name__) +SCAN_INTERVAL = timedelta(seconds=30) SCAN_INTERVAL_FIRMWARE = timedelta(hours=5) +SPEED_TEST_INTERVAL = timedelta(hours=2) @dataclass @@ -24,12 +25,12 @@ class NetgearRuntimeData: """Runtime data for the Netgear integration.""" router: NetgearRouter - coordinator_tracker: NetgearDataCoordinator[bool] - coordinator_traffic: NetgearDataCoordinator[dict[str, Any] | None] - coordinator_speed: NetgearDataCoordinator[dict[str, Any] | None] + coordinator_tracker: NetgearTrackerCoordinator + coordinator_traffic: NetgearTrafficMeterCoordinator + coordinator_speed: NetgearSpeedTestCoordinator coordinator_firmware: NetgearFirmwareCoordinator - coordinator_utilization: NetgearDataCoordinator[dict[str, Any] | None] - coordinator_link: NetgearDataCoordinator[dict[str, Any] | None] + coordinator_utilization: NetgearUtilizationCoordinator + coordinator_link: NetgearLinkCoordinator type NetgearConfigEntry = ConfigEntry[NetgearRuntimeData] @@ -48,7 +49,6 @@ class NetgearDataCoordinator[T](DataUpdateCoordinator[T]): *, name: str, update_interval: timedelta, - update_method: Callable[[], Coroutine[Any, Any, T]] | None = None, ) -> None: """Initialize the coordinator.""" super().__init__( @@ -57,11 +57,60 @@ class NetgearDataCoordinator[T](DataUpdateCoordinator[T]): config_entry=entry, name=f"{router.device_name} {name}", update_interval=update_interval, - update_method=update_method, ) self.router = router +class NetgearTrackerCoordinator(NetgearDataCoordinator[bool]): + """Coordinator for Netgear device tracking.""" + + def __init__( + self, hass: HomeAssistant, router: NetgearRouter, entry: NetgearConfigEntry + ) -> None: + """Initialize the coordinator.""" + super().__init__( + hass, router, entry, name="Devices", update_interval=SCAN_INTERVAL + ) + + async def _async_update_data(self) -> bool: + """Fetch data from the router.""" + if self.router.track_devices: + return await self.router.async_update_device_trackers() + return False + + +class NetgearTrafficMeterCoordinator(NetgearDataCoordinator[dict[str, Any] | None]): + """Coordinator for Netgear traffic meter data.""" + + def __init__( + self, hass: HomeAssistant, router: NetgearRouter, entry: NetgearConfigEntry + ) -> None: + """Initialize the coordinator.""" + super().__init__( + hass, router, entry, name="Traffic meter", update_interval=SCAN_INTERVAL + ) + + async def _async_update_data(self) -> dict[str, Any] | None: + """Fetch data from the router.""" + return await self.router.async_get_traffic_meter() + + +class NetgearSpeedTestCoordinator(NetgearDataCoordinator[dict[str, Any] | None]): + """Coordinator for Netgear speed test data.""" + + def __init__( + self, hass: HomeAssistant, router: NetgearRouter, entry: NetgearConfigEntry + ) -> None: + """Initialize the coordinator.""" + super().__init__( + hass, router, entry, name="Speed test", update_interval=SPEED_TEST_INTERVAL + ) + + async def _async_update_data(self) -> dict[str, Any] | None: + """Fetch data from the router.""" + return await self.router.async_get_speed_test() + + class NetgearFirmwareCoordinator(NetgearDataCoordinator[dict[str, Any] | None]): """Coordinator for Netgear firmware updates.""" @@ -70,13 +119,45 @@ class NetgearFirmwareCoordinator(NetgearDataCoordinator[dict[str, Any] | None]): ) -> None: """Initialize the coordinator.""" super().__init__( - hass, - router, - entry, - name="Firmware", - update_interval=SCAN_INTERVAL_FIRMWARE, + hass, router, entry, name="Firmware", update_interval=SCAN_INTERVAL_FIRMWARE ) async def _async_update_data(self) -> dict[str, Any] | None: """Check for new firmware of the router.""" return await self.router.async_check_new_firmware() + + +class NetgearUtilizationCoordinator(NetgearDataCoordinator[dict[str, Any] | None]): + """Coordinator for Netgear utilization data.""" + + def __init__( + self, hass: HomeAssistant, router: NetgearRouter, entry: NetgearConfigEntry + ) -> None: + """Initialize the coordinator.""" + super().__init__( + hass, router, entry, name="Utilization", update_interval=SCAN_INTERVAL + ) + + async def _async_update_data(self) -> dict[str, Any] | None: + """Fetch data from the router.""" + return await self.router.async_get_utilization() + + +class NetgearLinkCoordinator(NetgearDataCoordinator[dict[str, Any] | None]): + """Coordinator for Netgear Ethernet link status.""" + + def __init__( + self, hass: HomeAssistant, router: NetgearRouter, entry: NetgearConfigEntry + ) -> None: + """Initialize the coordinator.""" + super().__init__( + hass, + router, + entry, + name="Ethernet Link Status", + update_interval=SCAN_INTERVAL, + ) + + async def _async_update_data(self) -> dict[str, Any] | None: + """Fetch data from the router.""" + return await self.router.async_get_link_status() diff --git a/homeassistant/components/netgear/device_tracker.py b/homeassistant/components/netgear/device_tracker.py index 6e9df9618cde..e47964b82d36 100644 --- a/homeassistant/components/netgear/device_tracker.py +++ b/homeassistant/components/netgear/device_tracker.py @@ -9,7 +9,7 @@ from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from .const import DEVICE_ICONS -from .coordinator import NetgearConfigEntry, NetgearDataCoordinator +from .coordinator import NetgearConfigEntry, NetgearTrackerCoordinator from .entity import NetgearDeviceEntity from .router import NetgearRouter @@ -58,7 +58,7 @@ class NetgearScannerEntity(NetgearDeviceEntity, ScannerEntity): def __init__( self, - coordinator: NetgearDataCoordinator[bool], + coordinator: NetgearTrackerCoordinator, router: NetgearRouter, device: dict, ) -> None: diff --git a/homeassistant/components/netgear/entity.py b/homeassistant/components/netgear/entity.py index 67f52b3cc497..e56d507a72be 100644 --- a/homeassistant/components/netgear/entity.py +++ b/homeassistant/components/netgear/entity.py @@ -13,18 +13,18 @@ from homeassistant.helpers.entity import Entity from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import DOMAIN -from .coordinator import NetgearDataCoordinator +from .coordinator import NetgearDataCoordinator, NetgearTrackerCoordinator from .router import NetgearRouter -class NetgearDeviceEntity(CoordinatorEntity): +class NetgearDeviceEntity(CoordinatorEntity[NetgearTrackerCoordinator]): """Base class for a device connected to a Netgear router.""" _attr_has_entity_name = True def __init__( self, - coordinator: NetgearDataCoordinator[Any], + coordinator: NetgearTrackerCoordinator, router: NetgearRouter, device: dict, ) -> None: diff --git a/homeassistant/components/netgear/sensor.py b/homeassistant/components/netgear/sensor.py index cc39be817773..e404c7621ba4 100644 --- a/homeassistant/components/netgear/sensor.py +++ b/homeassistant/components/netgear/sensor.py @@ -27,7 +27,11 @@ from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from homeassistant.helpers.typing import StateType -from .coordinator import NetgearConfigEntry, NetgearDataCoordinator +from .coordinator import ( + NetgearConfigEntry, + NetgearDataCoordinator, + NetgearTrackerCoordinator, +) from .entity import NetgearDeviceEntity, NetgearRouterCoordinatorEntity from .router import NetgearRouter @@ -325,7 +329,7 @@ class NetgearSensorEntity(NetgearDeviceEntity, SensorEntity): def __init__( self, - coordinator: NetgearDataCoordinator[Any], + coordinator: NetgearTrackerCoordinator, router: NetgearRouter, device: dict, attribute: str, diff --git a/homeassistant/components/netgear/switch.py b/homeassistant/components/netgear/switch.py index d9e0fdf8f29e..9b5127fd0ba0 100644 --- a/homeassistant/components/netgear/switch.py +++ b/homeassistant/components/netgear/switch.py @@ -13,7 +13,7 @@ from homeassistant.const import EntityCategory from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback -from .coordinator import NetgearConfigEntry, NetgearDataCoordinator +from .coordinator import NetgearConfigEntry, NetgearTrackerCoordinator from .entity import NetgearDeviceEntity, NetgearRouterEntity from .router import NetgearRouter @@ -149,7 +149,7 @@ class NetgearAllowBlock(NetgearDeviceEntity, SwitchEntity): def __init__( self, - coordinator: NetgearDataCoordinator[bool], + coordinator: NetgearTrackerCoordinator, router: NetgearRouter, device: dict, entity_description: SwitchEntityDescription,