Operating StreamDeck from a terminal

pyproject.toml

@@ -40,6 +40,7 @@ types-pkg-resources = "^0.1.3"
 
 [tool.poetry.scripts]
 streamdeck = "streamdeck_ui.gui:start"
+streamdeckc = "streamdeck_ui.cli.server:execute"
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"

streamdeck_ui/cli/commands.py

@@ -0,0 +1,152 @@
+import typing as tp
+
+from streamdeck_ui.api import StreamDeckServer
+
+
+class Command(tp.Protocol):
+    def execute(self, api: StreamDeckServer, ui: tp.Any) -> None:
+        ...
+
+
+class SetPageCommand:
+    def __init__(self, cfg):
+        self.deck_index = cfg["deck"]
+        self.page_index = cfg["page"]
+
+    def execute(self, api: StreamDeckServer, ui):
+        deck_id = ui.device_list.itemData(ui.device_list.currentIndex()) if self.deck_index is None else self.deck_index
+        if api.get_page(deck_id) == self.page_index:
+            return
+        api.set_page(deck_id, self.page_index)
+        ui.pages.setCurrentIndex(self.page_index)
+
+
+class SetBrightnessCommand:
+    def __init__(self, cfg):
+        self.deck_index = cfg["deck"]
+        self.brightness = cfg["value"]
+
+    def execute(self, api: StreamDeckServer, ui):
+        deck_id = ui.device_list.itemData(ui.device_list.currentIndex()) if self.deck_index is None else self.deck_index
+        api.set_brightness(deck_id, self.brightness)
+
+
+class SetButtonTextCommand:
+    def __init__(self, cfg):
+        self.deck_index = cfg["deck"]
+        self.page_index = cfg["page"]
+        self.button_index = cfg["button"]
+        self.button_text = cfg["text"]
+
+    def execute(self, api: StreamDeckServer, ui):
+        deck_id = ui.device_list.itemData(ui.device_list.currentIndex()) if self.deck_index is None else self.deck_index
+        if self.page_index is None:
+            self.page_index = api.get_page(deck_id)
+        api.set_button_text(deck_id, self.page_index, self.button_index, self.button_text)
+
+
+class SetButtonTextAlignmentCommand:
+    def __init__(self, cfg):
+        self.deck_index = cfg["deck"]
+        self.page_index = cfg["page"]
+        self.button_index = cfg["button"]
+        self.button_text_alignment = cfg["alignment"]
+
+    def execute(self, api: StreamDeckServer, ui):
+        deck_id = ui.device_list.itemData(ui.device_list.currentIndex()) if self.deck_index is None else self.deck_index
+        if self.page_index is None:
+            self.page_index = api.get_page(deck_id)
+        api.set_text_vertical_align(deck_id, self.page_index, self.button_index, self.button_text_alignment)
+
+
+class SetButtonWriteCommand:
+    def __init__(self, cfg):
+        self.deck_index = cfg["deck"]
+        self.page_index = cfg["page"]
+        self.button_index = cfg["button"]
+        self.button_write = cfg["write"]
+
+    def execute(self, api: StreamDeckServer, ui):
+        deck_id = ui.device_list.itemData(ui.device_list.currentIndex()) if self.deck_index is None else self.deck_index
+        if self.page_index is None:
+            self.page_index = api.get_page(deck_id)
+        api.set_button_write(deck_id, self.page_index, self.button_index, self.button_write)
+
+
+class SetButtonCmdCommand:
+    def __init__(self, cfg):
+        self.deck_index = cfg["deck"]
+        self.page_index = cfg["page"]
+        self.button_index = cfg["button"]
+        self.button_cmd = cfg["button_cmd"]
+
+    def execute(self, api: StreamDeckServer, ui):
+        print(self.button_cmd)
+        deck_id = ui.device_list.itemData(ui.device_list.currentIndex()) if self.deck_index is None else self.deck_index
+        if self.page_index is None:
+            self.page_index = api.get_page(deck_id)
+        api.set_button_command(deck_id, self.page_index, self.button_index, self.button_cmd)
+
+
+class SetButtonKeysCommand:
+    def __init__(self, cfg):
+        self.deck_index = cfg["deck"]
+        self.page_index = cfg["page"]
+        self.button_index = cfg["button"]
+        self.button_keys = cfg["button_keys"]
+
+    def execute(self, api: StreamDeckServer, ui):
+        print(self.button_keys)
+        deck_id = ui.device_list.itemData(ui.device_list.currentIndex()) if self.deck_index is None else self.deck_index
+        if self.page_index is None:
+            self.page_index = api.get_page(deck_id)
+        api.set_button_keys(deck_id, self.page_index, self.button_index, self.button_keys)
+
+
+class SetButtonIconCommand:
+    def __init__(self, cfg):
+        self.deck_index = cfg["deck"]
+        self.page_index = cfg["page"]
+        self.button_index = cfg["button"]
+        self.icon_path = cfg["icon"]
+
+    def execute(self, api: StreamDeckServer, ui):
+        deck_id = ui.device_list.itemData(ui.device_list.currentIndex()) if self.deck_index is None else self.deck_index
+        if self.page_index is None:
+            self.page_index = api.get_page(deck_id)
+        api.set_button_icon(deck_id, self.page_index, self.button_index, self.icon_path)
+
+
+class ClearButtonIconCommand:
+    def __init__(self, cfg):
+        self.deck_index = cfg["deck"]
+        self.page_index = cfg["page"]
+        self.button_index = cfg["button"]
+
+    def execute(self, api: StreamDeckServer, ui):
+        deck_id = ui.device_list.itemData(ui.device_list.currentIndex()) if self.deck_index is None else self.deck_index
+        if self.page_index is None:
+            self.page_index = api.get_page(deck_id)
+        api.set_button_icon(deck_id, self.page_index, self.button_index, "")
+
+
+def create_command(cfg: dict) -> Command | None:
+    if cfg["command"] == "set_page":
+        return SetPageCommand(cfg)
+    elif cfg["command"] == "set_brightness":
+        return SetBrightnessCommand(cfg)
+    elif cfg["command"] == "set_text":
+        return SetButtonTextCommand(cfg)
+    elif cfg["command"] == "set_alignment":
+        return SetButtonTextAlignmentCommand(cfg)
+    elif cfg["command"] == "set_write":
+        return SetButtonWriteCommand(cfg)
+    elif cfg["command"] == "set_cmd":
+        return SetButtonCmdCommand(cfg)
+    elif cfg["command"] == "set_keys":
+        return SetButtonKeysCommand(cfg)
+    elif cfg["command"] == "set_icon":
+        return SetButtonIconCommand(cfg)
+    elif cfg["command"] == "clear_icon":
+        return ClearButtonIconCommand(cfg)
+    return None

