update fadein/fadeout logic

set mute_mics macro state on sound test off

mute game_pcs bus due to vban'ing mics over

update vban stream indexes for ws,tv broadcasts

README.md

@@ -22,7 +22,6 @@ 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
 
@@ -30,23 +29,22 @@ We use a triple pc streaming setup, one gaming pc for each of us and a third 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.
+- Voicemeeter is connected to Studio ONE daw for live processing. Any voice communication software (such as Discord) is therefore installed onto the workstation, separate of our gaming pcs.
 
 If you've ever attempted to setup a dual pc streaming setup, you may appreciate the challenges of a triple pc setup.
 
 ## Details about the code
 
-This package is for demonstration purposes only. Several of the interfaces on which it depends have been tightly coupled into a duckypad macros program.
+This package is for demonstration purposes only. Several of the interfaces on which it depends have been merged into a duckypad macros program.
 
 - The package entry point can be found at `duckypad_twitch.macros.duckypad`.
 - A base DuckyPad class in duckypad.py is used to connect the various layers of the driver.
 - 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.
+  - 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).
 - Logging is included to help with debugging but also to provide stream information in real time.
 
 ## License
@@ -59,6 +57,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
@@ -39,8 +39,8 @@ label = "Onyx Pc"
 A1 = false
 A2 = false
 A3 = false
-A4 = false
+A4 = true
-A5 = true
+A5 = false
 B1 = false
 B2 = false
 B3 = false
@@ -57,8 +57,8 @@ label = "Iris Pc"
 A1 = false
 A2 = false
 A3 = false
-A4 = false
+A4 = true
-A5 = true
+A5 = false
 B1 = false
 B2 = false
 B3 = false
@@ -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
@@ -90,8 +90,8 @@ gate.knob = 0
 
 [strip-5]
 label = "System"
-A1 = false
+A1 = true
-A2 = true
+A2 = false
 A3 = false
 A4 = false
 A5 = false
@@ -107,8 +107,8 @@ limit = 0
 [strip-6]
 label = "Comms"
 A1 = false
-A2 = false
+A2 = true
-A3 = true
+A3 = false
 A4 = false
 A5 = false
 B1 = false
@@ -125,8 +125,8 @@ k = 0
 label = "Pretzel"
 A1 = false
 A2 = false
-A3 = false
+A3 = true
-A4 = true
+A4 = false
 A5 = false
 B1 = false
 B2 = false
@@ -138,7 +138,7 @@ gain = 0.0
 limit = 0
 
 [bus-0]
-label = "MR18"
+label = "System"
 mono = false
 eq.on = false
 mute = false
@@ -146,7 +146,7 @@ gain = 0.0
 mode = "normal"
 
 [bus-1]
-label = "ASIO [1,2]"
+label = "Comms"
 mono = false
 eq.on = false
 mute = false
@@ -154,7 +154,7 @@ gain = 0.0
 mode = "normal"
 
 [bus-2]
-label = "ASIO [3,4]"
+label = "Pretzel"
 mono = false
 eq.on = false
 mute = false
@@ -162,7 +162,7 @@ gain = 0.0
 mode = "normal"
 
 [bus-3]
-label = "ASIO [5,6]"
+label = "GAME PCs"
 mono = false
 eq.on = false
 mute = false
@@ -170,7 +170,7 @@ gain = 0.0
 mode = "normal"
 
 [bus-4]
-label = "ASIO [7,8]"
+label = ""
 mono = false
 eq.on = false
 mute = false
@@ -194,7 +194,7 @@ gain = 0.0
 mode = "normal"
 
 [bus-7]
