jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 17:27:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Activate mypy for Vallox (#55874)

Browse Source
This commit is contained in:
Andre Richter 2021-09-23 19:59:28 +02:00 committed by GitHub
parent 2fe8c78811
commit a5c6a65161
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 151 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -113,6 +113,7 @@ homeassistant.components.upcloud.*
homeassistant.components.uptime.*
homeassistant.components.uptimerobot.*
homeassistant.components.vacuum.*
homeassistant.components.vallox.*
homeassistant.components.water_heater.*
homeassistant.components.weather.*
homeassistant.components.websocket_api.*

60
homeassistant/components/vallox/__init__.py
View File

@ -1,7 +1,10 @@
"""Support for Vallox ventilation units."""
from __future__ import annotations
from datetime import datetime
import ipaddress
import logging
from typing import Any
from vallox_websocket_api import PROFILE as VALLOX_PROFILE, Vallox
from vallox_websocket_api.constants import vlxDevConstants
@ -9,10 +12,12 @@ from vallox_websocket_api.exceptions import ValloxApiException
import voluptuous as vol
from homeassistant.const import CONF_HOST, CONF_NAME
import homeassistant.helpers.config_validation as cv
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.discovery import async_load_platform
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.typing import ConfigType, StateType
from .const import (
DEFAULT_FAN_SPEED_AWAY,
@ -95,7 +100,7 @@ SERVICE_TO_METHOD = {
}
async def async_setup(hass, config):
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the client and boot the platforms."""
conf = config[DOMAIN]
host = conf.get(CONF_HOST)
@ -113,13 +118,11 @@ async def async_setup(hass, config):
DOMAIN, vallox_service, service_handler.async_handle, schema=schema
)
# The vallox hardware expects quite strict timings for websocket
# requests. Timings that machines with less processing power, like
# Raspberries, cannot live up to during the busy start phase of Home
# Asssistant. Hence, async_add_entities() for fan and sensor in respective
# code will be called with update_before_add=False to intentionally delay
# the first request, increasing chance that it is issued only when the
# machine is less busy again.
# The vallox hardware expects quite strict timings for websocket requests. Timings that machines
# with less processing power, like Raspberries, cannot live up to during the busy start phase of
# Home Asssistant. Hence, async_add_entities() for fan and sensor in respective code will be
# called with update_before_add=False to intentionally delay the first request, increasing
# chance that it is issued only when the machine is less busy again.
hass.async_create_task(async_load_platform(hass, "sensor", DOMAIN, {}, config))
hass.async_create_task(async_load_platform(hass, "fan", DOMAIN, {}, config))
@ -131,15 +134,15 @@ async def async_setup(hass, config):
class ValloxStateProxy:
"""Helper class to reduce websocket API calls."""
def __init__(self, hass, client):
def __init__(self, hass: HomeAssistant, client: Vallox) -> None:
"""Initialize the proxy."""
self._hass = hass
self._client = client
self._metric_cache = {}
self._profile = None
self._metric_cache: dict[str, Any] = {}
self._profile = VALLOX_PROFILE.NONE
self._valid = False
def fetch_metric(self, metric_key):
def fetch_metric(self, metric_key: str) -> StateType:
"""Return cached state value."""
_LOGGER.debug("Fetching metric key: %s", metric_key)
@ -149,9 +152,18 @@ class ValloxStateProxy:
if metric_key not in vlxDevConstants.__dict__:
raise KeyError(f"Unknown metric key: {metric_key}")
return self._metric_cache[metric_key]
value = self._metric_cache[metric_key]
if value is None:
return None
def get_profile(self):
if not isinstance(value, (str, int, float)):
raise TypeError(
f"Return value of metric {metric_key} has unexpected type {type(value)}"
)
return value
def get_profile(self) -> str:
"""Return cached profile value."""
_LOGGER.debug("Returning profile")
@ -160,7 +172,7 @@ class ValloxStateProxy:
return PROFILE_TO_STR_REPORTABLE[self._profile]
async def async_update(self, event_time):
async def async_update(self, time: datetime | None = None) -> None:
"""Fetch state update."""
_LOGGER.debug("Updating Vallox state cache")
@ -180,7 +192,7 @@ class ValloxStateProxy:
class ValloxServiceHandler:
"""Services implementation."""
def __init__(self, client, state_proxy):
def __init__(self, client: Vallox, state_proxy: ValloxStateProxy) -> None:
"""Initialize the proxy."""
self._client = client
self._state_proxy = state_proxy
@ -245,10 +257,13 @@ class ValloxServiceHandler:
_LOGGER.error("Error setting fan speed for Boost profile: %s", err)
return False
async def async_handle(self, service):
async def async_handle(self, call: ServiceCall) -> None:
"""Dispatch a service call."""
method = SERVICE_TO_METHOD.get(service.service)
params = service.data.copy()
method = SERVICE_TO_METHOD.get(call.service)
params = call.data.copy()
if method is None:
return
if not hasattr(self, method["method"]):
_LOGGER.error("Service not implemented: %s", method["method"])
@ -256,7 +271,6 @@ class ValloxServiceHandler:
result = await getattr(self, method["method"])(**params)
# Force state_proxy to refresh device state, so that updates are
# propagated to platforms.
# Force state_proxy to refresh device state, so that updates are propagated to platforms.
if result:
await self._state_proxy.async_update(None)
await self._state_proxy.async_update()

99
homeassistant/components/vallox/fan.py
View File

@ -1,11 +1,19 @@
"""Support for the Vallox ventilation unit fan."""
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
from vallox_websocket_api import Vallox
from homeassistant.components.fan import FanEntity
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import ValloxStateProxy
from .const import (
DOMAIN,
METRIC_KEY_MODE,
@ -34,13 +42,17 @@ ATTR_PROFILE_FAN_SPEED_BOOST = {
}
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the fan device."""
if discovery_info is None:
return
client = hass.data[DOMAIN]["client"]
client.set_settable_address(METRIC_KEY_MODE, int)
device = ValloxFan(
@ -53,39 +65,41 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
class ValloxFan(FanEntity):
"""Representation of the fan."""
def __init__(self, name, client, state_proxy):
def __init__(
self, name: str, client: Vallox, state_proxy: ValloxStateProxy
) -> None:
"""Initialize the fan."""
self._name = name
self._client = client
self._state_proxy = state_proxy
self._available = False
self._state = None
self._fan_speed_home = None
self._fan_speed_away = None
self._fan_speed_boost = None
self._is_on = False
self._fan_speed_home: int | None = None
self._fan_speed_away: int | None = None
self._fan_speed_boost: int | None = None
@property
def should_poll(self):
def should_poll(self) -> bool:
"""Do not poll the device."""
return False
@property
def name(self):
def name(self) -> str:
"""Return the name of the device."""
return self._name
@property
def available(self):
def available(self) -> bool:
"""Return if state is known."""
return self._available
@property
def is_on(self):
def is_on(self) -> bool:
"""Return if device is on."""
return self._state
return self._is_on
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> Mapping[str, int | None]:
"""Return device specific state attributes."""
return {
ATTR_PROFILE_FAN_SPEED_HOME["description"]: self._fan_speed_home,
@ -93,7 +107,7 @@ class ValloxFan(FanEntity):
ATTR_PROFILE_FAN_SPEED_BOOST["description"]: self._fan_speed_boost,
}
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Call to update."""
self.async_on_remove(
async_dispatcher_connect(
@ -102,38 +116,42 @@ class ValloxFan(FanEntity):
)
@callback
def _update_callback(self):
def _update_callback(self) -> None:
"""Call update method."""
self.async_schedule_update_ha_state(True)
async def async_update(self):
async def async_update(self) -> None:
"""Fetch state from the device."""
try:
# Fetch if the whole device is in regular operation state.
self._state = self._state_proxy.fetch_metric(METRIC_KEY_MODE) == MODE_ON
self._is_on = self._state_proxy.fetch_metric(METRIC_KEY_MODE) == MODE_ON
# Fetch the profile fan speeds.
self._fan_speed_home = int(
self._state_proxy.fetch_metric(
ATTR_PROFILE_FAN_SPEED_HOME["metric_key"]
)
fan_speed_home = self._state_proxy.fetch_metric(
ATTR_PROFILE_FAN_SPEED_HOME["metric_key"]
)
self._fan_speed_away = int(
self._state_proxy.fetch_metric(
ATTR_PROFILE_FAN_SPEED_AWAY["metric_key"]
)
fan_speed_away = self._state_proxy.fetch_metric(
ATTR_PROFILE_FAN_SPEED_AWAY["metric_key"]
)
self._fan_speed_boost = int(
self._state_proxy.fetch_metric(
ATTR_PROFILE_FAN_SPEED_BOOST["metric_key"]
)
fan_speed_boost = self._state_proxy.fetch_metric(
ATTR_PROFILE_FAN_SPEED_BOOST["metric_key"]
)
except (OSError, KeyError) as err:
except (OSError, KeyError, TypeError) as err:
self._available = False
_LOGGER.error("Error updating fan: %s", err)
return
self._fan_speed_home = (
int(fan_speed_home) if isinstance(fan_speed_home, (int, float)) else None
)
self._fan_speed_away = (
int(fan_speed_away) if isinstance(fan_speed_away, (int, float)) else None
)
self._fan_speed_boost = (
int(fan_speed_boost) if isinstance(fan_speed_boost, (int, float)) else None
)
self._available = True
#
@ -145,20 +163,19 @@ class ValloxFan(FanEntity):
#
async def async_turn_on(
self,
speed: str = None,
percentage: int = None,
preset_mode: str = None,
**kwargs,
speed: str | None = None,
percentage: int | None = None,
preset_mode: str | None = None,
**kwargs: Any,
) -> None:
"""Turn the device on."""
_LOGGER.debug("Turn on: %s", speed)
# Only the case speed == None equals the GUI toggle switch being
# activated.
# Only the case speed == None equals the GUI toggle switch being activated.
if speed is not None:
return
if self._state is True:
if self._is_on:
_LOGGER.error("Already on")
return
@ -172,11 +189,11 @@ class ValloxFan(FanEntity):
# This state change affects other entities like sensors. Force an immediate update that can
# be observed by all parties involved.
await self._state_proxy.async_update(None)
await self._state_proxy.async_update()
async def async_turn_off(self, **kwargs) -> None:
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the device off."""
if self._state is False:
if not self._is_on:
_LOGGER.error("Already off")
return

75
homeassistant/components/vallox/sensor.py
View File

@ -19,8 +19,10 @@ from homeassistant.const import (
PERCENTAGE,
TEMP_CELSIUS,
)
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import ValloxStateProxy
from .const import DOMAIN, METRIC_KEY_MODE, MODE_ON, SIGNAL_VALLOX_STATE_UPDATE
@ -48,7 +50,7 @@ class ValloxSensor(SensorEntity):
self._attr_name = f"{name} {description.name}"
self._attr_available = False
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Call to update."""
self.async_on_remove(
async_dispatcher_connect(
@ -57,18 +59,23 @@ class ValloxSensor(SensorEntity):
)
@callback
def _update_callback(self):
def _update_callback(self) -> None:
"""Call update method."""
self.async_schedule_update_ha_state(True)
async def async_update(self):
async def async_update(self) -> None:
"""Fetch state from the ventilation unit."""
try:
self._attr_native_value = self._state_proxy.fetch_metric(
self.entity_description.metric_key
)
metric_key = self.entity_description.metric_key
except (OSError, KeyError) as err:
if metric_key is None:
self._attr_available = False
_LOGGER.error("Error updating sensor. Empty metric key")
return
try:
self._attr_native_value = self._state_proxy.fetch_metric(metric_key)
except (OSError, KeyError, TypeError) as err:
self._attr_available = False
_LOGGER.error("Error updating sensor: %s", err)
return
@ -79,7 +86,7 @@ class ValloxSensor(SensorEntity):
class ValloxProfileSensor(ValloxSensor):
"""Child class for profile reporting."""
async def async_update(self):
async def async_update(self) -> None:
"""Fetch state from the ventilation unit."""
try:
self._attr_native_value = self._state_proxy.get_profile()
@ -92,22 +99,21 @@ class ValloxProfileSensor(ValloxSensor):
self._attr_available = True
# There seems to be a quirk with respect to the fan speed reporting. The device
# keeps on reporting the last valid fan speed from when the device was in
# regular operation mode, even if it left that state and has been shut off in
# the meantime.
# There seems to be a quirk with respect to the fan speed reporting. The device keeps on reporting
# the last valid fan speed from when the device was in regular operation mode, even if it left that
# state and has been shut off in the meantime.
#
# Therefore, first query the overall state of the device, and report zero
# percent fan speed in case it is not in regular operation mode.
# Therefore, first query the overall state of the device, and report zero percent fan speed in case
# it is not in regular operation mode.
class ValloxFanSpeedSensor(ValloxSensor):
"""Child class for fan speed reporting."""
async def async_update(self):
async def async_update(self) -> None:
"""Fetch state from the ventilation unit."""
try:
fan_on = self._state_proxy.fetch_metric(METRIC_KEY_MODE) == MODE_ON
except (OSError, KeyError) as err:
except (OSError, KeyError, TypeError) as err:
self._attr_available = False
_LOGGER.error("Error updating sensor: %s", err)
return
@ -123,26 +129,28 @@ class ValloxFanSpeedSensor(ValloxSensor):
class ValloxFilterRemainingSensor(ValloxSensor):
"""Child class for filter remaining time reporting."""
async def async_update(self):
async def async_update(self) -> None:
"""Fetch state from the ventilation unit."""
try:
days_remaining = int(
self._state_proxy.fetch_metric(self.entity_description.metric_key)
)
await super().async_update()
except (OSError, KeyError) as err:
self._attr_available = False
_LOGGER.error("Error updating sensor: %s", err)
# Check if the update in the super call was a success.
if not self._attr_available:
return
days_remaining_delta = timedelta(days=days_remaining)
if not isinstance(self._attr_native_value, (int, float)):
self._attr_available = False
_LOGGER.error(
"Value has unexpected type: %s", type(self._attr_native_value)
)
return
# Since only a delta of days is received from the device, fix the
# time so the timestamp does not change with every update.
# Since only a delta of days is received from the device, fix the time so the timestamp does
# not change with every update.
days_remaining = float(self._attr_native_value)
days_remaining_delta = timedelta(days=days_remaining)
now = datetime.utcnow().replace(hour=13, minute=0, second=0, microsecond=0)
self._attr_native_value = (now + days_remaining_delta).isoformat()
self._attr_available = True
@dataclass
@ -235,7 +243,12 @@ SENSORS: tuple[ValloxSensorEntityDescription, ...] = (
)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the sensors."""
if discovery_info is None:
return

11
mypy.ini
View File

@ -1254,6 +1254,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.vallox.*]
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