initial commit

duckypad_twitch/__about__.py

@@ -0,0 +1,4 @@
+# SPDX-FileCopyrightText: 2023-present onyx-and-iris <75868496+onyx-and-iris@users.noreply.github.com>
+#
+# SPDX-License-Identifier: MIT
+__version__ = "1.0.0"

duckypad_twitch/__init__.py

@@ -0,0 +1,5 @@
+# SPDX-FileCopyrightText: 2023-present onyx-and-iris <75868496+onyx-and-iris@users.noreply.github.com>
+#
+# SPDX-License-Identifier: MIT
+
+from .duckypad import connect

duckypad_twitch/audio.py

@@ -0,0 +1,132 @@
+import logging
+from enum import IntEnum
+
+import vban_cmd
+
+from . import configuration
+from .layer import ILayer
+from .states import AudioState
+
+logger = logging.getLogger(__name__)
+
+
+Buttons = IntEnum("Buttons", "mute_mics only_discord only_stream", start=0)
+
+
+class Audio(ILayer):
+    """Audio concrete class"""
+
+    def __init__(self, duckypad, **kwargs):
+        super().__init__(duckypad)
+        for attr, val in kwargs.items():
+            setattr(self, attr, val)
+
+        self.reset_states()
+
+    @property
+    def identifier(self):
+        return type(self).__name__
+
+    @property
+    def state(self):
+        return self._state
+
+    @state.setter
+    def state(self, val):
+        self._state = val
+
+    def reset_states(self):
+        self.state = AudioState()
+        for button in Buttons:
+            self.vm.button[button].stateonly = getattr(AudioState, button.name)
+
+    def mute_mics(self):
+        self.state.mute_mics = not self.state.mute_mics
+        if self.state.mute_mics:
+            self.vm.strip[0].mute = True
+            self.vm.strip[1].mute = True
+            self.vm.strip[4].mute = True
+            self.logger.info("Mics Muted")
+        else:
+            self.vm.strip[0].mute = False
+            self.vm.strip[1].mute = False
+            self.vm.strip[4].mute = False
+            self.logger.info("Mics Unmuted")
+        self.vm.button[Buttons.mute_mics].stateonly = self.state.mute_mics
+
+    def only_discord(self):
+        self.state.only_discord = not self.state.only_discord
+        if self.state.only_discord:
+            self.mixer.dca[0].on = False
+            self.vm.strip[4].mute = True
+            self.logger.info("Only Discord Enabled")
+        else:
+            self.vm.strip[4].mute = False
+            self.mixer.dca[0].on = True
+            self.logger.info("Only Discord Disabled")
+        self.vm.button[Buttons.only_discord].stateonly = self.state.only_discord
+
+    def only_stream(self):
+        self.state.only_stream = not self.state.only_stream
+        if self.state.only_stream:
+            self.vm.bus[5].mute = True
+            self.vm.bus[6].mute = True
+            self.vm.strip[2].gain = -3
+            self.vm.strip[3].gain = -3
+            self.vm.strip[6].gain = -3
+            self.logger.info("Only Stream Enabled")
+        else:
+            self.vm.strip[2].gain = 0
+            self.vm.strip[3].gain = 0
+            self.vm.strip[6].gain = 0
+            self.vm.bus[5].mute = False
+            self.vm.bus[6].mute = False
+            self.logger.info("Only Stream Disabled")
+        self.vm.button[Buttons.only_stream].stateonly = self.state.only_stream
+
+    def sound_test(self):
+        def toggle_soundtest(script):
+            onyx_conn = configuration.get("vban_onyx")
+            iris_conn = configuration.get("vban_iris")
+            assert all(
+                [onyx_conn, iris_conn]
+            ), "expected configurations for onyx_conn, iris_conn"
+
+            with vban_cmd.api("potato", **onyx_conn) as vban:
+                vban.sendtext(script)
+            with vban_cmd.api("potato", **iris_conn) as vban:
+                vban.sendtext(script)
+
+        ENABLE_SOUNDTEST = "Strip(0).A1=1; Strip(0).A2=1; Strip(0).B1=0; Strip(0).B2=0; Strip(0).mono=1;"
+        DISABLE_SOUNDTEST = "Strip(0).A1=0; Strip(0).A2=0; Strip(0).B1=1; Strip(0).B2=1; Strip(0).mono=0;"
+
+        self.state.sound_test = not self.state.sound_test
+        if self.state.sound_test:
+            self.vm.strip[4].apply({"B3": False, "A1": True, "mute": False})
+            self.vm.vban.outstream[0].on = True
+            self.vm.vban.outstream[1].on = True
+            self.vm.vban.outstream[0].route = 0
+            self.vm.vban.outstream[1].route = 0
+            toggle_soundtest(ENABLE_SOUNDTEST)
+            self.logger.info("Sound Test Enabled")
+        else:
+            toggle_soundtest(DISABLE_SOUNDTEST)
+            self.vm.vban.outstream[0].route = 5
+            self.vm.vban.outstream[1].route = 6
+            self.vm.strip[4].apply({"B3": True, "A1": False, "mute": True})
+            self.logger.info("Sound Test Disabled")
+
+    def solo_onyx(self):
+        """placeholder method."""
+
+    def solo_iris(self):
+        """placeholder method."""
+
+    def toggle_workstation_to_onyx(self):
+        self.state.ws_to_onyx = not self.state.ws_to_onyx
+        if self.state.ws_to_onyx:
+            self.vm.strip[5].gain = -6
+            self.vm.vban.outstream[2].on = True
+        else:
+            self.vm.strip[5].gain = 0
+            self.vm.vban.outstream[2].on = False

