Migrate remaining netgear coordinators to separate module (#164826)

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()

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:

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()

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:

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:

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,

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,