Migrate all actions to ActionCore

This patch migrates each of the 13 action classes to derive from
either `OBSLiveSplitOneCore` or `TimerSourceActionCore` rather than
the older `ActionBase` base classes.

Author: pniedzielski

actions/Interact.py

@@ -1,34 +1,33 @@
-from .TimerSourceActionBase import TimerSourceActionBase
+from .TimerSourceActionCore import TimerSourceActionCore
+from src.backend.PluginManager.EventAssigner import EventAssigner
+from src.backend.DeckManagement.InputIdentifier import Input
 
 from loguru import logger as log
-import os
 from uuid import UUID
 
-class Interact(TimerSourceActionBase):
+
+class Interact(TimerSourceActionCore):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
-    def on_ready(self):
-        if self.plugin_base.backend is None:
-            return
-
-        # Connect to obs if not connected
-        if not self.plugin_base.get_connected():
-            self.reconnect_obs()
+    def get_icon_name(self) -> str:
+        return "interact.png"
 
-        image = "interact.png"
-        self.set_media(
-            media_path=os.path.join(self.plugin_base.PATH, "assets", image)
+    def create_event_assigners(self):
+        self.add_event_assigner(
+            EventAssigner(
+                id="interact",
+                ui_label="Interact",
+                default_event=Input.Key.Events.DOWN,
+                callback=self._on_interact,
+            )
         )
 
-    def on_key_down(self):
-        if not self.plugin_base.backend.get_connected():
-            # Try to reconnect once.
-            self._reconnect_obs()
-            if not self.plugin_base.backend.get_connected():
-                return
+    def _on_interact(self, data):
+        if not self.ensure_connection():
+            return
 
-        uuid_str = self.get_settings().get("source-uuid")
+        uuid_str = self.get_source_uuid()
         if uuid_str:
             self.plugin_base.backend.interact_with_livesplit_one_source(
                 UUID(uuid_str)

actions/NextComparison.py

@@ -1,27 +1,26 @@
-from .OBSLiveSplitOneActionBase import OBSLiveSplitOneActionBase
+from .OBSLiveSplitOneCore import OBSLiveSplitOneCore
+from src.backend.PluginManager.EventAssigner import EventAssigner
+from src.backend.DeckManagement.InputIdentifier import Input
 
-import os
 
-class NextComparison(OBSLiveSplitOneActionBase):
+class NextComparison(OBSLiveSplitOneCore):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
-    def on_ready(self):
-        # Connect to obs if not connected
-        if self.plugin_base.backend is not None:
-            if not self.plugin_base.get_connected():
-                self.reconnect_obs()
+    def get_icon_name(self) -> str:
+        return "next-comparison.png"
 
-        image = "next-comparison.png"
-        self.set_media(
-            media_path=os.path.join(self.plugin_base.PATH, "assets", image)
+    def create_event_assigners(self):
+        self.add_event_assigner(
+            EventAssigner(
+                id="next_comparison",
+                ui_label="Next Comparison",
+                default_event=Input.Key.Events.DOWN,
+                callback=self._on_next_comparison,
+            )
         )
 
-    def on_key_down(self):
-        if not self.plugin_base.backend.get_connected():
-            # Try to reconnect once.
-            self._reconnect_obs()
-            if not self.plugin_base.backend.get_connected():
-                return
-
+    def _on_next_comparison(self, data):
+        if not self.ensure_connection():
+            return
         self.plugin_base.backend.next_comparison()

actions/Pause.py

@@ -1,27 +1,26 @@
-from .OBSLiveSplitOneActionBase import OBSLiveSplitOneActionBase
+from .OBSLiveSplitOneCore import OBSLiveSplitOneCore
+from src.backend.PluginManager.EventAssigner import EventAssigner
+from src.backend.DeckManagement.InputIdentifier import Input
 
-import os
 
-class Pause(OBSLiveSplitOneActionBase):
+class Pause(OBSLiveSplitOneCore):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
-    def on_ready(self):
-        # Connect to obs if not connected
-        if self.plugin_base.backend is not None:
-            if not self.plugin_base.get_connected():
-                self.reconnect_obs()
+    def get_icon_name(self) -> str:
+        return "pause.png"
 
-        image = "pause.png"
-        self.set_media(
-            media_path=os.path.join(self.plugin_base.PATH, "assets", image)
+    def create_event_assigners(self):
+        self.add_event_assigner(
+            EventAssigner(
+                id="pause",
+                ui_label="Pause",
+                default_event=Input.Key.Events.DOWN,
+                callback=self._on_pause,
+            )
         )
 
-    def on_key_down(self):
-        if not self.plugin_base.backend.get_connected():
-            # Try to reconnect once.
-            self._reconnect_obs()
-            if not self.plugin_base.backend.get_connected():
-                return
-
+    def _on_pause(self, data):
+        if not self.ensure_connection():
+            return
         self.plugin_base.backend.pause()

actions/PrevComparison.py

@@ -1,27 +1,26 @@
-from .OBSLiveSplitOneActionBase import OBSLiveSplitOneActionBase
+from .OBSLiveSplitOneCore import OBSLiveSplitOneCore
+from src.backend.PluginManager.EventAssigner import EventAssigner
+from src.backend.DeckManagement.InputIdentifier import Input
 
-import os
 
-class PrevComparison(OBSLiveSplitOneActionBase):
+class PrevComparison(OBSLiveSplitOneCore):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
-    def on_ready(self):
-        # Connect to obs if not connected
-        if self.plugin_base.backend is not None:
-            if not self.plugin_base.get_connected():
-                self.reconnect_obs()
+    def get_icon_name(self) -> str:
+        return "prev-comparison.png"
 
-        image = "prev-comparison.png"
-        self.set_media(
-            media_path=os.path.join(self.plugin_base.PATH, "assets", image)
+    def create_event_assigners(self):
+        self.add_event_assigner(
+            EventAssigner(
+                id="prev_comparison",
+                ui_label="Previous Comparison",
+                default_event=Input.Key.Events.DOWN,
+                callback=self._on_prev_comparison,
+            )
         )
 
-    def on_key_down(self):
-        if not self.plugin_base.backend.get_connected():
-            # Try to reconnect once.
-            self._reconnect_obs()
-            if not self.plugin_base.backend.get_connected():
-                return
-
+    def _on_prev_comparison(self, data):
+        if not self.ensure_connection():
+            return
         self.plugin_base.backend.prev_comparison()

actions/Reset.py

@@ -1,27 +1,26 @@
-from .OBSLiveSplitOneActionBase import OBSLiveSplitOneActionBase
+from .OBSLiveSplitOneCore import OBSLiveSplitOneCore
+from src.backend.PluginManager.EventAssigner import EventAssigner
+from src.backend.DeckManagement.InputIdentifier import Input
 
-import os
 
-class Reset(OBSLiveSplitOneActionBase):
+class Reset(OBSLiveSplitOneCore):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
-    def on_ready(self):
-        # Connect to obs if not connected
-        if self.plugin_base.backend is not None:
-            if not self.plugin_base.get_connected():
-                self.reconnect_obs()
+    def get_icon_name(self) -> str:
+        return "reset.png"
 
-        image = "reset.png"
-        self.set_media(
-            media_path=os.path.join(self.plugin_base.PATH, "assets", image)
+    def create_event_assigners(self):
+        self.add_event_assigner(
+            EventAssigner(
+                id="reset",
+                ui_label="Reset",
+                default_event=Input.Key.Events.DOWN,
+                callback=self._on_reset,
+            )
         )
 
-    def on_key_down(self):
-        if not self.plugin_base.backend.get_connected():
-            # Try to reconnect once.
-            self._reconnect_obs()
-            if not self.plugin_base.backend.get_connected():
-                return
-
+    def _on_reset(self, data):
+        if not self.ensure_connection():
+            return
         self.plugin_base.backend.reset()

actions/SaveSplits.py

@@ -1,34 +1,33 @@
-from .TimerSourceActionBase import TimerSourceActionBase
+from .TimerSourceActionCore import TimerSourceActionCore
+from src.backend.PluginManager.EventAssigner import EventAssigner
+from src.backend.DeckManagement.InputIdentifier import Input
 
 from loguru import logger as log
-import os
 from uuid import UUID
 
-class SaveSplits(TimerSourceActionBase):
+
+class SaveSplits(TimerSourceActionCore):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
-    def on_ready(self):
-        if self.plugin_base.backend is None:
-            return
-
-        # Connect to obs if not connected
-        if not self.plugin_base.get_connected():
-            self.reconnect_obs()
+    def get_icon_name(self) -> str:
+        return "save-splits.png"
 
-        image = "save-splits.png"
-        self.set_media(
-            media_path=os.path.join(self.plugin_base.PATH, "assets", image)
+    def create_event_assigners(self):
+        self.add_event_assigner(
+            EventAssigner(
+                id="save_splits",
+                ui_label="Save Splits",
+                default_event=Input.Key.Events.DOWN,
+                callback=self._on_save_splits,
+            )
         )
 
-    def on_key_down(self):
-        if not self.plugin_base.backend.get_connected():
-            # Try to reconnect once.
-            self._reconnect_obs()
-            if not self.plugin_base.backend.get_connected():
-                return
+    def _on_save_splits(self, data):
+        if not self.ensure_connection():
+            return
 
-        uuid_str = self.get_settings().get("source-uuid")
+        uuid_str = self.get_source_uuid()
         if uuid_str:
             self.plugin_base.backend.save_splits(UUID(uuid_str))
         else:

actions/SetLayoutPath.py

@@ -1,19 +1,23 @@
-from .TimerSourceActionBase import TimerSourceActionBase
+from .TimerSourceActionCore import TimerSourceActionCore
+from src.backend.PluginManager.EventAssigner import EventAssigner
+from src.backend.DeckManagement.InputIdentifier import Input
 
 from loguru import logger as log
-import os
 from uuid import UUID
 
-# Import gtk modules
 import gi
 gi.require_version("Gtk", "4.0")
 gi.require_version("Adw", "1")
 from gi.repository import Gtk, Adw, Gio
 
-class SetLayoutPath(TimerSourceActionBase):
+
+class SetLayoutPath(TimerSourceActionCore):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
+    def get_icon_name(self) -> str:
+        return "set-layout-path.png"
+
     def get_config_rows(self) -> list:
         self.layout_path_settings = Adw.PreferencesGroup()
         self.layout_path_settings.set_title(
@@ -101,27 +105,21 @@ def on_layout_path_dialog_response(self, dialog, async_result):
             return
         self.layout_path_entry.set_text(path.get_path())
 
-    def on_ready(self):
-        if self.plugin_base.backend is None:
-            return
-
-        # Connect to obs if not connected
-        if not self.plugin_base.get_connected():
-            self.reconnect_obs()
-
-        image = "set-layout-path.png"
-        self.set_media(
-            media_path=os.path.join(self.plugin_base.PATH, "assets", image)
+    def create_event_assigners(self):
+        self.add_event_assigner(
+            EventAssigner(
+                id="set_layout_path",
+                ui_label="Set Layout Path",
+                default_event=Input.Key.Events.DOWN,
+                callback=self._on_set_layout_path,
+            )
         )
 
-    def on_key_down(self):
-        if not self.plugin_base.backend.get_connected():
-            # Try to reconnect once.
-            self._reconnect_obs()
-            if not self.plugin_base.backend.get_connected():
-                return
+    def _on_set_layout_path(self, data):
+        if not self.ensure_connection():
+            return
 
-        uuid_str = self.get_settings().get("source-uuid")
+        uuid_str = self.get_source_uuid()
         if not uuid_str:
             log.debug("No source UUID set")
             return