added files

.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 }}

.gitignore

@@ -0,0 +1 @@
+**/__pycache__/

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

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="<pip-stream-deck-api@hosting-rt.de>",
+    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",
+    ]
+)

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

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

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"

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"])