Add support for events

aionanoleaf/__init__.py

@@ -17,4 +17,5 @@
 
 """aioNanoleaf."""
 from .nanoleaf import *  # noqa: F401, F403
+from .events import *  # noqa: F401, F403
 from .exceptions import *  # noqa: F401, F403

aionanoleaf/events.py

@@ -0,0 +1,144 @@
+# Copyright 2021, Milan Meulemans.
+#
+# This file is part of aionanoleaf.
+#
+# aionanoleaf is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# aionanoleaf is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with aionanoleaf.  If not, see <https://www.gnu.org/licenses/>.
+
+"""Nanoleaf events."""
+from __future__ import annotations
+
+from abc import ABC
+
+from .typing import EffectsEventData, LayoutEventData, StateEventData, TouchEventData
+
+SINGLE_TAP = "Single Tap"
+DOUBLE_TAP = "Double Tap"
+SWIPE_UP = "Swipe Up"
+SWIPE_DOWN = "Swipe Down"
+SWIPE_LEFT = "Swipe Left"
+SWIPE_RIGHT = "Swipe Right"
+
+
+class Event(ABC):
+    """Abstract Nanoleaf event."""
+
+    EVENT_TYPE_ID: int
+
+
+class StateEvent(Event):
+    """Nanoleaf state event."""
+
+    EVENT_TYPE_ID = 1
+
+    def __init__(self, event_data: StateEventData) -> None:
+        """Init Nanoleaf state event."""
+        self._event_data = event_data
+
+    @property
+    def attribute_id(self) -> int:
+        """Return attribute ID."""
+        return self._event_data["attr"]
+
+    @property
+    def attribute(self) -> str:
+        """Return event attribute."""
+        return {
+            1: "is_on",
+            2: "brightness",
+            3: "hue",
+            4: "saturation",
+            5: "color_temperature",
+            6: "color_mode",
+        }[self.attribute_id]
+
+    @property
+    def value(self) -> str | int:
+        """Return event value, this is the new state of the attribute."""
+        return self._event_data["value"]
+
+
+class LayoutEvent(Event):
+    """Nanoleaf layout event."""
+
+    EVENT_TYPE_ID = 2
+
+    def __init__(self, event_data: LayoutEventData) -> None:
+        """Init Nanoleaf layout event."""
+        self._event_data = event_data
+
+    @property
+    def attribute_id(self) -> int:
+        """Return event attribute ID."""
+        return self._event_data["attr"]
+
+    @property
+    def attribute(self) -> str:
+        """Return event attribute."""
+        return {
+            1: "layout",
+            2: "global_orientation",
+        }[self.attribute_id]
+
+
+class EffectsEvent(Event):
+    """Nanoleaf effects event."""
+
+    EVENT_TYPE_ID = 3
+
+    def __init__(self, event_data: EffectsEventData) -> None:
+        """Init Nanoleaf effects event."""
+        self._event_data = event_data
+
+    @property
+    def attribute_id(self) -> int:
+        """Return event attribute ID."""
+        return self._event_data["attr"]
+
+    @property
+    def effect(self) -> str:
+        """Return the active effect."""
+        return self._event_data["value"]
+
+
+class TouchEvent(Event):
+    """Nanoleaf touch event."""
+
+    EVENT_TYPE_ID = 4
+
+    def __init__(self, event_data: TouchEventData) -> None:
+        """Init Nanoleaf touch event."""
+        self._event_data = event_data
+
+    @property
+    def gesture_id(self) -> int:
+        """Return gesture ID."""
+        return self._event_data["gesture"]
+
+    @property
+    def gesture(self) -> str:
+        """Return gesture."""
+        return {
+            0: SINGLE_TAP,
+            1: DOUBLE_TAP,
+            2: SWIPE_UP,
+            3: SWIPE_DOWN,
+            4: SWIPE_LEFT,
+            5: SWIPE_RIGHT,
+        }.get(self.gesture_id, str(self.gesture_id))
+
+    @property
+    def panel_id(self) -> int | None:
+        """Return panel ID if gesture has an associated panel else None."""
+        panel_id = self._event_data["panelId"]
+        return None if panel_id == -1 else panel_id

aionanoleaf/nanoleaf.py

@@ -20,16 +20,26 @@
 
 import asyncio
 import json
+from typing import Any, Callable, Coroutine
 
 from aiohttp import (
     ClientConnectorError,
+    ClientError,
     ClientResponse,
     ClientSession,
     ClientTimeout,
     ServerDisconnectedError,
 )
 
