From c5d608a40bc488663c48edf101d916ef31206f2a Mon Sep 17 00:00:00 2001 From: Milan Meulemans Date: Tue, 30 Nov 2021 23:19:29 +0100 Subject: [PATCH] Add support for events --- aionanoleaf/__init__.py | 1 + aionanoleaf/events.py | 144 ++++++++++++++++++++++++++++++++++++++++ aionanoleaf/nanoleaf.py | 75 +++++++++++++++++++-- aionanoleaf/typing.py | 27 ++++++++ 4 files changed, 243 insertions(+), 4 deletions(-) create mode 100644 aionanoleaf/events.py diff --git a/aionanoleaf/__init__.py b/aionanoleaf/__init__.py index aebb2c1..55e4fe2 100644 --- a/aionanoleaf/__init__.py +++ b/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 diff --git a/aionanoleaf/events.py b/aionanoleaf/events.py new file mode 100644 index 0000000..0716d56 --- /dev/null +++ b/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 . + +"""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 diff --git a/aionanoleaf/nanoleaf.py b/aionanoleaf/nanoleaf.py index cbabcc8..1de1c03 100644 --- a/aionanoleaf/nanoleaf.py +++ b/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) diff --git a/aionanoleaf/typing.py b/aionanoleaf/typing.py index dc4a5b4..e47b1cb 100644 --- a/aionanoleaf/typing.py +++ b/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