From 17470618206e05561bd381017a6476bb53356670 Mon Sep 17 00:00:00 2001 From: ollo69 <60491700+ollo69@users.noreply.github.com> Date: Tue, 17 May 2022 01:51:30 +0200 Subject: [PATCH] Enable NUT strict typing (#71913) --- .strict-typing | 1 + homeassistant/components/nut/__init__.py | 81 ++++++++++++--------- homeassistant/components/nut/config_flow.py | 62 +++++++++------- homeassistant/components/nut/sensor.py | 35 ++++++++- mypy.ini | 11 +++ 5 files changed, 126 insertions(+), 64 deletions(-) diff --git a/.strict-typing b/.strict-typing index 36ed1685e9f..4ec4fbff5ee 100644 --- a/.strict-typing +++ b/.strict-typing @@ -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.* diff --git a/homeassistant/components/nut/__init__.py b/homeassistant/components/nut/__init__.py index 775c512f946..28c41ccda3a 100644 --- a/homeassistant/components/nut/__init__.py +++ b/homeassistant/components/nut/__init__.py @@ -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: diff --git a/homeassistant/components/nut/config_flow.py b/homeassistant/components/nut/config_flow.py index fb0a2210a69..5ba8024878a 100644 --- a/homeassistant/components/nut/config_flow.py +++ b/homeassistant/components/nut/config_flow.py @@ -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) diff --git a/homeassistant/components/nut/sensor.py b/homeassistant/components/nut/sensor.py index 1992ba49635..6992e41366e 100644 --- a/homeassistant/components/nut/sensor.py +++ b/homeassistant/components/nut/sensor.py @@ -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()) diff --git a/mypy.ini b/mypy.ini index dc023da8d01..898c6b860f0 100644 --- a/mypy.ini +++ b/mypy.ini @@ -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