-label = "Both Mics"
+label = "Stream Mics"
 mono = false
 eq.on = false
 mute = 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, VBANChannels, VMBuses, VMStrips, XAirBuses, XAirStrips
 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"""
 
@@ -21,6 +19,7 @@ class Audio(ILayer):
         super().__init__(duckypad)
         for attr, val in kwargs.items():
             setattr(self, attr, val)
+        self.vm.observer.add(self.on_mdirty)
 
         self.reset_states()
 
@@ -41,47 +40,97 @@ class Audio(ILayer):
         for button in Buttons:
             self.vm.button[button].stateonly = getattr(AudioState, button.name)
 
+    def on_mdirty(self):
+        """Callback for Voicemeeter mdirty events.
+
+
+        This method keeps the DuckyPad state in sync with changes made from the Stream Deck"""
+        self.logger.debug('Voicemeeter state changed (mdirty event)')
+        for button in Buttons:
+            current_value = self.vm.button[button].stateonly
+            if getattr(self.state, button.name) != current_value:
+                match button.name:
+                    case 'mute_mics':
+                        if current_value:
+                            self.logger.info('Mics Muted')
+                        else:
+                            self.logger.info('Mics Unmuted')
+                    case 'only_discord':
+                        if current_value:
+                            self.mixer.strip[XAirStrips.comms].send[XAirBuses.stream_mix].level = -90
+                            self.logger.info('Only Discord Enabled')
+                        else:
+                            self.mixer.strip[XAirStrips.comms].send[XAirBuses.stream_mix].level = -24
+                            self.logger.info('Only Discord Disabled')
+                    case 'only_stream':
+                        if current_value:
+                            self.logger.info('Only Stream Enabled')
+                        else:
+                            self.logger.info('Only Stream Disabled')
+                    case 'sound_test':
+                        if current_value:
+                            self.logger.info('Sound Test Enabled')
+                        else:
+                            self.logger.info('Sound Test Disabled')
+                    case 'patch_onyx':
+                        if current_value:
+                            self.logger.info('Onyx mic has been patched')
+                        else:
+                            self.logger.info('Onyx mic has been unpatched')
+                    case 'patch_iris':
+                        if current_value:
+                            self.logger.info('Iris mic has been patched')
+                        else:
+                            self.logger.info('Iris mic has been unpatched')
+                    case 'mute_game_pcs':
+                        if current_value:
+                            self.logger.info('Game PCs Muted')
+                        else:
+                            self.logger.info('Game PCs Unmuted')
+
+            setattr(self.state, button.name, current_value)
+
     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[VMStrips.onyx_mic].mute = True
-            self.vm.strip[1].mute = True
+            self.vm.strip[VMStrips.iris_mic].mute = True
-            self.vm.strip[4].mute = True
             self.logger.info('Mics Muted')
         else:
-            self.vm.strip[0].mute = False
+            self.vm.strip[VMStrips.onyx_mic].mute = False
-            self.vm.strip[1].mute = False
+            self.vm.strip[VMStrips.iris_mic].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.bus[VMBuses.both_mics].mute = True
-            self.vm.strip[4].mute = True
+            self.mixer.strip[XAirStrips.comms].send[XAirBuses.stream_mix].level = -90
             self.logger.info('Only Discord Enabled')
         else:
-            self.vm.strip[4].mute = False
+            self.mixer.strip[XAirStrips.comms].send[XAirBuses.stream_mix].level = -24
-            self.mixer.dca[0].on = True
+            self.vm.bus[VMBuses.both_mics].mute = False
             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[VMBuses.onyx_mic].mute = True
-            self.vm.bus[6].mute = True
+            self.vm.bus[VMBuses.iris_mic].mute = True
-            self.vm.strip[2].gain = -3
+            self.vm.strip[VMStrips.onyx_pc].gain = -3
-            self.vm.strip[3].gain = -3
+            self.vm.strip[VMStrips.iris_pc].gain = -3
-            self.vm.strip[6].gain = -3
+            self.vm.strip[VMStrips.comms].gain = -6
+            self.vm.strip[VMStrips.pretzel].gain = -3
             self.logger.info('Only Stream Enabled')
         else:
-            self.vm.strip[2].gain = 0
+            self.vm.strip[VMStrips.onyx_pc].gain = 0
-            self.vm.strip[3].gain = 0
+            self.vm.strip[VMStrips.iris_pc].gain = 0
-            self.vm.strip[6].gain = 0
+            self.vm.strip[VMStrips.comms].gain = 0
-            self.vm.bus[5].mute = False
+            self.vm.strip[VMStrips.pretzel].gain = 0
-            self.vm.bus[6].mute = False
+            self.vm.bus[VMBuses.onyx_mic].mute = False
+            self.vm.bus[VMBuses.iris_mic].mute = False
             self.logger.info('Only Stream Disabled')
         self.vm.button[Buttons.only_stream].stateonly = self.state.only_stream
 
@@ -93,12 +142,15 @@ 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,
             'A2': True,
+            'A4': False,
             'B1': False,
             'B2': False,
             'mono': True,
@@ -106,6 +158,7 @@ class Audio(ILayer):
         DISABLE_SOUNDTEST = {
             'A1': False,
             'A2': False,
+            'A4': True,
             'B1': True,
             'B2': True,
             'mono': False,
@@ -113,50 +166,190 @@ 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[VMStrips.onyx_mic].apply({'A5': True, 'B1': False, 'B3': False, 'mute': False})
-            self.vm.vban.outstream[0].on = True
+            self.vm.strip[VMStrips.iris_mic].apply({'A5': True, 'B2': False, 'B3': False, 'mute': False})
-            self.vm.vban.outstream[1].on = True
+            self.vm.bus[VMBuses.game_pcs].mute = True
-            self.vm.vban.outstream[0].route = 0
+            self.vm.vban.outstream[VBANChannels.onyx_mic].apply({'on': True, 'route': 4})
-            self.vm.vban.outstream[1].route = 0
+            self.vm.vban.outstream[VBANChannels.iris_mic].apply({'on': True, 'route': 4})
             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[VBANChannels.onyx_mic].route = 5
-            self.vm.vban.outstream[1].route = 6
+            self.vm.vban.outstream[VBANChannels.iris_mic].route = 6
-            self.vm.strip[4].apply({'B3': True, 'A1': False, 'mute': True})
+            self.vm.bus[VMBuses.game_pcs].mute = False
+            self.vm.strip[VMStrips.onyx_mic].apply({'A5': False, 'B1': True, 'B3': True, 'mute': True})
+            self.vm.strip[VMStrips.iris_mic].apply({'A5': False, 'B2': True, 'B3': True, 'mute': True})
+            self.vm.button[Buttons.mute_mics].stateonly = True
             self.logger.info('Sound Test Disabled')
+        self.vm.button[Buttons.sound_test].stateonly = self.state.sound_test
 
-    def solo_onyx(self):
+    @ensure_mixer_fadeout
-        """placeholder method."""
+    def stage_onyx_mic(self):
+        """Gain stage onyx mic"""
+        config = configuration.mic('onyx')
 
-    def solo_iris(self):
+        self.mixer.headamp[XAirStrips.onyx_mic].phantom = config['phantom']
-        """placeholder method."""
+        for i in range(config['gain'] + 1):
+            self.mixer.headamp[XAirStrips.onyx_mic].gain = i
+            time.sleep(0.1)
+        self.logger.info('Onyx Mic Staged with Phantom Power')
 
-    def toggle_workstation_to_onyx(self):
+    @ensure_mixer_fadeout
-        self.state.ws_to_onyx = not self.state.ws_to_onyx
+    def stage_iris_mic(self):
-        onyx_conn = configuration.get('vban_onyx')
+        """Gain stage iris mic"""
-        if self.state.ws_to_onyx:
+        config = configuration.mic('iris')
-            with vban_cmd.api('potato', outbound=True, **onyx_conn) as vban:
+
-                vban.vban.instream[0].on = True
+        self.mixer.headamp[XAirStrips.iris_mic].phantom = config['phantom']
+        for i in range(config['gain'] + 1):
+            self.mixer.headamp[XAirStrips.iris_mic].gain = i
+            time.sleep(0.1)
+        self.logger.info('Iris Mic Staged with Phantom Power')
+
+    def unstage_onyx_mic(self):
+        """Unstage onyx mic, if phantom power was enabled, disable it"""
+        config = configuration.mic('onyx')
+
+        for i in reversed(range(config['gain'] + 1)):
+            self.mixer.headamp[XAirStrips.onyx_mic].gain = i
+            time.sleep(0.1)
+        if config['phantom']:
+            self.mixer.headamp[XAirStrips.onyx_mic].phantom = False
+            self.logger.info('Onyx Mic Unstaged and Phantom Power Disabled')
+        else:
+            self.logger.info('Onyx Mic Unstaged')
+
+    def unstage_iris_mic(self):
+        """Unstage iris mic, if phantom power was enabled, disable it"""
+        config = configuration.mic('iris')
+
+        for i in reversed(range(config['gain'] + 1)):
+            self.mixer.headamp[XAirStrips.iris_mic].gain = i
+            time.sleep(0.1)
+        if config['phantom']:
+            self.mixer.headamp[XAirStrips.iris_mic].phantom = False
+            self.logger.info('Iris Mic Unstaged and Phantom Power Disabled')
+        else:
+            self.logger.info('Iris Mic Unstaged')
+
+    def patch_onyx(self):
+        self.state.patch_onyx = not self.state.patch_onyx
+        if self.state.patch_onyx:
+            self.vm.patch.asio[0].set(11)
+            self.logger.info('Onyx mic has been patched')
+        else:
+            self.vm.patch.asio[0].set(0)
+            self.logger.info('Onyx mic has been unpatched')
+        self.vm.button[Buttons.patch_onyx].stateonly = self.state.patch_onyx
+
+    def patch_iris(self):
+        self.state.patch_iris = not self.state.patch_iris
+        if self.state.patch_iris:
+            self.vm.patch.asio[2].set(12)
+            self.logger.info('Iris mic has been patched')
+        else:
+            self.vm.patch.asio[2].set(0)
+            self.logger.info('Iris mic has been unpatched')
+        self.vm.button[Buttons.patch_iris].stateonly = self.state.patch_iris
+
+    def mute_game_pcs(self):
+        self.state.mute_game_pcs = not self.state.mute_game_pcs
+        if self.state.mute_game_pcs:
+            self.vm.bus[VMBuses.game_pcs].mute = True
+            self.logger.info('Game PCs Muted')
+        else:
+            self.vm.bus[VMBuses.game_pcs].mute = False
+            self.logger.info('Game PCs Unmuted')
+        self.vm.button[Buttons.mute_game_pcs].stateonly = self.state.mute_game_pcs
+
+    ### Workstation and TV Audio Routing via VBAN ###
+
+    def __fadein_main(self, target_level: float, duration: float = 5.0):
+        current_level = self.mixer.lr.mix.fader
+        level_difference = abs(target_level - current_level)
+        steps = max(10, min(100, int(level_difference)))
+        step_duration = duration / steps
+        level_step = (target_level - current_level) / steps
+
+        for _ in range(steps):
+            current_level += level_step
+            self.mixer.lr.mix.fader = current_level
+            time.sleep(step_duration)
+
+    def __fadeout_main(self, target_level: float, duration: float = 5.0):
+        current_level = self.mixer.lr.mix.fader
+        level_difference = abs(current_level - target_level)
+        steps = max(10, min(100, int(level_difference)))
+        step_duration = duration / steps
+        level_step = (current_level - target_level) / steps
+
+        for _ in range(steps):
+            current_level -= level_step
+            self.mixer.lr.mix.fader = current_level
+            time.sleep(step_duration)
+
+    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[6].on = True
             self.vm.strip[5].gain = -6
             self.vm.vban.outstream[2].on = True
-
+            self.__fadeout_main(-90)
-            # Fade out the workstation
+            self.logger.info(f'Workstation audio routed to {target_name}')
-            current_fader = self.mixer.lr.mix.fader
-            while current_fader > -90:
-                current_fader -= 1
-                self.mixer.lr.mix.fader = current_fader
-                time.sleep(0.05)
         else:
-            with vban_cmd.api('potato', outbound=True, **onyx_conn) as vban:
+            with vban_cmd.api('potato', outbound=True, **target_conn) as vban:
-                vban.vban.instream[0].on = False
+                vban.vban.instream[6].on = False
             self.vm.strip[5].gain = 0
             self.vm.vban.outstream[2].on = False
+            self.__fadein_main(-24)
+            self.logger.info('Workstation audio routed back to monitor speakers')
 
-            # Fade in the workstation
+    def toggle_workstation_to_onyx(self):
-            current_fader = self.mixer.lr.mix.fader
+        self._toggle_workstation_routing('ws_to_onyx', 'Onyx', 'vban_onyx')
-            while current_fader < -36:
+
-                current_fader += 1
+    def toggle_workstation_to_iris(self):
-                self.mixer.lr.mix.fader = current_fader
+        self._toggle_workstation_routing('ws_to_iris', 'Iris', 'vban_iris')
-                time.sleep(0.05)
+
+    def _toggle_tv_routing(self, state_attr, target_name, vban_config_key):
+        """Toggle routing of TV 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)
+        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[7].on = True
+            self.logger.info(f'TV audio routed to {target_name}')
+        else:
+            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 = True
+                vban_tv.strip[3].gain = 0
+                vban_tv.vban.outstream[0].on = False
+                vban_target.vban.instream[7].on = False
+            self.logger.info(f'TV audio routing to {target_name} disabled')
+
+    def toggle_tv_audio_to_onyx(self):
+        self._toggle_tv_routing('tv_to_onyx', 'Onyx', 'vban_onyx')
+
+    def toggle_tv_audio_to_iris(self):
+        self._toggle_tv_routing('tv_to_iris', 'Iris', 'vban_iris')

