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

.gitignore

@@ -161,8 +161,8 @@ cython_debug/
 #.idea/
 
 # configs
-configs/*
+duckypad.toml
-!configs/potato
+streaming_extender.toml
 
 # test
 quick.py

.pre-commit-config.yaml

@@ -0,0 +1,7 @@
+repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v2.3.0
+    hooks:
+      - id: check-yaml
+      - id: end-of-file-fixer
+      - id: trailing-whitespace

README.md

@@ -1,7 +1,7 @@
 # duckypad twitch
 
-[![PyPI - Version](https://img.shields.io/pypi/v/duckypad-twitch.svg)](https://pypi.org/project/duckypad-twitch)
+[![Hatch project](https://img.shields.io/badge/%F0%9F%A5%9A-Hatch-4051b5.svg)](https://github.com/pypa/hatch)
-[![PyPI - Python Version](https://img.shields.io/pypi/pyversions/duckypad-twitch.svg)](https://pypi.org/project/duckypad-twitch)
+[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)
 
 ---
 
@@ -10,11 +10,6 @@
 - [Installation](#installation)
 - [License](#license)
 
-## Installation
-
-```console
-pip install duckypad-twitch
-```
 
 ## About
 
@@ -27,27 +22,29 @@ 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 three 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](https://www.midasconsoles.com/product.html?modelCode=P0C8H) 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], which itself 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.
+We use a triple pc streaming setup, one gaming pc for each of us and a third pc that handles the stream.
 
-If you've ever attempted to setup a dual pc streaming setup, you may appreciate the audio challenges of a three pc setup.
+- 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 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 is a tightly coupled implementation meaning it is not designed for public use, it is purely a demonstration.
+This package is for demonstration purposes only. Several of the interfaces on which it depends have been merged into a duckypad macros program.
 
-- All keybindings are defined in `__main__.py`.
+- 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
@@ -60,5 +57,5 @@ This is a tightly coupled implementation meaning it is not designed for public u
 [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

__main__.py

@@ -1,77 +0,0 @@
-import logging
-
-import keyboard
-import voicemeeterlib
-import xair_api
-
-import duckypad_twitch
-from duckypad_twitch import configuration
-
-logging.basicConfig(level=logging.DEBUG)
-
-
-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_controller.begin_stream)
-        keyboard.add_hotkey("ctrl+F23", duckypad.streamlabs_controller.end_stream)
-        keyboard.add_hotkey(
-            "ctrl+alt+F23", duckypad.streamlabs_controller.launch, args=(8,)
-        )
-        keyboard.add_hotkey("ctrl+alt+F24", duckypad.streamlabs_controller.shutdown)
-
-    def duckypad_hotkeys():
-        keyboard.add_hotkey("ctrl+F21", duckypad.reset)
-
-    steps = (
-        audio_hotkeys,
-        scene_hotkeys,
-        streamlabs_controller_hotkeys,
-        obsws_hotkeys,
-        duckypad_hotkeys,
-    )
-    [step() for step in steps]
-
-
-def main():
-    xair_config = configuration.get("xair")
-
-    with voicemeeterlib.api("potato") as vm:
-        with xair_api.connect("MR18", **xair_config) as mixer:
-            with duckypad_twitch.connect(vm=vm, mixer=mixer) as duckypad:
-                vm.apply_config("streaming")
-
-                register_hotkeys(duckypad)
-
-                print("press ctrl+m to quit")
-                keyboard.wait("ctrl+m")
-
-
-if __name__ == "__main__":
-    main()

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
@@ -210,6 +210,36 @@ stateonly = false
 [button-2]
 stateonly = true
 
+[vban-in-0]
+on = false
+
+[vban-in-1]
+on = false
+
+[vban-in-2]
+on = false
+
+[vban-in-3]
+on = false
+
+[vban-in-4]
+on = false
+
+[vban-in-5]
+on = false
+
+[vban-in-6]
+on = false
+
+[vban-in-7]
+on = false
+
+[vban-in-8]
+on = false
+
+[vban-in-9]
+on = true
+
 [vban-out-0]
 on = false
 
@@ -233,3 +263,6 @@ on = false
 
 [vban-out-7]
 on = false
+
+[vban-out-8]
+on = false

duckypad_twitch/__about__.py

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

duckypad_twitch/audio.py

@@ -1,18 +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"""
 
@@ -20,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()
 
@@ -40,110 +40,316 @@ 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')
-            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.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")
+            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.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.logger.info("Only Stream Enabled")
+            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.logger.info("Only Stream Disabled")
+            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
 
     def sound_test(self):
         def toggle_soundtest(params):
-            onyx_conn = configuration.get("vban_onyx")
+            onyx_conn = configuration.get('vban_onyx')
-            iris_conn = configuration.get("vban_iris")
+            iris_conn = configuration.get('vban_iris')
-            assert all(
+            assert all([onyx_conn, iris_conn]), 'expected configurations for onyx_conn, iris_conn'
-                [onyx_conn, iris_conn]
-            ), "expected configurations for onyx_conn, iris_conn"
 
-            with vban_cmd.api("potato", **onyx_conn) as vban:
+            with vban_cmd.api('potato', outbound=True, **onyx_conn) as vban:
                 vban.strip[0].apply(params)
-            with vban_cmd.api("potato", **iris_conn) as vban:
+                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,
+            'A1': True,
-            "A2": True,
+            'A2': True,
-            "B1": False,
+            'A4': False,
-            "B2": False,
+            'B1': False,
-            "mono": True,
+            'B2': False,
+            'mono': True,
         }
         DISABLE_SOUNDTEST = {
-            "A1": False,
+            'A1': False,
-            "A2": False,
+            'A2': False,
-            "B1": True,
+            'A4': True,
-            "B2": True,
+            'B1': True,
-            "mono": False,
+            'B2': True,
+            'mono': False,
         }
 
         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")
+            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.logger.info("Sound Test Disabled")
+            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", **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)
+            self.logger.info(f'Workstation audio routed to {target_name}')
         else:
-            with vban_cmd.api("potato", **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')
+
+    def toggle_workstation_to_onyx(self):
+        self._toggle_workstation_routing('ws_to_onyx', 'Onyx', 'vban_onyx')
+
+    def toggle_workstation_to_iris(self):
+        self._toggle_workstation_routing('ws_to_iris', 'Iris', 'vban_iris')
+
+    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

@@ -7,14 +7,25 @@ except ModuleNotFoundError:
 
 configuration = {}
 
-configpath = Path.cwd() / "configs" / "duckypad.toml"
+configpath = Path.cwd() / 'configs' / 'duckypad.toml'
 if not configpath.exists():
-    raise OSError(f"unable to locate {configpath}")
+    raise OSError(f'unable to locate {configpath}')
 
-with open(configpath, "rb") as f:
+with open(configpath, 'rb') as f:
     configuration = tomllib.load(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,42 +18,40 @@ 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_controller = 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_controller.conn.disconnect()
+        self.obsws.disconnect()
 
     def reset(self):
-        '''
+        """
         apply streaming config,
         then apply current scene settings
         if stream is live enable both mics over vban
-        '''
+        """
-        self.vm.apply_config("streaming")
+        self.vm.apply_config('streaming')
         self.audio.reset_states()
         if self.stream.current_scene:
-            self.logger.debug(
+            self.logger.debug(f'Running function for current scene {self.stream.current_scene}')
-                f"Running function for current scene {self.stream.current_scene}"
+            try:
-            )
+                fn = getattr(
-            fn = getattr(
+                    self.scene,
-                self.scene,
+                    to_snakecase(self.stream.current_scene),
-                "_".join([word.lower() for word in self.stream.current_scene.split()]),
+                )
-            )
+                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.logger.debug('stream is live, enabling both mics over vban')
             self.vm.vban.outstream[0].on = True
             self.vm.vban.outstream[1].on = True
         else:
-            self.logger.debug(
+            self.logger.debug('stream is not live. Leaving both vban outstreams disabled')
-                "stream is not live. Leaving both vban outstreams disabled"
-            )
 
 
 def connect(*args, **kwargs):

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

@@ -0,0 +1 @@
+from .main import run

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,73 +29,84 @@ 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
 
     def obs_connect(self):
         try:
-            conn = configuration.get("obsws")
+            conn = configuration.get('obsws')
-            assert conn is not None, "expected configuration for obs"
+            assert conn is not None, 'expected configuration for obs'
             self.request = obsws.ReqClient(**conn)
             self.reset_states()
             self.event = obsws.EventClient(**conn)
             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,
                 ]
             )
         except (ConnectionRefusedError, TimeoutError) as e:
-            self.logger.error(f"{type(e).__name__}: {e}")
+            self.logger.error(f'{type(e).__name__}: {e}')
             raise
 
-    def on_current_program_scene_changed(self, data):
+    def disconnect(self):
-        self._duckypad.stream.current_scene = data.scene_name
+        for client in (self.request, self.event):
-        self.logger.info(f"scene switched to {self._duckypad.stream.current_scene}")
+            if client:
-        if self._duckypad.stream.current_scene in ("START", "BRB", "END"):
+                client.disconnect()
-            self.mute_mic_state(True)
+        self.request = self.event = None
 
-    def on_input_mute_state_changed(self, data):
+    ### Event Handlers ###
-        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_controller.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_controller.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_controller.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_controller.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_controller.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_controller.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_controller.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_controller.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

@@ -4,7 +4,7 @@ from dataclasses import dataclass
 @dataclass
 class StreamState:
     is_live: bool = False
-    current_scene: str = ""
+    current_scene: str = ''
 
 
 @dataclass
@@ -12,11 +12,15 @@ class AudioState:
     mute_mics: bool = True
     only_discord: bool = False
     only_stream: bool = True
-    sound_test: bool = True
+    sound_test: bool = False
-    solo_onyx: bool = True
+    patch_onyx: bool = True
-    solo_iris: bool = True
+    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,128 +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

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_controller.conn.ws is None:
-            try:
-                try:
-                    self.connect()
-                except AttributeError:
-                    self._duckypad.streamlabs_controller.connect()
-            except slobs_websocket.exceptions.ConnectionFailure:
-                self._duckypad.streamlabs_controller.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

@@ -1,44 +1,42 @@
 [build-system]
-requires = ["hatchling"]
 build-backend = "hatchling.build"
 
+requires = ["hatchling"]
+
 [project]
 name = "duckypad-twitch"
-dynamic = ["version"]
+description = "DuckyPad macros for streaming software"
-description = ''
 readme = "README.md"
-requires-python = ">=3.7"
-license = "MIT"
 keywords = []
-authors = [
+license = "MIT"
-  { name = "onyx-and-iris", email = "75868496+onyx-and-iris@users.noreply.github.com" },
+authors = [{ email = "code@onyxandiris.online", name = "onyx-and-iris" }]
-]
+requires-python = ">=3.12,<4.0"
 classifiers = [
   "Development Status :: 4 - Beta",
   "Programming Language :: Python",
-  "Programming Language :: Python :: 3.7",
+  "Programming Language :: Python :: 3 :: Only",
-  "Programming Language :: Python :: 3.8",
+  "Programming Language :: Python :: 3.12",
-  "Programming Language :: Python :: 3.9",
+  "Programming Language :: Python :: 3.13",
-  "Programming Language :: Python :: 3.10",
-  "Programming Language :: Python :: 3.11",
   "Programming Language :: Python :: Implementation :: CPython",
   "Programming Language :: Python :: Implementation :: PyPy",
 ]
+dynamic = ["version"]
 dependencies = [
-  "tomli >= 2.0.1;python_version < '3.11'",
-  "websocket-client",
   "keyboard",
-  "voicemeeter-api",
+  "obsws-python>=1.8.0",
-  "xair-api",
+  "vban-cmd>=2.5.2",
-  "slobs_websocket@git+https://git@github.com/onyx-and-iris/slobs_websocket@v0.1.4#egg=slobs_websocket",
+  "voicemeeter-api>=2.7.1",
-  "obsws-python",
+  "xair-api>=2.4.1",
-  "vban-cmd",
 ]
 
+[project.scripts]
+duckypad-macros = "duckypad_twitch.macros:run"
+
+
 [project.urls]
-Documentation = "https://github.com/unknown/duckypad-twitch#readme"
+Documentation = "https://github.com/onyx-and-iris/duckypad-twitch#readme"
-Issues = "https://github.com/unknown/duckypad-twitch/issues"
+Issues = "https://github.com/onyx-and-iris/duckypad-twitch/issues"
-Source = "https://github.com/unknown/duckypad-twitch"
+Source = "https://github.com/onyx-and-iris/duckypad-twitch"
 
 [tool.hatch.metadata]
 allow-direct-references = true
@@ -48,6 +46,7 @@ path = "duckypad_twitch/__about__.py"
 
 [tool.hatch.envs.default]
 dependencies = ["coverage[toml]>=6.5", "pytest"]
+
 [tool.hatch.envs.default.scripts]
 test = "pytest {args:tests}"
 test-cov = "coverage run -m pytest {args:tests}"
@@ -55,72 +54,98 @@ cov-report = ["- coverage combine", "coverage report"]
 cov = ["test-cov", "cov-report"]
 
 [[tool.hatch.envs.all.matrix]]
-python = ["3.7", "3.8", "3.9", "3.10", "3.11"]
+python = ["3.12", "3.13"]
 
 [tool.hatch.envs.lint]
 detached = true
 dependencies = ["black>=23.1.0", "mypy>=1.0.0", "ruff>=0.0.243"]
+
 [tool.hatch.envs.lint.scripts]
 typing = "mypy --install-types --non-interactive {args:duckypad_twitch tests}"
 style = ["ruff {args:.}", "black --check --diff {args:.}"]
 fmt = ["black {args:.}", "ruff --fix {args:.}", "style"]
 all = ["style", "typing"]
 
-[tool.black]
-target-version = ["py37"]
-line-length = 120
-skip-string-normalization = true
-
 [tool.ruff]
-target-version = "py37"
+exclude = [
+  ".bzr",
+  ".direnv",
+  ".eggs",
+  ".git",
+  ".git-rewrite",
+  ".hg",
+  ".mypy_cache",
+  ".nox",
+  ".pants.d",
+  ".pytype",
+  ".ruff_cache",
+  ".svn",
+  ".tox",
+  ".venv",
+  "__pypackages__",
+  "_build",
+  "buck-out",
+  "build",
+  "dist",
+  "node_modules",
+  "venv",
+]
+
 line-length = 120
-select = [
+indent-width = 4
-  "A",
+
-  "ARG",
+# Assume Python 3.10
-  "B",
+target-version = "py312"
-  "C",
+
-  "DTZ",
+[tool.ruff.lint]
-  "E",
+# Enable Pyflakes (`F`) and a subset of the pycodestyle (`E`)  codes by default.
-  "EM",
+# Unlike Flake8, Ruff doesn't enable pycodestyle warnings (`W`) or
-  "F",
+# McCabe complexity (`C901`) by default.
-  "FBT",
+select = ["E4", "E7", "E9", "F"]
-  "I",
+ignore = []
-  "ICN",
+
-  "ISC",
+# Allow fix for all enabled rules (when `--fix`) is provided.
-  "N",
+fixable = ["ALL"]
-  "PLC",
+unfixable = []
-  "PLE",
+
-  "PLR",
+# Allow unused variables when underscore-prefixed.
-  "PLW",
+dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
-  "Q",
+
-  "RUF",
+
-  "S",
+[tool.ruff.format]
-  "T",
+# Unlike Black, use single quotes for strings.
-  "TID",
+quote-style = "single"
-  "UP",
+
-  "W",
+# Like Black, indent with spaces, rather than tabs.
-  "YTT",
+indent-style = "space"
-]
+
-ignore = [
+# Like Black, respect magic trailing commas.
-  # Allow non-abstract empty methods in abstract base classes
+skip-magic-trailing-comma = false
-  "B027",
+
-  # Allow boolean positional values in function calls, like `dict.get(... True)`
+# Like Black, automatically detect the appropriate line ending.
-  "FBT003",
+line-ending = "auto"
-  # Ignore checks for possible passwords
+
-  "S105",
+# Enable auto-formatting of code examples in docstrings. Markdown,
-  "S106",
+# reStructuredText code/literal blocks and doctests are all supported.
-  "S107",
+#
-  # Ignore complexity
+# This is currently disabled by default, but it is planned for this
-  "C901",
+# to be opt-out in the future.
-  "PLR0911",
+docstring-code-format = false
-  "PLR0912",
+
-  "PLR0913",
+# Set the line length limit used when formatting code snippets in
-  "PLR0915",
+# docstrings.
-]
+#
-unfixable = [
+# This only has an effect when the `docstring-code-format` setting is
-  # Don't touch unused imports
+# enabled.
-  "F401",
+docstring-code-line-length = "dynamic"
-]
+
+[tool.ruff.lint.mccabe]
+max-complexity = 10
+
+[tool.ruff.lint.per-file-ignores]
+"__init__.py" = ["E402", "F401"]
+# Tests can use magic values, assertions, and relative imports
+"tests/**/*" = ["PLR2004", "S101", "TID252"]
 
 [tool.ruff.isort]
 known-first-party = ["duckypad_twitch"]
@@ -128,10 +153,6 @@ known-first-party = ["duckypad_twitch"]
 [tool.ruff.flake8-tidy-imports]
 ban-relative-imports = "all"
 
-[tool.ruff.per-file-ignores]
-# Tests can use magic values, assertions, and relative imports
-"tests/**/*" = ["PLR2004", "S101", "TID252"]
-
 [tool.coverage.run]
 source_pkgs = ["duckypad_twitch", "tests"]
 branch = true