feat(refactor): Refactor to new ActionCore

actions/AdSchedule.py

@@ -0,0 +1,125 @@
+from enum import StrEnum, Enum
+from datetime import datetime, timedelta
+from threading import Thread
+from time import sleep
+
+from GtkHelper.GenerativeUI.SwitchRow import SwitchRow
+from .TwitchCore import TwitchCore
+from src.backend.PluginManager.EventAssigner import EventAssigner
+from src.backend.PluginManager.InputBases import Input
+from src.backend.PluginManager.PluginSettings.Asset import Color
+
+from loguru import logger as log
+
+
+class Icons(StrEnum):
+    DELAY = "delay"
+
+
+class Colors(StrEnum):
+    DEFAULT = "default"
+    WARNING = "warning"
+    ALERT = "alert"
+
+
+class AdSchedule(TwitchCore):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.icon_keys = [Icons.DELAY]
+        self.current_icon = self.get_icon(Icons.DELAY)
+        self.icon_name = Icons.DELAY
+        self.color_keys = [Colors.DEFAULT, Colors.WARNING, Colors.ALERT]
+        self.current_color = self.get_color(Colors.DEFAULT)
+        self.has_configuration = True
+        self._next_ad: datetime = datetime.now()
+        self._snoozes: int = -1
+
+    def on_ready(self):
+        super().on_ready()
+        Thread(
+            target=self._get_ad_schedule, daemon=True, name="get_ad_schedule").start()
+        Thread(
+            target=self._update_ad_timer, daemon=True, name="update_ad_timer").start()
+
+    def create_event_assigners(self):
+        self.event_manager.add_event_assigner(
+            EventAssigner(
+                id="snooze-ad",
+                ui_label="Snooze Ad",
+                default_event=Input.Key.Events.DOWN,
+                callback=self._on_snooze_ad,
+            )
+        )
+
+    def create_generative_ui(self):
+        self._skip_ad_switch = SwitchRow(
+            action_core=self,
+            var_name="ad.snooze",
+            default_value=True,
+            title="ad-snooze",
+            subtitle="Snoozes ad for 5 minutes when pushed",
+            complex_var_name=True
+        )
+
+    def get_config_rows(self):
+        return [self._skip_ad_switch.widget]
+
+    def _update_background_color(self, color: str):
+        self.current_color = self.get_color(color)
+        self.display_color()
+
+    def _update_ad_timer(self):
+        while self.get_is_present():
+            self.display_color()
+            now = datetime.now()
+            self.set_bottom_label(
+                str(self._snoozes) if (self._snoozes >= 0 and self._skip_ad_switch.get_active()) else "")
+            try:
+                if self._next_ad < now:
+                    self._update_background_color(Colors.DEFAULT)
+                    self.set_center_label("")
+                    continue
+                diff = (self._next_ad - now).total_seconds()
+                self.set_center_label(self._convert_seconds_to_hh_mm_ss(diff))
+                if diff <= 60:
+                    self._update_background_color(Colors.ALERT)
+                    continue
+                if diff <= 300:
+                    self._update_background_color(Colors.WARNING)
+                    continue
+                self._update_background_color(Colors.DEFAULT)
+            except TypeError:
+                # There is a known issue where the default timestamp returned from
+                # the twitch API is an invalid datetime object and causes an error.
+                # Ignoring it here
+                pass
+            except Exception as ex:
+                log.error(ex)
+            sleep(1)
+
+    def _get_ad_schedule(self):
+        while self.get_is_present():
+            try:
+                schedule, snoozes = self.backend.get_next_ad()
+                self._next_ad = schedule
+                self._snoozes = snoozes
+                self._update_ad_timer()
+            except Exception as ex:
+                log.error(ex)
+                self.show_error(3)
+            sleep(30)
+
+    def _convert_seconds_to_hh_mm_ss(self, seconds) -> str:
+        hours = seconds // 3600
+        minutes = (seconds % 3600) // 60
+        remaining_seconds = seconds % 60
+        return f"{int(hours):02}:{int(minutes):02}:{int(remaining_seconds):02}"
+
+    def _on_snooze_ad(self, _):
+        if not self._skip_ad_switch.get_active():
+            return
+        try:
+            self.backend.snooze_ad()
+        except Exception as ex:
+            log.error(ex)
+            self.show_error(3)

