Procházet zdrojové kódy

require error sink state id for unstateless buttons + affects_button works when cmd is the same for each state

Noah Vogt před 2 měsíci
rodič
revize
7752247872
3 změnil soubory, kde provedl 109 přidání a 17 odebrání
  1. 24 10
      VulcanBoard.py
  2. 75 7
      config/load.py
  3. 10 0
      config/state.py

+ 24 - 10
VulcanBoard.py

@@ -19,7 +19,7 @@
 import threading
 import sys
 import asyncio
-
+from functools import partial
 import colorama
 
 from kivy.app import App
@@ -36,7 +36,6 @@ from config import (
     Config,
     get_state_from_id,
     get_state_id_from_exit_code,
-    contains_id,
     DEFAULT_BUTTON_BG_COLOR,
     DEFAULT_BUTTON_FG_COLOR,
     EMPTY_BUTTON_BG_COLOR,
@@ -49,6 +48,8 @@ from ui import AutoResizeButton
 class VulcanBoardApp(App):
     def build(self):
         self.loop = self.ensure_asyncio_loop_running()
+        self.button_grid = {}
+        self.button_config_map = {}
         self.icon = "icon.jpg"
         config_loader = ConfigLoader(get_config_path())
         config = config_loader.get_config()  # pyright: ignore
@@ -61,7 +62,7 @@ class VulcanBoardApp(App):
             kvConfig.set("kivy", "window_icon", "icon.ico")
             kvConfig.set("kivy", "exit_on_escape", "0")
 
-            button_map = {
+            self.button_config_map = {
                 (btn["position"][0], btn["position"][1]): btn
                 for btn in config.buttons
             }
@@ -76,7 +77,7 @@ class VulcanBoardApp(App):
             # Populate grid with buttons and placeholders
             for row in range(config.rows):
                 for col in range(config.columns):
-                    defined_button = button_map.get((row, col))
+                    defined_button = self.button_config_map.get((row, col))
                     if defined_button:
                         states = defined_button.get("states", [])
                         state_id = [DEFAULT_STATE_ID]
@@ -98,15 +99,15 @@ class VulcanBoardApp(App):
                         if defined_button.get("autostart", False):
                             self.async_task(
                                 self.execute_command_async(
-                                    states, state_id, btn
+                                    defined_button, state_id, btn
                                 )
                             )
 
                         # pylint: disable=no-member
                         btn.bind(  # pyright: ignore
-                            on_release=lambda btn_instance, states=states, state_id=state_id: self.async_task(
+                            on_release=lambda btn_instance, button=defined_button, state_id=state_id: self.async_task(
                                 self.execute_command_async(
-                                    states, state_id, btn_instance
+                                    button, state_id, btn_instance
                                 )
                             )
                         )
@@ -117,6 +118,7 @@ class VulcanBoardApp(App):
                                 EMPTY_BUTTON_BG_COLOR
                             ),
                         )
+                    self.button_grid[(row, col)] = btn
                     layout.add_widget(btn)
 
             return layout
@@ -142,21 +144,20 @@ class VulcanBoardApp(App):
         sys.exit(1)
 
     async def execute_command_async(
-        self, states: list, state_id: list[int], btn: AutoResizeButton
+        self, button: dict, state_id: list[int], btn: AutoResizeButton
     ):
         follow_up_state_loop = True
+        states = button["states"]
         while follow_up_state_loop:
             new_state_id = get_state_id_from_exit_code(states, state_id[0])
             state = get_state_from_id(states, new_state_id)
             follow_up_state_loop = False
 
             try:
-                print(states[new_state_id]["cmd"])
                 process = await asyncio.create_subprocess_shell(
                     state["cmd"], shell=True
                 )
                 exit_code = await process.wait()
-                print(f"EXIT {exit_code}")
                 log(f"Executed command: {state['cmd']}")
             except Exception as e:
                 exit_code = ERROR_SINK_STATE_ID
@@ -175,6 +176,19 @@ class VulcanBoardApp(App):
                         states, btn, exit_code  # pyright: ignore
                     )
                 )