duckypad_twitch/configuration.py

@@ -0,0 +1,20 @@
+from pathlib import Path
+
+try:
+    import tomllib
+except ModuleNotFoundError:
+    import tomli as tomllib  # type: ignore
+
+configuration = {}
+
+configpath = Path.cwd() / "configs" / "duckypad.toml"
+if not configpath.exists():
+    raise OSError(f"unable to locate {configpath}")
+
+with open(configpath, "rb") as f:
+    configuration = tomllib.load(f)
+
+
+def get(name):
+    if name in configuration:
+        return configuration[name]

duckypad_twitch/duckypad.py

@@ -0,0 +1,55 @@
+import logging
+
+from .audio import Audio
+from .obsws import OBSWS
+from .scene import Scene
+from .states import StreamState
+from .streamlabs import StreamlabsController
+
+logger = logging.getLogger(__name__)
+
+
+class DuckyPad:
+    """base DuckyPad class"""
+
+    def __init__(self, **kwargs):
+        self.logger = logger.getChild(__class__.__name__)
+        for attr, val in kwargs.items():
+            setattr(self, attr, val)
+
+        self.stream = StreamState()
+        self.audio = Audio(self, vm=self.vm, mixer=self.mixer)
+        self.scene = Scene(self, vm=self.vm)
+        self.obsws = OBSWS(self)
+        self.streamlabs_controller = StreamlabsController(self, conn=self.sl)
+
+    def reset(self):
+        '''
+        apply streaming config,
+        then apply current scene settings
+        if stream is live enable both mics over vban
+        '''
+        self.vm.apply_config("streaming")
+        self.audio.reset_states()
+        if self.stream.current_scene:
+            self.logger.debug(
+                f"Running function for current scene {self.stream.current_scene}"
+            )
+            fn = getattr(
+                self.scene,
+                "_".join([word.lower() for word in self.stream.current_scene.split()]),
+            )
+            fn()
+        if self.stream.is_live:
+            self.logger.debug("stream is live, enabling both mics over vban")
+            self.vm.vban.outstream[0].on = True
+            self.vm.vban.outstream[1].on = True
+        else:
+            self.logger.debug(
+                "stream is not live. Leaving both vban outstreams disabled"
+            )
+
+
+def connect(*args, **kwargs):
+    DuckyPad_cls = DuckyPad
+    return DuckyPad_cls(*args, **kwargs)

duckypad_twitch/layer.py