duckypad_twitch/configuration.py

@@ -18,3 +18,14 @@ with open(configpath, 'rb') as f:
 def get(name):
     if name in configuration:
         return configuration[name]
+
+
+def mic(name):
+    assert 'microphones' in configuration, 'No microphones defined in configuration'
+
+    try:
+        mic_key = configuration['microphones'][name]
+        mic_cfg = configuration['microphone'][mic_key]
+        return mic_cfg
+    except KeyError as e:
+        raise KeyError(f'Microphone configuration for "{name}" not found') from e

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,28 @@
+from enum import IntEnum
+
+Buttons = IntEnum(
+    'Buttons', 'mute_mics only_discord only_stream sound_test patch_onyx patch_iris mute_game_pcs', start=0
+)
+
+# Voicemeeter Channels
+VMStrips = IntEnum('Strips', 'onyx_mic iris_mic onyx_pc iris_pc st_input_5 system comms pretzel', start=0)
+VMBuses = IntEnum('Buses', 'system comms pretzel game_pcs output_5 onyx_mic iris_mic both_mics', start=0)
+
+# VBAN Channels
+VBANChannels = IntEnum('VBANChannels', 'onyx_mic iris_mic comms workstation', start=0)
+
+
+# XAir Channels
+class XAirStrips(IntEnum):
+    system = 0
+    comms = 2
+    pretzel = 4
+    game_pcs = 6
+    onyx_mic = 10
+    iris_mic = 11
+
+
+class XAirBuses(IntEnum):
+    onyx_mix = 0
+    iris_mix = 2
+    stream_mix = 4

