From 5819091af7a1f5dfd4d804aca2c289ceaa6a74b1 Mon Sep 17 00:00:00 2001 From: Jan-Philipp Benecke Date: Wed, 30 Aug 2023 22:31:51 +0200 Subject: [PATCH] Move octoprint coordinator to its own file (#99359) Move octoprint coordinator to own file --- .coveragerc | 1 + .../components/octoprint/__init__.py | 82 +--------------- .../components/octoprint/coordinator.py | 93 +++++++++++++++++++ 3 files changed, 96 insertions(+), 80 deletions(-) create mode 100644 homeassistant/components/octoprint/coordinator.py diff --git a/.coveragerc b/.coveragerc index 97ed97ef293..0532e5ed0a1 100644 --- a/.coveragerc +++ b/.coveragerc @@ -843,6 +843,7 @@ omit = homeassistant/components/obihai/connectivity.py homeassistant/components/obihai/sensor.py homeassistant/components/octoprint/__init__.py + homeassistant/components/octoprint/coordinator.py homeassistant/components/oem/climate.py homeassistant/components/ohmconnect/sensor.py homeassistant/components/ombi/* diff --git a/homeassistant/components/octoprint/__init__.py b/homeassistant/components/octoprint/__init__.py index 07b2fa1a15d..5fd2182ca00 100644 --- a/homeassistant/components/octoprint/__init__.py +++ b/homeassistant/components/octoprint/__init__.py @@ -1,15 +1,11 @@ """Support for monitoring OctoPrint 3D printers.""" from __future__ import annotations -from datetime import timedelta import logging -from typing import cast import aiohttp -from pyoctoprintapi import ApiError, OctoprintClient, PrinterOffline -from pyoctoprintapi.exceptions import UnauthorizedException +from pyoctoprintapi import OctoprintClient import voluptuous as vol -from yarl import URL from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry from homeassistant.const import ( @@ -27,15 +23,12 @@ from homeassistant.const import ( Platform, ) from homeassistant.core import Event, HomeAssistant, callback -from homeassistant.exceptions import ConfigEntryAuthFailed import homeassistant.helpers.config_validation as cv -from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.typing import ConfigType -from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.util import slugify as util_slugify -import homeassistant.util.dt as dt_util from .const import DOMAIN +from .coordinator import OctoprintDataUpdateCoordinator _LOGGER = logging.getLogger(__name__) @@ -209,74 +202,3 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: hass.data[DOMAIN].pop(entry.entry_id) return unload_ok - - -class OctoprintDataUpdateCoordinator(DataUpdateCoordinator): - """Class to manage fetching Octoprint data.""" - - config_entry: ConfigEntry - - def __init__( - self, - hass: HomeAssistant, - octoprint: OctoprintClient, - config_entry: ConfigEntry, - interval: int, - ) -> None: - """Initialize.""" - super().__init__( - hass, - _LOGGER, - name=f"octoprint-{config_entry.entry_id}", - update_interval=timedelta(seconds=interval), - ) - self.config_entry = config_entry - self._octoprint = octoprint - self._printer_offline = False - self.data = {"printer": None, "job": None, "last_read_time": None} - - async def _async_update_data(self): - """Update data via API.""" - printer = None - try: - job = await self._octoprint.get_job_info() - except UnauthorizedException as err: - raise ConfigEntryAuthFailed from err - except ApiError as err: - raise UpdateFailed(err) from err - - # If octoprint is on, but the printer is disconnected - # printer will return a 409, so continue using the last - # reading if there is one - try: - printer = await self._octoprint.get_printer_info() - except PrinterOffline: - if not self._printer_offline: - _LOGGER.debug("Unable to retrieve printer information: Printer offline") - self._printer_offline = True - except UnauthorizedException as err: - raise ConfigEntryAuthFailed from err - except ApiError as err: - raise UpdateFailed(err) from err - else: - self._printer_offline = False - - return {"job": job, "printer": printer, "last_read_time": dt_util.utcnow()} - - @property - def device_info(self) -> DeviceInfo: - """Device info.""" - unique_id = cast(str, self.config_entry.unique_id) - configuration_url = URL.build( - scheme=self.config_entry.data[CONF_SSL] and "https" or "http", - host=self.config_entry.data[CONF_HOST], - port=self.config_entry.data[CONF_PORT], - path=self.config_entry.data[CONF_PATH], - ) - - return DeviceInfo( - identifiers={(DOMAIN, unique_id)}, - manufacturer="OctoPrint", - name="OctoPrint", - configuration_url=str(configuration_url), - ) diff --git a/homeassistant/components/octoprint/coordinator.py b/homeassistant/components/octoprint/coordinator.py new file mode 100644 index 00000000000..c6ce8fa66b7 --- /dev/null +++ b/homeassistant/components/octoprint/coordinator.py @@ -0,0 +1,93 @@ +"""The data update coordinator for OctoPrint.""" +from __future__ import annotations + +from datetime import timedelta +import logging +from typing import cast + +from pyoctoprintapi import ApiError, OctoprintClient, PrinterOffline +from pyoctoprintapi.exceptions import UnauthorizedException +from yarl import URL + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import CONF_HOST, CONF_PATH, CONF_PORT, CONF_SSL +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import ConfigEntryAuthFailed +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed +import homeassistant.util.dt as dt_util + +from .const import DOMAIN + +_LOGGER = logging.getLogger(__name__) + + +class OctoprintDataUpdateCoordinator(DataUpdateCoordinator): + """Class to manage fetching Octoprint data.""" + + config_entry: ConfigEntry + + def __init__( + self, + hass: HomeAssistant, + octoprint: OctoprintClient, + config_entry: ConfigEntry, + interval: int, + ) -> None: + """Initialize.""" + super().__init__( + hass, + _LOGGER, + name=f"octoprint-{config_entry.entry_id}", + update_interval=timedelta(seconds=interval), + ) + self.config_entry = config_entry + self._octoprint = octoprint + self._printer_offline = False + self.data = {"printer": None, "job": None, "last_read_time": None} + + async def _async_update_data(self): + """Update data via API.""" + printer = None + try: + job = await self._octoprint.get_job_info() + except UnauthorizedException as err: + raise ConfigEntryAuthFailed from err + except ApiError as err: + raise UpdateFailed(err) from err + + # If octoprint is on, but the printer is disconnected + # printer will return a 409, so continue using the last + # reading if there is one + try: + printer = await self._octoprint.get_printer_info() + except PrinterOffline: + if not self._printer_offline: + _LOGGER.debug("Unable to retrieve printer information: Printer offline") + self._printer_offline = True + except UnauthorizedException as err: + raise ConfigEntryAuthFailed from err + except ApiError as err: + raise UpdateFailed(err) from err + else: + self._printer_offline = False + + return {"job": job, "printer": printer, "last_read_time": dt_util.utcnow()} + + @property + def device_info(self) -> DeviceInfo: + """Device info.""" + unique_id = cast(str, self.config_entry.unique_id) + configuration_url = URL.build( + scheme=self.config_entry.data[CONF_SSL] and "https" or "http", + host=self.config_entry.data[CONF_HOST], + port=self.config_entry.data[CONF_PORT], + path=self.config_entry.data[CONF_PATH], + ) + + return DeviceInfo( + identifiers={(DOMAIN, unique_id)}, + manufacturer="OctoPrint", + name="OctoPrint", + configuration_url=str(configuration_url), + )