@@ -0,0 +1,25 @@
+import abc
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class ILayer(abc.ABC):
+    """Abstract Base Class for Layers"""
+
+    def __init__(self, duckypad):
+        self.logger = logger.getChild(self.__class__.__name__)
+        self._duckypad = duckypad
+
+    @abc.abstractmethod
+    def identifier():
+        """a unique identifier for each class"""
+
+    @property
+    @abc.abstractmethod
+    def state(self):
+        """retrieve/update the states of a class"""
+
+    @abc.abstractmethod
+    def reset_states():
+        """reset states for a class"""

duckypad_twitch/obsws.py

@@ -0,0 +1,97 @@
+import logging
+
+import obsws_python as obsws
+
+from . import configuration
+from .layer import ILayer
+from .states import OBSWSState
+from .util import ensure_obsws
+
+logger = logging.getLogger(__name__)
+
+
+class OBSWS(ILayer):
+    def __init__(self, duckypad):
+        super().__init__(duckypad)
+        self.request = self.event = None
+        self._state = OBSWSState()
+
+    @property
+    def identifier(self):
+        return type(self).__name__
+
+    @property
+    def state(self):
+        return self._state
+
+    @state.setter
+    def state(self, val):
+        self._state = val
+
+    def reset_states(self):
+        resp = self.request.get_input_mute("Mic/Aux")
+        self.state.mute_mic = resp.input_muted
+        resp = self.request.get_stream_status()
+        self._duckypad.stream.is_live = resp.output_active
+
+    def obs_connect(self):
+        try:
+            conn = configuration.get("obsws")
+            assert conn is not None, "expected configuration for obs"
+            self.request = obsws.ReqClient(**conn)
+            self.reset_states()
+            self.event = obsws.EventClient(**conn)
+            self.event.callback.register(
+                [
+                    self.on_stream_state_changed,
+                    self.on_input_mute_state_changed,
+                    self.on_current_program_scene_changed,
+                ]
+            )
+        except (ConnectionRefusedError, TimeoutError) as e:
+            self.logger.error(f"{type(e).__name__}: {e}")
+            raise
+
+    def on_current_program_scene_changed(self, data):
+        self._duckypad.stream.current_scene = data.scene_name
+        self.logger.info(f"scene switched to {self._duckypad.stream.current_scene}")
+        if self._duckypad.stream.current_scene in ("START", "BRB", "END"):
+            self.mute_mic_state(True)
+
+    def on_input_mute_state_changed(self, data):
+        if data.input_name == "Mic/Aux":
+            self.state.mute_mic = data.input_muted
+        self.logger.info(f"mic was {'muted' if self.state.mute_mic else 'unmuted'}")
+
+    def on_stream_state_changed(self, data):
+        self._duckypad.stream.is_live = data.output_active
+        self.logger.info(
+            f"stream is {'live' if self._duckypad.stream.is_live else 'offline'}"
+        )
+
+    @ensure_obsws
+    def call(self, fn_name, *args):
+        fn = getattr(self.request, fn_name)
+        resp = fn(*args)
+        return resp
+
+    def start(self):
+        self.call("set_current_program_scene", "START")
+
+    def brb(self):
+        self.call("set_current_program_scene", "BRB")
+
+    def end(self):
+        self.call("set_current_program_scene", "END")
+
+    def live(self):
+        self.call("set_current_program_scene", "LIVE")
+
+    def mute_mic_state(self, val):
+        self.call("set_input_mute", "Mic/Aux", val)
+
+    def toggle_mute_mic(self):
+        self.call("toggle_input_mute", "Mic/Aux")
+
+    def toggle_stream(self):
+        self.call("toggle_stream")

duckypad_twitch/scene.py