duckypad_twitch/macros/__init__.py

@@ -1 +1 @@
-from .duckypad import run
+from .main import run

duckypad_twitch/macros/duckypad.py

@@ -1,73 +0,0 @@
-import logging
-
-import keyboard
-import voicemeeterlib
-import xair_api
-
-import duckypad_twitch
-from duckypad_twitch import configuration
-
-logging.basicConfig(level=logging.INFO)
-
-
-def register_hotkeys(duckypad):
-    def audio_hotkeys():
-        keyboard.add_hotkey('F13', duckypad.audio.mute_mics)
-        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('F18', duckypad.audio.solo_iris)
-        keyboard.add_hotkey('F19', duckypad.audio.toggle_workstation_to_onyx)
-
-    def scene_hotkeys():
-        keyboard.add_hotkey('ctrl+F13', duckypad.scene.onyx_only)
-        keyboard.add_hotkey('ctrl+F14', duckypad.scene.iris_only)
-        keyboard.add_hotkey('ctrl+F15', duckypad.scene.dual_scene)
-        keyboard.add_hotkey('ctrl+F16', duckypad.scene.onyx_big)
-        keyboard.add_hotkey('ctrl+F17', duckypad.scene.iris_big)
-        keyboard.add_hotkey('ctrl+F18', duckypad.scene.start)
-        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+F14', duckypad.obsws.brb)
-        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)
-
-    for step in (
-        audio_hotkeys,
-        scene_hotkeys,
-        obsws_hotkeys,
-        streamlabs_controller_hotkeys,
-        duckypad_hotkeys,
-    ):
-        step()
-
-
-def run():
-    xair_config = configuration.get('xair')
-
-    with (
-        voicemeeterlib.api('potato') as vm,
-        xair_api.connect('MR18', **xair_config) as mixer,
-        duckypad_twitch.connect(vm=vm, mixer=mixer) as duckypad,
-    ):
-        vm.apply_config('streaming_extender')  # extends the streaming config
-
-        register_hotkeys(duckypad)
-
-        print('press ctrl+m to quit')
-        keyboard.wait('ctrl+m')

