Implement push notifications for gateway

docs/device_docs/gateway.rst

@@ -0,0 +1,28 @@
+Gateway
+=======
+
+Adding support for new Zigbee devices
+-------------------------------------
+
+Once the event information is obtained as :ref:`described in the push server docs<obtain_event_info>`,
+a new event for a Zigbee device connected to a gateway can be implemented as follows:
+
+1. Open `miio/gateway/devices/subdevices.yaml` file and search for the target device for the new event.
+2. Add an entry for the new event:
+
+.. code-block:: yaml
+
+    properties:
+      - property: is_open # the new property of this device (optional)
+        default: False    # default value of the property when the device is initialized (optional)
+    push_properties:
+      open:               # the event you added, see the decoded packet capture `\"key\":\"event.lumi.sensor_magnet.aq2.open\"` take this equal to everything after the model
+        property: is_open # the property as listed above that this event will link to (optional)
+        value: True       # the value the property as listed above will be set to if this event is received (optional)
+        extra: "[1,6,1,0,[0,1],2,0]"  # the identification of this event, see the decoded packet capture `\"extra\":\"[1,6,1,0,[0,1],2,0]\"`
+      close:
+        property: is_open
+        value: False
+        extra: "[1,6,1,0,[0,0],2,0]"
+
+3. Create a pull request to get the event added to this library.

miio/gateway/alarm.py

@@ -1,9 +1,14 @@
 """Xiaomi Gateway Alarm implementation."""
 
+import logging
 from datetime import datetime
 
+from ..exceptions import DeviceException
+from ..push_server import EventInfo
 from .gatewaydevice import GatewayDevice
 
+_LOGGER = logging.getLogger(__name__)
+
 
 class Alarm(GatewayDevice):
     """Class representing the Xiaomi Gateway Alarm."""
@@ -61,3 +66,29 @@ def set_triggering_volume(self, volume):
     def last_status_change_time(self) -> datetime:
         """Return the last time the alarm changed status."""
         return datetime.fromtimestamp(self._gateway.send("get_arming_time").pop())
+
+    def subscribe_events(self):
+        """subscribe to the alarm events using the push server."""
+        if self._gateway._push_server is None:
+            raise DeviceException(
+                "Can not install push callback without a PushServer instance"
+            )
+
+        event_info = EventInfo(
+            action="alarm_triggering",
+            extra="[1,19,1,111,[0,1],2,0]",
+            trigger_token=self._gateway.token,
+        )
+
+        event_id = self._gateway._push_server.subscribe_event(self._gateway, event_info)
+        if event_id is None:
+            return False
+
+        self._event_ids.append(event_id)
+        return True
+
+    def unsubscribe_events(self):
+        """Unsubscibe from events registered in the gateway memory."""
+        for event_id in self._event_ids:
+            self._gateway._push_server.unsubscribe_event(self._gateway, event_id)
+            self._event_ids.remove(event_id)

miio/gateway/devices/subdevice.py

@@ -1,13 +1,20 @@
 """Xiaomi Gateway subdevice base class."""
 
 import logging
-from typing import TYPE_CHECKING, Dict, Optional
+from typing import TYPE_CHECKING, Dict, List, Optional
 
 import attr
 import click
 
 from ...click_common import command
-from ..gateway import GATEWAY_MODEL_EU, GATEWAY_MODEL_ZIG3, GatewayException
+from ...exceptions import DeviceException
+from ...push_server import EventInfo
+from ..gateway import (
+    GATEWAY_MODEL_EU,
+    GATEWAY_MODEL_ZIG3,
+    GatewayCallback,
+    GatewayException,
+)
 
 _LOGGER = logging.getLogger(__name__)
 if TYPE_CHECKING:
@@ -60,6 +67,10 @@ def __init__(
 
         self.setter = model_info.get("setter")
 
+        self.push_events = model_info.get("push_properties", [])
+        self._event_ids: List[str] = []
+        self._registered_callbacks: Dict[str, GatewayCallback] = {}
+
     def __repr__(self):
         return "<Subdevice {}: {}, model: {}, zigbee: {}, fw: {}, bat: {}, vol: {}, props: {}>".format(
             self.device_type,
@@ -260,3 +271,69 @@ def get_firmware_version(self) -> Optional[int]:
                 ex,
             )
         return self._fw_ver
+
+    def register_callback(self, id: str, callback: GatewayCallback):
+        """Register a external callback function for updates of this subdevice."""
+        if id in self._registered_callbacks:
+            _LOGGER.error(
+                "A callback with id '%s' was already registed, overwriting previous callback",
+                id,
+            )
+        self._registered_callbacks[id] = callback
+
+    def remove_callback(self, id: str):
+        """Remove a external callback using its id."""
+        self._registered_callbacks.pop(id)
+
+    def push_callback(self, action: str, params: str):
+        """Push callback received from the push server."""
+        if action not in self.push_events:
+            _LOGGER.error(
+                "Received unregistered action '%s' callback for sid '%s' model '%s'",
+                action,
+                self.sid,
+                self.model,
+            )
+
+        event = self.push_events[action]
+        prop = event.get("property")
+        value = event.get("value")
+        if prop is not None and value is not None:
+            self._props[prop] = value
+
+        for callback in self._registered_callbacks.values():
+            callback(action, params)
+
+    def subscribe_events(self):
+        """subscribe to all subdevice events using the push server."""
+        if self._gw._push_server is None:
+            raise DeviceException(
+                "Can not install push callback without a PushServer instance"
+            )
+
+        result = True
+        for action in self.push_events:
+            event_info = EventInfo(
+                action=action,
+                extra=self.push_events[action]["extra"],
+                source_sid=self.sid,
+                source_model=self.zigbee_model,
+                event=self.push_events[action].get("event", None),
+                command_extra=self.push_events[action].get("command_extra", ""),
+                trigger_value=self.push_events[action].get("trigger_value"),
+            )
+
+            event_id = self._gw._push_server.subscribe_event(self._gw, event_info)
+            if event_id is None:
+                result = False
+                continue
+
+            self._event_ids.append(event_id)
+
+        return result
+
+    def unsubscribe_events(self):
+        """Unsubscibe from events registered in the gateway memory."""
+        for event_id in self._event_ids:
+            self._gw._push_server.unsubscribe_event(self._gw, event_id)
+            self._event_ids.remove(event_id)