From 5fb1ffc431b08ff2a71fd7253b71c4b8db6db07b Mon Sep 17 00:00:00 2001 From: Patrick762 Date: Fri, 21 Apr 2023 17:41:41 +0200 Subject: [PATCH] added files --- .github/workflows/python-publish.yml | 39 +++++ .gitignore | 1 + README.md | 7 +- setup.py | 34 ++++ streamdeckapi/__init__.py | 5 + streamdeckapi/api.py | 245 +++++++++++++++++++++++++++ streamdeckapi/tools.py | 19 +++ streamdeckapi/types.py | 110 ++++++++++++ 8 files changed, 459 insertions(+), 1 deletion(-) create mode 100644 .github/workflows/python-publish.yml create mode 100644 .gitignore create mode 100644 setup.py create mode 100644 streamdeckapi/__init__.py create mode 100644 streamdeckapi/api.py create mode 100644 streamdeckapi/tools.py create mode 100644 streamdeckapi/types.py diff --git a/.github/workflows/python-publish.yml b/.github/workflows/python-publish.yml new file mode 100644 index 0000000..b43bdc0 --- /dev/null +++ b/.github/workflows/python-publish.yml @@ -0,0 +1,39 @@ +# This workflow will upload a Python Package using Twine when a release is created +# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python#publishing-to-package-registries + +# This workflow uses actions that are not certified by GitHub. +# They are provided by a third-party and are governed by +# separate terms of service, privacy policy, and support +# documentation. + +name: Upload Python Package + +on: + release: + types: [published] + +permissions: + contents: read + +jobs: + deploy: + + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v3 + - name: Set up Python + uses: actions/setup-python@v3 + with: + python-version: '3.x' + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install build + - name: Build package + run: python -m build + - name: Publish package + uses: pypa/gh-action-pypi-publish@27b31702a0e7fc50959f5ad993c78deac1bdfc29 + with: + user: __token__ + password: ${{ secrets.PYPI_TOKEN }} \ No newline at end of file diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..61f2dc9 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +**/__pycache__/ diff --git a/README.md b/README.md index ee6be4d..87ebc5c 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,7 @@ # streamdeckapi -Stream Deck API Library +Stream Deck API Library for Home Assistant Stream Deck Integration + +Only compatible with separate Stream Deck Plugin + +## Dependencies +- [websockets](https://pypi.org/project/websockets/) 11.0.2 diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..f3f13f8 --- /dev/null +++ b/setup.py @@ -0,0 +1,34 @@ +from setuptools import setup, find_packages +import codecs +import os + +here = os.path.abspath(os.path.dirname(__file__)) + +with codecs.open(os.path.join(here, "README.md"), encoding="utf-8") as fh: + long_description = "\n" + fh.read() + +VERSION = '0.0.1' +DESCRIPTION = 'Stream Deck API Library' + +# Setting up +setup( + name="streamdeckapi", + version=VERSION, + author="Patrick762", + author_email="", + description=DESCRIPTION, + long_description_content_type="text/markdown", + long_description=long_description, + url="https://github.com/Patrick762/streamdeckapi", + packages=find_packages(), + install_requires=["websockets==11.0.2"], + keywords=[], + classifiers=[ + "Development Status :: 1 - Planning", + "Intended Audience :: Developers", + "Programming Language :: Python :: 3", + "Operating System :: Unix", + "Operating System :: MacOS :: MacOS X", + "Operating System :: Microsoft :: Windows", + ] +) diff --git a/streamdeckapi/__init__.py b/streamdeckapi/__init__.py new file mode 100644 index 0000000..9bb9f9a --- /dev/null +++ b/streamdeckapi/__init__.py @@ -0,0 +1,5 @@ +"""Stream Deck API.""" + +from streamdeckapi.api import StreamDeckApi +from streamdeckapi.types import SDInfo, SDWebsocketMessage, SDSize, SDApplication, SDButton, SDButtonPosition, SDDevice +from streamdeckapi.tools import get_model diff --git a/streamdeckapi/api.py b/streamdeckapi/api.py new file mode 100644 index 0000000..feef946 --- /dev/null +++ b/streamdeckapi/api.py @@ -0,0 +1,245 @@ +"""Stream Deck API.""" + +import asyncio +from collections.abc import Callable +import json +import logging + +import requests +from websockets.client import connect +from websockets.exceptions import WebSocketException + +from .types import SDInfo, SDWebsocketMessage + +_PLUGIN_PORT = 6153 +_PLUGIN_INFO = "/sd/info" +_PLUGIN_ICON = "/sd/icon" + +_LOGGER = logging.getLogger(__name__) + + +class StreamDeckApi: + """Stream Deck API Class.""" + + def __init__( + self, + host: str, + on_button_press: Callable[[str], None] | None = None, + on_button_release: Callable[[str], None] | None = None, + on_status_update: Callable[[SDInfo], None] | None = None, + on_ws_message: Callable[[SDWebsocketMessage], None] | None = None, + ) -> None: + """Init Stream Deck API object.""" + self._host = host + self._on_button_press = on_button_press + self._on_button_release = on_button_release + self._on_status_update = on_status_update + self._on_ws_message = on_ws_message + self._loop = asyncio.get_event_loop() + self._running = False + self._task: asyncio.Task | None = None + + # + # Properties + # + + @property + def host(self) -> str: + """Stream Deck API host.""" + return self._host + + @property + def _info_url(self) -> str: + """URL to info endpoint.""" + return f"http://{self._host}:{_PLUGIN_PORT}{_PLUGIN_INFO}" + + @property + def _icon_url(self) -> str: + """URL to icon endpoint.""" + return f"http://{self._host}:{_PLUGIN_PORT}{_PLUGIN_ICON}/" + + @property + def _websocket_url(self) -> str: + """URL to websocket.""" + return f"ws://{self._host}:{_PLUGIN_PORT}" + + # + # API Methods + # + + @staticmethod + def _get_request(url: str) -> None | requests.Response: + """Handle GET requests.""" + try: + res = requests.get(url, timeout=5) + except requests.RequestException: + _LOGGER.debug( + "Error retrieving data from Stream Deck Plugin (exception). Is it offline?" + ) + return None + if res.status_code != 200: + _LOGGER.debug( + "Error retrieving data from Stream Deck Plugin (response code). Is it offline?" + ) + return None + return res + + @staticmethod + def _post_request(url: str, data: str, headers) -> None | requests.Response: + """Handle POST requests.""" + try: + res = requests.post(url, data, headers=headers, timeout=5) + except requests.RequestException: + _LOGGER.error("Error sending data to Stream Deck Plugin (exception)") + return None + if res.status_code != 200: + _LOGGER.info( + "Error sending data to Stream Deck Plugin (%s). Is the button currently visible?", + res.reason, + ) + return None + return res + + async def get_info(self, in_executor: bool = True) -> None | SDInfo: + """Get info about Stream Deck.""" + res: requests.Response | None = None + if in_executor: + res = await self._loop.run_in_executor( + None, self._get_request, self._info_url + ) + else: + res = self._get_request(self._info_url) + if res is None or res.status_code != 200: + return None + try: + rjson = res.json() + except requests.JSONDecodeError: + _LOGGER.error("Error decoding response from %s", self._info_url) + return None + try: + info = SDInfo(rjson) + except KeyError: + _LOGGER.error("Error parsing response from %s to SDInfo", self._info_url) + return None + return info + + async def get_icon(self, btn: str) -> None | str: + """Get svg icon from Stream Deck button.""" + url = f"{self._icon_url}{btn}" + res = await self._loop.run_in_executor(None, self._get_request, url) + if res is None or res.status_code != 200: + return None + if res.headers.get("Content-Type", "") != "image/svg+xml": + _LOGGER.error("Invalid content type received from %s", url) + return None + return res.text + + async def update_icon(self, btn: str, svg: str) -> bool: + """Update svg icon of Stream Deck button.""" + url = f"{self._icon_url}{btn}" + res = await self._loop.run_in_executor( + None, + self._post_request, + url, + svg.encode("utf-8"), + {"Content-Type": "image/svg+xml"}, + ) + return isinstance(res, requests.Response) and res.status_code == 200 + + # + # Websocket Methods + # + + def _on_button_change(self, uuid: str | dict, state: bool): + """Handle button down event.""" + if not isinstance(uuid, str): + _LOGGER.debug("Method _on_button_change: uuid is not str") + return + if state is True and self._on_button_press is not None: + self._on_button_press(uuid) + elif state is False and self._on_button_release is not None: + self._on_button_release(uuid) + + def _on_ws_status_update(self, info: SDInfo | str | dict): + """Handle Stream Deck status update event.""" + if not isinstance(info, SDInfo): + _LOGGER.debug("Method _on_ws_status_update: info is not SDInfo") + return + if self._on_status_update is not None: + self._on_status_update(info) + + def _on_message(self, msg: str): + """Handle websocket messages.""" + if not isinstance(msg, str): + return + + _LOGGER.debug(msg) + + try: + datajson = json.loads(msg) + except json.JSONDecodeError: + _LOGGER.warning("Method _on_message: Websocket message couldn't get parsed") + return + try: + data = SDWebsocketMessage(datajson) + except KeyError: + _LOGGER.warning( + "Method _on_message: Websocket message couldn't get parsed to SDWebsocketMessage" + ) + return + + _LOGGER.debug("Method _on_message: Got event %s", data.event) + + if self._on_ws_message is not None: + self._on_ws_message(data) + + match data.event: + case "keyDown": + self._on_button_change(data.args, True) + case "keyUp": + self._on_button_change(data.args, False) + case "status": + self._on_ws_status_update(data.args) + case _: + _LOGGER.debug( + "Method _on_message: Unknown event from Stream Deck Plugin received (Event: %s)", + data.event, + ) + + async def _websocket_loop(self): + """Start the websocket client loop.""" + self._running = True + while self._running: + info = await self.get_info() + if isinstance(info, SDInfo): + _LOGGER.info("Method _websocket_loop: Streamdeck online") + try: + async with connect(self._websocket_url) as websocket: + try: + while self._running: + data = await asyncio.wait_for( + websocket.recv(), timeout=60 + ) + self._on_message(data) + await websocket.close() + _LOGGER.info("Method _websocket_loop: Websocket closed") + except WebSocketException: + _LOGGER.warning( + "Method _websocket_loop: Websocket client crashed. Restarting it" + ) + except asyncio.TimeoutError: + _LOGGER.warning( + "Method _websocket_loop: Websocket client timed out. Restarting it" + ) + except WebSocketException: + _LOGGER.warning( + "Method _websocket_loop: Websocket client not connecting. Restarting it" + ) + + def start_websocket_loop(self): + """Start the websocket client.""" + self._task = asyncio.create_task(self._websocket_loop()) + + def stop_websocket_loop(self): + """Stop the websocket client.""" + self._running = False diff --git a/streamdeckapi/tools.py b/streamdeckapi/tools.py new file mode 100644 index 0000000..acb80b1 --- /dev/null +++ b/streamdeckapi/tools.py @@ -0,0 +1,19 @@ +"""Stream Deck API Tools.""" + +from .types import SDInfo + + +def get_model(info: SDInfo) -> str: + """Get Stream Deck model.""" + if len(info.devices) == 0: + return "None" + size = info.devices[0].size + if size.columns == 3 and size.rows == 2: + return "Stream Deck Mini" + if size.columns == 5 and size.rows == 3: + return "Stream Deck MK.2" + if size.columns == 4 and size.rows == 2: + return "Stream Deck +" + if size.columns == 8 and size.rows == 4: + return "Stream Deck XL" + return "Unknown" diff --git a/streamdeckapi/types.py b/streamdeckapi/types.py new file mode 100644 index 0000000..34d103b --- /dev/null +++ b/streamdeckapi/types.py @@ -0,0 +1,110 @@ +"""Stream Deck API types.""" + + +class SDApplication: + """Stream Deck Application Type.""" + + font: str + language: str + platform: str + platform_version: str + version: str + + def __init__(self, obj: dict) -> None: + """Init Stream Deck Application object.""" + self.font = obj["font"] + self.language = obj["language"] + self.platform = obj["platform"] + self.platform_version = obj["platformVersion"] + self.version = obj["version"] + + +class SDSize: + """Stream Deck Size Type.""" + + columns: int + rows: int + + def __init__(self, obj: dict) -> None: + """Init Stream Deck Size object.""" + self.columns = obj["columns"] + self.rows = obj["rows"] + + +class SDDevice: + """Stream Deck Device Type.""" + + id: str + name: str + type: int + size: SDSize + + def __init__(self, obj: dict) -> None: + """Init Stream Deck Device object.""" + self.id = obj["id"] + self.name = obj["name"] + self.type = obj["type"] + self.size = SDSize(obj["size"]) + + +class SDButtonPosition: + """Stream Deck Button Position Type.""" + + x_pos: int + y_pos: int + + def __init__(self, obj: dict) -> None: + """Init Stream Deck Button Position object.""" + self.x_pos = obj["x"] + self.y_pos = obj["y"] + + +class SDButton: + """Stream Deck Button Type.""" + + uuid: str + device: str + position: SDButtonPosition + svg: str + + def __init__(self, obj: dict) -> None: + """Init Stream Deck Button object.""" + self.uuid = obj["uuid"] + self.device = obj["device"] + self.svg = obj["svg"] + self.position = SDButtonPosition(obj["position"]) + + +class SDInfo(dict): + """Stream Deck Info Type.""" + + application: SDApplication + devices: list[SDDevice] = [] + buttons: dict[str, SDButton] = {} + + def __init__(self, obj: dict) -> None: + """Init Stream Deck Info object.""" + dict.__init__(self, obj) + self.application = SDApplication(obj["application"]) + for device in obj["devices"]: + self.devices.append(SDDevice(device)) + for _id in obj["buttons"]: + self.buttons.update({_id: SDButton(obj["buttons"][_id])}) + + +class SDWebsocketMessage: + """Stream Deck Websocket Message Type.""" + + event: str + args: SDInfo | str | dict + + def __init__(self, obj: dict) -> None: + """Init Stream Deck Websocket Message object.""" + self.event = obj["event"] + if obj["args"] == {}: + self.args = {} + return + if isinstance(obj["args"], str): + self.args = obj["args"] + return + self.args = SDInfo(obj["args"])