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
This commit is contained in:
Mick Vleeshouwer 2022-01-08 01:53:15 -08:00 committed by GitHub
parent 943aaaeb3f
commit b4187540c0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 67 additions and 64 deletions

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@ -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": [
{

View File

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

View File

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

View File

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

View File

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