From 38a7c7438ee5abf55773408b316d529f09924d08 Mon Sep 17 00:00:00 2001 From: Marc Mueller <30130371+cdce8p@users.noreply.github.com> Date: Mon, 28 Mar 2022 23:58:33 +0200 Subject: [PATCH] Add zha typing [core.device] (1) (#68686) --- homeassistant/components/zha/core/device.py | 57 ++++++++++---------- homeassistant/components/zha/core/helpers.py | 9 +++- 2 files changed, 37 insertions(+), 29 deletions(-) diff --git a/homeassistant/components/zha/core/device.py b/homeassistant/components/zha/core/device.py index e8b38fb9699..91ac905707f 100644 --- a/homeassistant/components/zha/core/device.py +++ b/homeassistant/components/zha/core/device.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio +from collections.abc import Callable from datetime import timedelta from enum import Enum import logging @@ -13,6 +14,7 @@ from zigpy import types import zigpy.exceptions from zigpy.profiles import PROFILES import zigpy.quirks +from zigpy.types.named import EUI64, NWK from zigpy.zcl.clusters.general import Groups import zigpy.zdo.types as zdo_types @@ -88,6 +90,8 @@ class DeviceStatus(Enum): class ZHADevice(LogMixin): """ZHA Zigbee device object.""" + _ha_device_id: str + def __init__( self, hass: HomeAssistant, @@ -101,7 +105,7 @@ class ZHADevice(LogMixin): self._available = False self._available_signal = f"{self.name}_{self.ieee}_{SIGNAL_AVAILABLE}" self._checkins_missed_count = 0 - self.unsubs = [] + self.unsubs: list[Callable[[], None]] = [] self.quirk_applied = isinstance(self._zigpy_device, zigpy.quirks.CustomDevice) self.quirk_class = ( f"{self._zigpy_device.__class__.__module__}." @@ -129,16 +133,15 @@ class ZHADevice(LogMixin): self.hass, self._check_available, timedelta(seconds=keep_alive_interval) ) ) - self._ha_device_id = None - self.status = DeviceStatus.CREATED + self.status: DeviceStatus = DeviceStatus.CREATED self._channels = channels.Channels(self) @property - def device_id(self): + def device_id(self) -> str: """Return the HA device registry device id.""" return self._ha_device_id - def set_device_id(self, device_id): + def set_device_id(self, device_id: str) -> None: """Set the HA device registry device id.""" self._ha_device_id = device_id @@ -159,24 +162,24 @@ class ZHADevice(LogMixin): self._channels = value @property - def name(self): + def name(self) -> str: """Return device name.""" return f"{self.manufacturer} {self.model}" @property - def ieee(self): + def ieee(self) -> EUI64: """Return ieee address for device.""" return self._zigpy_device.ieee @property - def manufacturer(self): + def manufacturer(self) -> str: """Return manufacturer for device.""" if self._zigpy_device.manufacturer is None: return UNKNOWN_MANUFACTURER return self._zigpy_device.manufacturer @property - def model(self): + def model(self) -> str: """Return model for device.""" if self._zigpy_device.model is None: return UNKNOWN_MODEL @@ -191,7 +194,7 @@ class ZHADevice(LogMixin): return self._zigpy_device.node_desc.manufacturer_code @property - def nwk(self): + def nwk(self) -> NWK: """Return nwk for device.""" return self._zigpy_device.nwk @@ -206,7 +209,7 @@ class ZHADevice(LogMixin): return self._zigpy_device.rssi @property - def last_seen(self): + def last_seen(self) -> float | None: """Return last_seen for device.""" return self._zigpy_device.last_seen @@ -227,7 +230,7 @@ class ZHADevice(LogMixin): return self._zigpy_device.node_desc.logical_type.name @property - def power_source(self): + def power_source(self) -> str: """Return the power source for the device.""" return ( POWER_MAINS_POWERED if self.is_mains_powered else POWER_BATTERY_OR_UNKNOWN @@ -258,14 +261,14 @@ class ZHADevice(LogMixin): return self._zigpy_device.node_desc.is_end_device @property - def is_groupable(self): + def is_groupable(self) -> bool: """Return true if this device has a group cluster.""" return self.is_coordinator or ( - self.available and self.async_get_groupable_endpoints() + self.available and bool(self.async_get_groupable_endpoints()) ) @property - def skip_configuration(self): + def skip_configuration(self) -> bool: """Return true if the device should not issue configuration related commands.""" return self._zigpy_device.skip_configuration @@ -275,7 +278,7 @@ class ZHADevice(LogMixin): return self._zha_gateway @property - def device_automation_triggers(self): + def device_automation_triggers(self) -> dict[tuple[str, str], dict[str, str]]: """Return the device automation triggers for this device.""" triggers = { ("device_offline", "device_offline"): { @@ -289,7 +292,7 @@ class ZHADevice(LogMixin): return triggers @property - def available_signal(self): + def available_signal(self) -> str: """Signal to use to subscribe to device availability changes.""" return self._available_signal @@ -332,7 +335,7 @@ class ZHADevice(LogMixin): return zha_dev @callback - def async_update_sw_build_id(self, sw_version: int): + def async_update_sw_build_id(self, sw_version: int) -> None: """Update device sw version.""" if self.device_id is None: return @@ -340,7 +343,7 @@ class ZHADevice(LogMixin): self.device_id, sw_version=f"0x{sw_version:08x}" ) - async def _check_available(self, *_): + async def _check_available(self, *_: Any) -> None: # don't flip the availability state of the coordinator if self.is_coordinator: return @@ -400,7 +403,7 @@ class ZHADevice(LogMixin): async_dispatcher_send(self.hass, f"{self._available_signal}_entity") @property - def device_info(self): + def device_info(self) -> dict[str, Any]: """Return a device description for device.""" ieee = str(self.ieee) time_struct = time.localtime(self.last_seen) @@ -423,7 +426,7 @@ class ZHADevice(LogMixin): ATTR_SIGNATURE: self.zigbee_signature, } - async def async_configure(self): + async def async_configure(self) -> None: """Configure the device.""" should_identify = async_get_zha_config_value( self._zha_gateway.config_entry, @@ -446,7 +449,7 @@ class ZHADevice(LogMixin): EFFECT_OKAY, EFFECT_DEFAULT_VARIANT ) - async def async_initialize(self, from_cache=False): + async def async_initialize(self, from_cache: bool = False) -> None: """Initialize channels.""" self.debug("started initialization") await self._channels.async_initialize(from_cache) @@ -461,15 +464,15 @@ class ZHADevice(LogMixin): unsubscribe() @callback - def async_update_last_seen(self, last_seen): + def async_update_last_seen(self, last_seen: float | None) -> None: """Set last seen on the zigpy device.""" if self._zigpy_device.last_seen is None and last_seen is not None: self._zigpy_device.last_seen = last_seen @property - def zha_device_info(self): + def zha_device_info(self) -> dict[str, Any]: """Get ZHA device information.""" - device_info = {} + device_info: dict[str, Any] = {} device_info.update(self.device_info) device_info["entities"] = [ { @@ -496,7 +499,7 @@ class ZHADevice(LogMixin): ] # Return endpoint device type Names - names = [] + names: list[dict[str, str]] = [] for endpoint in (ep for epid, ep in self.device.endpoints.items() if epid): profile = PROFILES.get(endpoint.profile_id) if profile and endpoint.device_type is not None: @@ -768,7 +771,7 @@ class ZHADevice(LogMixin): fmt = f"{log_msg[1]} completed: %s" zdo.debug(fmt, *(log_msg[2] + (outcome,))) - def log(self, level, msg, *args): + def log(self, level: int, msg: str, *args: Any) -> None: """Log a message.""" msg = f"[%s](%s): {msg}" args = (self.nwk, self.model) + args diff --git a/homeassistant/components/zha/core/helpers.py b/homeassistant/components/zha/core/helpers.py index d150ab9df45..5e98799e387 100644 --- a/homeassistant/components/zha/core/helpers.py +++ b/homeassistant/components/zha/core/helpers.py @@ -15,7 +15,7 @@ import itertools import logging from random import uniform import re -from typing import Any +from typing import Any, TypeVar import voluptuous as vol import zigpy.exceptions @@ -23,6 +23,7 @@ import zigpy.types import zigpy.util import zigpy.zdo.types as zdo_types +from homeassistant.config_entries import ConfigEntry from homeassistant.core import State, callback from .const import ( @@ -35,6 +36,8 @@ from .const import ( from .registries import BINDABLE_CLUSTERS from .typing import ZhaDeviceType, ZigpyClusterType +_T = TypeVar("_T") + @dataclass class BindingPair: @@ -130,7 +133,9 @@ def async_is_bindable_target(source_zha_device, target_zha_device): @callback -def async_get_zha_config_value(config_entry, section, config_key, default): +def async_get_zha_config_value( + config_entry: ConfigEntry, section: str, config_key: str, default: _T +) -> _T: """Get the value for the specified configuration from the zha config entry.""" return ( config_entry.options.get(CUSTOM_CONFIGURATION, {})