-from .exceptions import InvalidEffect, InvalidToken, NoAuthToken, Unauthorized, Unavailable
+from .events import EffectsEvent, LayoutEvent, StateEvent, TouchEvent
+from .exceptions import (
+    InvalidEffect,
+    InvalidToken,
+    NanoleafException,
+    NoAuthToken,
+    Unauthorized,
+    Unavailable,
+)
 from .typing import InfoData
 
 
@@ -189,13 +199,17 @@ async def _request(
     ) -> ClientResponse:
         """Make an authorized request to Nanoleaf with an auth_token."""
         url = f"{self._api_url}/{self.auth_token}/{path}"
-        data = json.dumps(data)
+        json_data = json.dumps(data)
         try:
             try:
-                resp = await self._session.request(method, url, data=data, timeout=self._REQUEST_TIMEOUT)
+                resp = await self._session.request(
+                    method, url, data=json_data, timeout=self._REQUEST_TIMEOUT
+                )
             except ServerDisconnectedError:
                 # Retry request once if the device disconnected
-                resp = await self._session.request(method, url, data=data, timeout=self._REQUEST_TIMEOUT)
+                resp = await self._session.request(
+                    method, url, data=json_data, timeout=self._REQUEST_TIMEOUT
+                )
         except ClientConnectorError as err:
             raise Unavailable from err
         except asyncio.TimeoutError as err:
@@ -346,3 +360,56 @@ async def turn_off(self, transition: int | None = None) -> None:
     async def identify(self) -> None:
         """Identify the Nanoleaf."""
         await self._request("put", "identify")
+
+    async def listen_events(
+        self,
+        state_callback: Callable[[StateEvent], Any] | None = None,
+        layout_callback: Callable[[LayoutEvent], Any] | None = None,
+        effects_callback: Callable[[EffectsEvent], Any] | None = None,
+        touch_callback: Callable[[TouchEvent], Any] | None = None,
+    ) -> Callable[[], Coroutine[Any, Any, None]]:
+        """Listen to events, apply changes to object and call callback with event."""
+        path = f"events?id={StateEvent.EVENT_TYPE_ID}, {EffectsEvent.EVENT_TYPE_ID}"
+        if layout_callback is not None:
+            path += f",{LayoutEvent.EVENT_TYPE_ID}"
+        if touch_callback is not None:
+            path += f",{TouchEvent.EVENT_TYPE_ID}"
+        while True:
+            try:
+                async with self._session.get(
+                    f"{self._api_url}/{self.auth_token}/{path}"
+                ) as resp:
+                    while True:
+                        id_line = await resp.content.readline()
+                        data_line = await resp.content.readline()
+                        await resp.content.readline()  # Empty line
+                        if resp.closed:
+                            return
+                        event_type_id = int(str(id_line)[6:-3])
+                        data = json.loads(str(data_line)[8:-3])
+                        events: list = data["events"]
+                        for event_data in events:
+                            if event_type_id == StateEvent.EVENT_TYPE_ID:
+                                event = StateEvent(event_data)
+                                setattr(self, f"_{event.attribute}", event.value)
+                                if state_callback is not None:
+                                    asyncio.create_task(state_callback(event))
+                            elif event_type_id == LayoutEvent.EVENT_TYPE_ID:
+                                layout_event = LayoutEvent(event_data)
+                                if layout_callback is not None:
+                                    asyncio.create_task(layout_callback(layout_event))
+                            elif event_type_id == EffectsEvent.EVENT_TYPE_ID:
+                                effects_event = EffectsEvent(event_data)
+                                self._effect = effects_event.effect
+                                if effects_callback is not None:
+                                    asyncio.create_task(effects_callback(effects_event))
+                            elif event_type_id == TouchEvent.EVENT_TYPE_ID:
+                                touch_event = TouchEvent(event_data)
+                                if touch_callback is not None:
+                                    asyncio.create_task(touch_callback(touch_event))
+                            else:
+                                raise NanoleafException(
+                                    f"Unknown event type id {event_type_id}"
+                                )
+            except ClientError:
+                await asyncio.sleep(5)

aionanoleaf/typing.py

@@ -101,3 +101,30 @@ class LightPanelsInfoData(InfoData):
     """Nanoleaf API Light Panels info."""
 
     rhythm: dict
+
+
+class StateEventData(TypedDict):
+    """Nanoleaf API State event data."""
+
+    attr: int
+    value: str | int
+
+
+class LayoutEventData(TypedDict):
+    """Nanoleaf API Layout event data."""
+
+    attr: int
+
+
+class EffectsEventData(TypedDict):
+    """Nanoleaf API Effects event data."""
+
+    attr: int
+    value: str
+
+
+class TouchEventData(TypedDict):
+    """Nanoleaf API Touch event data."""
+
+    gesture: int
+    panelId: int