Support door/window sensor and rockers (switches)

tests/test_light.py

@@ -3,7 +3,7 @@
 import json
 from xcomfort.bridge import Bridge
 from xcomfort.devices import (Light, LightState)
-from xcomfort.messages import Messages
+from xcomfort.constants import Messages
 
 
 class MockBridge(Bridge):

xcomfort/bridge.py

@@ -1,17 +1,13 @@
-from unicodedata import numeric
+from typing import Optional
 import aiohttp
 import asyncio
-import string
-import time
-import rx
-import rx.operators as ops
 from enum import Enum
 from .connection import SecureBridgeConnection, setup_secure_connection
-from .messages import Messages
-from .devices import (BridgeDevice, Light, RcTouch, Heater, Shade)
-from .room import Room, RoomState, RctMode, RctState, RctModeRange
-from .comp import Comp, CompState
-
+from .constants import ComponentTypes, DeviceTypes, Messages
+from .devices import (BridgeDevice, DoorSensor, Light, RcTouch, Heater, Rocker, Shade, WindowSensor)
+# Some HA code relies on bridge having imported these:
+from .room import Room, RoomState, RctMode, RctState, RctModeRange # noqa
+from .comp import Comp, CompState # noqa
 
 class State(Enum):
     Uninitialized = 0
@@ -130,20 +126,34 @@ def _create_device_from_payload(self, payload):
         name = payload['name']
         dev_type = payload["devType"]
         comp_id = payload["compId"]
-
-        if dev_type == 100 or dev_type == 101:
-            dimmable = payload['dimmable']
-            return Light(self, device_id, name, dimmable)
-
-        if dev_type == 102:
+        if dev_type in (DeviceTypes.ACTUATOR_SWITCH, DeviceTypes.ACTUATOR_DIMM):
+            if payload.get("usage") == 0:
+                # If usage = 1 then it's configured as a "load",
+                # and not as a light.
+                dimmable = payload["dimmable"]
+                return Light(self, device_id, name, dimmable)
+
+        elif dev_type == DeviceTypes.SHADING_ACTUATOR:
             return Shade(self, device_id, name, comp_id, payload)
 
-        if dev_type == 440:
+        elif dev_type == DeviceTypes.HEATING_ACTUATOR:
             return Heater(self, device_id, name, comp_id)
 
-        if dev_type == 450:
+        elif dev_type == DeviceTypes.RC_TOUCH:
             return RcTouch(self, device_id, name, comp_id)
 
+        elif dev_type == DeviceTypes.SWITCH:
+            component: Optional[Comp] = self._comps.get(comp_id)
+            if component and component.comp_type == ComponentTypes.DOOR_WINDOW_SENSOR:
+                if component.payload.get("mode") == "1310":
+                    return DoorSensor(self, device_id, name, comp_id, payload)
+                return WindowSensor(self, device_id, name, comp_id, payload)
+
+        elif dev_type == DeviceTypes.ROCKER:
+            # What Xcomfort calls a rocker HomeAssistant (and most humans) call a
+            # switch
+            return Rocker(self, device_id, name, comp_id, payload)
+
         return BridgeDevice(self, device_id, name)
 
     def _create_room_from_payload(self, payload):

xcomfort/connection.py

@@ -6,7 +6,7 @@
 import time
 import rx
 from enum import IntEnum
-from .messages import Messages
+from .constants import Messages
 from Crypto.Hash import SHA256
 from Crypto.PublicKey import RSA
 from Crypto.Cipher import PKCS1_OAEP, PKCS1_v1_5, AES

xcomfort/messages.py → xcomfort/constants.py

@@ -110,3 +110,42 @@ class ShadeOperationState(IntEnum):
     LOCK = 11
     UNLOCK = 12
     QUIT = 13