actions/ChatMode.py

@@ -0,0 +1,97 @@
+from enum import StrEnum, Enum
+from threading import Thread
+from time import sleep
+
+from .TwitchCore import TwitchCore
+from src.backend.PluginManager.EventAssigner import EventAssigner
+from src.backend.PluginManager.InputBases import Input
+from GtkHelper.GenerativeUI.ComboRow import ComboRow
+from GtkHelper.ComboRow import SimpleComboRowItem, BaseComboRowItem
+
+from loguru import logger as log
+
+
+class Icons(StrEnum):
+    FOLLOWER = "follower_mode"
+    SUBSCRIBER = "subscriber_mode"
+    EMOTE = "emote_mode"
+    SLOW = "slow_mode"
+
+
+class ChatModeOptions(Enum):
+    FOLLOWER = SimpleComboRowItem("follower_mode", "Follower Only")
+    SUBSCRIBER = SimpleComboRowItem("subscriber_mode", "Subscriber Only")
+    EMOTE = SimpleComboRowItem("emote_mode", "Emote Only")
+    SLOW = SimpleComboRowItem("slow_mode", "Slow Mode")
+
+
+class ChatMode(TwitchCore):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.icon_keys = [Icons.FOLLOWER,
+                          Icons.SUBSCRIBER, Icons.EMOTE, Icons.SLOW]
+        self.current_icon = self.get_icon(Icons.FOLLOWER)
+        self.icon_name = Icons.FOLLOWER
+
+    def create_event_assigners(self):
+        self.event_manager.add_event_assigner(
+            EventAssigner(
+                id="chat-toggle",
+                ui_label="Chat Toggle",
+                default_event=Input.Key.Events.DOWN,
+                callback=self._on_toggle_chat,
+            )
+        )
+
+    def create_generative_ui(self):
+        self._chat_select_row = ComboRow(
+            action_core=self,
+            var_name="chat.mode",
+            default_value=ChatModeOptions.FOLLOWER.value,
+            items=[
+                ChatModeOptions.FOLLOWER.value,
+                ChatModeOptions.SUBSCRIBER.value,
+                ChatModeOptions.EMOTE.value,
+                ChatModeOptions.SLOW.value,
+            ],
+            title="chat-toggle-dropdown",
+            complex_var_name=True,
+            on_change=self._change_chat_mode
+        )
+
+    def on_ready(self):
+        Thread(
+            target=self._update_chat_mode, daemon=True, name="update_chat_mode").start()
+
+    def get_config_rows(self):
+        return [self._chat_select_row.widget]
+
+    def _change_chat_mode(self, _, new, __):
+        self.icon_name = Icons(new)
+        self.current_icon = self.get_icon(self.icon_name)
+        self.display_icon()
+
+    def _update_icon(self, mode: str, enabled: bool):
+        # TODO: Custom icons for enabled/disabled
+        self.set_center_label("Enabled" if enabled else "Disabled")
+
+    def _update_chat_mode(self):
+        while self.get_is_present():
+            try:
+                chat_settings = self.backend.get_chat_settings()
+                mode = self._chat_select_row.get_selected_item().get_value()
+                enabled = chat_settings.get(mode)
+                self._update_icon(mode, enabled)
+            except Exception as ex:
+                log.error(ex)
+                self.show_error(3)
+            sleep(5)
+
+    def _on_toggle_chat(self, _):
+        item = self._chat_select_row.get_selected_item().get_value()
+        try:
+            resp = self.backend.toggle_chat_mode(item)
+            self._update_icon(item, resp)
+        except Exception as ex:
+            log.error(ex)
+            self.show_error(3)

