fixes, typing, enums and more (#65)

- Remove variables that are covered by other sensors (CCU only)
- Remove dummy from service message (HmIP-RF always sends 0001D3C98DD4B6:3 unreach)
- Rename Bidcos thermostats to SimpleRfThermostat and RfThermostat
- Use more Enums (like HA does): HmPlatform, HmEventType
- Use assignment expressions
- Add more type hints (fix most mypy errors)

changelog.txt

@@ -1,3 +1,11 @@
+Version 0.0.17 (2021-12-05)
+- Remove variables that are covered by other sensors (CCU only)
+- Remove dummy from service message (HmIP-RF always sends 0001D3C98DD4B6:3 unreach)
+- Rename Bidcos thermostats to SimpleRfThermostat and RfThermostat
+- Use more Enums (like HA does): HmPlatform, HmEventType
+- Use assignment expressions
+- Add more type hints (fix most mypy errors)
+
 Version 0.0.16 (2021-12-02)
 - Don't use default entities for climate groups (already included in device)
 

hahomematic/central_unit.py

@@ -32,6 +32,7 @@
     HM_VIRTUAL_REMOTE_HMIP,
     LOCALHOST,
     PRIMARY_PORTS,
+    HmPlatform,
 )
 from hahomematic.data import INSTANCES
 from hahomematic.device import HmDevice, create_devices
@@ -112,14 +113,14 @@ def __init__(self, central_config):
         self._load_caches()
         self.init_address_parameter_list()
         self._connection_checker = ConnectionChecker(self)
-        self.hub = None
+        self.hub: HmHub | None = None
 
     async def init_hub(self):
         """Init the hub."""
         if self.model is not BACKEND_PYDEVCCU:
             self.hub = HmHub(
                 self,
-                use_entities=self.central_config.enable_sensors_for_own_system_variables,
+                use_entities=self.central_config.enable_sensors_for_system_variables,
             )
             await self.hub.fetch_data()
         else:
@@ -147,8 +148,7 @@ def has_multiple_channels(self, address, parameter) -> bool:
         if ":" not in address:
             return False
         d_address = address.split(":")[0]
-        channels = self.address_parameter_cache.get((d_address, parameter))
-        if channels:
+        if channels := self.address_parameter_cache.get((d_address, parameter)):
             return len(set(channels)) > 1
         return False
 
@@ -303,20 +303,26 @@ async def set_system_variable(self, name, value):
 
     async def get_service_messages(self):
         """Get service messages from CCU / Homegear."""
-        await self.get_primary_client().get_service_messages()
+        service_messages = []
+        for client in self.clients.values():
+            if client.port in PRIMARY_PORTS:
+                if client_messages := await client.get_service_messages():
+                    service_messages.append(client_messages)
+        return _remove_dummy_service_message(service_messages)
 
     # pylint: disable=invalid-name
     async def set_install_mode(
         self, interface_id, on=True, t=60, mode=1, address=None
     ) -> None:
         """Activate or deactivate install-mode on CCU / Homegear."""
-        await self.get_primary_client(interface_id).set_install_mode(
-            on=on, t=t, mode=mode, address=address
-        )
+        if client := self.get_primary_client(interface_id):
+            await client.set_install_mode(on=on, t=t, mode=mode, address=address)
 
     async def get_install_mode(self, interface_id) -> int:
         """Get remaining time in seconds install mode is active from CCU / Homegear."""
-        return await self.get_primary_client(interface_id).get_install_mode()
+        if client := self.get_primary_client(interface_id):
+            return await client.get_install_mode()
+        return 0
 
     async def put_paramset(self, interface_id, address, paramset, value, rx_mode=None):
         """Set paramsets manually."""
@@ -349,14 +355,13 @@ async def press_virtual_remote_key(self, address, parameter):
             )
 
         device_address = address.split(":")[0]
-        virtual_remote: HmDevice = self._get_virtual_remote(device_address)
-        if virtual_remote:
+        if virtual_remote := self._get_virtual_remote(device_address):
             virtual_remote_channel = virtual_remote.action_events.get(
                 (address, parameter)
             )
             await virtual_remote_channel.send_value(True)
 
-    def get_hm_entities_by_platform(self, platform):
+    def get_hm_entities_by_hmplatform(self, platform: HmPlatform):
         """
         Return all hm-entities by platform
         """
@@ -367,7 +372,7 @@ def get_hm_entities_by_platform(self, platform):
 
         return hm_entities
 