streamdeck_ui/cli/server.py

@@ -0,0 +1,189 @@
+import json
+import optparse
+import os
+import socket
+import sys
+import tempfile
+from threading import Event, Thread
+
+from streamdeck_ui.api import StreamDeckServer
+from streamdeck_ui.cli.commands import create_command
+
+
+def read_json(sock: socket.socket) -> dict:
+    header = sock.recv(4)
+    num_bytes = int.from_bytes(header, "little")
+
+    return json.loads(sock.recv(num_bytes))
+
+
+def write_json(sock: socket.socket, data: dict) -> None:
+    binary_data = json.dumps(data).encode("utf-8")
+    num_bytes = len(binary_data)
+
+    sock.send(num_bytes.to_bytes(4, "little"))
+    sock.send(binary_data)
+
+
+class CLIStreamDeckServer:
+    SOCKET_CONNECTION_TIMEOUT_SECOND = 0.5
+
+    def __init__(self, api: StreamDeckServer, ui):
+        self.quit = Event()
+        self.cli_thread = None
+
+        self.api = api
+        self.ui = ui
+
+        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+
+    def start(self):
+        if not self.quit.is_set:
+            return
+
+        self.cli_thread = Thread(target=self._run)
+        self.quit.clear()
+        self.cli_thread.start()
+
+    def stop(self):
+        if self.quit.is_set():
+            return
+
+        self.quit.set()
+        try:
+            self.cli_thread.join()
+        except RuntimeError:
+            pass
+
+    def _run(self):
+        try:
+            tmpdir = tempfile.gettempdir()
+            filename = "streamdeck_ui.sock"
+
+            saved_umask = os.umask(0o077)
+            path = os.path.join(tmpdir, filename)
+        except OSError:
+            pass
+        self.sock.bind(path)
+        self.sock.listen(1)
+        self.sock.settimeout(CLIStreamDeckServer.SOCKET_CONNECTION_TIMEOUT_SECOND)
+
+        while not self.quit.is_set():
+            try:
+                conn, _ = self.sock.accept()
+                cfg = read_json(conn)
+                cmd = create_command(cfg)
+                cmd.execute(self.api, self.ui)
+                conn.close()
+            except BaseException:
+                pass
+        try:
+            os.remove(path)
+        except OSError:
+            pass
+        finally:
+            os.umask(saved_umask)
+
+
+def execute():
+    parser = optparse.OptionParser()
+
+    parser.add_option(
+        "-a",
+        "--action",
+        type="string",
+        dest="action",
+        help="the action to be performed. valid options (case-insensitive): " + "SET_PAGE, SET_BRIGHTNESS, SET_TEXT, SET_ALIGNMENT, SET_CMD, SET_KEYS, SET_WRITE, SET_ICON, CLEAR_ICON",
+        metavar="NAME",
+    )
+
+    parser.add_option("-d", "--deck", type="int", dest="deck_index", help="the deck to be manipulated. defaults to the currently selected deck in the ui", metavar="INDEX")
+    parser.add_option("-p", "--page", type="int", dest="page_index", help="the page to be manipulated. defaults to the currently active page", metavar="INDEX")
+    parser.add_option("-b", "--button", type="int", dest="button_index", help="the button to be manipulated", metavar="INDEX")
+
+    parser.add_option("--icon", type="string", dest="icon_path", help="path to an icon. used with SET_ICON", metavar="PATH")
+    parser.add_option("--brightness", type="int", dest="brightness", help="brightness to set, 0-100. used with SET_BRIGHTNESS", metavar="VALUE")
+    parser.add_option("--text", type="string", dest="button_text", help="button text to set. used with SET_TEXT", metavar="VALUE")
+    parser.add_option("--write", type="string", dest="button_write", help="text to be written when the button is pressed. used with SET_WRITE", metavar="VALUE")
+    parser.add_option("--command", type="string", dest="button_cmd", help="button command to set. used with SET_CMD", metavar="VALUE")
+    parser.add_option("--keys", type="string", dest="button_keys", help="button keys to set. used with SET_KEYS", metavar="VALUE")
+    parser.add_option(
+        "--alignment", type="string", dest="button_text_alignment", help="button text alignment. used with SET_ALIGNMENT. valid values: top, middle-top, middle, middle-bottom, bottom", metavar="VALUE"
+    )
+
+    (options, args) = parser.parse_args(sys.argv)
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    tmpdir = tempfile.gettempdir()
+    file = "streamdeck_ui.sock"
+    path = os.path.join(tmpdir, file)
+    sock.connect(path)
+    data = None
+
+    if options.action is not None:
+        action_name = options.action.lower()
+        if action_name == "set_page":
+            if options.page_index is None:
+                print("error: --page not set...")
+                return
+            data = {"command": "set_page", "deck": options.deck_index, "page": options.page_index}
+        elif action_name == "set_brightness":
+            if options.brightness is None:
+                print("error: --brightness not set...")
+                return
+            data = {"command": "set_brightness", "deck": options.deck_index, "value": options.brightness}
+        elif action_name == "set_text":
+            if options.button_text is None:
+                print("error: --text not set...")
+                return
+            if options.button_index is None:
+                print("error: --button not set...")
+                return
+            data = {"command": "set_button_text", "deck": options.deck_index, "page": options.page_index, "button": options.button_index, "text": options.button_text}
+        elif action_name == "set_write":
+            if options.button_write is None:
+                print("error: --write not set...")
+                return
+            if options.button_index is None:
+                print("error: --button not set...")
+                return
+            data = {"command": "set_button_write", "deck": options.deck_index, "page": options.page_index, "button": options.button_index, "write": options.button_write}
+        elif action_name == "set_alignment":
+            if options.button_text_alignment is None:
+                print("error: --alignment not set...")
+                return
+            if options.button_index is None:
+                print("error: --button not set...")
+                return
+            data = {"command": "set_alignment", "deck": options.deck_index, "page": options.page_index, "button": options.button_index, "alignment": options.button_text_alignment}
+        elif action_name == "set_cmd":
+            if options.button_cmd is None:
+                print("error: --command not set...")
+                return
+            if options.button_index is None:
+                print("error: --button not set...")
+                return
+            data = {"command": "set_button_cmd", "deck": options.deck_index, "page": options.page_index, "button": options.button_index, "button_cmd": options.button_cmd}
+        elif action_name == "set_keys":
+            if options.button_keys is None:
+                print("error: --keys not set...")
+                return
+            if options.button_index is None:
+                print("error: --button not set...")
+                return
+            data = {"command": "set_button_keys", "deck": options.deck_index, "page": options.page_index, "button": options.button_index, "button_keys": options.button_keys}
+        elif action_name == "set_icon":
+            if options.icon_path is None:
+                print("error: --icon not set...")
+                return
+            if options.button_index is None:
+                print("error: --button not set...")
+                return
+            data = {"command": "set_button_icon", "deck": options.deck_index, "page": options.page_index, "button": options.button_index, "icon": options.icon_path}
+        elif action_name == "clear_icon":
+            if options.button_index is None:
+                print("error: --button not set...")
+                return
+            data = {"command": "clear_button_icon", "deck": options.deck_index, "page": options.page_index, "button": options.button_index}
+
+    if data is not None:
+        write_json(sock, data)

streamdeck_ui/gui.py

@@ -15,6 +15,7 @@
 from PySide6.QtWidgets import QApplication, QDialog, QFileDialog, QMainWindow, QMenu, QMessageBox, QSizePolicy, QSystemTrayIcon
 
 from streamdeck_ui.api import StreamDeckServer
+from streamdeck_ui.cli.server import CLIStreamDeckServer
 from streamdeck_ui.config import LOGO, STATE_FILE
 from streamdeck_ui.semaphore import Semaphore, SemaphoreAcquireError
 from streamdeck_ui.ui_main import Ui_MainWindow
@@ -874,6 +875,9 @@ def start(_exit: bool = False) -> None:
 
             api.start()
 
+            cli = CLIStreamDeckServer(api, ui)
+            cli.start()
+
             # Configure signal hanlders
             # https://stackoverflow.com/a/4939113/192815
             timer = QTimer()
@@ -895,6 +899,7 @@ def start(_exit: bool = False) -> None:
             else:
                 app.exec()
                 api.stop()
+                cli.stop()
                 sys.exit()
 
     except SemaphoreAcquireError: