diff --git a/homeassistant/components/prusalink/__init__.py b/homeassistant/components/prusalink/__init__.py index 2582a920102620f34bbbeb240af442661d498772..9d6096748ddd9847c4d08e74a8ccef2d21f8f962 100644 --- a/homeassistant/components/prusalink/__init__.py +++ b/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.""" diff --git a/homeassistant/components/prusalink/button.py b/homeassistant/components/prusalink/button.py index d70356f04d1a90cdd2b9897d81d7572ef8fb03b1..0ad7e531d4617b7703a98af514a41bd950859960 100644 --- a/homeassistant/components/prusalink/button.py +++ b/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) diff --git a/homeassistant/components/prusalink/camera.py b/homeassistant/components/prusalink/camera.py index cc625b7ef57f9ec826f74b1ed771f2bf5cd675af..2185c5f3cf69e7d6e0b98268642416f477728511 100644 --- a/homeassistant/components/prusalink/camera.py +++ b/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( diff --git a/homeassistant/components/prusalink/coordinator.py b/homeassistant/components/prusalink/coordinator.py new file mode 100644 index 0000000000000000000000000000000000000000..7d4526a8b454aaa68ba4707a13df40127d609e27 --- /dev/null +++ b/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() diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index e8d357726bc3ca2e8ab993b64bd6a2c6d89839e2..80998d680d22f08df6374159df952f6d4bbe8c4e 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/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)