-    def get_primary_client(self, interface_id=None) -> hm_client.Client:
+    def get_primary_client(self, interface_id=None) -> hm_client.Client | None:
         """Return the client by interface_id or the first with a primary port."""
         try:
             if interface_id:
@@ -382,15 +387,14 @@ def get_primary_client(self, interface_id=None) -> hm_client.Client:
             )
             _LOGGER.warning(message)
             raise hm_client.ClientException(message) from err
+        return None
 
     def get_hm_entity_by_parameter(self, address, parameter) -> GenericEntity | None:
         """Get entity by address and parameter."""
         if ":" in address:
             device_address = address.split(":")[0]
-            device = self.hm_devices.get(device_address)
-            if device:
-                entity = device.entities.get((address, parameter))
-                if entity:
+            if device := self.hm_devices.get(device_address):
+                if entity := device.entities.get((address, parameter)):
                     return entity
         return None
 
@@ -430,20 +434,17 @@ def get_all_used_parameters(self):
         parameters = set()
         for entity in self.hm_entities.values():
             if isinstance(entity, GenericEntity):
-                parameter = getattr(entity, "parameter", None)
-                if parameter:
+                if getattr(entity, "parameter", None):
                     parameters.add(entity.parameter)
 
         return sorted(parameters)
 
     def get_used_parameters(self, address):
         """Return used parameters"""
         parameters = set()
-        device = self.hm_devices.get(address)
-        if device:
+        if device := self.hm_devices.get(address):
             for entity in device.entities.values():
-                parameter = getattr(entity, "parameter", None)
-                if parameter:
+                if getattr(entity, "parameter", None):
                     parameters.add(entity.parameter)
 
         return sorted(parameters)
@@ -713,7 +714,7 @@ def __init__(
         json_port=None,
         json_tls=DEFAULT_TLS,
         enable_virtual_channels=False,
-        enable_sensors_for_own_system_variables=False,
+        enable_sensors_for_system_variables=False,
     ):
         self.entry_id = entry_id
         self.loop = loop
