VulcanBoard/VulcanBoard.py

#!/usr/bin/env python3

# Copyright © 2024 Noah Vogt <noah@noahvogt.com>

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program 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 General Public License for more details.

# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

# pylint: disable=invalid-name
import threading
import sys
import asyncio
import time
import colorama
import uvicorn
from fastapi import FastAPI, HTTPException

from kivy.app import App
from kivy.uix.gridlayout import GridLayout
from kivy.utils import get_color_from_hex
from kivy.config import Config as kvConfig
from kivy.core.window import Window
from kivy.clock import Clock

from util import log, error_exit_gui
from config import (
    get_config_path,
    ConfigLoader,
    Config,
    get_state_from_id,
    get_state_id_from_exit_code,
    DEFAULT_BUTTON_BG_COLOR,
    DEFAULT_BUTTON_FG_COLOR,
    EMPTY_BUTTON_BG_COLOR,
    DEFAULT_STATE_ID,
    ERROR_SINK_STATE_ID,
)
from ui import AutoResizeButton


class VulcanBoardApp(App):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.last_touch_time = 0  # For debounce handling

    def build(self):
        self.loop = self.ensure_asyncio_loop_running()
        self.button_grid = {}
        self.button_config_map = {}
        self.icon = "icon.png"
        config_loader = ConfigLoader(get_config_path())
        config = config_loader.get_config()  # pyright: ignore

        if isinstance(config, str):
            error_exit_gui(config)
        else:
            config: Config = config

            Window.borderless = config.borderless
            if config.set_window_pos:
                Window.left = config.window_pos_x
                Window.top = config.window_pos_y
            if config.use_auto_fullscreen_mode:
                Window.fullscreen = "auto"

            kvConfig.set("kivy", "exit_on_escape", "0")

            self.button_config_map = {
                (btn["position"][0], btn["position"][1]): btn
                for btn in config.buttons
            }

            layout = GridLayout(
                cols=config.columns,
                rows=config.rows,
                spacing=config.spacing,
                padding=config.padding,
            )

            # Populate grid with buttons and placeholders
            for row in range(config.rows):
                for col in range(config.columns):
                    defined_button = self.button_config_map.get((row, col))
                    if defined_button:
                        states = defined_button.get("states", [])
                        state = get_state_from_id(states, DEFAULT_STATE_ID)

                        btn = AutoResizeButton(
                            text=state.get("txt", ""),
                            background_color=get_color_from_hex(
                                state.get("bg_color", DEFAULT_BUTTON_BG_COLOR)
                            ),
                            color=get_color_from_hex(
                                state.get("fg_color", DEFAULT_BUTTON_FG_COLOR)
                            ),
                            halign="center",
                            valign="middle",
                            background_normal="",
                            state_id=DEFAULT_STATE_ID,
                        )

                        if defined_button.get("autostart", False):
                            self.async_task(
                                self.execute_command_async(defined_button, btn)
                            )

                        # Use debounce wrapper instead of raw on_release to
                        # avoid double execution on single taps on touchscreens
                        btn.bind(  # pyright: ignore pylint: disable=no-member
                            on_release=lambda btn_instance, button=defined_button: self.on_button_pressed_once(
                                button, btn_instance
                            )
                        )

                    else:
                        btn = AutoResizeButton(
                            background_color=get_color_from_hex(
                                EMPTY_BUTTON_BG_COLOR
                            ),
                        )
                    self.button_grid[(row, col)] = btn
                    layout.add_widget(btn)

            self.api_app = FastAPI(title="VulcanBoard API")
            self._setup_api_routes()

            uvicorn_config = uvicorn.Config(
                app=self.api_app,
                host="0.0.0.0",
                port=config.api_port,
                log_level="info",
            )
            server = uvicorn.Server(uvicorn_config)
            self.async_task(server.serve())

            return layout

    def on_button_pressed_once(self, button, btn_instance):
        now = time.time()
        if now - self.last_touch_time > 0.3:  # 300 ms debounce
            self.last_touch_time = now
            self.async_task(self.execute_command_async(button, btn_instance))

    def async_task(self, coroutine):
        asyncio.run_coroutine_threadsafe(coroutine, self.loop)

    def ensure_asyncio_loop_running(self):
        if hasattr(self, "loop"):
            return self.loop

        loop = asyncio.new_event_loop()

        def run_loop():
            asyncio.set_event_loop(loop)
            loop.run_forever()

        threading.Thread(target=run_loop, daemon=True).start()
        return loop

    def config_error_exit(self, popup):
        popup.dismiss()
        sys.exit(1)

    async def execute_command_async(self, button: dict, btn: AutoResizeButton):
        follow_up_state_loop = True
        states = button["states"]
        while follow_up_state_loop:
            state = get_state_from_id(states, btn.state_id)
            follow_up_state_loop = False

            try:
                process = await asyncio.create_subprocess_shell(
                    state["cmd"], shell=True
                )
                exit_code = await process.wait()
                log(f"Executed command: {state['cmd']}")
            except Exception as e:
                exit_code = ERROR_SINK_STATE_ID
                log(f"Error executing command: {e}", color="yellow")

            if len(states) != 1:
                if isinstance(
                    follow_up_state := state.get("follow_up_state"), int
                ):
                    follow_up_state_loop = True
                    exit_code = follow_up_state

                Clock.schedule_once(
                    lambda _: self.update_button_feedback(
                        states, btn, exit_code
                    )
                )
                affects_buttons = button.get("affects_buttons", None)
                if affects_buttons:
                    for affected_btn_dims in affects_buttons:
                        btn_pos = (affected_btn_dims[0], affected_btn_dims[1])
                        affected_button = self.button_grid[btn_pos]
                        Clock.schedule_once(
                            lambda _, btn_pos=btn_pos, affected_button=affected_button: self.update_button_feedback(
                                self.button_config_map[btn_pos]["states"],
                                affected_button,
                                exit_code,
                            )
                        )

    def update_button_feedback(
        self, states: list, btn: AutoResizeButton, exit_code: int
    ):
        state_id = get_state_id_from_exit_code(states, exit_code)
        state = get_state_from_id(states, state_id)

        btn.text = state.get("txt", "")
        btn.state_id = state_id
        btn.background_color = get_color_from_hex(
            state.get("bg_color", DEFAULT_BUTTON_BG_COLOR)
        )
        btn.color = get_color_from_hex(
            state.get("fg_color", DEFAULT_BUTTON_FG_COLOR)
        )

    def update_button_state_from_api(
        self, dt, row: int, col: int, state_id: int
    ):
        btn = self.button_grid.get((row, col))
        btn_config = self.button_config_map.get((row, col))
        if not btn or not btn_config:
            return

        states = btn_config.get("states", [])
        state = get_state_from_id(states, state_id)

        btn.text = state.get("txt", "")
        btn.state_id = state_id
        btn.background_color = get_color_from_hex(
            state.get("bg_color", DEFAULT_BUTTON_BG_COLOR)
        )
        btn.color = get_color_from_hex(
            state.get("fg_color", DEFAULT_BUTTON_FG_COLOR)
        )

    def _setup_api_routes(self):
        @self.api_app.get("/get_states")
        def get_states(x: int, y: int):
            btn_config = self.button_config_map.get((x, y))
            if not btn_config:
                raise HTTPException(status_code=404, detail="Button not found")
            return {"states": btn_config.get("states", [])}

        @self.api_app.get("/get_current_state")
        def get_current_state(x: int, y: int):
            btn = self.button_grid.get((x, y))
            if not btn:
                raise HTTPException(status_code=404, detail="Button not found")
            return {"state_id": getattr(btn, "state_id", DEFAULT_STATE_ID)}

        @self.api_app.get("/set_state")
        @self.api_app.post("/set_state")
        def set_state(x: int, y: int, state: int):
            btn_config = self.button_config_map.get((x, y))
            btn = self.button_grid.get((x, y))
            if not btn_config or not btn:
                raise HTTPException(status_code=404, detail="Button not found")

            states = btn_config.get("states", [])
            valid_state = None
            for s in states:
                if s.get("id", DEFAULT_STATE_ID) == state:
                    valid_state = s
                    break
            if not valid_state:
                raise HTTPException(
                    status_code=400,
                    detail=f"State {state} not found for this button",
                )

            Clock.schedule_once(
                lambda dt: self.update_button_state_from_api(dt, x, y, state)
            )
            return {"status": "success", "state_id": state}


def start_asyncio_loop():
    asyncio.set_event_loop(asyncio.new_event_loop())
    asyncio.get_event_loop().run_forever()


if __name__ == "__main__":
    colorama.init()
    VulcanBoardApp().run()