@@ -0,0 +1,80 @@
+import logging
+
+from .layer import ILayer
+from .states import SceneState
+
+logger = logging.getLogger(__name__)
+
+
+class Scene(ILayer):
+    """Scene concrete class"""
+
+    def __init__(self, duckypad, **kwargs):
+        super().__init__(duckypad)
+        for attr, val in kwargs.items():
+            setattr(self, attr, val)
+
+        self.reset_states()
+
+    @property
+    def identifier(self):
+        return type(self).__name__
+
+    @property
+    def state(self):
+        return self._state
+
+    @state.setter
+    def state(self, val):
+        self._state = val
+
+    def reset_states(self):
+        self._state = SceneState()
+
+    def onyx_only(self):
+        if self._duckypad.streamlabs_controller.switch_scene("onyx_only"):
+            self.vm.strip[2].mute = False
+            self.vm.strip[3].mute = True
+            self.logger.info("Only Onyx Scene enabled, Iris game pc muted")
+
+    def iris_only(self):
+        if self._duckypad.streamlabs_controller.switch_scene("iris_only"):
+            self.vm.strip[2].mute = True
+            self.vm.strip[3].mute = False
+            self.logger.info("Only Iris Scene enabled, Onyx game pc muted")
+
+    def dual_scene(self):
+        if self._duckypad.streamlabs_controller.switch_scene("dual_scene"):
+            self.vm.strip[2].apply({"mute": False, "gain": 0})
+            self.vm.strip[3].apply({"A5": True, "mute": False, "gain": 0})
+            self.logger.info("Dual Scene enabled")
+
+    def onyx_big(self):
+        if self._duckypad.streamlabs_controller.switch_scene("onyx_big"):
+            self.vm.strip[2].apply({"mute": False, "gain": 0})
+            self.vm.strip[3].apply({"mute": False, "gain": -3})
+            self.logger.info("Onyx Big scene enabled")
+
+    def iris_big(self):
+        if self._duckypad.streamlabs_controller.switch_scene("iris_big"):
+            self.vm.strip[2].apply({"mute": False, "gain": -3})
+            self.vm.strip[3].apply({"mute": False, "gain": 0})
+            self.logger.info("Iris Big enabled")
+
+    def start(self):
+        if self._duckypad.streamlabs_controller.switch_scene("start"):
+            self.vm.strip[2].mute = True
+            self.vm.strip[3].mute = True
+            self.logger.info("Start scene enabled.. ready to go live!")
+
+    def brb(self):
+        if self._duckypad.streamlabs_controller.switch_scene("brb"):
+            self.vm.strip[2].mute = True
+            self.vm.strip[3].mute = True
+            self.logger.info("BRB: game pcs muted")
+
+    def end(self):
+        if self._duckypad.streamlabs_controller.switch_scene("end"):
+            self.vm.strip[2].mute = True
+            self.vm.strip[3].mute = True
+            self.logger.info("End scene enabled.")

duckypad_twitch/states.py

@@ -0,0 +1,36 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class StreamState:
+    is_live: bool = False
+    current_scene: str = ""
+
+
+@dataclass
+class AudioState:
+    mute_mics: bool = True
+    only_discord: bool = False
+    only_stream: bool = True
+    sound_test: bool = True
+    solo_onyx: bool = True
+    solo_iris: bool = True
+
+    ws_to_onyx: bool = False
+
+
+@dataclass
+class SceneState:
+    onyx_only: bool = False
+    iris_only: bool = False
+    dual_scene: bool = False
+    onyx_big: bool = False
+    iris_big: bool = False
+    start: bool = False
+    brb: bool = False
+    end: bool = False
+
+
+@dataclass
+class OBSWSState:
+    mute_mic: bool = True

duckypad_twitch/streamlabs.py

