From b4187540c03c9cff2c044e02344ba8b904cb8932 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Sat, 8 Jan 2022 01:53:15 -0800 Subject: [PATCH] Improve typing for Overkiz integration and address late feedback (#63483) * Bump pyoverkiz to 1.0.2 * Remove cast for str enum. * Address feedback on coordinator * Change datatype to Callable * Address feedback * Move scenarios to seperate list * Cast Device to avoid issues with DataUpdateCoordinator default * Remove unnecessary casts and improve type notation * Check if state.value exists * Fix last mypy error (thanks @epenet) * Remove extra string cast * Improve sensor typing * Update pyoverkiz and remove typing * Small code improvement * Fix assert to reflect real world * Properly type Callable to not return Any * Remove unnecessary cast * Add OverkizStateType * Bugfix * Address feedback - multiline lambda * Pylint fix * Remove added binary sensor --- homeassistant/components/overkiz/__init__.py | 10 ++++---- .../components/overkiz/binary_sensor.py | 23 +++++++++---------- homeassistant/components/overkiz/const.py | 4 +++- .../components/overkiz/coordinator.py | 23 ++++++++++--------- homeassistant/components/overkiz/entity.py | 16 +++++++------ homeassistant/components/overkiz/executor.py | 11 +++++---- homeassistant/components/overkiz/light.py | 9 ++++---- homeassistant/components/overkiz/lock.py | 7 +++--- .../components/overkiz/manifest.json | 2 +- homeassistant/components/overkiz/scene.py | 4 +--- homeassistant/components/overkiz/sensor.py | 18 +++++++-------- requirements_all.txt | 2 +- requirements_test_all.txt | 2 +- 13 files changed, 67 insertions(+), 64 deletions(-) diff --git a/homeassistant/components/overkiz/__init__.py b/homeassistant/components/overkiz/__init__.py index 282d3e8d295..8baad82d378 100644 --- a/homeassistant/components/overkiz/__init__.py +++ b/homeassistant/components/overkiz/__init__.py @@ -41,7 +41,8 @@ class HomeAssistantOverkizData: """Overkiz data stored in the Home Assistant data object.""" coordinator: OverkizDataUpdateCoordinator - platforms: defaultdict[Platform, list[Device | Scenario]] + platforms: defaultdict[Platform, list[Device]] + scenarios: list[Scenario] async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: @@ -95,16 +96,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: ) coordinator.update_interval = UPDATE_INTERVAL_ALL_ASSUMED_STATE - platforms: defaultdict[Platform, list[Device | Scenario]] = defaultdict(list) + platforms: defaultdict[Platform, list[Device]] = defaultdict(list) hass.data.setdefault(DOMAIN, {})[entry.entry_id] = HomeAssistantOverkizData( - coordinator=coordinator, - platforms=platforms, + coordinator=coordinator, platforms=platforms, scenarios=scenarios ) # Map Overkiz entities to Home Assistant platform - platforms[Platform.SCENE] = scenarios - for device in coordinator.data.values(): _LOGGER.debug( "The following device has been retrieved. Report an issue if not supported correctly (%s)", diff --git a/homeassistant/components/overkiz/binary_sensor.py b/homeassistant/components/overkiz/binary_sensor.py index 507c36f88a9..69a2bc60f6f 100644 --- a/homeassistant/components/overkiz/binary_sensor.py +++ b/homeassistant/components/overkiz/binary_sensor.py @@ -3,7 +3,6 @@ from __future__ import annotations from collections.abc import Callable from dataclasses import dataclass -from typing import cast from pyoverkiz.enums import OverkizCommandParam, OverkizState @@ -17,7 +16,7 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback from . import HomeAssistantOverkizData -from .const import DOMAIN, IGNORED_OVERKIZ_DEVICES +from .const import DOMAIN, IGNORED_OVERKIZ_DEVICES, OverkizStateType from .entity import OverkizDescriptiveEntity @@ -25,7 +24,7 @@ from .entity import OverkizDescriptiveEntity class OverkizBinarySensorDescriptionMixin: """Define an entity description mixin for binary sensor entities.""" - value_fn: Callable[[str], bool] + value_fn: Callable[[OverkizStateType], bool] @dataclass @@ -41,28 +40,28 @@ BINARY_SENSOR_DESCRIPTIONS: list[OverkizBinarySensorDescription] = [ key=OverkizState.CORE_RAIN, name="Rain", icon="mdi:weather-rainy", - value_fn=lambda state: state == cast(str, OverkizCommandParam.DETECTED), + value_fn=lambda state: state == OverkizCommandParam.DETECTED, ), # SmokeSensor/SmokeSensor OverkizBinarySensorDescription( key=OverkizState.CORE_SMOKE, name="Smoke", device_class=BinarySensorDeviceClass.SMOKE, - value_fn=lambda state: state == cast(str, OverkizCommandParam.DETECTED), + value_fn=lambda state: state == OverkizCommandParam.DETECTED, ), # WaterSensor/WaterDetectionSensor OverkizBinarySensorDescription( key=OverkizState.CORE_WATER_DETECTION, name="Water", icon="mdi:water", - value_fn=lambda state: state == cast(str, OverkizCommandParam.DETECTED), + value_fn=lambda state: state == OverkizCommandParam.DETECTED, ), # AirSensor/AirFlowSensor OverkizBinarySensorDescription( key=OverkizState.CORE_GAS_DETECTION, name="Gas", device_class=BinarySensorDeviceClass.GAS, - value_fn=lambda state: state == cast(str, OverkizCommandParam.DETECTED), + value_fn=lambda state: state == OverkizCommandParam.DETECTED, ), # OccupancySensor/OccupancySensor # OccupancySensor/MotionSensor @@ -70,35 +69,35 @@ BINARY_SENSOR_DESCRIPTIONS: list[OverkizBinarySensorDescription] = [ key=OverkizState.CORE_OCCUPANCY, name="Occupancy", device_class=BinarySensorDeviceClass.OCCUPANCY, - value_fn=lambda state: state == cast(str, OverkizCommandParam.PERSON_INSIDE), + value_fn=lambda state: state == OverkizCommandParam.PERSON_INSIDE, ), # ContactSensor/WindowWithTiltSensor OverkizBinarySensorDescription( key=OverkizState.CORE_VIBRATION, name="Vibration", device_class=BinarySensorDeviceClass.VIBRATION, - value_fn=lambda state: state == cast(str, OverkizCommandParam.DETECTED), + value_fn=lambda state: state == OverkizCommandParam.DETECTED, ), # ContactSensor/ContactSensor OverkizBinarySensorDescription( key=OverkizState.CORE_CONTACT, name="Contact", device_class=BinarySensorDeviceClass.DOOR, - value_fn=lambda state: state == cast(str, OverkizCommandParam.OPEN), + value_fn=lambda state: state == OverkizCommandParam.OPEN, ), # Siren/SirenStatus OverkizBinarySensorDescription( key=OverkizState.CORE_ASSEMBLY, name="Assembly", device_class=BinarySensorDeviceClass.PROBLEM, - value_fn=lambda state: state == cast(str, OverkizCommandParam.OPEN), + value_fn=lambda state: state == OverkizCommandParam.OPEN, ), # Unknown OverkizBinarySensorDescription( key=OverkizState.IO_VIBRATION_DETECTED, name="Vibration", device_class=BinarySensorDeviceClass.VIBRATION, - value_fn=lambda state: state == cast(str, OverkizCommandParam.DETECTED), + value_fn=lambda state: state == OverkizCommandParam.DETECTED, ), ] diff --git a/homeassistant/components/overkiz/const.py b/homeassistant/components/overkiz/const.py index 09eccdd30d9..5d29358c3f4 100644 --- a/homeassistant/components/overkiz/const.py +++ b/homeassistant/components/overkiz/const.py @@ -2,7 +2,7 @@ from __future__ import annotations from datetime import timedelta -from typing import Final +from typing import Any, Dict, Final, List, Union from pyoverkiz.enums import UIClass from pyoverkiz.enums.ui import UIWidget @@ -37,3 +37,5 @@ OVERKIZ_DEVICE_TO_PLATFORM: dict[UIClass | UIWidget, Platform] = { UIClass.DOOR_LOCK: Platform.LOCK, UIClass.LIGHT: Platform.LIGHT, } + +OverkizStateType = Union[str, int, float, bool, Dict[Any, Any], List[Any], None] diff --git a/homeassistant/components/overkiz/coordinator.py b/homeassistant/components/overkiz/coordinator.py index 96ee918b750..d0d1baadff0 100644 --- a/homeassistant/components/overkiz/coordinator.py +++ b/homeassistant/components/overkiz/coordinator.py @@ -5,7 +5,7 @@ from collections.abc import Callable from datetime import timedelta import json import logging -from typing import Any, cast +from typing import Any, Dict from aiohttp import ServerDisconnectedError from pyoverkiz.client import OverkizClient @@ -22,9 +22,9 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers import device_registry as dr from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed -from .const import DOMAIN, UPDATE_INTERVAL +from .const import DOMAIN, UPDATE_INTERVAL, OverkizStateType -DATA_TYPE_TO_PYTHON: dict[DataType, Callable[[DataType], Any]] = { +DATA_TYPE_TO_PYTHON: dict[DataType, Callable[[Any], OverkizStateType]] = { DataType.INTEGER: int, DataType.DATE: int, DataType.STRING: str, @@ -37,7 +37,7 @@ DATA_TYPE_TO_PYTHON: dict[DataType, Callable[[DataType], Any]] = { _LOGGER = logging.getLogger(__name__) -class OverkizDataUpdateCoordinator(DataUpdateCoordinator): +class OverkizDataUpdateCoordinator(DataUpdateCoordinator[Dict[str, Device]]): """Class to manage fetching data from Overkiz platform.""" def __init__( @@ -132,9 +132,12 @@ class OverkizDataUpdateCoordinator(DataUpdateCoordinator): elif event.name == EventName.DEVICE_STATE_CHANGED: for state in event.device_states: device = self.devices[event.device_url] - if state.name not in device.states: - device.states[state.name] = state - device.states[state.name].value = self._get_state(state) + + if (device_state := device.states[state.name]) is None: + device_state = state + device.states[state.name] = device_state + + device_state.value = self._get_state(state) elif event.name == EventName.EXECUTION_REGISTERED: if event.exec_id not in self.executions: @@ -163,18 +166,16 @@ class OverkizDataUpdateCoordinator(DataUpdateCoordinator): @staticmethod def _get_state( state: State, - ) -> dict[Any, Any] | list[Any] | float | int | bool | str | None: + ) -> OverkizStateType: """Cast string value to the right type.""" data_type = DataType(state.type) if data_type == DataType.NONE: - return cast(None, state.value) + return state.value cast_to_python = DATA_TYPE_TO_PYTHON[data_type] value = cast_to_python(state.value) - assert isinstance(value, (str, float, int, bool)) - return value def places_to_area(self, place: Place) -> dict[str, str]: diff --git a/homeassistant/components/overkiz/entity.py b/homeassistant/components/overkiz/entity.py index 99f553d3474..2a0f529c502 100644 --- a/homeassistant/components/overkiz/entity.py +++ b/homeassistant/components/overkiz/entity.py @@ -56,10 +56,12 @@ class OverkizEntity(CoordinatorEntity): ) model = ( - self.executor.select_state( - OverkizState.CORE_MODEL, - OverkizState.CORE_PRODUCT_MODEL_NAME, - OverkizState.IO_MODEL, + str( + self.executor.select_state( + OverkizState.CORE_MODEL, + OverkizState.CORE_PRODUCT_MODEL_NAME, + OverkizState.IO_MODEL, + ), ) or self.device.widget ) @@ -67,10 +69,10 @@ class OverkizEntity(CoordinatorEntity): return DeviceInfo( identifiers={(DOMAIN, self.executor.base_device_url)}, name=self.device.label, - manufacturer=manufacturer, + manufacturer=str(manufacturer), model=model, - sw_version=self.executor.select_attribute( - OverkizAttribute.CORE_FIRMWARE_REVISION + sw_version=str( + self.executor.select_attribute(OverkizAttribute.CORE_FIRMWARE_REVISION) ), hw_version=self.device.controllable_name, suggested_area=self.coordinator.areas[self.device.place_oid], diff --git a/homeassistant/components/overkiz/executor.py b/homeassistant/components/overkiz/executor.py index 71ed9b0b7d7..e4b306cb836 100644 --- a/homeassistant/components/overkiz/executor.py +++ b/homeassistant/components/overkiz/executor.py @@ -2,11 +2,12 @@ from __future__ import annotations import logging -from typing import Any, cast +from typing import Any from urllib.parse import urlparse from pyoverkiz.models import Command, Device +from .const import OverkizStateType from .coordinator import OverkizDataUpdateCoordinator _LOGGER = logging.getLogger(__name__) @@ -37,11 +38,11 @@ class OverkizExecutor: """Return True if a command exists in a list of commands.""" return self.select_command(*commands) is not None - def select_state(self, *states: str) -> str | None: + def select_state(self, *states: str) -> OverkizStateType: """Select first existing active state in a list of states.""" for state in states: if current_state := self.device.states[state]: - return cast(str, current_state.value) + return current_state.value return None @@ -49,11 +50,11 @@ class OverkizExecutor: """Return True if a state exists in self.""" return self.select_state(*states) is not None - def select_attribute(self, *attributes: str) -> str | None: + def select_attribute(self, *attributes: str) -> OverkizStateType: """Select first existing active state in a list of states.""" for attribute in attributes: if current_attribute := self.device.attributes[attribute]: - return cast(str, current_attribute.value) + return current_attribute.value return None diff --git a/homeassistant/components/overkiz/light.py b/homeassistant/components/overkiz/light.py index 4e8aa820710..6075267b8e6 100644 --- a/homeassistant/components/overkiz/light.py +++ b/homeassistant/components/overkiz/light.py @@ -59,8 +59,9 @@ class OverkizLight(OverkizEntity, LightEntity): @property def is_on(self) -> bool: """Return true if light is on.""" - return self.executor.select_state(OverkizState.CORE_ON_OFF) == cast( - str, OverkizCommandParam.ON + return ( + self.executor.select_state(OverkizState.CORE_ON_OFF) + == OverkizCommandParam.ON ) @property @@ -73,13 +74,13 @@ class OverkizLight(OverkizEntity, LightEntity): if red is None or green is None or blue is None: return None - return (int(red), int(green), int(blue)) + return (cast(int, red), cast(int, green), cast(int, blue)) @property def brightness(self) -> int | None: """Return the brightness of this light (0-255).""" if brightness := self.executor.select_state(OverkizState.CORE_LIGHT_INTENSITY): - return round(int(brightness) * 255 / 100) + return round(cast(int, brightness) * 255 / 100) return None diff --git a/homeassistant/components/overkiz/lock.py b/homeassistant/components/overkiz/lock.py index b4e92586d57..8a333652b75 100644 --- a/homeassistant/components/overkiz/lock.py +++ b/homeassistant/components/overkiz/lock.py @@ -1,7 +1,7 @@ """Support for Overkiz locks.""" from __future__ import annotations -from typing import Any, cast +from typing import Any from pyoverkiz.enums import OverkizCommand, OverkizCommandParam, OverkizState @@ -44,6 +44,7 @@ class OverkizLock(OverkizEntity, LockEntity): @property def is_locked(self) -> bool | None: """Return a boolean for the state of the lock.""" - return self.executor.select_state(OverkizState.CORE_LOCKED_UNLOCKED) == cast( - str, OverkizCommandParam.LOCKED + return ( + self.executor.select_state(OverkizState.CORE_LOCKED_UNLOCKED) + == OverkizCommandParam.LOCKED ) diff --git a/homeassistant/components/overkiz/manifest.json b/homeassistant/components/overkiz/manifest.json index 1278028d5e8..4780be228ba 100644 --- a/homeassistant/components/overkiz/manifest.json +++ b/homeassistant/components/overkiz/manifest.json @@ -4,7 +4,7 @@ "config_flow": true, "documentation": "https://www.home-assistant.io/integrations/overkiz", "requirements": [ - "pyoverkiz==1.0.0" + "pyoverkiz==1.0.3" ], "dhcp": [ { diff --git a/homeassistant/components/overkiz/scene.py b/homeassistant/components/overkiz/scene.py index 4cddd88e40d..464b19d87e6 100644 --- a/homeassistant/components/overkiz/scene.py +++ b/homeassistant/components/overkiz/scene.py @@ -8,7 +8,6 @@ from pyoverkiz.models import Scenario from homeassistant.components.scene import Scene from homeassistant.config_entries import ConfigEntry -from homeassistant.const import Platform from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback @@ -25,8 +24,7 @@ async def async_setup_entry( data: HomeAssistantOverkizData = hass.data[DOMAIN][entry.entry_id] async_add_entities( - OverkizScene(scene, data.coordinator.client) - for scene in data.platforms[Platform.SCENE] + OverkizScene(scene, data.coordinator.client) for scene in data.scenarios ) diff --git a/homeassistant/components/overkiz/sensor.py b/homeassistant/components/overkiz/sensor.py index 3283151ee7d..7bde815a171 100644 --- a/homeassistant/components/overkiz/sensor.py +++ b/homeassistant/components/overkiz/sensor.py @@ -32,7 +32,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.typing import StateType from . import HomeAssistantOverkizData -from .const import DOMAIN, IGNORED_OVERKIZ_DEVICES +from .const import DOMAIN, IGNORED_OVERKIZ_DEVICES, OverkizStateType from .coordinator import OverkizDataUpdateCoordinator from .entity import OverkizDescriptiveEntity, OverkizEntity @@ -41,7 +41,7 @@ from .entity import OverkizDescriptiveEntity, OverkizEntity class OverkizSensorDescription(SensorEntityDescription): """Class to describe an Overkiz sensor.""" - native_value: Callable[[str | int | float], str | int | float] | None = None + native_value: Callable[[OverkizStateType], StateType] | None = None SENSOR_DESCRIPTIONS: list[OverkizSensorDescription] = [ @@ -65,7 +65,7 @@ SENSOR_DESCRIPTIONS: list[OverkizSensorDescription] = [ native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS, device_class=SensorDeviceClass.SIGNAL_STRENGTH, state_class=SensorStateClass.MEASUREMENT, - native_value=lambda value: round(float(value)), + native_value=lambda value: round(cast(float, value)), entity_category=EntityCategory.DIAGNOSTIC, entity_registry_enabled_default=False, ), @@ -236,7 +236,7 @@ SENSOR_DESCRIPTIONS: list[OverkizSensorDescription] = [ OverkizSensorDescription( key=OverkizState.CORE_RELATIVE_HUMIDITY, name="Relative Humidity", - native_value=lambda value: round(float(value), 2), + native_value=lambda value: round(cast(float, value), 2), device_class=SensorDeviceClass.HUMIDITY, native_unit_of_measurement=PERCENTAGE, # core:MeasuredValueType = core:RelativeValueInPercentage state_class=SensorStateClass.MEASUREMENT, @@ -245,7 +245,7 @@ SENSOR_DESCRIPTIONS: list[OverkizSensorDescription] = [ OverkizSensorDescription( key=OverkizState.CORE_TEMPERATURE, name="Temperature", - native_value=lambda value: round(float(value), 2), + native_value=lambda value: round(cast(float, value), 2), device_class=SensorDeviceClass.TEMPERATURE, native_unit_of_measurement=TEMP_CELSIUS, # core:MeasuredValueType = core:TemperatureInCelcius state_class=SensorStateClass.MEASUREMENT, @@ -292,7 +292,7 @@ SENSOR_DESCRIPTIONS: list[OverkizSensorDescription] = [ OverkizSensorDescription( key=OverkizState.CORE_SUN_ENERGY, name="Sun Energy", - native_value=lambda value: round(float(value), 2), + native_value=lambda value: round(cast(float, value), 2), icon="mdi:solar-power", state_class=SensorStateClass.MEASUREMENT, ), @@ -300,7 +300,7 @@ SENSOR_DESCRIPTIONS: list[OverkizSensorDescription] = [ OverkizSensorDescription( key=OverkizState.CORE_WIND_SPEED, name="Wind Speed", - native_value=lambda value: round(float(value), 2), + native_value=lambda value: round(cast(float, value), 2), icon="mdi:weather-windy", state_class=SensorStateClass.MEASUREMENT, ), @@ -395,14 +395,14 @@ class OverkizStateSensor(OverkizDescriptiveEntity, SensorEntity): """Return the value of the sensor.""" state = self.device.states.get(self.entity_description.key) - if not state: + if not state or not state.value: return None # Transform the value with a lambda function if self.entity_description.native_value: return self.entity_description.native_value(state.value) - return cast(str, state.value) + return state.value class OverkizHomeKitSetupCodeSensor(OverkizEntity, SensorEntity): diff --git a/requirements_all.txt b/requirements_all.txt index c8986b16a54..33d32547209 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -1734,7 +1734,7 @@ pyotgw==1.1b1 pyotp==2.6.0 # homeassistant.components.overkiz -pyoverkiz==1.0.0 +pyoverkiz==1.0.3 # homeassistant.components.openweathermap pyowm==3.2.0 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index bbbdd624809..6f104944679 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -1088,7 +1088,7 @@ pyotgw==1.1b1 pyotp==2.6.0 # homeassistant.components.overkiz -pyoverkiz==1.0.0 +pyoverkiz==1.0.3 # homeassistant.components.openweathermap pyowm==3.2.0