steps.

when toggling ws_to_onyx toggle receiver instream

__main__.py

@@ -3,7 +3,6 @@ import logging
 import keyboard
 import voicemeeterlib
 import xair_api
-from slobs_websocket import StreamlabsOBS
 
 import duckypad_twitch
 from duckypad_twitch import configuration
@@ -39,7 +38,25 @@ 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)
 
-    [step() for step in (audio_hotkeys, scene_hotkeys, obsws_hotkeys)]
+    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]
 
 
 def main():
@@ -47,26 +64,13 @@ def main():
 
     with voicemeeterlib.api("potato") as vm:
         with xair_api.connect("MR18", **xair_config) as mixer:
-            sl = StreamlabsOBS()
+            with duckypad_twitch.connect(vm=vm, mixer=mixer) as duckypad:
+                vm.apply_config("streaming")
 
-            duckypad = duckypad_twitch.connect(vm=vm, mixer=mixer, sl=sl)
+                register_hotkeys(duckypad)
 
-            vm.apply_config("streaming")
+                print("press ctrl+m to quit")
-
+                keyboard.wait("ctrl+m")
-            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.0"
+__version__ = "1.0.1"

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(script):
+        def toggle_soundtest(params):
             onyx_conn = configuration.get("vban_onyx")
             iris_conn = configuration.get("vban_iris")
             assert all(
@@ -93,12 +93,24 @@ class Audio(ILayer):
             ), "expected configurations for onyx_conn, iris_conn"
 
             with vban_cmd.api("potato", **onyx_conn) as vban:
-                vban.sendtext(script)
+                vban.strip[0].apply(params)
             with vban_cmd.api("potato", **iris_conn) as vban:
-                vban.sendtext(script)
+                vban.strip[0].apply(params)
 
-        ENABLE_SOUNDTEST = "Strip(0).A1=1; Strip(0).A2=1; Strip(0).B1=0; Strip(0).B2=0; Strip(0).mono=1;"
+        ENABLE_SOUNDTEST = {
-        DISABLE_SOUNDTEST = "Strip(0).A1=0; Strip(0).A2=0; Strip(0).B1=1; Strip(0).B2=1; Strip(0).mono=0;"
+            "A1": True,
+            "A2": True,
+            "B1": False,
+            "B2": False,
+            "mono": True,
+        }
+        DISABLE_SOUNDTEST = {
+            "A1": False,
+            "A2": False,
+            "B1": True,
+            "B2": True,
+            "mono": False,
+        }
 
         self.state.sound_test = not self.state.sound_test
         if self.state.sound_test:
@@ -124,9 +136,14 @@ 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,7 +21,13 @@ 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, conn=self.sl)
+        self.streamlabs_controller = StreamlabsController(self)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_value, exc_type, traceback):
+        self.streamlabs_controller.conn.disconnect()
 
     def reset(self):
         '''

duckypad_twitch/obsws.py

@@ -46,6 +46,7 @@ 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:
@@ -69,6 +70,9 @@ 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,9 +3,11 @@ 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
@@ -14,14 +16,13 @@ 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
 
     ####################################################################################
@@ -98,24 +99,24 @@ class StreamlabsController:
     #   LAUNCH/SHUTDOWN the streamlabs process
     ####################################################################################
 
-    def launch(self, delay=5):
+    @cached_property
-        def get_slpath():
+    def sl_fullpath(self) -> Path:
+        try:
+            self.logger.debug("fetching sl_fullpath from the registry")
             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]
+                slpath = winreg.QueryValueEx(regpath, r"InstallLocation")[0]
-
+            return Path(slpath) / "Streamlabs OBS.exe"
-        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(self.SL_FULLPATH, shell=False, stdout=DEVNULL)
+            self.proc = sp.Popen(str(self.sl_fullpath), shell=False, stdout=DEVNULL)
             time.sleep(delay)
             self.connect()