+                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]
+                        # TODO: check if also works if cmd is not the same for each state
+                        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,  # pyright: ignore
+                            )
+                        )
 
     def update_button_feedback(
         self, states: list, btn: AutoResizeButton, exit_code: int

+ 75 - 7
config/load.py

@@ -25,7 +25,9 @@ from .const import (
     DEFAULT_BUTTON_BG_COLOR,
     DEFAULT_BUTTON_FG_COLOR,
     DEFAULT_STATE_ID,
+    ERROR_SINK_STATE_ID,
 )
+from .state import get_state_ids
 
 
 @dataclass
@@ -75,19 +77,16 @@ class ConfigLoader:
             raise CustomException(
                 "invalid button config. needs to be a list of dicts."
             )
+        buttons_that_affect_others = set()
+        button_grid = {}
         for button in self.buttons:
             if not isinstance(button, dict):
                 raise CustomException(
                     "invalid button config. needs to be a list of dicts."
                 )
 
-            if (
-                not isinstance(dimensions := button.get("position", ""), list)
-                or (not isinstance(dimensions[0], int))
-                or (not isinstance(dimensions[1], int))
-                or (0 > dimensions[0] or dimensions[0] > self.rows - 1)
-                or (0 > dimensions[1] or dimensions[1] > self.columns - 1)
-            ):
+            dimensions = button.get("position", "")
+            if not self.is_valid_dimension(dimensions):
                 raise CustomException(
                     f"invalid 'position' subentry: '{dimensions}'"
                 )
@@ -156,11 +155,38 @@ class ConfigLoader:
                     )
                 to_follow_up_state_ids.add(follow_up_state)
 
+            button_grid[(dimensions[0], dimensions[1])] = button
+
+            affects_buttons = button.get("affects_buttons", None)
+            if isinstance(affects_buttons, list):
+                if len(affects_buttons) == 0:
+                    raise CustomException(
+                        f"invalid {btn_dims}: 'affects_buttons' entry: must be"
+                        + "a non-empty list"
+                    )
+
+            if affects_buttons:
+                for affected_button_dimension in affects_buttons:
+                    if not self.is_valid_dimension(affected_button_dimension):
+                        raise CustomException(
+                            f"invalid {btn_dims}: 'affects_buttons' entry: "
+                            + "invalid dimensions: "
+                            + f"'{affected_button_dimension}'"
+                        )
+                buttons_that_affect_others.add(str(dimensions))
+
             if not DEFAULT_STATE_ID in defined_state_ids:
                 raise CustomException(
                     f"invalid {btn_dims}: missing default state id "
                     + f"'{DEFAULT_STATE_ID}'"
                 )
+            if (len(defined_state_ids) > 1) and (
+                not ERROR_SINK_STATE_ID in defined_state_ids
+            ):
+                raise CustomException(
+                    f"invalid {btn_dims}: missing error sink state id "
+                    + f"'{ERROR_SINK_STATE_ID}' for unstateless button"
+                )
 
             for follow_up_state_id in to_follow_up_state_ids:
                 if follow_up_state_id not in defined_state_ids:
@@ -170,6 +196,48 @@ class ConfigLoader:
                         + "not exist"
                     )
 
+        for btn_dims in buttons_that_affect_others:
+            row = int(btn_dims[btn_dims.find("[") + 1 : btn_dims.find(",")])
+            col = int(btn_dims[btn_dims.find(" ") + 1 : btn_dims.find("]")])
+            button_dimensions = (row, col)
+
+            button = button_grid[button_dimensions]
+            affects_buttons = button["affects_buttons"]
+            ids = []
+            ids.append(get_state_ids(button["states"]))
+            for affected_btn_dims in affects_buttons:
+                try:
+                    affected_button = button_grid[
+                        (affected_btn_dims[0], affected_btn_dims[1])
+                    ]
+                except KeyError as e:
+                    raise CustomException(
+                        f"invalid button ({row}, {col}): 'affects_buttons' "
+                        + "buttons must be defined"
+                    ) from e
+                ids.append(get_state_ids(affected_button["states"]))
+
+            for id_listing in ids[1:]:
+                if len(id_listing) == 1:
+                    raise CustomException(
+                        f"invalid button ({row}, {col}): 'affects_buttons' "
+                        + "buttons cannot be stateless"
+                    )
+                if id_listing != ids[0]:
+                    raise CustomException(
+                        f"invalid button ({row}, {col}): 'affects_buttons' "
+                        + "buttons must have the same state id's"
+                    )
+
+    def is_valid_dimension(self, dimensions):
+        return not (
+            not isinstance(dimensions, list)
+            or (not isinstance(dimensions[0], int))
+            or (not isinstance(dimensions[1], int))
+            or (0 > dimensions[0] or dimensions[0] > self.rows - 1)
+            or (0 > dimensions[1] or dimensions[1] > self.columns - 1)
+        )
+
     def __validate_dimensions(self) -> None:
         for dimension in (self.columns, self.rows):
             if not isinstance(dimension, int) or (dimension <= 0):

+ 10 - 0
config/state.py

@@ -41,3 +41,13 @@ def get_state_id_from_exit_code(states: list, exit_code: int) -> int:
         exit_code = ERROR_SINK_STATE_ID
 
     return exit_code
+
+
+def get_state_ids(states: list) -> list:
+    output = set()
+    for state in states:
+        possible_id = state.get("id", None)
+        if isinstance(possible_id, int):
+            output.add(possible_id)
+
+    return list(output)