From 6089aef0726da47d039ca4eb04de80ffa38c33d3 Mon Sep 17 00:00:00 2001 From: Marc Mueller <30130371+cdce8p@users.noreply.github.com> Date: Tue, 23 Nov 2021 22:30:22 +0100 Subject: [PATCH] Enable strict typing - wallbox (#59301) --- .strict-typing | 1 + homeassistant/components/wallbox/__init__.py | 28 +++++++-------- .../components/wallbox/config_flow.py | 21 ++++++++--- homeassistant/components/wallbox/number.py | 36 ++++++++++++------- homeassistant/components/wallbox/sensor.py | 35 ++++++++++++------ mypy.ini | 11 ++++++ 6 files changed, 91 insertions(+), 41 deletions(-) diff --git a/.strict-typing b/.strict-typing index daed6f3434a..318a1367bef 100644 --- a/.strict-typing +++ b/.strict-typing @@ -141,6 +141,7 @@ homeassistant.components.vacuum.* homeassistant.components.vallox.* homeassistant.components.velbus.* homeassistant.components.vlc_telnet.* +homeassistant.components.wallbox.* homeassistant.components.water_heater.* homeassistant.components.watttime.* homeassistant.components.weather.* diff --git a/homeassistant/components/wallbox/__init__.py b/homeassistant/components/wallbox/__init__.py index b1604a37c6f..c40d3fb37a8 100644 --- a/homeassistant/components/wallbox/__init__.py +++ b/homeassistant/components/wallbox/__init__.py @@ -1,7 +1,10 @@ """The Wallbox integration.""" +from __future__ import annotations + from datetime import timedelta from http import HTTPStatus import logging +from typing import Any, Dict import requests from wallbox import Wallbox @@ -20,10 +23,10 @@ PLATFORMS = ["sensor", "number"] UPDATE_INTERVAL = 30 -class WallboxCoordinator(DataUpdateCoordinator): +class WallboxCoordinator(DataUpdateCoordinator[Dict[str, Any]]): """Wallbox Coordinator class.""" - def __init__(self, station, wallbox, hass): + def __init__(self, station: str, wallbox: Wallbox, hass: HomeAssistant) -> None: """Initialize.""" self._station = station self._wallbox = wallbox @@ -35,31 +38,29 @@ class WallboxCoordinator(DataUpdateCoordinator): update_interval=timedelta(seconds=UPDATE_INTERVAL), ) - def _authenticate(self): + def _authenticate(self) -> None: """Authenticate using Wallbox API.""" try: self._wallbox.authenticate() - return True except requests.exceptions.HTTPError as wallbox_connection_error: if wallbox_connection_error.response.status_code == HTTPStatus.FORBIDDEN: raise ConfigEntryAuthFailed from wallbox_connection_error raise ConnectionError from wallbox_connection_error - def _validate(self): + def _validate(self) -> None: """Authenticate using Wallbox API.""" try: self._wallbox.authenticate() - return True except requests.exceptions.HTTPError as wallbox_connection_error: if wallbox_connection_error.response.status_code == 403: raise InvalidAuth from wallbox_connection_error raise ConnectionError from wallbox_connection_error - def _get_data(self): + def _get_data(self) -> dict[str, Any]: """Get new sensor data for Wallbox component.""" try: self._authenticate() - data = self._wallbox.getChargerStatus(self._station) + data: dict[str, Any] = self._wallbox.getChargerStatus(self._station) data[CONF_MAX_CHARGING_CURRENT_KEY] = data[CONF_DATA_KEY][ CONF_MAX_CHARGING_CURRENT_KEY ] @@ -69,7 +70,7 @@ class WallboxCoordinator(DataUpdateCoordinator): except requests.exceptions.HTTPError as wallbox_connection_error: raise ConnectionError from wallbox_connection_error - def _set_charging_current(self, charging_current): + def _set_charging_current(self, charging_current: float) -> None: """Set maximum charging current for Wallbox.""" try: self._authenticate() @@ -79,22 +80,21 @@ class WallboxCoordinator(DataUpdateCoordinator): raise InvalidAuth from wallbox_connection_error raise ConnectionError from wallbox_connection_error - async def async_set_charging_current(self, charging_current): + async def async_set_charging_current(self, charging_current: float) -> None: """Set maximum charging current for Wallbox.""" await self.hass.async_add_executor_job( self._set_charging_current, charging_current ) await self.async_request_refresh() - async def _async_update_data(self) -> bool: + async def _async_update_data(self) -> dict[str, Any]: """Get new sensor data for Wallbox component.""" data = await self.hass.async_add_executor_job(self._get_data) return data - async def async_validate_input(self) -> bool: + async def async_validate_input(self) -> None: """Get new sensor data for Wallbox component.""" - data = await self.hass.async_add_executor_job(self._validate) - return data + await self.hass.async_add_executor_job(self._validate) async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: diff --git a/homeassistant/components/wallbox/config_flow.py b/homeassistant/components/wallbox/config_flow.py index 8559c29c9aa..d2c0a048fa1 100644 --- a/homeassistant/components/wallbox/config_flow.py +++ b/homeassistant/components/wallbox/config_flow.py @@ -1,9 +1,14 @@ """Config flow for Wallbox integration.""" +from __future__ import annotations + +from typing import Any + import voluptuous as vol from wallbox import Wallbox from homeassistant import config_entries, core from homeassistant.const import CONF_PASSWORD, CONF_USERNAME +from homeassistant.data_entry_flow import FlowResult from . import InvalidAuth, WallboxCoordinator from .const import CONF_STATION, DOMAIN @@ -19,7 +24,9 @@ STEP_USER_DATA_SCHEMA = vol.Schema( ) -async def validate_input(hass: core.HomeAssistant, data): +async def validate_input( + hass: core.HomeAssistant, data: dict[str, Any] +) -> dict[str, str]: """Validate the user input allows to connect. Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user. @@ -36,11 +43,13 @@ async def validate_input(hass: core.HomeAssistant, data): class ConfigFlow(config_entries.ConfigFlow, domain=COMPONENT_DOMAIN): """Handle a config flow for Wallbox.""" - def __init__(self): + def __init__(self) -> None: """Start the Wallbox config flow.""" - self._reauth_entry = None + self._reauth_entry: config_entries.ConfigEntry | None = None - async def async_step_reauth(self, user_input=None): + async def async_step_reauth( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """Perform reauth upon an API authentication error.""" self._reauth_entry = self.hass.config_entries.async_get_entry( self.context["entry_id"] @@ -48,7 +57,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=COMPONENT_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 initial step.""" if user_input is None: return self.async_show_form( diff --git a/homeassistant/components/wallbox/number.py b/homeassistant/components/wallbox/number.py index b94351f4353..64c1f1e1abb 100644 --- a/homeassistant/components/wallbox/number.py +++ b/homeassistant/components/wallbox/number.py @@ -2,12 +2,16 @@ from __future__ import annotations from dataclasses import dataclass +from typing import Optional, cast from homeassistant.components.number import NumberEntity, NumberEntityDescription +from homeassistant.config_entries import ConfigEntry from homeassistant.const import DEVICE_CLASS_CURRENT +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.update_coordinator import CoordinatorEntity -from . import InvalidAuth +from . import InvalidAuth, WallboxCoordinator from .const import CONF_MAX_AVAILABLE_POWER_KEY, CONF_MAX_CHARGING_CURRENT_KEY, DOMAIN @@ -28,9 +32,11 @@ NUMBER_TYPES: dict[str, WallboxNumberEntityDescription] = { } -async def async_setup_entry(hass, config, async_add_entities): +async def async_setup_entry( + hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback +) -> None: """Create wallbox sensor entities in HASS.""" - coordinator = hass.data[DOMAIN][config.entry_id] + coordinator: WallboxCoordinator = hass.data[DOMAIN][entry.entry_id] # Check if the user is authorized to change current, if so, add number component: try: await coordinator.async_set_charging_current( @@ -41,7 +47,7 @@ async def async_setup_entry(hass, config, async_add_entities): async_add_entities( [ - WallboxNumber(coordinator, config, description) + WallboxNumber(coordinator, entry, description) for ent in coordinator.data if (description := NUMBER_TYPES.get(ent)) ] @@ -52,27 +58,33 @@ class WallboxNumber(CoordinatorEntity, NumberEntity): """Representation of the Wallbox portal.""" entity_description: WallboxNumberEntityDescription + coordinator: WallboxCoordinator def __init__( - self, coordinator, config, description: WallboxNumberEntityDescription - ): + self, + coordinator: WallboxCoordinator, + entry: ConfigEntry, + description: WallboxNumberEntityDescription, + ) -> None: """Initialize a Wallbox sensor.""" super().__init__(coordinator) self.entity_description = description self._coordinator = coordinator - self._attr_name = f"{config.title} {description.name}" + self._attr_name = f"{entry.title} {description.name}" self._attr_min_value = description.min_value @property - def max_value(self): + def max_value(self) -> float: """Return the maximum available current.""" - return self._coordinator.data[CONF_MAX_AVAILABLE_POWER_KEY] + return cast(float, self._coordinator.data[CONF_MAX_AVAILABLE_POWER_KEY]) @property - def value(self): + def value(self) -> float | None: """Return the state of the sensor.""" - return self._coordinator.data[CONF_MAX_CHARGING_CURRENT_KEY] + return cast( + Optional[float], self._coordinator.data[CONF_MAX_CHARGING_CURRENT_KEY] + ) - async def async_set_value(self, value: float): + async def async_set_value(self, value: float) -> None: """Set the value of the entity.""" await self._coordinator.async_set_charging_current(value) diff --git a/homeassistant/components/wallbox/sensor.py b/homeassistant/components/wallbox/sensor.py index 4571ed08725..835a8c405da 100644 --- a/homeassistant/components/wallbox/sensor.py +++ b/homeassistant/components/wallbox/sensor.py @@ -3,6 +3,7 @@ from __future__ import annotations from dataclasses import dataclass import logging +from typing import cast from homeassistant.components.sensor import ( STATE_CLASS_MEASUREMENT, @@ -10,6 +11,8 @@ from homeassistant.components.sensor import ( SensorEntity, SensorEntityDescription, ) +from homeassistant.components.wallbox import WallboxCoordinator +from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( DEVICE_CLASS_BATTERY, DEVICE_CLASS_CURRENT, @@ -21,6 +24,9 @@ from homeassistant.const import ( PERCENTAGE, POWER_KILO_WATT, ) +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.typing import StateType from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import ( @@ -130,13 +136,15 @@ SENSOR_TYPES: dict[str, WallboxSensorEntityDescription] = { } -async def async_setup_entry(hass, config, async_add_entities): +async def async_setup_entry( + hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback +) -> None: """Create wallbox sensor entities in HASS.""" - coordinator = hass.data[DOMAIN][config.entry_id] + coordinator: WallboxCoordinator = hass.data[DOMAIN][entry.entry_id] async_add_entities( [ - WallboxSensor(coordinator, config, description) + WallboxSensor(coordinator, entry, description) for ent in coordinator.data if (description := SENSOR_TYPES.get(ent)) ] @@ -147,24 +155,31 @@ class WallboxSensor(CoordinatorEntity, SensorEntity): """Representation of the Wallbox portal.""" entity_description: WallboxSensorEntityDescription + coordinator: WallboxCoordinator def __init__( - self, coordinator, config, description: WallboxSensorEntityDescription - ): + self, + coordinator: WallboxCoordinator, + entry: ConfigEntry, + description: WallboxSensorEntityDescription, + ) -> None: """Initialize a Wallbox sensor.""" super().__init__(coordinator) self.entity_description = description - self._attr_name = f"{config.title} {description.name}" + self._attr_name = f"{entry.title} {description.name}" @property - def native_value(self): + def native_value(self) -> StateType: """Return the state of the sensor.""" if (sensor_round := self.entity_description.precision) is not None: try: - return round( - self.coordinator.data[self.entity_description.key], sensor_round + return cast( + StateType, + round( + self.coordinator.data[self.entity_description.key], sensor_round + ), ) except TypeError: _LOGGER.debug("Cannot format %s", self._attr_name) return None - return self.coordinator.data[self.entity_description.key] + return cast(StateType, self.coordinator.data[self.entity_description.key]) diff --git a/mypy.ini b/mypy.ini index 9c7c481e986..88b10fe06ab 100644 --- a/mypy.ini +++ b/mypy.ini @@ -1562,6 +1562,17 @@ no_implicit_optional = true warn_return_any = true warn_unreachable = true +[mypy-homeassistant.components.wallbox.*] +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.water_heater.*] check_untyped_defs = true disallow_incomplete_defs = true