Move prusalink coordinators to separate module (#117495)

homeassistant/components/prusalink/__init__.py

@@ -2,15 +2,8 @@
 
 from __future__ import annotations
 
-from abc import ABC, abstractmethod
-import asyncio
-from datetime import timedelta
-import logging
-from time import monotonic
-from typing import TypeVar
-
-from pyprusalink import JobInfo, LegacyPrinterStatus, PrinterStatus, PrusaLink
-from pyprusalink.types import InvalidAuth, PrusaLinkError
+from pyprusalink import PrusaLink
+from pyprusalink.types import InvalidAuth
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import (
@@ -20,22 +13,23 @@ from homeassistant.const import (
     CONF_USERNAME,
     Platform,
 )
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryError
 from homeassistant.helpers import issue_registry as ir
 from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.httpx_client import get_async_client
-from homeassistant.helpers.update_coordinator import (
-    CoordinatorEntity,
-    DataUpdateCoordinator,
-    UpdateFailed,
-)
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .config_flow import ConfigFlow
 from .const import DOMAIN
+from .coordinator import (
+    JobUpdateCoordinator,
+    LegacyStatusCoordinator,
+    PrusaLinkUpdateCoordinator,
+    StatusCoordinator,
+)
 
 PLATFORMS: list[Platform] = [Platform.BUTTON, Platform.CAMERA, Platform.SENSOR]
-_LOGGER = logging.getLogger(__name__)
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@@ -129,78 +123,6 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     return unload_ok
 
 
-T = TypeVar("T", PrinterStatus, LegacyPrinterStatus, JobInfo)
-
-
-class PrusaLinkUpdateCoordinator(DataUpdateCoordinator[T], ABC):  # pylint: disable=hass-enforce-coordinator-module
-    """Update coordinator for the printer."""
-
-    config_entry: ConfigEntry
-    expect_change_until = 0.0
-
-    def __init__(self, hass: HomeAssistant, api: PrusaLink) -> None:
-        """Initialize the update coordinator."""
-        self.api = api
-
-        super().__init__(
-            hass, _LOGGER, name=DOMAIN, update_interval=self._get_update_interval(None)
-        )
-
-    async def _async_update_data(self) -> T:
-        """Update the data."""
-        try:
-            async with asyncio.timeout(5):
-                data = await self._fetch_data()
-        except InvalidAuth:
-            raise UpdateFailed("Invalid authentication") from None
-        except PrusaLinkError as err:
-            raise UpdateFailed(str(err)) from err
-
-        self.update_interval = self._get_update_interval(data)
-        return data
-
-    @abstractmethod
-    async def _fetch_data(self) -> T:
-        """Fetch the actual data."""
-        raise NotImplementedError
-
-    @callback
-    def expect_change(self) -> None:
-        """Expect a change."""
-        self.expect_change_until = monotonic() + 30
-
-    def _get_update_interval(self, data: T) -> timedelta:
-        """Get new update interval."""
-        if self.expect_change_until > monotonic():
-            return timedelta(seconds=5)
-
-        return timedelta(seconds=30)
-
-
-class StatusCoordinator(PrusaLinkUpdateCoordinator[PrinterStatus]):  # pylint: disable=hass-enforce-coordinator-module
-    """Printer update coordinator."""
-
-    async def _fetch_data(self) -> PrinterStatus:
-        """Fetch the printer data."""
-        return await self.api.get_status()
-
-
-class LegacyStatusCoordinator(PrusaLinkUpdateCoordinator[LegacyPrinterStatus]):  # pylint: disable=hass-enforce-coordinator-module
-    """Printer legacy update coordinator."""
-
-    async def _fetch_data(self) -> LegacyPrinterStatus:
-        """Fetch the printer data."""
-        return await self.api.get_legacy_printer()
-
-
-class JobUpdateCoordinator(PrusaLinkUpdateCoordinator[JobInfo]):  # pylint: disable=hass-enforce-coordinator-module
-    """Job update coordinator."""
-
-    async def _fetch_data(self) -> JobInfo:
-        """Fetch the printer data."""
-        return await self.api.get_job()
-
-
 class PrusaLinkEntity(CoordinatorEntity[PrusaLinkUpdateCoordinator]):
     """Defines a base PrusaLink entity."""
 

homeassistant/components/prusalink/button.py

@@ -15,7 +15,9 @@ from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from . import DOMAIN, PrusaLinkEntity, PrusaLinkUpdateCoordinator
+from . import PrusaLinkEntity
+from .const import DOMAIN
+from .coordinator import PrusaLinkUpdateCoordinator
 
 T = TypeVar("T", PrinterStatus, LegacyPrinterStatus, JobInfo)
 

homeassistant/components/prusalink/camera.py

@@ -9,7 +9,9 @@ from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from . import DOMAIN, JobUpdateCoordinator, PrusaLinkEntity
+from . import PrusaLinkEntity
+from .const import DOMAIN
+from .coordinator import JobUpdateCoordinator
 
 
 async def async_setup_entry(

homeassistant/components/prusalink/coordinator.py

@@ -0,0 +1,93 @@
+"""Coordinators for the PrusaLink integration."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+import asyncio
+from datetime import timedelta
+import logging
+from time import monotonic
+from typing import TypeVar
+
+from pyprusalink import JobInfo, LegacyPrinterStatus, PrinterStatus, PrusaLink
+from pyprusalink.types import InvalidAuth, PrusaLinkError
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+T = TypeVar("T", PrinterStatus, LegacyPrinterStatus, JobInfo)
+
+
+class PrusaLinkUpdateCoordinator(DataUpdateCoordinator[T], ABC):
+    """Update coordinator for the printer."""
+
+    config_entry: ConfigEntry
+    expect_change_until = 0.0
+
+    def __init__(self, hass: HomeAssistant, api: PrusaLink) -> None:
+        """Initialize the update coordinator."""
+        self.api = api
+
+        super().__init__(
+            hass, _LOGGER, name=DOMAIN, update_interval=self._get_update_interval(None)
+        )
+
+    async def _async_update_data(self) -> T:
+        """Update the data."""
+        try:
+            async with asyncio.timeout(5):
+                data = await self._fetch_data()
+        except InvalidAuth:
+            raise UpdateFailed("Invalid authentication") from None
+        except PrusaLinkError as err:
+            raise UpdateFailed(str(err)) from err
+
+        self.update_interval = self._get_update_interval(data)
+        return data
+
+    @abstractmethod
+    async def _fetch_data(self) -> T:
+        """Fetch the actual data."""
+        raise NotImplementedError
+
+    @callback
+    def expect_change(self) -> None:
+        """Expect a change."""
+        self.expect_change_until = monotonic() + 30
+
+    def _get_update_interval(self, data: T) -> timedelta:
+        """Get new update interval."""
+        if self.expect_change_until > monotonic():
+            return timedelta(seconds=5)
+
+        return timedelta(seconds=30)
+
+
+class StatusCoordinator(PrusaLinkUpdateCoordinator[PrinterStatus]):
+    """Printer update coordinator."""
+
+    async def _fetch_data(self) -> PrinterStatus:
+        """Fetch the printer data."""
+        return await self.api.get_status()
+
+
+class LegacyStatusCoordinator(PrusaLinkUpdateCoordinator[LegacyPrinterStatus]):
+    """Printer legacy update coordinator."""
+
+    async def _fetch_data(self) -> LegacyPrinterStatus:
+        """Fetch the printer data."""
+        return await self.api.get_legacy_printer()
+
+
+class JobUpdateCoordinator(PrusaLinkUpdateCoordinator[JobInfo]):
+    """Job update coordinator."""
+
+    async def _fetch_data(self) -> JobInfo:
+        """Fetch the printer data."""
+        return await self.api.get_job()

homeassistant/components/prusalink/sensor.py

@@ -29,7 +29,9 @@ from homeassistant.helpers.typing import StateType
 from homeassistant.util.dt import utcnow
 from homeassistant.util.variance import ignore_variance
 
-from . import DOMAIN, PrusaLinkEntity, PrusaLinkUpdateCoordinator
+from . import PrusaLinkEntity
+from .const import DOMAIN
+from .coordinator import PrusaLinkUpdateCoordinator
 
 T = TypeVar("T", PrinterStatus, LegacyPrinterStatus, JobInfo)