@@ -730,10 +731,19 @@ def __init__(
         self.json_port = json_port
         self.json_tls = json_tls
         self.enable_virtual_channels = enable_virtual_channels
-        self.enable_sensors_for_own_system_variables = (
-            enable_sensors_for_own_system_variables
-        )
+        self.enable_sensors_for_system_variables = enable_sensors_for_system_variables
 
     def get_central(self) -> CentralUnit:
         """Identify the used client."""
         return CentralUnit(self)
+
+
+def _remove_dummy_service_message(service_messages):
+    """Remove dummy SM, that hmip server always sends."""
+    new_service_messages = []
+    for client_messages in service_messages:
+        if "0001D3C98DD4B6:3" not in [
+            client_message[0] for client_message in client_messages
+        ]:
+            new_service_messages.append(client_messages)
+    return new_service_messages

hahomematic/client.py

@@ -177,6 +177,7 @@ async def proxy_re_init(self) -> int:
         de_init_status = await self.proxy_de_init()
         if de_init_status is not PROXY_DE_INIT_FAILED:
             return await self.proxy_init()
+        return PROXY_DE_INIT_FAILED
 
     def stop(self):
         """Stop depending services."""
@@ -268,12 +269,13 @@ async def set_install_mode(self, on=True, t=60, mode=1, address=None) -> None:
         except ProxyException:
             _LOGGER.exception("set_install_mode: ProxyException")
 
-    async def get_install_mode(self):
+    async def get_install_mode(self) -> int:
         """Get remaining time in seconds install mode is active from CCU / Homegear."""
         try:
             return await self.proxy.getInstallMode()
         except ProxyException:
             _LOGGER.exception("get_install_mode: ProxyException")
+        return 0
 
     async def get_all_metadata(self, address):
         """Get all metadata of device."""

hahomematic/const.py

@@ -3,8 +3,12 @@
 """
 from __future__ import annotations
 
-DEFAULT_ENCODING = "UTF-8"
+from datetime import datetime
+from enum import Enum
 
+DEFAULT_ENCODING = "UTF-8"
+HA_DOMAIN = "hahm"
+INIT_DATETIME = datetime.strptime("01.01.1970 00:00:00", "%d.%m.%Y %H:%M:%S")
 LOCALHOST = "localhost"
 IP_LOCALHOST_V4 = "127.0.0.1"
 IP_LOCALHOST_V6 = "::1"
@@ -57,8 +61,6 @@
     # PARAMSET_MASTER,
 ]
 
-HA_DOMAIN = "hahm"
-
 HH_EVENT_DELETE_DEVICES = "deleteDevices"
 HH_EVENT_DEVICES_CREATED = "devicesCreated"
 HH_EVENT_ERROR = "error"
@@ -85,10 +87,6 @@
 EVENT_SEQUENCE_OK = "SEQUENCE_OK"
 EVENT_UN_REACH = "UNREACH"
 
-EVENT_ALARM = "homematic.alarm"
-EVENT_KEYPRESS = "homematic.keypress"
-EVENT_IMPULSE = "homematic.impulse"
-
 CLICK_EVENTS = [
     EVENT_PRESS,
     EVENT_PRESS_SHORT,
@@ -274,35 +272,56 @@
 DEFAULT_TLS = False
 DEFAULT_VERIFY_TLS = False
 
-HA_PLATFORM_ACTION = "action"
-HA_PLATFORM_BINARY_SENSOR = "binary_sensor"
-HA_PLATFORM_BUTTON = "button"
-HA_PLATFORM_CLIMATE = "climate"
-HA_PLATFORM_COVER = "cover"
-HA_PLATFORM_EVENT = "event"
-HA_PLATFORM_LIGHT = "light"
-HA_PLATFORM_LOCK = "lock"
-HA_PLATFORM_NUMBER = "number"
-HA_PLATFORM_SELECT = "select"
-HA_PLATFORM_SENSOR = "sensor"
-HA_PLATFORM_SWITCH = "switch"
-HA_PLATFORM_TEXT = "text"
-
-HA_PLATFORMS = [
-    HA_PLATFORM_BINARY_SENSOR,
-    HA_PLATFORM_BUTTON,
-    HA_PLATFORM_CLIMATE,
-    HA_PLATFORM_COVER,
-    HA_PLATFORM_LIGHT,
-    HA_PLATFORM_LOCK,
-    HA_PLATFORM_NUMBER,
-    HA_PLATFORM_SELECT,
-    HA_PLATFORM_SENSOR,
-    HA_PLATFORM_SWITCH,
-]
-
 HM_ENTITY_UNIT_REPLACE = {'"': "", "100%": "%", "% rF": "%"}
 
 HM_VIRTUAL_REMOTE_HM = "BidCoS-RF"
 HM_VIRTUAL_REMOTE_HMIP = "HmIP-RCV-1"
 HM_VIRTUAL_REMOTES = [HM_VIRTUAL_REMOTE_HM, HM_VIRTUAL_REMOTE_HMIP]
+
+
+class HmPlatform(Enum):
+    """Enum with platforms relevant for Home Assistant."""
+
+    ACTION = "action"
+    BINARY_SENSOR = "binary_sensor"
+    BUTTON = "button"
+    CLIMATE = "climate"
+    COVER = "cover"
+    EVENT = "event"
+    LIGHT = "light"
+    LOCK = "lock"
+    NUMBER = "number"
+    SELECT = "select"
+    SENSOR = "sensor"
+    SWITCH = "switch"
+    TEXT = "text"
+
+    def __str__(self) -> str:
+        """Return self.value."""
+        return str(self.value)
+
+
+class HmEventType(Enum):
+    """Enum with hahm event types."""
+
+    ALARM = "homematic.alarm"
+    KEYPRESS = "homematic.keypress"
+    IMPULSE = "homematic.impulse"
+
+    def __str__(self) -> str:
+        """Return self.value."""
+        return str(self.value)
+
+
+AVAILABLE_HM_PLATFORMS = [
+    HmPlatform.BINARY_SENSOR,
+    HmPlatform.BUTTON,
+    HmPlatform.CLIMATE,
+    HmPlatform.COVER,
+    HmPlatform.LIGHT,
+    HmPlatform.LOCK,
+    HmPlatform.NUMBER,
+    HmPlatform.SELECT,
+    HmPlatform.SENSOR,
+    HmPlatform.SWITCH,
+]

hahomematic/data.py

@@ -3,13 +3,14 @@
 """
 from __future__ import annotations
 
+import hahomematic.central_unit as hm_central
+
 # {instance_name, central_unit}
-INSTANCES = {}
+INSTANCES: dict[str, hm_central.CentralUnit] = {}
 
 
 def get_client_by_interface_id(interface_id):
     """Return client by interface_id"""
     for central in INSTANCES.values():
-        client = central.clients.get(interface_id)
-        if client:
+        if client := central.clients.get(interface_id):
             return client