Move coordinator and getting data closer together in devolo Home Network (#144814)

homeassistant/components/devolo_home_network/__init__.py

@@ -2,27 +2,13 @@
 
 from __future__ import annotations
 
-from asyncio import Semaphore
-from dataclasses import dataclass
 import logging
 from typing import Any
 
 from devolo_plc_api import Device
-from devolo_plc_api.device_api import (
+from devolo_plc_api.exceptions.device import DeviceNotFound
-    ConnectedStationInfo,
-    NeighborAPInfo,
-    UpdateFirmwareCheck,
-    WifiGuestAccessGet,
-)
-from devolo_plc_api.exceptions.device import (
-    DeviceNotFound,
-    DevicePasswordProtected,
-    DeviceUnavailable,
-)
-from devolo_plc_api.plcnet_api import LogicalNetwork
 
 from homeassistant.components import zeroconf
-from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import (
     CONF_IP_ADDRESS,
     CONF_PASSWORD,
@@ -30,38 +16,34 @@ from homeassistant.const import (
     Platform,
 )
 from homeassistant.core import Event, HomeAssistant, callback
-from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
+from homeassistant.exceptions import ConfigEntryNotReady
-from homeassistant.helpers import device_registry as dr
 from homeassistant.helpers.httpx_client import get_async_client
-from homeassistant.helpers.update_coordinator import UpdateFailed
 
 from .const import (
     CONNECTED_PLC_DEVICES,
     CONNECTED_WIFI_CLIENTS,
     DOMAIN,
-    FIRMWARE_UPDATE_INTERVAL,
     LAST_RESTART,
-    LONG_UPDATE_INTERVAL,
     NEIGHBORING_WIFI_NETWORKS,
     REGULAR_FIRMWARE,
-    SHORT_UPDATE_INTERVAL,
     SWITCH_GUEST_WIFI,
     SWITCH_LEDS,
 )
-from .coordinator import DevoloDataUpdateCoordinator
+from .coordinator import (
+    DevoloDataUpdateCoordinator,
+    DevoloFirmwareUpdateCoordinator,
+    DevoloHomeNetworkConfigEntry,
+    DevoloHomeNetworkData,
+    DevoloLedSettingsGetCoordinator,
+    DevoloLogicalNetworkCoordinator,
+    DevoloUptimeGetCoordinator,
+    DevoloWifiConnectedStationsGetCoordinator,
+    DevoloWifiGuestAccessGetCoordinator,
+    DevoloWifiNeighborAPsGetCoordinator,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
-type DevoloHomeNetworkConfigEntry = ConfigEntry[DevoloHomeNetworkData]
-
-
-@dataclass
-class DevoloHomeNetworkData:
-    """The devolo Home Network data."""
-
-    device: Device
-    coordinators: dict[str, DevoloDataUpdateCoordinator[Any]]
-
 
 async def async_setup_entry(
     hass: HomeAssistant, entry: DevoloHomeNetworkConfigEntry
@@ -69,8 +51,6 @@ async def async_setup_entry(
     """Set up devolo Home Network from a config entry."""
     zeroconf_instance = await zeroconf.async_get_async_instance(hass)
     async_client = get_async_client(hass)
-    device_registry = dr.async_get(hass)
-    semaphore = Semaphore(1)
 
     try:
         device = Device(
@@ -90,177 +70,52 @@ async def async_setup_entry(
 
     entry.runtime_data = DevoloHomeNetworkData(device=device, coordinators={})
 
-    async def async_update_firmware_available() -> UpdateFirmwareCheck:
-        """Fetch data from API endpoint."""
-        assert device.device
-        update_sw_version(device_registry, device)
-        try:
-            return await device.device.async_check_firmware_available()
-        except DeviceUnavailable as err:
-            raise UpdateFailed(
-                translation_domain=DOMAIN,
-                translation_key="update_failed",
-                translation_placeholders={"error": str(err)},
-            ) from err
-
-    async def async_update_connected_plc_devices() -> LogicalNetwork:
-        """Fetch data from API endpoint."""
-        assert device.plcnet
-        update_sw_version(device_registry, device)
-        try:
-            return await device.plcnet.async_get_network_overview()
-        except DeviceUnavailable as err:
-            raise UpdateFailed(
-                translation_domain=DOMAIN,
-                translation_key="update_failed",
-                translation_placeholders={"error": str(err)},
-            ) from err
-
-    async def async_update_guest_wifi_status() -> WifiGuestAccessGet:
-        """Fetch data from API endpoint."""
-        assert device.device
-        update_sw_version(device_registry, device)
-        try:
-            return await device.device.async_get_wifi_guest_access()
-        except DeviceUnavailable as err:
-            raise UpdateFailed(
-                translation_domain=DOMAIN,
-                translation_key="update_failed",
-                translation_placeholders={"error": str(err)},
-            ) from err
-        except DevicePasswordProtected as err:
-            raise ConfigEntryAuthFailed(
-                translation_domain=DOMAIN, translation_key="password_wrong"
-            ) from err
-
-    async def async_update_led_status() -> bool:
-        """Fetch data from API endpoint."""
-        assert device.device
-        update_sw_version(device_registry, device)
-        try:
-            return await device.device.async_get_led_setting()
-        except DeviceUnavailable as err:
-            raise UpdateFailed(
-                translation_domain=DOMAIN,
-                translation_key="update_failed",
-                translation_placeholders={"error": str(err)},
-            ) from err
-
-    async def async_update_last_restart() -> int:
-        """Fetch data from API endpoint."""
-        assert device.device
-        update_sw_version(device_registry, device)
-        try:
-            return await device.device.async_uptime()
-        except DeviceUnavailable as err:
-            raise UpdateFailed(
-                translation_domain=DOMAIN,
-                translation_key="update_failed",
-                translation_placeholders={"error": str(err)},
-            ) from err
-        except DevicePasswordProtected as err:
-            raise ConfigEntryAuthFailed(
-                translation_domain=DOMAIN, translation_key="password_wrong"
-            ) from err
-
-    async def async_update_wifi_connected_station() -> list[ConnectedStationInfo]:
-        """Fetch data from API endpoint."""
-        assert device.device
-        update_sw_version(device_registry, device)
-        try:
-            return await device.device.async_get_wifi_connected_station()
-        except DeviceUnavailable as err:
-            raise UpdateFailed(
-                translation_domain=DOMAIN,
-                translation_key="update_failed",
-                translation_placeholders={"error": str(err)},
-            ) from err
-
-    async def async_update_wifi_neighbor_access_points() -> list[NeighborAPInfo]:
-        """Fetch data from API endpoint."""
-        assert device.device
-        update_sw_version(device_registry, device)
-        try:
-            return await device.device.async_get_wifi_neighbor_access_points()
-        except DeviceUnavailable as err:
-            raise UpdateFailed(
-                translation_domain=DOMAIN,
-                translation_key="update_failed",
-                translation_placeholders={"error": str(err)},
-            ) from err
-
     async def disconnect(event: Event) -> None:
         """Disconnect from device."""
         await device.async_disconnect()
 
     coordinators: dict[str, DevoloDataUpdateCoordinator[Any]] = {}
     if device.plcnet:
-        coordinators[CONNECTED_PLC_DEVICES] = DevoloDataUpdateCoordinator(
+        coordinators[CONNECTED_PLC_DEVICES] = DevoloLogicalNetworkCoordinator(
             hass,
             _LOGGER,
             config_entry=entry,
-            name=CONNECTED_PLC_DEVICES,
-            semaphore=semaphore,
-            update_method=async_update_connected_plc_devices,
-            update_interval=LONG_UPDATE_INTERVAL,
         )
     if device.device and "led" in device.device.features:
-        coordinators[SWITCH_LEDS] = DevoloDataUpdateCoordinator(
+        coordinators[SWITCH_LEDS] = DevoloLedSettingsGetCoordinator(
             hass,
             _LOGGER,
             config_entry=entry,
-            name=SWITCH_LEDS,
-            semaphore=semaphore,
-            update_method=async_update_led_status,
-            update_interval=SHORT_UPDATE_INTERVAL,
         )
     if device.device and "restart" in device.device.features:
-        coordinators[LAST_RESTART] = DevoloDataUpdateCoordinator(
+        coordinators[LAST_RESTART] = DevoloUptimeGetCoordinator(
             hass,
             _LOGGER,
             config_entry=entry,
-            name=LAST_RESTART,
-            semaphore=semaphore,
-            update_method=async_update_last_restart,
-            update_interval=SHORT_UPDATE_INTERVAL,
         )
     if device.device and "update" in device.device.features:
-        coordinators[REGULAR_FIRMWARE] = DevoloDataUpdateCoordinator(
+        coordinators[REGULAR_FIRMWARE] = DevoloFirmwareUpdateCoordinator(
             hass,
             _LOGGER,
             config_entry=entry,
-            name=REGULAR_FIRMWARE,
-            semaphore=semaphore,
-            update_method=async_update_firmware_available,
-            update_interval=FIRMWARE_UPDATE_INTERVAL,
         )
     if device.device and "wifi1" in device.device.features:
-        coordinators[CONNECTED_WIFI_CLIENTS] = DevoloDataUpdateCoordinator(
+        coordinators[CONNECTED_WIFI_CLIENTS] = (
-            hass,
+            DevoloWifiConnectedStationsGetCoordinator(
-            _LOGGER,
+                hass,
-            config_entry=entry,
+                _LOGGER,
-            name=CONNECTED_WIFI_CLIENTS,
+                config_entry=entry,
-            semaphore=semaphore,
+            )
-            update_method=async_update_wifi_connected_station,
-            update_interval=SHORT_UPDATE_INTERVAL,
         )
-        coordinators[NEIGHBORING_WIFI_NETWORKS] = DevoloDataUpdateCoordinator(
+        coordinators[NEIGHBORING_WIFI_NETWORKS] = DevoloWifiNeighborAPsGetCoordinator(
             hass,
             _LOGGER,
             config_entry=entry,
-            name=NEIGHBORING_WIFI_NETWORKS,
-            semaphore=semaphore,
-            update_method=async_update_wifi_neighbor_access_points,
-            update_interval=LONG_UPDATE_INTERVAL,
         )
-        coordinators[SWITCH_GUEST_WIFI] = DevoloDataUpdateCoordinator(
+        coordinators[SWITCH_GUEST_WIFI] = DevoloWifiGuestAccessGetCoordinator(
             hass,
             _LOGGER,
             config_entry=entry,
-            name=SWITCH_GUEST_WIFI,
-            semaphore=semaphore,
-            update_method=async_update_guest_wifi_status,
-            update_interval=SHORT_UPDATE_INTERVAL,
         )
 
     for coordinator in coordinators.values():
@@ -303,16 +158,3 @@ def platforms(device: Device) -> set[Platform]:
     if device.device and "update" in device.device.features:
         supported_platforms.add(Platform.UPDATE)
     return supported_platforms
-
-
-@callback
-def update_sw_version(device_registry: dr.DeviceRegistry, device: Device) -> None:
-    """Update device registry with new firmware version."""
-    if (
-        device_entry := device_registry.async_get_device(
-            identifiers={(DOMAIN, str(device.serial_number))}
-        )
-    ) and device_entry.sw_version != device.firmware_version:
-        device_registry.async_update_device(
-            device_id=device_entry.id, sw_version=device.firmware_version
-        )

homeassistant/components/devolo_home_network/binary_sensor.py

@@ -16,9 +16,8 @@ from homeassistant.const import EntityCategory
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
-from . import DevoloHomeNetworkConfigEntry
 from .const import CONNECTED_PLC_DEVICES, CONNECTED_TO_ROUTER
-from .coordinator import DevoloDataUpdateCoordinator
+from .coordinator import DevoloDataUpdateCoordinator, DevoloHomeNetworkConfigEntry
 from .entity import DevoloCoordinatorEntity
 
 PARALLEL_UPDATES = 0

homeassistant/components/devolo_home_network/button.py

@@ -18,8 +18,8 @@ from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
-from . import DevoloHomeNetworkConfigEntry
 from .const import DOMAIN, IDENTIFY, PAIRING, RESTART, START_WPS
+from .coordinator import DevoloHomeNetworkConfigEntry
 from .entity import DevoloEntity
 
 PARALLEL_UPDATES = 0

homeassistant/components/devolo_home_network/config_flow.py

@@ -17,8 +17,8 @@ from homeassistant.core import HomeAssistant
 from homeassistant.helpers.httpx_client import get_async_client
 from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
 
-from . import DevoloHomeNetworkConfigEntry
 from .const import DOMAIN, PRODUCT, SERIAL_NUMBER, TITLE
+from .coordinator import DevoloHomeNetworkConfigEntry
 
 _LOGGER = logging.getLogger(__name__)
 

homeassistant/components/devolo_home_network/coordinator.py

@@ -1,13 +1,44 @@
 """Base coordinator."""
 
 from asyncio import Semaphore
-from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
 from datetime import timedelta
 from logging import Logger
+from typing import Any
+
+from devolo_plc_api import Device
+from devolo_plc_api.device_api import (
+    ConnectedStationInfo,
+    NeighborAPInfo,
+    UpdateFirmwareCheck,
+    WifiGuestAccessGet,
+)
+from devolo_plc_api.exceptions.device import DevicePasswordProtected, DeviceUnavailable
+from devolo_plc_api.plcnet_api import LogicalNetwork
 
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.core import HomeAssistant
+from homeassistant.core import HomeAssistant, callback
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+from homeassistant.exceptions import ConfigEntryAuthFailed
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import (
+    CONNECTED_PLC_DEVICES,
+    CONNECTED_WIFI_CLIENTS,
+    DOMAIN,
+    FIRMWARE_UPDATE_INTERVAL,
+    LAST_RESTART,
+    LONG_UPDATE_INTERVAL,
+    NEIGHBORING_WIFI_NETWORKS,
+    REGULAR_FIRMWARE,
+    SHORT_UPDATE_INTERVAL,
+    SWITCH_GUEST_WIFI,
+    SWITCH_LEDS,
+)
+
+SEMAPHORE = Semaphore(1)
+
+type DevoloHomeNetworkConfigEntry = ConfigEntry[DevoloHomeNetworkData]
 
 
 class DevoloDataUpdateCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
@@ -18,11 +49,62 @@ class DevoloDataUpdateCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
         hass: HomeAssistant,
         logger: Logger,
         *,
-        config_entry: ConfigEntry,
+        config_entry: DevoloHomeNetworkConfigEntry,
         name: str,
-        semaphore: Semaphore,
+        update_interval: timedelta | None = None,
-        update_interval: timedelta,
+    ) -> None:
-        update_method: Callable[[], Awaitable[_DataT]],
+        """Initialize global data updater."""
+        self.device = config_entry.runtime_data.device
+        super().__init__(
+            hass,
+            logger,
+            config_entry=config_entry,
+            name=name,
+            update_interval=update_interval,
+        )
+
+    async def _async_update_data(self) -> _DataT:
+        """Fetch the latest data from the source."""
+        self.update_sw_version()
+        async with SEMAPHORE:
+            try:
+                return await super()._async_update_data()
+            except DeviceUnavailable as err:
+                raise UpdateFailed(
+                    translation_domain=DOMAIN,
+                    translation_key="update_failed",
+                    translation_placeholders={"error": str(err)},
+                ) from err
+            except DevicePasswordProtected as err:
+                raise ConfigEntryAuthFailed(
+                    translation_domain=DOMAIN, translation_key="password_wrong"
+                ) from err
+
+    @callback
+    def update_sw_version(self) -> None:
+        """Update device registry with new firmware version, if it changed at runtime."""
+        device_registry = dr.async_get(self.hass)
+        if (
+            device_entry := device_registry.async_get_device(
+                identifiers={(DOMAIN, self.device.serial_number)}
+            )
+        ) and device_entry.sw_version != self.device.firmware_version:
+            device_registry.async_update_device(
+                device_id=device_entry.id, sw_version=self.device.firmware_version
+            )
+
+
+class DevoloFirmwareUpdateCoordinator(DevoloDataUpdateCoordinator[UpdateFirmwareCheck]):
+    """Class to manage fetching data from the UpdateFirmwareCheck endpoint."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        logger: Logger,
+        *,
+        config_entry: ConfigEntry,
+        name: str = REGULAR_FIRMWARE,
+        update_interval: timedelta | None = FIRMWARE_UPDATE_INTERVAL,
     ) -> None:
         """Initialize global data updater."""
         super().__init__(
@@ -31,11 +113,192 @@ class DevoloDataUpdateCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
             config_entry=config_entry,
             name=name,
             update_interval=update_interval,
-            update_method=update_method,
         )
-        self._semaphore = semaphore
+        self.update_method = self.async_update_firmware_available
 
-    async def _async_update_data(self) -> _DataT:
+    async def async_update_firmware_available(self) -> UpdateFirmwareCheck:
-        """Fetch the latest data from the source."""
+        """Fetch data from API endpoint."""
-        async with self._semaphore:
+        assert self.device.device
-            return await super()._async_update_data()
+        return await self.device.device.async_check_firmware_available()
+
+
+class DevoloLedSettingsGetCoordinator(DevoloDataUpdateCoordinator[bool]):
+    """Class to manage fetching data from the LedSettingsGet endpoint."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        logger: Logger,
+        *,
+        config_entry: ConfigEntry,
+        name: str = SWITCH_LEDS,
+        update_interval: timedelta | None = SHORT_UPDATE_INTERVAL,
+    ) -> None:
+        """Initialize global data updater."""
+        super().__init__(
+            hass,
+            logger,
+            config_entry=config_entry,
+            name=name,
+            update_interval=update_interval,
+        )
+        self.update_method = self.async_update_led_status
+
+    async def async_update_led_status(self) -> bool:
+        """Fetch data from API endpoint."""
+        assert self.device.device
+        return await self.device.device.async_get_led_setting()
+
+
+class DevoloLogicalNetworkCoordinator(DevoloDataUpdateCoordinator[LogicalNetwork]):
+    """Class to manage fetching data from the GetNetworkOverview endpoint."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        logger: Logger,
+        *,
+        config_entry: ConfigEntry,
+        name: str = CONNECTED_PLC_DEVICES,
+        update_interval: timedelta | None = LONG_UPDATE_INTERVAL,
+    ) -> None:
+        """Initialize global data updater."""
+        super().__init__(
+            hass,
+            logger,
+            config_entry=config_entry,
+            name=name,
+            update_interval=update_interval,
+        )
+        self.update_method = self.async_update_connected_plc_devices
+
+    async def async_update_connected_plc_devices(self) -> LogicalNetwork:
+        """Fetch data from API endpoint."""
+        assert self.device.plcnet
+        return await self.device.plcnet.async_get_network_overview()
+
+
+class DevoloUptimeGetCoordinator(DevoloDataUpdateCoordinator[int]):
+    """Class to manage fetching data from the UptimeGet endpoint."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        logger: Logger,
+        *,
+        config_entry: ConfigEntry,
+        name: str = LAST_RESTART,
+        update_interval: timedelta | None = SHORT_UPDATE_INTERVAL,
+    ) -> None:
+        """Initialize global data updater."""
+        super().__init__(
+            hass,
+            logger,
+            config_entry=config_entry,
+            name=name,
+            update_interval=update_interval,
+        )
+        self.update_method = self.async_update_last_restart
+
+    async def async_update_last_restart(self) -> int:
+        """Fetch data from API endpoint."""
+        assert self.device.device
+        return await self.device.device.async_uptime()
+
+
+class DevoloWifiConnectedStationsGetCoordinator(
+    DevoloDataUpdateCoordinator[list[ConnectedStationInfo]]
+):
+    """Class to manage fetching data from the WifiGuestAccessGet endpoint."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        logger: Logger,
+        *,
+        config_entry: ConfigEntry,
+        name: str = CONNECTED_WIFI_CLIENTS,
+        update_interval: timedelta | None = SHORT_UPDATE_INTERVAL,
+    ) -> None:
+        """Initialize global data updater."""
+        super().__init__(
+            hass,
+            logger,
+            config_entry=config_entry,
+            name=name,
+            update_interval=update_interval,
+        )
+        self.update_method = self.async_get_wifi_connected_station
+
+    async def async_get_wifi_connected_station(self) -> list[ConnectedStationInfo]:
+        """Fetch data from API endpoint."""
+        assert self.device.device
+        return await self.device.device.async_get_wifi_connected_station()
+
+
+class DevoloWifiGuestAccessGetCoordinator(
+    DevoloDataUpdateCoordinator[WifiGuestAccessGet]
+):
+    """Class to manage fetching data from the WifiGuestAccessGet endpoint."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        logger: Logger,
+        *,
+        config_entry: ConfigEntry,
+        name: str = SWITCH_GUEST_WIFI,
+        update_interval: timedelta | None = SHORT_UPDATE_INTERVAL,
+    ) -> None:
+        """Initialize global data updater."""
+        super().__init__(
+            hass,
+            logger,
+            config_entry=config_entry,
+            name=name,
+            update_interval=update_interval,
+        )
+        self.update_method = self.async_update_guest_wifi_status
+
+    async def async_update_guest_wifi_status(self) -> WifiGuestAccessGet:
+        """Fetch data from API endpoint."""
+        assert self.device.device
+        return await self.device.device.async_get_wifi_guest_access()
+
+
+class DevoloWifiNeighborAPsGetCoordinator(
+    DevoloDataUpdateCoordinator[list[NeighborAPInfo]]
+):
+    """Class to manage fetching data from the WifiNeighborAPsGet endpoint."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        logger: Logger,
+        *,
+        config_entry: ConfigEntry,
+        name: str = NEIGHBORING_WIFI_NETWORKS,
+        update_interval: timedelta | None = LONG_UPDATE_INTERVAL,
+    ) -> None:
+        """Initialize global data updater."""
+        super().__init__(
+            hass,
+            logger,
+            config_entry=config_entry,
+            name=name,
+            update_interval=update_interval,
+        )
+        self.update_method = self.async_update_wifi_neighbor_access_points
+
+    async def async_update_wifi_neighbor_access_points(self) -> list[NeighborAPInfo]:
+        """Fetch data from API endpoint."""
+        assert self.device.device
+        return await self.device.device.async_get_wifi_neighbor_access_points()
+
+
+@dataclass
+class DevoloHomeNetworkData:
+    """The devolo Home Network data."""
+
+    device: Device
+    coordinators: dict[str, DevoloDataUpdateCoordinator[Any]]

homeassistant/components/devolo_home_network/device_tracker.py

@@ -15,9 +15,8 @@ from homeassistant.helpers import entity_registry as er
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
-from . import DevoloHomeNetworkConfigEntry
 from .const import CONNECTED_WIFI_CLIENTS, DOMAIN, WIFI_APTYPE, WIFI_BANDS
-from .coordinator import DevoloDataUpdateCoordinator
+from .coordinator import DevoloDataUpdateCoordinator, DevoloHomeNetworkConfigEntry
 
 PARALLEL_UPDATES = 0
 

homeassistant/components/devolo_home_network/diagnostics.py

@@ -8,7 +8,7 @@ from homeassistant.components.diagnostics import async_redact_data
 from homeassistant.const import CONF_PASSWORD
 from homeassistant.core import HomeAssistant
 
-from . import DevoloHomeNetworkConfigEntry
+from .coordinator import DevoloHomeNetworkConfigEntry
 
 TO_REDACT = {CONF_PASSWORD}
 

homeassistant/components/devolo_home_network/entity.py

@@ -15,9 +15,8 @@ from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, Device
 from homeassistant.helpers.entity import Entity
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
-from . import DevoloHomeNetworkConfigEntry
 from .const import DOMAIN
-from .coordinator import DevoloDataUpdateCoordinator
+from .coordinator import DevoloDataUpdateCoordinator, DevoloHomeNetworkConfigEntry
 
 type _DataType = (
     LogicalNetwork

homeassistant/components/devolo_home_network/image.py

@@ -15,9 +15,8 @@ from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 from homeassistant.util import dt as dt_util
 
-from . import DevoloHomeNetworkConfigEntry
 from .const import IMAGE_GUEST_WIFI, SWITCH_GUEST_WIFI
-from .coordinator import DevoloDataUpdateCoordinator
+from .coordinator import DevoloDataUpdateCoordinator, DevoloHomeNetworkConfigEntry
 from .entity import DevoloCoordinatorEntity
 
 PARALLEL_UPDATES = 0

homeassistant/components/devolo_home_network/sensor.py

@@ -22,7 +22,6 @@ from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 from homeassistant.util.dt import utcnow
 
-from . import DevoloHomeNetworkConfigEntry
 from .const import (
     CONNECTED_PLC_DEVICES,
     CONNECTED_WIFI_CLIENTS,
@@ -31,7 +30,7 @@ from .const import (
     PLC_RX_RATE,
     PLC_TX_RATE,
 )
-from .coordinator import DevoloDataUpdateCoordinator
+from .coordinator import DevoloDataUpdateCoordinator, DevoloHomeNetworkConfigEntry
 from .entity import DevoloCoordinatorEntity
 
 PARALLEL_UPDATES = 0

homeassistant/components/devolo_home_network/switch.py

@@ -16,9 +16,8 @@ from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
-from . import DevoloHomeNetworkConfigEntry
 from .const import DOMAIN, SWITCH_GUEST_WIFI, SWITCH_LEDS
-from .coordinator import DevoloDataUpdateCoordinator
+from .coordinator import DevoloDataUpdateCoordinator, DevoloHomeNetworkConfigEntry
 from .entity import DevoloCoordinatorEntity
 
 PARALLEL_UPDATES = 0

homeassistant/components/devolo_home_network/update.py

@@ -21,9 +21,8 @@ from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
-from . import DevoloHomeNetworkConfigEntry
 from .const import DOMAIN, REGULAR_FIRMWARE
-from .coordinator import DevoloDataUpdateCoordinator
+from .coordinator import DevoloDataUpdateCoordinator, DevoloHomeNetworkConfigEntry
 from .entity import DevoloCoordinatorEntity
 
 PARALLEL_UPDATES = 0