actions/Clip.py

@@ -0,0 +1,37 @@
+from enum import StrEnum
+
+from loguru import logger as log
+
+from .TwitchCore import TwitchCore
+from src.backend.PluginManager.EventAssigner import EventAssigner
+from src.backend.PluginManager.InputBases import Input
+
+
+class Icons(StrEnum):
+    CLIP = "camera"
+
+
+class Clip(TwitchCore):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.icon_keys = [Icons.CLIP]
+        self.current_icon = self.get_icon(Icons.CLIP)
+        self.icon_name = Icons.CLIP
+        self.has_configuration = False
+
+    def create_event_assigners(self):
+        self.event_manager.add_event_assigner(
+            EventAssigner(
+                id="clip",
+                ui_label="Clip",
+                default_event=Input.Key.Events.DOWN,
+                callback=self._on_clip
+            )
+        )
+
+    def _on_clip(self, _):
+        try:
+            self.backend.create_clip()
+        except Exception as ex:
+            log.error(ex)
+            self.show_error(3)

actions/Marker.py

@@ -0,0 +1,37 @@
+from enum import StrEnum
+
+from loguru import logger as log
+
+from .TwitchCore import TwitchCore
+from src.backend.PluginManager.EventAssigner import EventAssigner
+from src.backend.PluginManager.InputBases import Input
+
+
+class Icons(StrEnum):
+    MARKER = "bookmark"
+
+
+class Marker(TwitchCore):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.icon_keys = [Icons.MARKER]
+        self.current_icon = self.get_icon(Icons.MARKER)
+        self.icon_name = Icons.MARKER
+        self.has_configuration = False
+
+    def create_event_assigners(self):
+        self.event_manager.add_event_assigner(
+            EventAssigner(
+                id="marker",
+                ui_label="Marker",
+                default_event=Input.Key.Events.DOWN,
+                callback=self._on_marker
+            )
+        )
+
+    def _on_marker(self, _):
+        try:
+            self.backend.create_marker()
+        except Exception as ex:
+            log.error(ex)
+            self.show_error(3)

actions/PlayAd.py

@@ -0,0 +1,54 @@
+from enum import StrEnum, Enum
+
+from loguru import logger as log
+
+from .TwitchCore import TwitchCore
+from src.backend.PluginManager.EventAssigner import EventAssigner
+from src.backend.PluginManager.InputBases import Input
+from GtkHelper.GenerativeUI.ComboRow import ComboRow
+from GtkHelper.ComboRow import SimpleComboRowItem, BaseComboRowItem
+
+
+class Icons(StrEnum):
+    AD = "money"
+
+
+class PlayAd(TwitchCore):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.icon_keys = [Icons.AD]
+        self.current_icon = self.get_icon(Icons.AD)
+        self.icon_name = Icons.AD
+        self.has_configuration = True
+
+    def create_generative_ui(self):
+        options = [SimpleComboRowItem(x, f"{x} seconds")
+                   for x in [30, 60, 90, 120]]
+        self._time_row = ComboRow(
+            action_core=self,
+            var_name="ad.duration",
+            default_value=options[0],
+            items=options,
+            title="ad-options-dropdown",
+            complex_var_name=True,
+        )
+
+    def get_config_rows(self):
+        return [self._time_row.widget]
+
+    def create_event_assigners(self):
+        self.event_manager.add_event_assigner(
+            EventAssigner(
+                id="play-ad",
+                ui_label="Play Ad",
+                default_event=Input.Key.Events.DOWN,
+                callback=self._on_play_ad
+            )
+        )
+
+    def _on_play_ad(self, _):
+        try:
+            self.backend.play_ad(10)
+        except Exception as ex:
+            log.error(ex)
+            self.show_error(3)