__main__.py

@@ -3,6 +3,7 @@ import logging
 import keyboard
 import voicemeeterlib
 import xair_api
+from slobs_websocket import StreamlabsOBS
 
 import duckypad_twitch
 from duckypad_twitch import configuration
@@ -38,25 +39,7 @@ def register_hotkeys(duckypad):
         keyboard.add_hotkey("ctrl+alt+F17", duckypad.obsws.toggle_mute_mic)
         keyboard.add_hotkey("ctrl+alt+F18", duckypad.obsws.toggle_stream)
 
-    def streamlabs_controller_hotkeys():
-        keyboard.add_hotkey("ctrl+F22", duckypad.streamlabs_controller.begin_stream)
-        keyboard.add_hotkey("ctrl+F23", duckypad.streamlabs_controller.end_stream)
-        keyboard.add_hotkey(
-            "ctrl+alt+F23", duckypad.streamlabs_controller.launch, args=(8,)
-        )
-        keyboard.add_hotkey("ctrl+alt+F24", duckypad.streamlabs_controller.shutdown)
-
-    def duckypad_hotkeys():
-        keyboard.add_hotkey("ctrl+F21", duckypad.reset)
-
-    steps = (
-        audio_hotkeys,
-        scene_hotkeys,
-        streamlabs_controller_hotkeys,
-        obsws_hotkeys,
-        duckypad_hotkeys,
-    )
-    [step() for step in steps]
+    [step() for step in (audio_hotkeys, scene_hotkeys, obsws_hotkeys)]
 
 
 def main():
@@ -64,13 +47,26 @@ def main():
 
     with voicemeeterlib.api("potato") as vm:
         with xair_api.connect("MR18", **xair_config) as mixer:
-            with duckypad_twitch.connect(vm=vm, mixer=mixer) as duckypad:
-                vm.apply_config("streaming")
+            sl = StreamlabsOBS()
 
-                register_hotkeys(duckypad)
+            duckypad = duckypad_twitch.connect(vm=vm, mixer=mixer, sl=sl)
 
-                print("press ctrl+m to quit")
-                keyboard.wait("ctrl+m")
+            vm.apply_config("streaming")
+
+            register_hotkeys(duckypad)
+
+            keyboard.add_hotkey("ctrl+F21", duckypad.reset)
+            keyboard.add_hotkey("ctrl+F22", duckypad.streamlabs_controller.begin_stream)
+            keyboard.add_hotkey("ctrl+F23", duckypad.streamlabs_controller.end_stream)
+            keyboard.add_hotkey(
+                "ctrl+alt+F23", duckypad.streamlabs_controller.launch, args=(5,)
+            )
+            keyboard.add_hotkey("ctrl+alt+F24", duckypad.streamlabs_controller.shutdown)
+
+            print("press ctrl+m to quit")
+            keyboard.wait("ctrl+m")
+
+            sl.disconnect()
 
 
 if __name__ == "__main__":

duckypad_twitch/__about__.py

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

duckypad_twitch/audio.py

@@ -85,7 +85,7 @@ class Audio(ILayer):
         self.vm.button[Buttons.only_stream].stateonly = self.state.only_stream
 
     def sound_test(self):
-        def toggle_soundtest(params):
+        def toggle_soundtest(script):
             onyx_conn = configuration.get("vban_onyx")
             iris_conn = configuration.get("vban_iris")
             assert all(
@@ -93,24 +93,12 @@ class Audio(ILayer):
             ), "expected configurations for onyx_conn, iris_conn"
 
             with vban_cmd.api("potato", **onyx_conn) as vban:
-                vban.strip[0].apply(params)
+                vban.sendtext(script)
             with vban_cmd.api("potato", **iris_conn) as vban:
-                vban.strip[0].apply(params)
+                vban.sendtext(script)
 
-        ENABLE_SOUNDTEST = {
-            "A1": True,
-            "A2": True,
-            "B1": False,
-            "B2": False,
-            "mono": True,
-        }
-        DISABLE_SOUNDTEST = {
-            "A1": False,
-            "A2": False,
-            "B1": True,
-            "B2": True,
-            "mono": False,
-        }
+        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:
@@ -136,14 +124,9 @@ class Audio(ILayer):
 
     def toggle_workstation_to_onyx(self):
         self.state.ws_to_onyx = not self.state.ws_to_onyx
-        onyx_conn = configuration.get("vban_onyx")
         if self.state.ws_to_onyx:
-            with vban_cmd.api("potato", **onyx_conn) as vban:
-                vban.vban.instream[0].on = True
             self.vm.strip[5].gain = -6
             self.vm.vban.outstream[2].on = True
         else:
-            with vban_cmd.api("potato", **onyx_conn) as vban:
-                vban.vban.instream[0].on = False
             self.vm.strip[5].gain = 0
             self.vm.vban.outstream[2].on = False

duckypad_twitch/duckypad.py

@@ -21,13 +21,7 @@ class DuckyPad:
         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)
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, exc_value, exc_type, traceback):
-        self.streamlabs_controller.conn.disconnect()
+        self.streamlabs_controller = StreamlabsController(self, conn=self.sl)
 
     def reset(self):
         '''

duckypad_twitch/obsws.py

@@ -46,7 +46,6 @@ class OBSWS(ILayer):
                     self.on_stream_state_changed,
                     self.on_input_mute_state_changed,
                     self.on_current_program_scene_changed,
-                    self.on_exit_started,
                 ]
             )
         except (ConnectionRefusedError, TimeoutError) as e:
@@ -70,9 +69,6 @@ class OBSWS(ILayer):
             f"stream is {'live' if self._duckypad.stream.is_live else 'offline'}"
         )
 
-    def on_exit_started(self, _):
-        self.event.unsubscribe()
-
     @ensure_obsws
     def call(self, fn_name, *args):
         fn = getattr(self.request, fn_name)

duckypad_twitch/streamlabs.py

@@ -3,11 +3,9 @@ import subprocess as sp
 import time
 import winreg
 from asyncio.subprocess import DEVNULL
-from functools import cached_property
 from pathlib import Path
 
 import slobs_websocket
-from slobs_websocket import StreamlabsOBS
 
 from . import configuration
 from .util import ensure_sl
@@ -16,13 +14,14 @@ 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.conn = StreamlabsOBS()
         self.proc = None
 
     ####################################################################################
@@ -99,24 +98,24 @@ class StreamlabsController:
     #   LAUNCH/SHUTDOWN the streamlabs process
     ####################################################################################
 
-    @cached_property
-    def sl_fullpath(self) -> Path:
-        try:
-            self.logger.debug("fetching sl_fullpath from the registry")
+    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:
-                slpath = winreg.QueryValueEx(regpath, r"InstallLocation")[0]
-            return Path(slpath) / "Streamlabs OBS.exe"
+                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
 
-    def launch(self, delay=5):
         if self.proc is None:
-            self.proc = sp.Popen(str(self.sl_fullpath), shell=False, stdout=DEVNULL)
+            self.proc = sp.Popen(self.SL_FULLPATH, shell=False, stdout=DEVNULL)
             time.sleep(delay)
             self.connect()