Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Enable NUT strict typing (#71913)

pull/71978/head
ollo69 2022-05-17 01:51:30 +02:00 committed by GitHub
parent 9092dcacea
commit 1747061820
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 126 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -165,6 +165,7 @@ homeassistant.components.no_ip.*
homeassistant.components.notify.*
homeassistant.components.notion.*
homeassistant.components.number.*
homeassistant.components.nut.*
homeassistant.components.oncue.*
homeassistant.components.onewire.*
homeassistant.components.open_meteo.*

81
homeassistant/components/nut/__init__.py
View File

@ -1,4 +1,7 @@
"""The nut component."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
import logging
@ -7,9 +10,6 @@ from pynut2.nut2 import PyNUTClient, PyNUTError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_MANUFACTURER,
ATTR_MODEL,
ATTR_SW_VERSION,
CONF_ALIAS,
CONF_HOST,
CONF_PASSWORD,
@ -59,7 +59,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
data = PyNUTData(host, port, alias, username, password)
async def async_update_data():
async def async_update_data() -> dict[str, str]:
"""Fetch data from NUT."""
async with async_timeout.timeout(10):
await hass.async_add_executor_job(data.update)
@ -98,9 +98,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, unique_id)},
name=data.name.title(),
manufacturer=data.device_info.get(ATTR_MANUFACTURER),
model=data.device_info.get(ATTR_MODEL),
sw_version=data.device_info.get(ATTR_SW_VERSION),
manufacturer=data.device_info.manufacturer,
model=data.device_info.model,
sw_version=data.device_info.firmware,
)
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
@ -120,7 +120,7 @@ async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> Non
await hass.config_entries.async_reload(entry.entry_id)
def _manufacturer_from_status(status):
def _manufacturer_from_status(status: dict[str, str]) -> str | None:
"""Find the best manufacturer value from the status."""
return (
status.get("device.mfr")
@ -130,7 +130,7 @@ def _manufacturer_from_status(status):
)
def _model_from_status(status):
def _model_from_status(status: dict[str, str]) -> str | None:
"""Find the best model value from the status."""
return (
status.get("device.model")
@ -139,12 +139,12 @@ def _model_from_status(status):
)
def _firmware_from_status(status):
def _firmware_from_status(status: dict[str, str]) -> str | None:
"""Find the best firmware value from the status."""
return status.get("ups.firmware") or status.get("ups.firmware.aux")
def _serial_from_status(status):
def _serial_from_status(status: dict[str, str]) -> str | None:
"""Find the best serialvalue from the status."""
serial = status.get("device.serial") or status.get("ups.serial")
if serial and (
@ -154,7 +154,7 @@ def _serial_from_status(status):
return serial
def _unique_id_from_status(status):
def _unique_id_from_status(status: dict[str, str]) -> str | None:
"""Find the best unique id value from the status."""
serial = _serial_from_status(status)
# We must have a serial for this to be unique
@ -174,6 +174,15 @@ def _unique_id_from_status(status):
return "_".join(unique_id_group)
@dataclass
class NUTDeviceInfo:
"""Device information for NUT."""
manufacturer: str | None = None
model: str | None = None
firmware: str | None = None
class PyNUTData:
"""Stores the data retrieved from NUT.
@ -181,7 +190,14 @@ class PyNUTData:
updates from the server.
"""
def __init__(self, host, port, alias, username, password):
def __init__(
self,
host: str,
port: int,
alias: str | None,
username: str | None,
password: str | None,
) -> None:
"""Initialize the data object."""
self._host = host
@ -190,29 +206,29 @@ class PyNUTData:
# Establish client with persistent=False to open/close connection on
# each update call. This is more reliable with async.
self._client = PyNUTClient(self._host, port, username, password, 5, False)
self.ups_list = None
self._status = None
self._device_info = None
self.ups_list: dict[str, str] | None = None
self._status: dict[str, str] | None = None
self._device_info: NUTDeviceInfo | None = None
@property
def status(self):
def status(self) -> dict[str, str] | None:
"""Get latest update if throttle allows. Return status."""
return self._status
@property
def name(self):
def name(self) -> str:
"""Return the name of the ups."""
return self._alias
return self._alias or f"Nut-{self._host}"
@property
def device_info(self):
def device_info(self) -> NUTDeviceInfo:
"""Return the device info for the ups."""
return self._device_info or {}
return self._device_info or NUTDeviceInfo()
def _get_alias(self):
def _get_alias(self) -> str | None:
"""Get the ups alias from NUT."""
try:
ups_list = self._client.list_ups()
ups_list: dict[str, str] = self._client.list_ups()
except PyNUTError as err:
_LOGGER.error("Failure getting NUT ups alias, %s", err)
return None
@ -224,7 +240,7 @@ class PyNUTData:
self.ups_list = ups_list
return list(ups_list)[0]
def _get_device_info(self):
def _get_device_info(self) -> NUTDeviceInfo | None:
"""Get the ups device info from NUT."""
if not self._status:
return None
@ -232,27 +248,24 @@ class PyNUTData:
manufacturer = _manufacturer_from_status(self._status)
model = _model_from_status(self._status)
firmware = _firmware_from_status(self._status)
device_info = {}
if model:
device_info[ATTR_MODEL] = model
if manufacturer:
device_info[ATTR_MANUFACTURER] = manufacturer
if firmware:
device_info[ATTR_SW_VERSION] = firmware
device_info = NUTDeviceInfo(manufacturer, model, firmware)
return device_info
def _get_status(self):
def _get_status(self) -> dict[str, str] | None:
"""Get the ups status from NUT."""
if self._alias is None:
self._alias = self._get_alias()
try:
return self._client.list_vars(self._alias)
status: dict[str, str] = self._client.list_vars(self._alias)
except (PyNUTError, ConnectionResetError) as err:
_LOGGER.debug("Error getting NUT vars for host %s: %s", self._host, err)
return None
def update(self):
return status
def update(self) -> None:
"""Fetch the latest status from NUT."""
self._status = self._get_status()
if self._device_info is None:

62
homeassistant/components/nut/config_flow.py
View File

@ -2,11 +2,13 @@
from __future__ import annotations
import logging
from typing import Any
import voluptuous as vol
from homeassistant import config_entries, core, exceptions
from homeassistant import exceptions
from homeassistant.components import zeroconf
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
from homeassistant.const import (
CONF_ALIAS,
CONF_BASE,
@ -16,7 +18,7 @@ from homeassistant.const import (
CONF_SCAN_INTERVAL,
CONF_USERNAME,
)
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
from . import PyNUTData
@ -42,12 +44,12 @@ def _base_schema(discovery_info: zeroconf.ZeroconfServiceInfo | None) -> vol.Sch
return vol.Schema(base_schema)
def _ups_schema(ups_list):
def _ups_schema(ups_list: dict[str, str]) -> vol.Schema:
"""UPS selection schema."""
return vol.Schema({vol.Required(CONF_ALIAS): vol.In(ups_list)})
async def validate_input(hass: core.HomeAssistant, data):
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
"""Validate the user input allows us to connect.
Data has the keys from _base_schema with values provided by the user.
@ -59,15 +61,15 @@ async def validate_input(hass: core.HomeAssistant, data):
username = data.get(CONF_USERNAME)
password = data.get(CONF_PASSWORD)
data = PyNUTData(host, port, alias, username, password)
await hass.async_add_executor_job(data.update)
if not (status := data.status):
nut_data = PyNUTData(host, port, alias, username, password)
await hass.async_add_executor_job(nut_data.update)
if not (status := nut_data.status):
raise CannotConnect
return {"ups_list": data.ups_list, "available_resources": status}
return {"ups_list": nut_data.ups_list, "available_resources": status}
def _format_host_port_alias(user_input):
def _format_host_port_alias(user_input: dict[str, Any]) -> str:
"""Format a host, port, and alias so it can be used for comparison or display."""
host = user_input[CONF_HOST]
port = user_input[CONF_PORT]
@ -77,17 +79,17 @@ def _format_host_port_alias(user_input):
return f"{host}:{port}"
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
class NutConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Network UPS Tools (NUT)."""
VERSION = 1
def __init__(self):
def __init__(self) -> None:
"""Initialize the nut config flow."""
self.nut_config = {}
self.nut_config: dict[str, Any] = {}
self.discovery_info: zeroconf.ZeroconfServiceInfo | None = None
self.ups_list = None
self.title = None
self.ups_list: dict[str, str] | None = None
self.title: str | None = None
async def async_step_zeroconf(
self, discovery_info: zeroconf.ZeroconfServiceInfo
@ -101,9 +103,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
}
return await self.async_step_user()
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the user input."""
errors = {}
errors: dict[str, str] = {}
if user_input is not None:
if self.discovery_info:
user_input.update(
@ -129,9 +133,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
step_id="user", data_schema=_base_schema(self.discovery_info), errors=errors
)
async def async_step_ups(self, user_input=None):
async def async_step_ups(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the picking the ups."""
errors = {}
errors: dict[str, str] = {}
if user_input is not None:
self.nut_config.update(user_input)
@ -144,20 +150,22 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_show_form(
step_id="ups",
data_schema=_ups_schema(self.ups_list),
data_schema=_ups_schema(self.ups_list or {}),
errors=errors,
)
def _host_port_alias_already_configured(self, user_input):
def _host_port_alias_already_configured(self, user_input: dict[str, Any]) -> bool:
"""See if we already have a nut entry matching user input configured."""
existing_host_port_aliases = {
_format_host_port_alias(entry.data)
_format_host_port_alias(dict(entry.data))
for entry in self._async_current_entries()
if CONF_HOST in entry.data
}
return _format_host_port_alias(user_input) in existing_host_port_aliases
async def _async_validate_or_error(self, config):
async def _async_validate_or_error(
self, config: dict[str, Any]
) -> tuple[dict[str, Any], dict[str, str]]:
errors = {}
info = {}
try:
@ -171,19 +179,21 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@staticmethod
@callback
def async_get_options_flow(config_entry):
def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
"""Get the options flow for this handler."""
return OptionsFlowHandler(config_entry)
class OptionsFlowHandler(config_entries.OptionsFlow):
class OptionsFlowHandler(OptionsFlow):
"""Handle a option flow for nut."""
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(self, user_input=None):
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle options flow."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

35
homeassistant/components/nut/sensor.py
View File

@ -1,11 +1,18 @@
"""Provides a sensor to track various status aspects of a UPS."""
from __future__ import annotations
from dataclasses import asdict
import logging
from typing import cast
from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_UNKNOWN
from homeassistant.const import (
ATTR_MANUFACTURER,
ATTR_MODEL,
ATTR_SW_VERSION,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
@ -26,9 +33,27 @@ from .const import (
STATE_TYPES,
)
NUT_DEV_INFO_TO_DEV_INFO: dict[str, str] = {
"manufacturer": ATTR_MANUFACTURER,
"model": ATTR_MODEL,
"firmware": ATTR_SW_VERSION,
}
_LOGGER = logging.getLogger(__name__)
def _get_nut_device_info(data: PyNUTData) -> DeviceInfo:
"""Return a DeviceInfo object filled with NUT device info."""
nut_dev_infos = asdict(data.device_info)
nut_infos = {
info_key: nut_dev_infos[nut_key]
for nut_key, info_key in NUT_DEV_INFO_TO_DEV_INFO.items()
if nut_dev_infos[nut_key] is not None
}
return cast(DeviceInfo, nut_infos)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
@ -64,6 +89,8 @@ async def async_setup_entry(
class NUTSensor(CoordinatorEntity, SensorEntity):
"""Representation of a sensor entity for NUT status values."""
coordinator: DataUpdateCoordinator[dict[str, str]]
def __init__(
self,
coordinator: DataUpdateCoordinator,
@ -82,10 +109,10 @@ class NUTSensor(CoordinatorEntity, SensorEntity):
identifiers={(DOMAIN, unique_id)},
name=device_name,
)
self._attr_device_info.update(data.device_info)
self._attr_device_info.update(_get_nut_device_info(data))
@property
def native_value(self):
def native_value(self) -> str | None:
"""Return entity state from ups."""
status = self.coordinator.data
if self.entity_description.key == KEY_STATUS_DISPLAY:
@ -93,7 +120,7 @@ class NUTSensor(CoordinatorEntity, SensorEntity):
return status.get(self.entity_description.key)
def _format_display_state(status):
def _format_display_state(status: dict[str, str]) -> str:
"""Return UPS display state."""
try:
return " ".join(STATE_TYPES[state] for state in status[KEY_STATUS].split())

11
mypy.ini
View File

@ -1578,6 +1578,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.nut.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.oncue.*]
check_untyped_defs = true
disallow_incomplete_defs = true