upd workstation routing

added methods to Audio class for:
stage gaining microphones
toggling audio to/from ws
toggling audio to/from tv

README.md

@@ -22,13 +22,12 @@ Packages used in this codebase:
 - [`vban-cmd`][vban-cmd]
 - [`xair-api`][xair-api]
 - [`obsws-python`][obsws-python]
-- [`slobs-websocket`][slobs-websocket]
 
 ## Need for a custom driver
 
-We use a triple pc streaming setup, one gaming pc for each of us and a third pc that handles the stream. 
+We use a triple pc streaming setup, one gaming pc for each of us and a third pc that handles the stream.
 
-- Both of our microphones, as well as both gaming pc are wired into an [MR18 mixer][mr18] which itself is connected to the streaming pc. 
+- Both of our microphones, as well as both gaming pc are wired into an [MR18 mixer][mr18] which itself is connected to the streaming pc.
 - Then we vban our microphones from the workstation off to each of our pcs in order to talk in-game. All audio is routed through [Voicemeeter][voicemeeter].
 - Voicemeeter is connected to Studio ONE daw for background noise removal. Any voice communication software (such as Discord) is therefore installed onto the workstation, separate of our gaming pcs.
 
@@ -43,7 +42,7 @@ This package is for demonstration purposes only. Several of the interfaces on wh
 - Most of the audio routing for the dual stream is handled in the `Audio class` in audio.py with the aid of Voicemeeter's Remote API.
   - Some communication with the Xair mixer and the vban protocol can also be found in this class.
 - Scene switching and some audio routing are handled in the `Scene class` in scene.py.
-  - A `StreamlabsController` class is used to communicate with the Streamlabs API.
+  - A `OBSWS` class is used to communicate with OBS websocket.
 - Dataclasses are used to hold internal states and states are updated using event callbacks.
 - Decorators are used to confirm websocket connections.
 - A separate OBSWS class is used to handle scenes and mic muting (for a single pc stream).
@@ -59,6 +58,5 @@ This package is for demonstration purposes only. Several of the interfaces on wh
 [vban-cmd]: https://github.com/onyx-and-iris/vban-cmd-python
 [xair-api]: https://github.com/onyx-and-iris/xair-api-python
 [obsws-python]: https://github.com/aatikturk/obsws-python
-[slobs-websocket]: https://github.com/onyx-and-iris/slobs_websocket
 [voicemeeter]: https://voicemeeter.com/
 [mr18]: https://www.midasconsoles.com/product.html?modelCode=P0C8H

configs/potato/streaming.toml

@@ -7,7 +7,7 @@ A4 = false
 A5 = false
 B1 = true
 B2 = false
-B3 = false
+B3 = true
 mono = false
 solo = false
 mute = true
@@ -25,7 +25,7 @@ A4 = false
 A5 = false
 B1 = false
 B2 = true
-B3 = false
+B3 = true
 mono = false
 solo = false
 mute = true
@@ -71,7 +71,7 @@ comp.knob = 0
 gate.knob = 0
 
 [strip-4]
-label = "Mics to Stream"
+label = ""
 A1 = false
 A2 = false
 A3 = false
@@ -79,10 +79,10 @@ A4 = false
 A5 = false
 B1 = false
 B2 = false
-B3 = true
+B3 = false
 mono = false
 solo = false
-mute = true
+mute = false
 gain = 0.0
 limit = 0
 comp.knob = 0
@@ -265,4 +265,4 @@ on = false
 on = false
 
 [vban-out-8]
-on = false
+on = false

duckypad_twitch/audio.py

@@ -1,19 +1,17 @@
 import logging
-from enum import IntEnum
-
 import time
+
 import vban_cmd
 
 from . import configuration
+from .enums import Buttons, Strips
 from .layer import ILayer
 from .states import AudioState
+from .util import ensure_mixer_fadeout
 
 logger = logging.getLogger(__name__)
 
 
-Buttons = IntEnum('Buttons', 'mute_mics only_discord only_stream', start=0)
-
-
 class Audio(ILayer):
     """Audio concrete class"""
 
@@ -93,8 +91,10 @@ class Audio(ILayer):
 
             with vban_cmd.api('potato', outbound=True, **onyx_conn) as vban:
                 vban.strip[0].apply(params)
+                vban.vban.instream[0].on = True
             with vban_cmd.api('potato', outbound=True, **iris_conn) as vban:
                 vban.strip[0].apply(params)
