From e3599bc26f155fe079d74fa11e93d37e0d406d5a Mon Sep 17 00:00:00 2001 From: Joost Lekkerkerker Date: Sun, 26 Nov 2023 13:04:52 +0100 Subject: [PATCH] Move APCUPSd coordinator to separate file (#104540) --- homeassistant/components/apcupsd/__init__.py | 95 +--------------- homeassistant/components/apcupsd/const.py | 4 + .../components/apcupsd/coordinator.py | 102 ++++++++++++++++++ tests/components/apcupsd/test_init.py | 3 +- tests/components/apcupsd/test_sensor.py | 2 +- 5 files changed, 112 insertions(+), 94 deletions(-) create mode 100644 homeassistant/components/apcupsd/const.py create mode 100644 homeassistant/components/apcupsd/coordinator.py diff --git a/homeassistant/components/apcupsd/__init__.py b/homeassistant/components/apcupsd/__init__.py index 8431d282e7d..550e1014d2a 100644 --- a/homeassistant/components/apcupsd/__init__.py +++ b/homeassistant/components/apcupsd/__init__.py @@ -1,32 +1,20 @@ """Support for APCUPSd via its Network Information Server (NIS).""" from __future__ import annotations -import asyncio -from collections import OrderedDict -from datetime import timedelta import logging from typing import Final -from apcaccess import status - from homeassistant.config_entries import ConfigEntry from homeassistant.const import CONF_HOST, CONF_PORT, Platform from homeassistant.core import HomeAssistant import homeassistant.helpers.config_validation as cv -from homeassistant.helpers.debounce import Debouncer -from homeassistant.helpers.device_registry import DeviceInfo -from homeassistant.helpers.update_coordinator import ( - REQUEST_REFRESH_DEFAULT_IMMEDIATE, - DataUpdateCoordinator, - UpdateFailed, -) + +from .const import DOMAIN +from .coordinator import APCUPSdCoordinator _LOGGER = logging.getLogger(__name__) -DOMAIN: Final = "apcupsd" PLATFORMS: Final = (Platform.BINARY_SENSOR, Platform.SENSOR) -UPDATE_INTERVAL: Final = timedelta(seconds=60) -REQUEST_REFRESH_COOLDOWN: Final = 5 CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False) @@ -53,80 +41,3 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: if unload_ok and DOMAIN in hass.data: hass.data[DOMAIN].pop(entry.entry_id) return unload_ok - - -class APCUPSdCoordinator(DataUpdateCoordinator[OrderedDict[str, str]]): - """Store and coordinate the data retrieved from APCUPSd for all sensors. - - For each entity to use, acts as the single point responsible for fetching - updates from the server. - """ - - def __init__(self, hass: HomeAssistant, host: str, port: int) -> None: - """Initialize the data object.""" - super().__init__( - hass, - _LOGGER, - name=DOMAIN, - update_interval=UPDATE_INTERVAL, - request_refresh_debouncer=Debouncer( - hass, - _LOGGER, - cooldown=REQUEST_REFRESH_COOLDOWN, - immediate=REQUEST_REFRESH_DEFAULT_IMMEDIATE, - ), - ) - self._host = host - self._port = port - - @property - def ups_name(self) -> str | None: - """Return the name of the UPS, if available.""" - return self.data.get("UPSNAME") - - @property - def ups_model(self) -> str | None: - """Return the model of the UPS, if available.""" - # Different UPS models may report slightly different keys for model, here we - # try them all. - for model_key in ("APCMODEL", "MODEL"): - if model_key in self.data: - return self.data[model_key] - return None - - @property - def ups_serial_no(self) -> str | None: - """Return the unique serial number of the UPS, if available.""" - return self.data.get("SERIALNO") - - @property - def device_info(self) -> DeviceInfo | None: - """Return the DeviceInfo of this APC UPS, if serial number is available.""" - if not self.ups_serial_no: - return None - - return DeviceInfo( - identifiers={(DOMAIN, self.ups_serial_no)}, - model=self.ups_model, - manufacturer="APC", - name=self.ups_name if self.ups_name else "APC UPS", - hw_version=self.data.get("FIRMWARE"), - sw_version=self.data.get("VERSION"), - ) - - async def _async_update_data(self) -> OrderedDict[str, str]: - """Fetch the latest status from APCUPSd. - - Note that the result dict uses upper case for each resource, where our - integration uses lower cases as keys internally. - """ - - async with asyncio.timeout(10): - try: - raw = await self.hass.async_add_executor_job( - status.get, self._host, self._port - ) - result: OrderedDict[str, str] = status.parse(raw) - return result - except OSError as error: - raise UpdateFailed(error) from error diff --git a/homeassistant/components/apcupsd/const.py b/homeassistant/components/apcupsd/const.py new file mode 100644 index 00000000000..cacc9e29369 --- /dev/null +++ b/homeassistant/components/apcupsd/const.py @@ -0,0 +1,4 @@ +"""Constants for APCUPSd component.""" +from typing import Final + +DOMAIN: Final = "apcupsd" diff --git a/homeassistant/components/apcupsd/coordinator.py b/homeassistant/components/apcupsd/coordinator.py new file mode 100644 index 00000000000..321da56095a --- /dev/null +++ b/homeassistant/components/apcupsd/coordinator.py @@ -0,0 +1,102 @@ +"""Support for APCUPSd via its Network Information Server (NIS).""" +from __future__ import annotations + +import asyncio +from collections import OrderedDict +from datetime import timedelta +import logging +from typing import Final + +from apcaccess import status + +from homeassistant.core import HomeAssistant +from homeassistant.helpers.debounce import Debouncer +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.update_coordinator import ( + REQUEST_REFRESH_DEFAULT_IMMEDIATE, + DataUpdateCoordinator, + UpdateFailed, +) + +from .const import DOMAIN + +_LOGGER = logging.getLogger(__name__) +UPDATE_INTERVAL: Final = timedelta(seconds=60) +REQUEST_REFRESH_COOLDOWN: Final = 5 + + +class APCUPSdCoordinator(DataUpdateCoordinator[OrderedDict[str, str]]): + """Store and coordinate the data retrieved from APCUPSd for all sensors. + + For each entity to use, acts as the single point responsible for fetching + updates from the server. + """ + + def __init__(self, hass: HomeAssistant, host: str, port: int) -> None: + """Initialize the data object.""" + super().__init__( + hass, + _LOGGER, + name=DOMAIN, + update_interval=UPDATE_INTERVAL, + request_refresh_debouncer=Debouncer( + hass, + _LOGGER, + cooldown=REQUEST_REFRESH_COOLDOWN, + immediate=REQUEST_REFRESH_DEFAULT_IMMEDIATE, + ), + ) + self._host = host + self._port = port + + @property + def ups_name(self) -> str | None: + """Return the name of the UPS, if available.""" + return self.data.get("UPSNAME") + + @property + def ups_model(self) -> str | None: + """Return the model of the UPS, if available.""" + # Different UPS models may report slightly different keys for model, here we + # try them all. + for model_key in ("APCMODEL", "MODEL"): + if model_key in self.data: + return self.data[model_key] + return None + + @property + def ups_serial_no(self) -> str | None: + """Return the unique serial number of the UPS, if available.""" + return self.data.get("SERIALNO") + + @property + def device_info(self) -> DeviceInfo | None: + """Return the DeviceInfo of this APC UPS, if serial number is available.""" + if not self.ups_serial_no: + return None + + return DeviceInfo( + identifiers={(DOMAIN, self.ups_serial_no)}, + model=self.ups_model, + manufacturer="APC", + name=self.ups_name if self.ups_name else "APC UPS", + hw_version=self.data.get("FIRMWARE"), + sw_version=self.data.get("VERSION"), + ) + + async def _async_update_data(self) -> OrderedDict[str, str]: + """Fetch the latest status from APCUPSd. + + Note that the result dict uses upper case for each resource, where our + integration uses lower cases as keys internally. + """ + + async with asyncio.timeout(10): + try: + raw = await self.hass.async_add_executor_job( + status.get, self._host, self._port + ) + result: OrderedDict[str, str] = status.parse(raw) + return result + except OSError as error: + raise UpdateFailed(error) from error diff --git a/tests/components/apcupsd/test_init.py b/tests/components/apcupsd/test_init.py index 43eab28eb46..756fa07f120 100644 --- a/tests/components/apcupsd/test_init.py +++ b/tests/components/apcupsd/test_init.py @@ -4,7 +4,8 @@ from unittest.mock import patch import pytest -from homeassistant.components.apcupsd import DOMAIN, UPDATE_INTERVAL +from homeassistant.components.apcupsd.const import DOMAIN +from homeassistant.components.apcupsd.coordinator import UPDATE_INTERVAL from homeassistant.config_entries import SOURCE_USER, ConfigEntryState from homeassistant.const import STATE_UNAVAILABLE from homeassistant.core import HomeAssistant diff --git a/tests/components/apcupsd/test_sensor.py b/tests/components/apcupsd/test_sensor.py index 73546613002..bff1b858216 100644 --- a/tests/components/apcupsd/test_sensor.py +++ b/tests/components/apcupsd/test_sensor.py @@ -3,7 +3,7 @@ from datetime import timedelta from unittest.mock import patch -from homeassistant.components.apcupsd import REQUEST_REFRESH_COOLDOWN +from homeassistant.components.apcupsd.coordinator import REQUEST_REFRESH_COOLDOWN from homeassistant.components.sensor import ( ATTR_STATE_CLASS, SensorDeviceClass,