@@ -0,0 +1,127 @@
+import logging
+import subprocess as sp
+import time
+import winreg
+from asyncio.subprocess import DEVNULL
+from pathlib import Path
+
+import slobs_websocket
+
+from . import configuration
+from .util import ensure_sl
+
+logger = logging.getLogger(__name__)
+
+
+class StreamlabsController:
+    SL_FULLPATH = ""
+
+    def __init__(self, duckypad, **kwargs):
+        self.logger = logger.getChild(__class__.__name__)
+        self._duckypad = duckypad
+        for attr, val in kwargs.items():
+            setattr(self, attr, val)
+
+        self.proc = None
+
+    ####################################################################################
+    #   CONNECT/DISCONNECT from the API
+    ####################################################################################
+
+    def connect(self):
+        try:
+            conn = configuration.get("streamlabs")
+            assert conn is not None, "expected configuration for streamlabs"
+            self.conn.connect(**conn)
+        except slobs_websocket.exceptions.ConnectionFailure as e:
+            self.logger.error(f"{type(e).__name__}: {e}")
+            raise
+
+        self._duckypad.scene.scenes = {
+            scene.name: scene.id for scene in self.conn.ScenesService.getScenes()
+        }
+        self.logger.debug(f"registered scenes: {self._duckypad.scene.scenes}")
+        self.conn.ScenesService.sceneSwitched += self.on_scene_switched
+        self.conn.StreamingService.streamingStatusChange += (
+            self.on_streaming_status_change
+        )
+
+    def disconnect(self):
+        self.conn.disconnect()
+
+    ####################################################################################
+    #   EVENTS
+    ####################################################################################
+
+    def on_streaming_status_change(self, data):
+        self.logger.debug(f"streaming status changed, now: {data}")
+        if data in ("live", "starting"):
+            self._duckypad.stream.is_live = True
+        else:
+            self._duckypad.stream.is_live = False
+
+    def on_scene_switched(self, data):
+        self._duckypad.stream.current_scene = data.name
+        self.logger.debug(
+            f"stream.current_scene updated to {self._duckypad.stream.current_scene}"
+        )
+
+    ####################################################################################
+    #   START/STOP the stream
+    ####################################################################################
+
+    @ensure_sl
+    def begin_stream(self):
+        if self._duckypad.stream.is_live:
+            self.logger.info("Stream is already online")
+            return
+        self.conn.StreamingService.toggleStreaming()
+
+    @ensure_sl
+    def end_stream(self):
+        if not self._duckypad.stream.is_live:
+            self.logger.info("Stream is already offline")
+            return
+        self.conn.StreamingService.toggleStreaming()
+
+    ####################################################################################
+    #   CONTROL the stream
+    ####################################################################################
+
+    @ensure_sl
+    def switch_scene(self, name):
+        return self.conn.ScenesService.makeSceneActive(
+            self._duckypad.scene.scenes[name.upper()]
+        )
+
+    ####################################################################################
+    #   LAUNCH/SHUTDOWN the streamlabs process
+    ####################################################################################
+
+    def launch(self, delay=5):
+        def get_slpath():
+            SL_KEY = "029c4619-0385-5543-9426-46f9987161d9"
+
+            with winreg.OpenKey(
+                winreg.HKEY_LOCAL_MACHINE, r"{}".format("SOFTWARE" + "\\" + SL_KEY)
+            ) as regpath:
+                return winreg.QueryValueEx(regpath, r"InstallLocation")[0]
+
+        try:
+            if not self.SL_FULLPATH:  # so we only read from registry once.
+                self.SL_FULLPATH = Path(get_slpath()) / "Streamlabs OBS.exe"
+        except FileNotFoundError as e:
+            self.logger.exception(f"{type(e).__name__}: {e}")
+            raise
+
+        if self.proc is None:
+            self.proc = sp.Popen(self.SL_FULLPATH, shell=False, stdout=DEVNULL)
+            time.sleep(delay)
+            self.connect()
+
+    def shutdown(self):
+        self.disconnect()
+        time.sleep(1)
+        if self.proc is not None:
+            self.proc.terminate()
+            self.proc = None

duckypad_twitch/util.py

@@ -0,0 +1,33 @@
+import slobs_websocket
+
+
+def ensure_sl(func):
+    """ensure a streamlabs websocket connection has been established"""
+
+    def wrapper(self, *args):
+        if self._duckypad.streamlabs_controller.conn.ws is None:
+            try:
+                try:
+                    self.connect()
+                except AttributeError:
+                    self._duckypad.streamlabs_controller.connect()
+            except slobs_websocket.exceptions.ConnectionFailure:
+                self._duckypad.streamlabs_controller.conn.ws = None
+                return
+        return func(self, *args)
+
+    return wrapper
+
+
+def ensure_obsws(func):
+    """ensure an obs websocket connection has been established"""
+
+    def wrapper(self, *args):
+        if self.request is None:
+            try:
+                self.obs_connect()
+            except (ConnectionRefusedError, TimeoutError):
+                return
+        return func(self, *args)
+
+    return wrapper