+                vban.vban.instream[0].on = True
 
         ENABLE_SOUNDTEST = {
             'A1': True,
@@ -113,7 +113,8 @@ class Audio(ILayer):
 
         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.strip[Strips.onyx_mic].apply({'A1': True, 'B1': False, 'B3': False, 'mute': False})
+            self.vm.strip[Strips.iris_mic].apply({'A1': True, 'B2': False, 'B3': False, 'mute': False})
             self.vm.vban.outstream[0].on = True
             self.vm.vban.outstream[1].on = True
             self.vm.vban.outstream[0].route = 0
@@ -124,39 +125,121 @@ class Audio(ILayer):
             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.vm.strip[Strips.onyx_mic].apply({'A1': False, 'B1': True, 'B3': True, 'mute': True})
+            self.vm.strip[Strips.iris_mic].apply({'A1': False, 'B2': True, 'B3': True, 'mute': True})
             self.logger.info('Sound Test Disabled')
 
+    @ensure_mixer_fadeout
+    def stage_onyx_mic(self):
+        """Gain stage SE Electronics DCM8 with phantom power"""
+        self.mixer.headamp[10].phantom = True
+        for i in range(21):
+            self.mixer.headamp[10].gain = i
+            time.sleep(0.1)
+        self.logger.info('Onyx Mic Staged with Phantom Power')
+
+    @ensure_mixer_fadeout
+    def stage_iris_mic(self):
+        """Gain stage TLM102 with phantom power"""
+        self.mixer.headamp[11].phantom = True
+        for i in range(31):
+            self.mixer.headamp[11].gain = i
+            time.sleep(0.1)
+        self.logger.info('Iris Mic Staged with Phantom Power')
+
+    def unstage_onyx_mic(self):
+        """Unstage SE Electronics DCM8 and disable phantom power"""
+        for i in reversed(range(21)):
+            self.mixer.headamp[10].gain = i
+            time.sleep(0.1)
+        self.mixer.headamp[10].phantom = False
+        self.logger.info('Onyx Mic Unstaged and Phantom Power Disabled')
+
+    def unstage_iris_mic(self):
+        """Unstage TLM102 and disable phantom power"""
+        for i in reversed(range(31)):
+            self.mixer.headamp[11].gain = i
+            time.sleep(0.1)
+        self.mixer.headamp[11].phantom = False
+        self.logger.info('Iris Mic Unstaged and Phantom Power Disabled')
+
     def solo_onyx(self):
-        """placeholder method."""
+        """placeholder method"""
 
     def solo_iris(self):
-        """placeholder method."""
+        """placeholder method"""
+
+    def _fade_mixer(self, target_fader, fade_in=True):
+        """Fade the mixer's fader to the target level."""
+        current_fader = self.mixer.lr.mix.fader
+        step = 1 if fade_in else -1
+        while (fade_in and current_fader < target_fader) or (not fade_in and current_fader > target_fader):
+            current_fader += step
+            self.mixer.lr.mix.fader = current_fader
+            time.sleep(0.05)
+
+    def _toggle_workstation_routing(self, state_attr, target_name, vban_config_key):
+        """Toggle routing of workstation audio to either Onyx or Iris via VBAN."""
+
+        current_state = getattr(self.state, state_attr)
+        new_state = not current_state
+        setattr(self.state, state_attr, new_state)
+
+        target_conn = configuration.get(vban_config_key)
+
+        if new_state:
+            with vban_cmd.api('potato', outbound=True, **target_conn) as vban:
+                vban.vban.instream[2].on = True
+            self.vm.strip[5].gain = -6
+            self.vm.vban.outstream[3].on = True
+            self._fade_mixer(-90, fade_in=False)
+            self.logger.info(f'Workstation audio routed to {target_name}')
+        else:
+            with vban_cmd.api('potato', outbound=True, **target_conn) as vban:
+                vban.vban.instream[2].on = False
+            self.vm.strip[5].gain = 0
+            self.vm.vban.outstream[3].on = False
+            self._fade_mixer(-36, fade_in=True)
+            self.logger.info('Workstation audio routed back to monitor speakers')
 
     def toggle_workstation_to_onyx(self):
-        self.state.ws_to_onyx = not self.state.ws_to_onyx
+        self._toggle_workstation_routing('ws_to_onyx', 'Onyx', 'vban_onyx')
-        onyx_conn = configuration.get('vban_onyx')
-        if self.state.ws_to_onyx:
-            with vban_cmd.api('potato', outbound=True, **onyx_conn) as vban:
-                vban.vban.instream[0].on = True
-            self.vm.strip[5].gain = -6
-            self.vm.vban.outstream[2].on = True
 
-            # Fade out the workstation
+    def toggle_workstation_to_iris(self):
-            current_fader = self.mixer.lr.mix.fader
+        self._toggle_workstation_routing('ws_to_iris', 'Iris', 'vban_iris')
-            while current_fader > -90:
+
-                current_fader -= 1
+    def _toggle_tv_routing(self, state_attr, target_name, vban_config_key):
-                self.mixer.lr.mix.fader = current_fader
+        """Toggle routing of TV audio to either Onyx or Iris via VBAN."""
-                time.sleep(0.05)
+        current_state = getattr(self.state, state_attr)
+        new_state = not current_state
+        setattr(self.state, state_attr, new_state)
+
+        target_conn = configuration.get(vban_config_key)
+        tv_conn = configuration.get('vban_tv')
+
+        if new_state:
+            with (
+                vban_cmd.api('banana', outbound=True, **tv_conn) as vban_tv,
+                vban_cmd.api('potato', outbound=True, **target_conn) as vban_target,
+            ):
+                vban_tv.strip[3].A1 = False
+                vban_tv.strip[3].gain = -6
+                vban_tv.vban.outstream[0].on = True
+                vban_target.vban.instream[2].on = True
+            self.logger.info(f'TV audio routed to {target_name}')
         else:
-            with vban_cmd.api('potato', outbound=True, **onyx_conn) as vban:
+            with (
-                vban.vban.instream[0].on = False
+                vban_cmd.api('banana', outbound=True, **tv_conn) as vban_tv,
-            self.vm.strip[5].gain = 0
+                vban_cmd.api('potato', outbound=True, **target_conn) as vban_target,
-            self.vm.vban.outstream[2].on = False
+            ):
+                vban_tv.strip[3].A1 = True
+                vban_tv.strip[3].gain = 0
+                vban_tv.vban.outstream[0].on = False
+                vban_target.vban.instream[2].on = False
+            self.logger.info(f'TV audio routing to {target_name} disabled')
 
-            # Fade in the workstation
+    def toggle_tv_audio_to_onyx(self):
-            current_fader = self.mixer.lr.mix.fader
+        self._toggle_tv_routing('tv_to_onyx', 'Onyx', 'vban_onyx')
-            while current_fader < -36:
+
-                current_fader += 1
+    def toggle_tv_audio_to_iris(self):
-                self.mixer.lr.mix.fader = current_fader
+        self._toggle_tv_routing('tv_to_iris', 'Iris', 'vban_iris')
-                time.sleep(0.05)

duckypad_twitch/duckypad.py

@@ -4,7 +4,7 @@ from .audio import Audio
 from .obsws import OBSWS
 from .scene import Scene
 from .states import StreamState
-from .streamlabs import StreamlabsController
+from .util import to_snakecase
 
 logger = logging.getLogger(__name__)
 
@@ -18,16 +18,14 @@ class DuckyPad:
             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 = StreamlabsController(self)
+        self.audio = Audio(self, vm=self.vm, mixer=self.mixer)
+        self.scene = Scene(self, vm=self.vm, obsws=self.obsws)
 
     def __enter__(self):
         return self
 
-    def __exit__(self, exc_value, exc_type, traceback):
+    def __exit__(self, exc_value, exc_type, exc_tb):
-        self.streamlabs.disconnect()
         self.obsws.disconnect()
 
     def reset(self):
@@ -40,11 +38,14 @@ class DuckyPad:
         self.audio.reset_states()
         if self.stream.current_scene:
             self.logger.debug(f'Running function for current scene {self.stream.current_scene}')
-            fn = getattr(
+            try:
-                self.scene,
+                fn = getattr(
-                '_'.join([word.lower() for word in self.stream.current_scene.split()]),
+                    self.scene,
-            )
+                    to_snakecase(self.stream.current_scene),
-            fn()
+                )
+                fn()
+            except AttributeError:
+                self.logger.warning(f'No function found for scene {self.stream.current_scene}')
         if self.stream.is_live:
             self.logger.debug('stream is live, enabling both mics over vban')
             self.vm.vban.outstream[0].on = True

duckypad_twitch/enums.py

@@ -0,0 +1,6 @@
+from enum import IntEnum
+
+Buttons = IntEnum('Buttons', 'mute_mics only_discord only_stream', start=0)
+
+Strips = IntEnum('Strips', 'onyx_mic iris_mic onyx_pc iris_pc', start=0)
+Buses = IntEnum('Buses', 'MR18 ASIO[1,2] ASIO[3,4] ASIO[5,6] ASIO[7,8] onyx_mic iris_mic both_mics', start=5)

duckypad_twitch/macros/duckypad.py

@@ -16,33 +16,28 @@ def register_hotkeys(duckypad):
         keyboard.add_hotkey('F14', duckypad.audio.only_discord)
         keyboard.add_hotkey('F15', duckypad.audio.only_stream)
         keyboard.add_hotkey('F16', duckypad.audio.sound_test)
-        keyboard.add_hotkey('F17', duckypad.audio.solo_onyx)
+        keyboard.add_hotkey('F17', duckypad.audio.stage_onyx_mic)
-        keyboard.add_hotkey('F18', duckypad.audio.solo_iris)
+        keyboard.add_hotkey('F18', duckypad.audio.stage_iris_mic)
-        keyboard.add_hotkey('F19', duckypad.audio.toggle_workstation_to_onyx)
+        keyboard.add_hotkey('shift+F17', duckypad.audio.unstage_onyx_mic)
+        keyboard.add_hotkey('shift+F18', duckypad.audio.unstage_iris_mic)
+        keyboard.add_hotkey('F19', duckypad.audio.solo_onyx)
+        keyboard.add_hotkey('F20', duckypad.audio.solo_iris)
+        keyboard.add_hotkey('F21', duckypad.audio.toggle_workstation_to_onyx)
+        keyboard.add_hotkey('F22', duckypad.audio.toggle_workstation_to_iris)
+        keyboard.add_hotkey('F23', duckypad.audio.toggle_tv_audio_to_onyx)
+        keyboard.add_hotkey('F24', duckypad.audio.toggle_tv_audio_to_iris)
 
     def scene_hotkeys():
-        keyboard.add_hotkey('ctrl+F13', duckypad.scene.onyx_only)
+        keyboard.add_hotkey('ctrl+F13', duckypad.scene.start)
-        keyboard.add_hotkey('ctrl+F14', duckypad.scene.iris_only)
+        keyboard.add_hotkey('ctrl+F14', duckypad.scene.dual_stream)
-        keyboard.add_hotkey('ctrl+F15', duckypad.scene.dual_scene)
+        keyboard.add_hotkey('ctrl+F15', duckypad.scene.brb)
-        keyboard.add_hotkey('ctrl+F16', duckypad.scene.onyx_big)
+        keyboard.add_hotkey('ctrl+F16', duckypad.scene.end)
-        keyboard.add_hotkey('ctrl+F17', duckypad.scene.iris_big)
+        keyboard.add_hotkey('ctrl+F17', duckypad.scene.onyx_solo)
-        keyboard.add_hotkey('ctrl+F18', duckypad.scene.start)
+        keyboard.add_hotkey('ctrl+F18', duckypad.scene.iris_solo)
-        keyboard.add_hotkey('ctrl+F19', duckypad.scene.brb)
-        keyboard.add_hotkey('ctrl+F20', duckypad.scene.end)
 
     def obsws_hotkeys():
-        keyboard.add_hotkey('ctrl+alt+F13', duckypad.obsws.start)
+        keyboard.add_hotkey('ctrl+alt+F13', duckypad.obsws.start_stream)
-        keyboard.add_hotkey('ctrl+alt+F14', duckypad.obsws.brb)
+        keyboard.add_hotkey('ctrl+alt+F14', duckypad.obsws.stop_stream)
-        keyboard.add_hotkey('ctrl+alt+F15', duckypad.obsws.end)
-        keyboard.add_hotkey('ctrl+alt+F16', duckypad.obsws.live)
-        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.begin_stream)
-        keyboard.add_hotkey('ctrl+F23', duckypad.streamlabs.end_stream)
-        keyboard.add_hotkey('ctrl+alt+F23', duckypad.streamlabs.launch, args=(10,))
-        keyboard.add_hotkey('ctrl+alt+F24', duckypad.streamlabs.shutdown)
 
     def duckypad_hotkeys():
         keyboard.add_hotkey('ctrl+F21', duckypad.reset)
