custom-streamdeck/backend/services/actions.py

from __future__ import annotations

import asyncio
from typing import Any

from backend.services.apps import AppDiscovery
from backend.services.plugins import PluginContext


SPECIAL_KEYS = {
    "ctrl": "ctrl",
    "control": "ctrl",
    "shift": "shift",
    "alt": "alt",
    "cmd": "cmd",
    "win": "cmd",
    "enter": "enter",
    "return": "enter",
    "esc": "esc",
    "escape": "esc",
    "tab": "tab",
    "space": "space",
    "backspace": "backspace",
    "delete": "delete",
    "up": "up",
    "down": "down",
    "left": "left",
    "right": "right",
}


class KeyPresser:
    def __init__(self) -> None:
        self._controller = None
        self._key = None

    def _ensure(self) -> None:
        if self._controller is not None:
            return
        try:
            from pynput.keyboard import Controller, Key
        except ImportError as exc:
            raise RuntimeError("pynput is not installed. Run: python -m pip install -r requirements.txt") from exc
        self._controller = Controller()
        self._key = Key

    async def press_combo(self, combo: str) -> None:
        await asyncio.to_thread(self._press_combo_sync, combo)

    def _press_combo_sync(self, combo: str) -> None:
        self._ensure()
        parts = [part.strip().lower() for part in combo.replace(" ", "").split("+") if part.strip()]
        if not parts:
            raise ValueError("Key combo is empty.")
        keys = [self._resolve_key(part) for part in parts]
        for key in keys[:-1]:
            self._controller.press(key)
        self._controller.press(keys[-1])
        self._controller.release(keys[-1])
        for key in reversed(keys[:-1]):
            self._controller.release(key)

    def _resolve_key(self, name: str) -> Any:
        mapped = SPECIAL_KEYS.get(name, name)
        if len(mapped) == 1:
            return mapped
        key = getattr(self._key, mapped, None)
        if key is None:
            raise ValueError(f"Unknown key '{name}'.")
        return key


class ActionEngine:
    def __init__(self, app: Any):
        self.app = app
        self.keys = KeyPresser()
        self.apps = AppDiscovery(app.db)

    async def handle_button_event(self, event: dict[str, Any]) -> None:
        settings = self.app.db.settings()
        profile_id = settings.get("active_profile_id")
        folder_id = settings.get("active_folder_id")
        if not profile_id or not folder_id:
            return

        button = self.app.db.find_button_for_event(profile_id, folder_id, int(event["button"]))
        payload = {"event": event, "button": button}
        await self.app.ws.broadcast(f"button.{event['event']}", payload)

        await self.app.plugins.on_event(PluginContext(self.app), {"type": f"button.{event['event']}", "payload": payload})

        if settings.get("click_check"):
            await self.app.ws.broadcast("click_check.event", payload)
            return

        if not button or button["trigger_mode"] != event["event"]:
            return
        await self.execute(button["action_type"], button["action_config"], event, source_button=button)

    async def execute(self, action_type: str, config: dict[str, Any], event: dict[str, Any] | None = None, source_button: dict[str, Any] | None = None) -> None:
        action_id = config.get("id") or action_type
        payload = {"action_type": action_type, "action_config": config, "button": source_button, "event": event}
        await self.app.ws.broadcast("action.started", payload)
        try:
            await self._execute(action_type, config, event)
        except Exception as exc:
            await self.app.ws.broadcast("action.failed", payload | {"error": str(exc)})
            raise
        await self.app.ws.broadcast("action.finished", payload | {"id": action_id})

    async def _execute(self, action_type: str, config: dict[str, Any], event: dict[str, Any] | None) -> None:
        if action_type == "noop":
            return
        if action_type == "key_combo":
            await self.keys.press_combo(str(config.get("combo", "")))
            return
        if action_type == "chain":
            for step in config.get("steps", []):
                await self._execute(step.get("action_type", "noop"), step.get("action_config", {}), event)
                delay_ms = int(step.get("delay_ms", 0) or 0)
                if delay_ms > 0:
                    await asyncio.sleep(delay_ms / 1000)
            return
        if action_type == "app_launch":
            await asyncio.to_thread(self.apps.launch, config)
            return
        if action_type == "folder":
            folder_id = config.get("folder_id")
            if not folder_id:
                raise ValueError("Folder action requires folder_id.")
            self.app.db.set_setting("active_folder_id", folder_id)
            await self.app.ws.broadcast("state.updated", self.app.public_state())
            return
        if action_type == "folder_rotation":
            settings = self.app.db.settings()
            profile_id = settings.get("active_profile_id")
            folder_id = settings.get("active_folder_id")
            if not profile_id or not folder_id:
                raise ValueError("Folder rotation requires an active profile and folder.")
            next_folder_id = self.app.db.next_folder_id(profile_id, folder_id, str(config.get("direction", "next")))
            if not next_folder_id:
                raise ValueError("No folders are available for rotation.")
            self.app.db.set_setting("active_folder_id", next_folder_id)
            await self.app.ws.broadcast("state.updated", self.app.public_state())
            return
        if action_type == "plugin":
            await self.app.plugins.execute_action(
                PluginContext(self.app),
                str(config.get("plugin_id", "")),
                str(config.get("action_id", "")),
                config.get("fields", {}),
                event,
            )
            return
        raise ValueError(f"Unknown action type '{action_type}'.")