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

@@ -29,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 `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

configs/potato/streaming.toml

@@ -39,8 +39,8 @@ label = "Onyx Pc"
 A1 = false
 A2 = false
 A3 = false
-A4 = false
-A5 = true
+A4 = true
+A5 = false
 B1 = false
 B2 = false
 B3 = false
@@ -57,8 +57,8 @@ label = "Iris Pc"
 A1 = false
 A2 = false
 A3 = false
-A4 = false
-A5 = true
+A4 = true
+A5 = false
 B1 = false
 B2 = false
 B3 = false
@@ -90,8 +90,8 @@ gate.knob = 0
 
 [strip-5]
 label = "System"
-A1 = false
-A2 = true
+A1 = true
+A2 = false
 A3 = false
 A4 = false
 A5 = false
@@ -107,8 +107,8 @@ limit = 0
 [strip-6]
 label = "Comms"
 A1 = false
-A2 = false
-A3 = true
+A2 = true
+A3 = false
 A4 = false
 A5 = false
 B1 = false
@@ -125,8 +125,8 @@ k = 0
 label = "Pretzel"
 A1 = false
 A2 = false
-A3 = false
-A4 = true
+A3 = 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

duckypad_twitch/audio.py

@@ -15,9 +15,6 @@ logger = logging.getLogger(__name__)
 class Audio(ILayer):
     """Audio concrete class"""
 
-    DCM8_MAX_GAIN = 20  # SE Electronics DCM8 max gain
-    TLM102_MAX_GAIN = 30  # Neumann TLM102 max gain
-
     def __init__(self, duckypad, **kwargs):
         super().__init__(duckypad)
         for attr, val in kwargs.items():
@@ -70,6 +67,26 @@ class Audio(ILayer):
                             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)
 
@@ -133,6 +150,7 @@ class Audio(ILayer):
         ENABLE_SOUNDTEST = {
             'A1': True,
             'A2': True,
+            'A4': False,
             'B1': False,
             'B2': False,
             'mono': True,
@@ -140,6 +158,7 @@ class Audio(ILayer):
         DISABLE_SOUNDTEST = {
             'A1': False,
             'A2': False,
+            'A4': True,
             'B1': True,
             'B2': True,
             'mono': False,
@@ -147,55 +166,71 @@ class Audio(ILayer):
 
         self.state.sound_test = not self.state.sound_test
         if self.state.sound_test:
-            self.vm.strip[VMStrips.onyx_mic].apply({'A1': True, 'B1': False, 'B3': False, 'mute': False})
-            self.vm.strip[VMStrips.iris_mic].apply({'A1': True, 'B2': False, 'B3': False, 'mute': False})
-            self.vm.vban.outstream[VBANChannels.onyx_mic].on = True
-            self.vm.vban.outstream[VBANChannels.iris_mic].on = True
-            self.vm.vban.outstream[VBANChannels.onyx_mic].route = 0
-            self.vm.vban.outstream[VBANChannels.iris_mic].route = 0
+            self.vm.strip[VMStrips.onyx_mic].apply({'A5': True, 'B1': False, 'B3': False, 'mute': False})
+            self.vm.strip[VMStrips.iris_mic].apply({'A5': True, 'B2': False, 'B3': False, 'mute': False})
+            self.vm.bus[VMBuses.game_pcs].mute = True
+            self.vm.vban.outstream[VBANChannels.onyx_mic].apply({'on': True, 'route': 4})
+            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[VBANChannels.onyx_mic].route = 5
             self.vm.vban.outstream[VBANChannels.iris_mic].route = 6
-            self.vm.strip[VMStrips.onyx_mic].apply({'A1': False, 'B1': True, 'B3': True, 'mute': True})
-            self.vm.strip[VMStrips.iris_mic].apply({'A1': False, 'B2': True, 'B3': True, '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
 
     @ensure_mixer_fadeout
     def stage_onyx_mic(self):
-        """Gain stage SE Electronics DCM8 with phantom power"""
-        self.mixer.headamp[XAirStrips.onyx_mic].phantom = True
-        for i in range(Audio.DCM8_MAX_GAIN + 1):
+        """Gain stage onyx mic"""
+        config = configuration.mic('onyx')
+
+        self.mixer.headamp[XAirStrips.onyx_mic].phantom = config['phantom']
+        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')
 
     @ensure_mixer_fadeout
     def stage_iris_mic(self):
-        """Gain stage TLM102 with phantom power"""
-        self.mixer.headamp[XAirStrips.iris_mic].phantom = True
-        for i in range(Audio.TLM102_MAX_GAIN + 1):
+        """Gain stage iris mic"""
+        config = configuration.mic('iris')
+
+        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 SE Electronics DCM8 and disable phantom power"""
-        for i in reversed(range(Audio.DCM8_MAX_GAIN + 1)):
+        """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)
-        self.mixer.headamp[XAirStrips.onyx_mic].phantom = False
-        self.logger.info('Onyx Mic Unstaged and Phantom Power Disabled')
+        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 TLM102 and disable phantom power"""
-        for i in reversed(range(Audio.TLM102_MAX_GAIN + 1)):
+        """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)
-        self.mixer.headamp[XAirStrips.iris_mic].phantom = False
-        self.logger.info('Iris Mic Unstaged and Phantom Power Disabled')
+        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
@@ -205,6 +240,7 @@ class Audio(ILayer):
         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
@@ -214,26 +250,43 @@ class Audio(ILayer):
         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.mixer.strip[XAirStrips.game_pcs].send[XAirBuses.stream_mix].level = -90
+            self.vm.bus[VMBuses.game_pcs].mute = True
             self.logger.info('Game PCs Muted')
         else:
-            self.mixer.strip[XAirStrips.game_pcs].send[XAirBuses.stream_mix].level = -24
+            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 _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 __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."""
@@ -246,17 +299,17 @@ class Audio(ILayer):
 
         if new_state:
             with vban_cmd.api('potato', outbound=True, **target_conn) as vban:
-                vban.vban.instream[2].on = True
+                vban.vban.instream[6].on = True
             self.vm.strip[5].gain = -6
-            self.vm.vban.outstream[3].on = True
-            self._fade_mixer(-90, fade_in=False)
+            self.vm.vban.outstream[2].on = True
+            self.__fadeout_main(-90)
             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
+                vban.vban.instream[6].on = False
             self.vm.strip[5].gain = 0
-            self.vm.vban.outstream[3].on = False
-            self._fade_mixer(-36, fade_in=True)
+            self.vm.vban.outstream[2].on = False
+            self.__fadein_main(-24)
             self.logger.info('Workstation audio routed back to monitor speakers')
 
     def toggle_workstation_to_onyx(self):
@@ -282,7 +335,7 @@ class Audio(ILayer):
                 vban_tv.strip[3].A1 = False
                 vban_tv.strip[3].gain = -6
                 vban_tv.vban.outstream[0].on = True
-                vban_target.vban.instream[3].on = True
+                vban_target.vban.instream[7].on = True
             self.logger.info(f'TV audio routed to {target_name}')
         else:
             with (
@@ -292,7 +345,7 @@ class Audio(ILayer):
                 vban_tv.strip[3].A1 = True
                 vban_tv.strip[3].gain = 0
                 vban_tv.vban.outstream[0].on = False
-                vban_target.vban.instream[3].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):

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/enums.py

@@ -1,10 +1,12 @@
 from enum import IntEnum
 
-Buttons = IntEnum('Buttons', 'mute_mics only_discord only_stream', start=0)
+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', 'onyx_mic iris_mic both_mics', start=5)
+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)

duckypad_twitch/scene.py

@@ -1,6 +1,6 @@
 import logging
 
-from .enums import VMStrips
+from .enums import VMBuses, VMStrips
 from .layer import ILayer
 from .states import SceneState
 
@@ -40,8 +40,7 @@ class Scene(ILayer):
     def dual_stream(self):
         ENABLE_PC = {
             'mute': False,
-            'A5': True,
-            'gain': 0,
+            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)