@@ -51,7 +46,6 @@ def register_hotkeys(duckypad):
         audio_hotkeys,
         scene_hotkeys,
         obsws_hotkeys,
-        streamlabs_controller_hotkeys,
         duckypad_hotkeys,
     ):
         step()
@@ -69,5 +63,5 @@ def run():
 
         register_hotkeys(duckypad)
 
-        print('press ctrl+m to quit')
+        print('press ctrl+shift+F24 to quit')
-        keyboard.wait('ctrl+m')
+        keyboard.wait('ctrl+shift+F24')

duckypad_twitch/obsws.py

@@ -4,7 +4,6 @@ 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__)
@@ -14,12 +13,13 @@ 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__
 
+    ### State Management ###
+
     @property
     def state(self):
         return self._state
@@ -29,8 +29,6 @@ class OBSWS(ILayer):
         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
 
@@ -44,7 +42,6 @@ class OBSWS(ILayer):
             self.event.callback.register(
                 [
                     self.on_stream_state_changed,
-                    self.on_input_mute_state_changed,
                     self.on_current_program_scene_changed,
                     self.on_exit_started,
                 ]
@@ -58,47 +55,54 @@ class OBSWS(ILayer):
             if client:
                 client.disconnect()
 
-    def on_current_program_scene_changed(self, data):
+    ### Event Handlers ###
-        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"}')
 
+    def on_current_program_scene_changed(self, data):
+        self._duckypad.stream.current_scene = data.scene_name
+        match data.scene_name:
+            case 'START':
+                self.logger.info('Start scene enabled.. ready to go live!')
+            case 'DUAL STREAM':
+                self.logger.info('Dual Stream Scene enabled')
+            case 'BRB':
+                self.logger.info('BRB: game pcs muted')
+            case 'END':
+                self.logger.info('End Scene enabled.')
+            case 'ONYX SOLO':
+                self.logger.info('Onyx Solo Scene enabled, Iris game pc muted')
+            case 'IRIS SOLO':
+                self.logger.info('Iris Solo Scene enabled, Onyx game pc muted')
+
     def on_exit_started(self, _):
         self.event.unsubscribe()
 
+    ### OBSWS Request Wrappers ###
+
     @ensure_obsws
-    def call(self, fn_name, *args):
+    def _call(self, fn_name, *args):
         fn = getattr(self.request, fn_name)
         resp = fn(*args)
         return resp
 
-    def start(self):
+    def switch_to_scene(self, scene_name):
-        self.call('set_current_program_scene', 'START')
+        self._call('set_current_program_scene', scene_name)
 
-    def brb(self):
+    def start_stream(self):
-        self.call('set_current_program_scene', 'BRB')
+        resp = self._call('get_stream_status')
+        if resp.output_active:
+            self.logger.info("stream is already live, can't start stream")
+            return
 
-    def end(self):
+        self._call('start_stream')
-        self.call('set_current_program_scene', 'END')
 
-    def live(self):
+    def stop_stream(self):
-        self.call('set_current_program_scene', 'LIVE')
+        resp = self._call('get_stream_status')
+        if not resp.output_active:
+            self.logger.info("stream is not live, can't stop stream")
+            return
 
-    def mute_mic_state(self, val):
+        self._call('stop_stream')
-        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

@@ -1,5 +1,6 @@
 import logging
 
+from .enums import Strips
 from .layer import ILayer
 from .states import SceneState
 
@@ -31,50 +32,37 @@ class Scene(ILayer):
     def reset_states(self):
         self._state = SceneState()
 
-    def onyx_only(self):
-        if self._duckypad.streamlabs.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.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.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.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.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.switch_scene('start'):
+        self.vm.strip[Strips.onyx_pc].mute = True
-            self.vm.strip[2].mute = True
+        self.vm.strip[Strips.iris_pc].mute = True
-            self.vm.strip[3].mute = True
+        self.obsws.switch_to_scene('START')
-            self.logger.info('Start scene enabled.. ready to go live!')
+
+    def dual_stream(self):
+        ENABLE_PC = {
+            'mute': False,
+            'A5': True,
+            'gain': 0,
+        }
+        self.vm.strip[Strips.onyx_pc].apply(ENABLE_PC)
+        self.vm.strip[Strips.iris_pc].apply(ENABLE_PC)
+        self.obsws.switch_to_scene('DUAL STREAM')
 
     def brb(self):
-        if self._duckypad.streamlabs.switch_scene('brb'):
+        self.vm.strip[Strips.onyx_pc].mute = True
-            self.vm.strip[2].mute = True
+        self.vm.strip[Strips.iris_pc].mute = True
-            self.vm.strip[3].mute = True
+        self.obsws.switch_to_scene('BRB')
-            self.logger.info('BRB: game pcs muted')
 
     def end(self):
-        if self._duckypad.streamlabs.switch_scene('end'):
+        self.vm.strip[Strips.onyx_pc].mute = True
-            self.vm.strip[2].mute = True
+        self.vm.strip[Strips.iris_pc].mute = True
-            self.vm.strip[3].mute = True
+        self.obsws.switch_to_scene('END')
-            self.logger.info('End scene enabled.')
+
+    def onyx_solo(self):
+        self.vm.strip[Strips.onyx_pc].mute = False
+        self.vm.strip[Strips.iris_pc].mute = True
+        self.obsws.switch_to_scene('ONYX SOLO')
+
+    def iris_solo(self):
+        self.vm.strip[Strips.onyx_pc].mute = True
+        self.vm.strip[Strips.iris_pc].mute = False
+        self.obsws.switch_to_scene('IRIS SOLO')

duckypad_twitch/states.py

@@ -17,6 +17,9 @@ class AudioState:
     solo_iris: bool = False
 
     ws_to_onyx: bool = False
+    ws_to_iris: bool = False
+    tv_to_onyx: bool = False
+    tv_to_iris: bool = False
 
 
 @dataclass
@@ -29,8 +32,3 @@ class SceneState:
     start: bool = False
     brb: bool = False
     end: bool = False
-
-
-@dataclass
-class OBSWSState:
-    mute_mic: bool = True

duckypad_twitch/streamlabs.py

@@ -1,119 +0,0 @@
-import logging
-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
-
-logger = logging.getLogger(__name__)
-
-
-class StreamlabsController:
-    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
-
-    ####################################################################################
-    #   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
-    ####################################################################################
-
-    @cached_property
-    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:
-                slpath = winreg.QueryValueEx(regpath, r'InstallLocation')[0]
-            return Path(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)
-            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
-            self._duckypad.stream.current_scene = ''

duckypad_twitch/util.py

@@ -1,24 +1,3 @@
-import slobs_websocket
-
-
-def ensure_sl(func):
-    """ensure a streamlabs websocket connection has been established"""
-
-    def wrapper(self, *args):
-        if self._duckypad.streamlabs.conn.ws is None:
-            try:
-                try:
-                    self.connect()
-                except AttributeError:
-                    self._duckypad.streamlabs.connect()
-            except slobs_websocket.exceptions.ConnectionFailure:
-                self._duckypad.streamlabs.conn.ws = None
-                return
-        return func(self, *args)
-
-    return wrapper
-
-
 def ensure_obsws(func):
     """ensure an obs websocket connection has been established"""
 
@@ -31,3 +10,17 @@ def ensure_obsws(func):
         return func(self, *args)
 
     return wrapper
+
+def ensure_mixer_fadeout(func):
+    """ensure mixer fadeout is stopped before proceeding"""
+
+    def wrapper(self, *args):
+        if self.mixer.lr.mix.fader > -90:
+            self._fade_mixer(-90, fade_in=False)
+        return func(self, *args)
+
+    return wrapper
+
+def to_snakecase(scene_name: str) -> str:
+    """Convert caplitalized words to lowercase snake_case"""
+    return '_'.join(word.lower() for word in scene_name.split())

pyproject.toml

@@ -24,11 +24,8 @@ dynamic = ["version"]
 dependencies = [
   "keyboard",
   "obsws-python>=1.7",
-  "slobs-websocket @ git+https://git@github.com/onyx-and-iris/slobs_websocket@v0.1.4#egg=slobs_websocket",
-  "tomli>=2.0.1; python_version<'3.11'",
   "vban-cmd>=2.5.2",
   "voicemeeter-api>=2.6.1",
-  "websocket-client",
   "xair-api>=2.4.1",
 ]