duckypad_twitch/macros/main.py

@@ -0,0 +1,86 @@
+import logging
+
+import keyboard
+import voicemeeterlib
+import xair_api
+
+import duckypad_twitch
+from duckypad_twitch import configuration
+
+logging.basicConfig(level=logging.INFO)
+
+logger = logging.getLogger(__name__)
+
+
+def register_hotkeys(duckypad):
+    def audio_hotkeys():
+        keyboard.add_hotkey('F13', duckypad.audio.mute_mics)
+        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.stage_onyx_mic)
+        keyboard.add_hotkey('F18', duckypad.audio.stage_iris_mic)
+        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.patch_onyx)
+        keyboard.add_hotkey('F20', duckypad.audio.patch_iris)
+        keyboard.add_hotkey('F21', duckypad.audio.mute_game_pcs)
+        keyboard.add_hotkey('alt+F13', duckypad.audio.toggle_workstation_to_onyx)
+        keyboard.add_hotkey('alt+F14', duckypad.audio.toggle_workstation_to_iris)
+        keyboard.add_hotkey('alt+F15', duckypad.audio.toggle_tv_audio_to_onyx)
+        keyboard.add_hotkey('alt+F16', duckypad.audio.toggle_tv_audio_to_iris)
+
+    def scene_hotkeys():
+        keyboard.add_hotkey('ctrl+F13', duckypad.scene.start)
+        keyboard.add_hotkey('ctrl+F14', duckypad.scene.dual_stream)
+        keyboard.add_hotkey('ctrl+F15', duckypad.scene.brb)
+        keyboard.add_hotkey('ctrl+F16', duckypad.scene.end)
+        keyboard.add_hotkey('ctrl+F17', duckypad.scene.onyx_solo)
+        keyboard.add_hotkey('ctrl+F18', duckypad.scene.iris_solo)
+
+    def obsws_hotkeys():
+        keyboard.add_hotkey('ctrl+F22', duckypad.obsws.start_stream)
+        keyboard.add_hotkey('ctrl+F23', duckypad.obsws.stop_stream)
+
+    def duckypad_hotkeys():
+        keyboard.add_hotkey('ctrl+F24', duckypad.reset)
+
+    for step in (
+        audio_hotkeys,
+        scene_hotkeys,
+        obsws_hotkeys,
+        duckypad_hotkeys,
+    ):
+        step()
+
+
+def run():
+    xair_config = configuration.get('xair')
+
+    with (
+        voicemeeterlib.api('potato', mdirty=True) as vm,
+        xair_api.connect('MR18', **xair_config) as mixer,
+        duckypad_twitch.connect(vm=vm, mixer=mixer) as duckypad,
+    ):
+        vm.apply_config('streaming_extender')  # extends the streaming config
+
+        register_hotkeys(duckypad)
+
+        banner_width = 80
+        logger.info(
+            '\n'.join(
+                (
+                    '\n' + '#' * banner_width,
+                    'Duckypad Twitch is running. ',
+                    'Run sound test and gain stage mics to verify audio setup.',
+                    'Then start the stream.',
+                    "Don't forget Voicemeeter starts in Only Stream mode!",
+                    'So first unmute mics, then give stream introduction, then disable Only Stream mode.',
+                    'Now you are live with mics unmuted!',
+                    '#' * banner_width,
+                )
+            )
+        )
+
+        print('press ctrl+shift+F24 to quit')
+        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,
                 ]
