evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Avoid using coordinator in config flow of APCUPSD (#112121)

Browse source 
* Separate data class out of coordinator

* Further fix the imports

* Update homeassistant/components/apcupsd/coordinator.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Use `or` to make it a bit cleaner when trying to find the UPS model

Co-authored-by: Robert Svensson <Kane610@users.noreply.github.com>

* Use or to make it a bit cleaner when trying to find the UPS model

Co-authored-by: Robert Svensson <Kane610@users.noreply.github.com>

* Use plain dict instead of `OrderedDict`

---------

Co-authored-by: J. Nick Koston <nick@koston.org>
Co-authored-by: Robert Svensson <Kane610@users.noreply.github.com>
This commit is contained in:
Yuxin Wang 2024-03-04 03:40:59 -05:00 • committed by GitHub
parent 38f9285bd6
commit 2c5510df30
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 49 additions and 53 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
homeassistant/components/apcupsd/binary_sensor.py
View file

@ -13,7 +13,8 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import DOMAIN, APCUPSdCoordinator
from .const import DOMAIN
from .coordinator import APCUPSdCoordinator
_LOGGER = logging.getLogger(__name__)
_DESCRIPTION = BinarySensorEntityDescription(
@ -53,7 +54,7 @@ class OnlineStatus(CoordinatorEntity[APCUPSdCoordinator], BinarySensorEntity):
super().__init__(coordinator, context=description.key.upper())
# Set up unique id and device info if serial number is available.
if (serial_no := coordinator.ups_serial_no) is not None:
if (serial_no := coordinator.data.serial_no) is not None:
self._attr_unique_id = f"{serial_no}_{description.key}"
self.entity_description = description
self._attr_device_info = coordinator.device_info

32
homeassistant/components/apcupsd/config_flow.py
View file

@ -1,17 +1,19 @@
"""Config flow for APCUPSd integration."""
from __future__ import annotations
import asyncio
from typing import Any
import aioapcaccess
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.helpers import selector
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.update_coordinator import UpdateFailed
from . import DOMAIN, APCUPSdCoordinator
from .const import CONNECTION_TIMEOUT, DOMAIN
from .coordinator import APCUPSdData
_PORT_SELECTOR = vol.All(
selector.NumberSelector(
@ -49,32 +51,22 @@ class ConfigFlowHandler(ConfigFlow, domain=DOMAIN):
self._async_abort_entries_match({CONF_HOST: host, CONF_PORT: port})
# Test the connection to the host and get the current status for serial number.
coordinator = APCUPSdCoordinator(self.hass, host, port)
await coordinator.async_request_refresh()
if isinstance(coordinator.last_exception, (UpdateFailed, TimeoutError)):
try:
async with asyncio.timeout(CONNECTION_TIMEOUT):
data = APCUPSdData(await aioapcaccess.request_status(host, port))
except (OSError, asyncio.IncompleteReadError, TimeoutError):
errors = {"base": "cannot_connect"}
return self.async_show_form(
step_id="user", data_schema=_SCHEMA, errors=errors
)
if not coordinator.data:
if not data:
return self.async_abort(reason="no_status")
# We _try_ to use the serial number of the UPS as the unique id since this field
# is not guaranteed to exist on all APC UPS models.
await self.async_set_unique_id(coordinator.ups_serial_no)
await self.async_set_unique_id(data.serial_no)
self._abort_if_unique_id_configured()
title = "APC UPS"
if coordinator.ups_name is not None:
title = coordinator.ups_name
elif coordinator.ups_model is not None:
title = coordinator.ups_model
elif coordinator.ups_serial_no is not None:
title = coordinator.ups_serial_no
return self.async_create_entry(
title=title,
data=user_input,
)
title = data.name or data.model or data.serial_no or "APC UPS"
return self.async_create_entry(title=title, data=user_input)

1
homeassistant/components/apcupsd/const.py
View file

@ -2,3 +2,4 @@
from typing import Final
DOMAIN: Final = "apcupsd"
CONNECTION_TIMEOUT: int = 10

59
homeassistant/components/apcupsd/coordinator.py
View file

@ -2,7 +2,6 @@
from __future__ import annotations
import asyncio
from collections import OrderedDict
from datetime import timedelta
import logging
from typing import Final
@ -19,14 +18,35 @@ from homeassistant.helpers.update_coordinator import (
UpdateFailed,
)
from .const import DOMAIN
from .const import CONNECTION_TIMEOUT, DOMAIN
_LOGGER = logging.getLogger(__name__)
UPDATE_INTERVAL: Final = timedelta(seconds=60)
REQUEST_REFRESH_COOLDOWN: Final = 5
class APCUPSdCoordinator(DataUpdateCoordinator[OrderedDict[str, str]]):
class APCUPSdData(dict[str, str]):
"""Store data about an APCUPSd and provide a few helper methods for easier accesses."""
@property
def name(self) -> str | None:
"""Return the name of the UPS, if available."""
return self.get("UPSNAME")
@property
def 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.
return self.get("APCMODEL") or self.get("MODEL")
@property
def serial_no(self) -> str | None:
"""Return the unique serial number of the UPS, if available."""
return self.get("SERIALNO")
class APCUPSdCoordinator(DataUpdateCoordinator[APCUPSdData]):
"""Store and coordinate the data retrieved from APCUPSd for all sensors.
For each entity to use, acts as the single point responsible for fetching
@ -52,46 +72,27 @@ class APCUPSdCoordinator(DataUpdateCoordinator[OrderedDict[str, str]]):
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:
"""Return the DeviceInfo of this APC UPS, if serial number is available."""
return DeviceInfo(
identifiers={(DOMAIN, self.ups_serial_no or self.config_entry.entry_id)},
model=self.ups_model,
identifiers={(DOMAIN, self.data.serial_no or self.config_entry.entry_id)},
model=self.data.model,
manufacturer="APC",
name=self.ups_name if self.ups_name else "APC UPS",
name=self.data.name or "APC UPS",
hw_version=self.data.get("FIRMWARE"),
sw_version=self.data.get("VERSION"),
)
async def _async_update_data(self) -> OrderedDict[str, str]:
async def _async_update_data(self) -> APCUPSdData:
"""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):
async with asyncio.timeout(CONNECTION_TIMEOUT):
try:
return await aioapcaccess.request_status(self._host, self._port)
data = await aioapcaccess.request_status(self._host, self._port)
return APCUPSdData(data)
except (OSError, asyncio.IncompleteReadError) as error:
raise UpdateFailed(error) from error

5
homeassistant/components/apcupsd/sensor.py
View file

@ -24,7 +24,8 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import DOMAIN, APCUPSdCoordinator
from .const import DOMAIN
from .coordinator import APCUPSdCoordinator
_LOGGER = logging.getLogger(__name__)
@ -493,7 +494,7 @@ class APCUPSdSensor(CoordinatorEntity[APCUPSdCoordinator], SensorEntity):
super().__init__(coordinator=coordinator, context=description.key.upper())
# Set up unique id and device info if serial number is available.
if (serial_no := coordinator.ups_serial_no) is not None:
if (serial_no := coordinator.data.serial_no) is not None:
self._attr_unique_id = f"{serial_no}_{description.key}"
self.entity_description = description