Move TP-Link Omada update coordinator into coordinator module (#159943)

homeassistant/components/tplink_omada/coordinator.py

@@ -5,14 +5,19 @@ from __future__ import annotations
 import asyncio
 from datetime import timedelta
 import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, NamedTuple
 
 from tplink_omada_client import OmadaSiteClient, OmadaSwitchPortDetails
 from tplink_omada_client.clients import OmadaWirelessClient
-from tplink_omada_client.devices import OmadaGateway, OmadaListDevice, OmadaSwitch
+from tplink_omada_client.devices import (
+    OmadaFirmwareUpdate,
+    OmadaGateway,
+    OmadaListDevice,
+    OmadaSwitch,
+)
 from tplink_omada_client.exceptions import OmadaClientException
 
-from homeassistant.core import HomeAssistant
+from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
 if TYPE_CHECKING:
@@ -24,6 +29,7 @@ POLL_SWITCH_PORT = 300
 POLL_GATEWAY = 300
 POLL_CLIENTS = 300
 POLL_DEVICES = 300
+POLL_UPGRADE = 60
 
 
 class OmadaCoordinator[_T](DataUpdateCoordinator[dict[str, _T]]):
@@ -144,3 +150,68 @@ class OmadaClientsCoordinator(OmadaCoordinator[OmadaWirelessClient]):
             async for c in self.omada_client.get_connected_clients()
             if isinstance(c, OmadaWirelessClient)
         }
+
+
+class FirmwareUpdateStatus(NamedTuple):
+    """Firmware update information for Omada SDN devices."""
+
+    device: OmadaListDevice
+    firmware: OmadaFirmwareUpdate | None
+
+
+class OmadaFirmwareUpdateCoordinator(OmadaCoordinator[FirmwareUpdateStatus]):  # pylint: disable=hass-enforce-class-module
+    """Coordinator for getting details about available firmware updates for Omada devices."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        config_entry: OmadaConfigEntry,
+        omada_client: OmadaSiteClient,
+        devices_coordinator: OmadaDevicesCoordinator,
+    ) -> None:
+        """Initialize my coordinator."""
+        super().__init__(
+            hass, config_entry, omada_client, "Firmware Updates", poll_delay=None
+        )
+
+        self._devices_coordinator = devices_coordinator
+        self._config_entry = config_entry
+
+        config_entry.async_on_unload(
+            devices_coordinator.async_add_listener(self._handle_devices_update)
+        )
+
+    async def _get_firmware_updates(self) -> list[FirmwareUpdateStatus]:
+        devices = self._devices_coordinator.data.values()
+
+        updates = [
+            FirmwareUpdateStatus(
+                device=d,
+                firmware=None
+                if not d.need_upgrade
+                else await self.omada_client.get_firmware_details(d),
+            )
+            for d in devices
+        ]
+
+        # During a firmware upgrade, poll device list more frequently
+        self._devices_coordinator.update_interval = timedelta(
+            seconds=(
+                POLL_UPGRADE
+                if any(u.device.fw_download for u in updates)
+                else POLL_DEVICES
+            )
+        )
+        return updates
+
+    async def poll_update(self) -> dict[str, FirmwareUpdateStatus]:
+        """Poll the state of Omada Devices firmware update availability."""
+        return {d.device.mac: d for d in await self._get_firmware_updates()}
+
+    @callback
+    def _handle_devices_update(self) -> None:
+        """Handle updated data from the devices coordinator."""
+        # Trigger a refresh of our data, based on the updated device list
+        self._config_entry.async_create_background_task(
+            self.hass, self.async_request_refresh(), "Omada Firmware Update Refresh"
+        )

homeassistant/components/tplink_omada/quality_scale.yaml

@@ -5,9 +5,7 @@ rules:
     status: done
     comment: Service data APIs are polled every 5 minutes
   brands: done
-  common-modules:
-    status: todo
-    comment: The coordinator for the update platform should be moved to common module.
+  common-modules: done
   config-flow-test-coverage:
     status: todo
     comment: "test_form_single_site is patching config flow internals, and should only patch external APIs. Must address feedback from #156697."

homeassistant/components/tplink_omada/update.py

@@ -2,11 +2,9 @@
 
 from __future__ import annotations
 
-from datetime import timedelta
-from typing import Any, NamedTuple
+from typing import Any
 
-from tplink_omada_client import OmadaSiteClient
-from tplink_omada_client.devices import OmadaFirmwareUpdate, OmadaListDevice
+from tplink_omada_client.devices import OmadaListDevice
 from tplink_omada_client.exceptions import OmadaClientException, RequestFailed
 
 from homeassistant.components.update import (
@@ -19,76 +17,9 @@ from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
 from . import OmadaConfigEntry
-from .coordinator import POLL_DEVICES, OmadaCoordinator, OmadaDevicesCoordinator
+from .coordinator import OmadaFirmwareUpdateCoordinator
 from .entity import OmadaDeviceEntity
 
-POLL_DELAY_UPGRADE = 60
-
-
-class FirmwareUpdateStatus(NamedTuple):
-    """Firmware update information for Omada SDN devices."""
-
-    device: OmadaListDevice
-    firmware: OmadaFirmwareUpdate | None
-
-
-class OmadaFirmwareUpdateCoordinator(OmadaCoordinator[FirmwareUpdateStatus]):  # pylint: disable=hass-enforce-class-module
-    """Coordinator for getting details about available firmware updates for Omada devices."""
-
-    def __init__(
-        self,
-        hass: HomeAssistant,
-        config_entry: OmadaConfigEntry,
-        omada_client: OmadaSiteClient,
-        devices_coordinator: OmadaDevicesCoordinator,
-    ) -> None:
-        """Initialize my coordinator."""
-        super().__init__(
-            hass, config_entry, omada_client, "Firmware Updates", poll_delay=None
-        )
-
-        self._devices_coordinator = devices_coordinator
-        self._config_entry = config_entry
-
-        config_entry.async_on_unload(
-            devices_coordinator.async_add_listener(self._handle_devices_update)
-        )
-
-    async def _get_firmware_updates(self) -> list[FirmwareUpdateStatus]:
-        devices = self._devices_coordinator.data.values()
-
-        updates = [
-            FirmwareUpdateStatus(
-                device=d,
-                firmware=None
-                if not d.need_upgrade
-                else await self.omada_client.get_firmware_details(d),
-            )
-            for d in devices
-        ]
-
-        # During a firmware upgrade, poll device list more frequently
-        self._devices_coordinator.update_interval = timedelta(
-            seconds=(
-                POLL_DELAY_UPGRADE
-                if any(u.device.fw_download for u in updates)
-                else POLL_DEVICES
-            )
-        )
-        return updates
-
-    async def poll_update(self) -> dict[str, FirmwareUpdateStatus]:
-        """Poll the state of Omada Devices firmware update availability."""
-        return {d.device.mac: d for d in await self._get_firmware_updates()}
-
-    @callback
-    def _handle_devices_update(self) -> None:
-        """Handle updated data from the devices coordinator."""
-        # Trigger a refresh of our data, based on the updated device list
-        self._config_entry.async_create_background_task(
-            self.hass, self.async_request_refresh(), "Omada Firmware Update Refresh"
-        )
-
 
 async def async_setup_entry(
     hass: HomeAssistant,