Add type hints based on HA mypy config (#36)

- Add type hints based on HA mypy config
- Rename impulse to special event
- Add persistant notification

custom_components/changelog.txt

@@ -1,3 +1,9 @@
+Version 0.0.18 (2021-12-12)
+- Add type hints based on HA mypy config
+- Rename impulse to special event
+- Add persistant notification
+
+
 Version 0.0.17 (2021-12-05)
 - Add translation for HmIP-SRH states
 

custom_components/hahm/__init__.py

@@ -24,7 +24,7 @@
 
 async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
     """Set up HA-Homematic from a config entry."""
-    control_unit = ControlConfig(
+    control = ControlConfig(
         hass=hass,
         entry_id=config_entry.entry_id,
         data=config_entry.data,
@@ -36,18 +36,18 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
         ),
     ).get_control_unit()
     hass.data.setdefault(DOMAIN, {})
-    hass.data[DOMAIN][config_entry.entry_id] = control_unit
+    hass.data[DOMAIN][config_entry.entry_id] = control
     hass.config_entries.async_setup_platforms(config_entry, HAHM_PLATFORMS)
-    await control_unit.start()
+    await control.start()
     await async_setup_services(hass)
     return True
 
 
 async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
     """Unload a config entry."""
-    control_unit = hass.data[DOMAIN][config_entry.entry_id]
-    await control_unit.stop()
-    control_unit.central.clear_all()
+    control = hass.data[DOMAIN][config_entry.entry_id]
+    await control.stop()
+    control.central.clear_all()
     if unload_ok := await hass.config_entries.async_unload_platforms(
         config_entry, HAHM_PLATFORMS
     ):

custom_components/hahm/binary_sensor.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import logging
+from typing import Any
 
 from hahomematic.const import HmPlatform
 from hahomematic.platforms.binary_sensor import HmBinarySensor
@@ -28,9 +29,9 @@ async def async_setup_entry(
     control_unit: ControlUnit = hass.data[DOMAIN][config_entry.entry_id]
 
     @callback
-    def async_add_binary_sensor(args):
+    def async_add_binary_sensor(args: Any) -> None:
         """Add binary_sensor from HAHM."""
-        entities = []
+        entities: list[HaHomematicGenericEntity] = []
 
         for hm_entity in args[0]:
             entities.append(HaHomematicBinarySensor(control_unit, hm_entity))
@@ -53,12 +54,12 @@ def async_add_binary_sensor(args):
     )
 
 
-class HaHomematicBinarySensor(HaHomematicGenericEntity, BinarySensorEntity):
+class HaHomematicBinarySensor(
+    HaHomematicGenericEntity[HmBinarySensor], BinarySensorEntity
+):
     """Representation of the Homematic binary sensor."""
 
-    _hm_entity: HmBinarySensor
-
     @property
     def is_on(self) -> bool:
         """Return true if motion is detected."""
-        return self._hm_entity.state
+        return self._hm_entity.state is True

custom_components/hahm/button.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import logging
+from typing import Any
 
 from hahomematic.const import HmPlatform
 from hahomematic.platforms.button import HmButton
@@ -28,9 +29,9 @@ async def async_setup_entry(
     control_unit: ControlUnit = hass.data[DOMAIN][config_entry.entry_id]
 
     @callback
-    def async_add_button(args):
+    def async_add_button(args: Any) -> None:
         """Add button from HAHM."""
-        entities = []
+        entities: list[HaHomematicGenericEntity] = []
 
         for hm_entity in args[0]:
             entities.append(HaHomematicButton(control_unit, hm_entity))
@@ -51,10 +52,8 @@ def async_add_button(args):
     async_add_button([control_unit.get_hm_entities_by_platform(HmPlatform.BUTTON)])
 
 
-class HaHomematicButton(HaHomematicGenericEntity, ButtonEntity):
+class HaHomematicButton(HaHomematicGenericEntity[HmButton], ButtonEntity):
     """Representation of the Homematic button."""
 
-    _hm_entity: HmButton
-
     async def async_press(self) -> None:
         await self._hm_entity.press()

custom_components/hahm/climate.py

@@ -5,7 +5,7 @@
 from typing import Any
 
 from hahomematic.const import HmPlatform
-from hahomematic.devices.climate import IPThermostat, RfThermostat, SimpleRfThermostat
+from hahomematic.devices.climate import BaseClimateEntity
 
 from homeassistant.components.climate import ClimateEntity
 from homeassistant.config_entries import ConfigEntry
@@ -29,9 +29,9 @@ async def async_setup_entry(
     control_unit: ControlUnit = hass.data[DOMAIN][config_entry.entry_id]
 
     @callback
-    def async_add_climate(args):
+    def async_add_climate(args: Any) -> None:
         """Add climate from HAHM."""
-        entities = []
+        entities: list[HaHomematicGenericEntity] = []
 
         for hm_entity in args[0]:
             entities.append(HaHomematicClimate(control_unit, hm_entity))
@@ -52,11 +52,9 @@ def async_add_climate(args):
     async_add_climate([control_unit.get_hm_entities_by_platform(HmPlatform.CLIMATE)])
 
 
-class HaHomematicClimate(HaHomematicGenericEntity, ClimateEntity):
+class HaHomematicClimate(HaHomematicGenericEntity[BaseClimateEntity], ClimateEntity):
     """Representation of the HomematicIP climate entity."""
 
-    _hm_entity: SimpleRfThermostat | RfThermostat | IPThermostat
-
     @property
     def temperature_unit(self) -> str:
         """Return the unit of measurement."""
@@ -68,7 +66,7 @@ def supported_features(self) -> int:
         return self._hm_entity.supported_features
 
     @property
-    def target_temperature(self) -> float:
+    def target_temperature(self) -> float | None:
         """Return the temperature we try to reach."""
         return self._hm_entity.target_temperature
 
@@ -78,12 +76,12 @@ def target_temperature_step(self) -> float:
         return self._hm_entity.target_temperature_step
 
     @property
-    def current_temperature(self) -> float:
+    def current_temperature(self) -> float | None:
         """Return the current temperature."""
         return self._hm_entity.current_temperature
 
     @property
-    def current_humidity(self) -> int:
+    def current_humidity(self) -> int | None:
         """Return the current humidity."""
         return self._hm_entity.current_humidity
 
@@ -117,7 +115,7 @@ def max_temp(self) -> float:
         """Return the maximum temperature."""
         return float(self._hm_entity.max_temp)
 
-    async def async_set_temperature(self, **kwargs) -> None:
+    async def async_set_temperature(self, **kwargs: Any) -> None:
         """Set new target temperature."""
         await self._hm_entity.set_temperature(**kwargs)
 

custom_components/hahm/config_flow.py

@@ -2,8 +2,6 @@
 from __future__ import annotations
 
 import logging
-from types import MappingProxyType
-from typing import Any
 from xmlrpc.client import ProtocolError
 
 from hahomematic import config
@@ -26,9 +24,11 @@
 import voluptuous as vol
 
 from homeassistant import config_entries
+from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.data_entry_flow import FlowResult
 from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.typing import ConfigType
 
 from .const import (
     ATTR_ADD_ANOTHER_INTERFACE,
@@ -70,9 +70,7 @@
 )
 
 
-async def validate_input(
-    hass: HomeAssistant, data: MappingProxyType[str, Any], interface_name: str
-) -> bool:
+async def validate_input(hass: HomeAssistant, data: ConfigType) -> bool:
     """
     Validate the user input allows us to connect.
     Data has the keys with values provided by the user.
@@ -88,8 +86,8 @@ async def validate_input(
     control_unit.create_central()
     try:
         await control_unit.create_clients()
-        first_client: Client = control_unit.central.get_primary_client()
-        return await first_client.is_connected()
+        if first_client := control_unit.central.get_primary_client():
+            return await first_client.is_connected()
     except ConnectionError as cex:
         _LOGGER.exception(cex)
         raise CannotConnect from cex
@@ -107,10 +105,10 @@ class DomainConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
     VERSION = 1
     CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_PUSH
 
-    def __init__(self):
-        self.data = {}
+    def __init__(self) -> None:
+        self.data: ConfigType = {}
 
-    async def async_step_user(self, user_input: dict[str, Any] = None) -> FlowResult:
+    async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
         """Handle the initial step."""
         if user_input is not None:
             await self.async_set_unique_id(user_input[ATTR_INSTANCE_NAME])
@@ -134,7 +132,7 @@ async def async_step_user(self, user_input: dict[str, Any] = None) -> FlowResult
         return self.async_show_form(step_id="user", data_schema=DOMAIN_SCHEMA)
 
     async def async_step_interface(
-        self, user_input: dict[str, Any] = None
+        self, user_input: ConfigType | None = None
     ) -> FlowResult:
         """Handle the initial step."""
         if user_input is None:
@@ -156,7 +154,7 @@ async def async_step_interface(
         errors = {}
 
         try:
-            await validate_input(self.hass, self.data, user_input[ATTR_INTERFACE_NAME])
+            await validate_input(self.hass, self.data)
         except CannotConnect:
             errors["base"] = "cannot_connect"
         except InvalidAuth:
@@ -180,26 +178,27 @@ async def async_step_interface(
 
     @staticmethod
     @callback
-    def async_get_options_flow(config_entry):
+    def async_get_options_flow(config_entry: ConfigEntry) -> config_entries.OptionsFlow:
         """Get the options flow for this handler."""
         return HahmOptionsFlowHandler(config_entry)
 
 
 class HahmOptionsFlowHandler(config_entries.OptionsFlow):
     """Handle hahm options."""
 
-    def __init__(self, config_entry):
+    def __init__(self, config_entry: ConfigEntry) -> None:
         """Initialize hahm options flow."""
         self.config_entry = config_entry
         self.options = dict(config_entry.options)
-        self._cu = None
+        self._cu = self.hass.data[DOMAIN][self.config_entry.entry_id]
 
-    async def async_step_init(self, user_input=None):
+    async def async_step_init(self, user_input: ConfigType | None = None) -> FlowResult:
         """Manage the hahm options."""
-        self._cu = self.hass.data[DOMAIN][self.config_entry.entry_id]
         return await self.async_step_hahm_devices()
 
-    async def async_step_hahm_devices(self, user_input=None):
+    async def async_step_hahm_devices(
+        self, user_input: ConfigType | None = None
+    ) -> FlowResult:
         """Manage the hahm devices options."""
         if user_input is not None:
             self.options.update(user_input)

custom_components/hahm/const.py

@@ -1,12 +1,20 @@
 """Constants."""
+from __future__ import annotations
+
+from typing import Union, TypeVar
 from hahomematic.const import AVAILABLE_HM_PLATFORMS
 
 from homeassistant.const import Platform
+from hahomematic.entity import CustomEntity, GenericEntity, BaseParameterEntity
+from hahomematic.hub import BaseHubEntity
 
 DOMAIN = "hahm"
 
+HM_ENTITIES = Union[BaseHubEntity, BaseParameterEntity, CustomEntity, GenericEntity]
+HmCallbackEntity = (CustomEntity, GenericEntity)
+HMEntityType = TypeVar("HMEntityType", bound=HM_ENTITIES)
+
 ATTR_ADD_ANOTHER_INTERFACE = "add_another_interface"
-ATTR_ADDRESS = "address"
 ATTR_CHANNEL = "channel"
 ATTR_INSTANCE_NAME = "instance_name"
 ATTR_INTERFACE = "interface"
@@ -29,16 +37,16 @@
 SERVICE_VIRTUAL_KEY = "virtual_key"
 
 
-def _get_hahm_platforms():
+def _get_hahm_platforms() -> list[str]:
     """Return relevant hahm platforms."""
     platforms = [entry.value for entry in Platform]
     hm_platforms = [entry.value for entry in AVAILABLE_HM_PLATFORMS]
-    hahm_platforms = []
+    hahm_platforms: list[str] = []
     for hm_platform in hm_platforms:
         if hm_platform in platforms:
             hahm_platforms.append(hm_platform)
 
     return hahm_platforms
 
 
-HAHM_PLATFORMS = _get_hahm_platforms()
+HAHM_PLATFORMS: list[str] = _get_hahm_platforms()