@@ -57,48 +54,59 @@ class OBSWS(ILayer):
         for client in (self.request, self.event):
             if client:
                 client.disconnect()
+        self.request = self.event = None
 
-    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_exit_started(self, _):
+    def on_current_program_scene_changed(self, data):
-        self.event.unsubscribe()
+        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, data):
+        self.logger.info('OBS is exiting, disconnecting...')
+        self.request.disconnect()
+        self.request = self.event = None
+
+    ### 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')
+        self.logger.info('stream started')
 
-    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)
+        self.logger.info('stream stopped')
-
-    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 VMBuses, VMStrips
 from .layer import ILayer
 from .states import SceneState
 
@@ -31,50 +32,36 @@ 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[VMStrips.onyx_pc].mute = True
-            self.vm.strip[2].mute = True
+        self.vm.strip[VMStrips.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,
+            f'A{VMBuses.game_pcs + 1}': True,  # Voicemeeter A output is 1-indexed
+        }
+        self.vm.strip[VMStrips.onyx_pc].apply(ENABLE_PC)
+        self.vm.strip[VMStrips.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[VMStrips.onyx_pc].mute = True
-            self.vm.strip[2].mute = True
+        self.vm.strip[VMStrips.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[VMStrips.onyx_pc].mute = True
-            self.vm.strip[2].mute = True
+        self.vm.strip[VMStrips.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[VMStrips.onyx_pc].mute = False
+        self.vm.strip[VMStrips.iris_pc].mute = True
+        self.obsws.switch_to_scene('ONYX SOLO')
+
+    def iris_solo(self):
+        self.vm.strip[VMStrips.onyx_pc].mute = True
+        self.vm.strip[VMStrips.iris_pc].mute = False
+        self.obsws.switch_to_scene('IRIS SOLO')

duckypad_twitch/states.py

@@ -13,10 +13,14 @@ class AudioState:
     only_discord: bool = False
     only_stream: bool = True
     sound_test: bool = False
-    solo_onyx: bool = False
+    patch_onyx: bool = True
-    solo_iris: bool = False
+    patch_iris: bool = True
+    mute_game_pcs: 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 +33,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,26 +1,7 @@
-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"""
+    """ensure an obs websocket connection has been established
+
+    Used as a decorator for functions that require an obs websocket connection"""
 
     def wrapper(self, *args):
         if self.request is None:
@@ -31,3 +12,21 @@ def ensure_obsws(func):
         return func(self, *args)
 
     return wrapper
+
+
+def ensure_mixer_fadeout(func):
+    """ensure mixer is faded out before proceeding (disable monitor speaker)
+
+    Used as a decorator for functions that require the mixer to be faded out"""
+
+    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

@@ -23,12 +23,9 @@ classifiers = [
 dynamic = ["version"]
 dependencies = [
   "keyboard",
-  "obsws-python>=1.7",
+  "obsws-python>=1.8.0",
-  "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",
+  "voicemeeter-api>=2.7.1",
-  "websocket-client",
   "xair-api>=2.4.1",
 ]