+
+
+class DeviceTypes(IntEnum):
+    ACTUATOR_SWITCH = 100
+    ACTUATOR_DIMM = 101
+    SHADING_ACTUATOR = 102
+    SWITCH = 202
+    ROCKER = 220
+    TEMP_SENSOR = 410
+    HEATING_ACTUATOR = 440
+    HEATING_VALVE = 441
+    HEATING_WATER_VALVE = 442
+    RC_TOUCH = 450
+    TEMP_HUMIDITY_SENSOR = 451
+    WATER_GUARD = 497
+    WATER_SENSOR = 499
+
+
+class HeatingTypes(IntEnum):
+    ELECTRIC_FLOOR_FOIL = 1
+    ELECTRIC_FLOOR_CABLE = 2
+    WATER_FLOOR = 3
+    ELECTRIC_RADIATOR = 4
+    ELECTRIC_INFRARED = 5
+    WATER_RADIATOR = 6
+
+
+class ComponentTypes(IntEnum):
+    PUSH_BUTTON_1 = 1
+    PUSH_BUTTON_2 = 2
+    MULTI_HEATING_ACTUATOR = 71
+    LIGHT_SWITCH_ACTUATOR = 74
+    DOOR_WINDOW_SENSOR = 76
+    DIMMING_ACTUATOR = 77
+    RC_TOUCH = 78
+    BRIDGE = 83
+    WATER_GUARD = 84
+    WATER_SENSOR = 85
+    SHADING_ACTUATOR = 86

xcomfort/devices.py

@@ -1,6 +1,7 @@
-from contextlib import nullcontext
+from typing import Optional
+
 import rx
-from .messages import Messages, ShadeOperationState
+from .constants import Messages, ShadeOperationState
 
 class DeviceState:
     def __init__(self, payload):
@@ -200,3 +201,58 @@ async def move_to_position(self, position: int):
 
     def __str__(self) -> str:
         return f"<Shade device_id={self.device_id} name={self.name} state={self.state} supports_go_to={self.supports_go_to}>"
+
+
+class DoorWindowSensor(BridgeDevice):
+    def __init__(self, bridge, device_id, name, comp_id, payload):
+        BridgeDevice.__init__(self, bridge, device_id, name)
+
+        self.comp_id = comp_id
+        self.payload = payload
+        self.is_open: Optional[bool] = None
+        self.is_closed: Optional[bool] = None
+
+    def handle_state(self, payload):
+        if (state := payload.get("curstate")) is not None:
+            self.is_closed = state == 1
+            self.is_open = not self.is_closed
+
+        self.state.on_next(self.is_closed)
+
+
+class WindowSensor(DoorWindowSensor):
+    pass
+
+
+class DoorSensor(DoorWindowSensor):
+    pass
+
+
+class Rocker(BridgeDevice):
+    def __init__(self, bridge, device_id, name, comp_id, payload):
+        BridgeDevice.__init__(self, bridge, device_id, name)
+        self.comp_id = comp_id
+        self.payload = payload
+        self.is_on: bool | None = None
+        if "curstate" in payload:
+            self.is_on = bool(payload["curstate"])
+
+    @property
+    def name_with_controlled(self) -> str:
+        """Name of Rocker, with the names of controlled devices in parens."""
+        names_of_controlled: set[str] = set()
+        for device_id in self.payload.get("controlId", []):
+            device = self.bridge._devices.get(device_id)
+            if device:
+                names_of_controlled.add(device.name)
+
+        return f"{self.name} ({', '.join(sorted(names_of_controlled))})"
+
+    def handle_state(self, payload, broadcast: bool = True) -> None:
+        self.payload.update(payload)
+        self.is_on = bool(payload["curstate"])
+        if broadcast:
+            self.state.on_next(self.is_on)
+
+    def __str__(self):
+        return f'Rocker({self.device_id}, "{self.name}", is_on: {self.is_on} payload: {self.payload})'

xcomfort/room.py

@@ -7,7 +7,7 @@
 import rx.operators as ops
 from enum import Enum
 from .connection import SecureBridgeConnection, setup_secure_connection
-from .messages import Messages
+from .constants import Messages
 from .devices import (BridgeDevice, Light, RcTouch, Heater, Shade)
 
 class RctMode(Enum):