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

Enable strict typing - wallbox (#59301)

Browse source
This commit is contained in:
Marc Mueller 2021-11-23 22:30:22 +01:00 • committed by GitHub
parent ce369bb336
commit 6089aef072
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 91 additions and 41 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.strict-typing
View file

@ -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.*

28
homeassistant/components/wallbox/__init__.py
View file

@ -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:

21
homeassistant/components/wallbox/config_flow.py
View file

@ -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(

36
homeassistant/components/wallbox/number.py
View file

@ -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)

35
homeassistant/components/wallbox/sensor.py
View file

@ -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])

11
mypy.ini
View file

@ -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