diff --git a/duckypad_twitch/audio.py b/duckypad_twitch/audio.py index b7e3105..2fa3d02 100644 --- a/duckypad_twitch/audio.py +++ b/duckypad_twitch/audio.py @@ -1,5 +1,4 @@ import logging -from enum import IntEnum import time import vban_cmd @@ -7,13 +6,12 @@ import vban_cmd from . import configuration from .layer import ILayer from .states import AudioState +from .enums import Buttons, Strips +from .util import ensure_mixer_fadeout logger = logging.getLogger(__name__) -Buttons = IntEnum('Buttons', 'mute_mics only_discord only_stream', start=0) - - class Audio(ILayer): """Audio concrete class""" @@ -93,8 +91,10 @@ class Audio(ILayer): with vban_cmd.api('potato', outbound=True, **onyx_conn) as vban: vban.strip[0].apply(params) + vban.vban.instream[0].on = True with vban_cmd.api('potato', outbound=True, **iris_conn) as vban: vban.strip[0].apply(params) + vban.vban.instream[0].on = True ENABLE_SOUNDTEST = { 'A1': True, @@ -113,7 +113,8 @@ class Audio(ILayer): self.state.sound_test = not self.state.sound_test if self.state.sound_test: - self.vm.strip[4].apply({'B3': False, 'A1': True, 'mute': False}) + self.vm.strip[Strips.onyx_mic].apply({'A1': True, 'B1': False, 'B3': False, 'mute': False}) + self.vm.strip[Strips.iris_mic].apply({'A1': True, 'B2': False, 'B3': False, 'mute': False}) self.vm.vban.outstream[0].on = True self.vm.vban.outstream[1].on = True self.vm.vban.outstream[0].route = 0 @@ -124,39 +125,121 @@ class Audio(ILayer): toggle_soundtest(DISABLE_SOUNDTEST) self.vm.vban.outstream[0].route = 5 self.vm.vban.outstream[1].route = 6 - self.vm.strip[4].apply({'B3': True, 'A1': False, 'mute': True}) + self.vm.strip[Strips.onyx_mic].apply({'A1': False, 'B1': True, 'B3': True, 'mute': True}) + self.vm.strip[Strips.iris_mic].apply({'A1': False, 'B2': True, 'B3': True, 'mute': True}) self.logger.info('Sound Test Disabled') + @ensure_mixer_fadeout + def stage_onyx_mic(self): + """Gain stage SE Electronics DCM8 with phantom power""" + self.mixer.headamp[10].phantom = True + for i in range(21): + self.mixer.headamp[10].gain = i + time.sleep(0.1) + self.logger.info('Onyx Mic Staged with Phantom Power') + + @ensure_mixer_fadeout + def stage_iris_mic(self): + """Gain stage TLM102 with phantom power""" + self.mixer.headamp[11].phantom = True + for i in range(31): + self.mixer.headamp[11].gain = i + time.sleep(0.1) + self.logger.info('Iris Mic Staged with Phantom Power') + + def unstage_onyx_mic(self): + """Unstage SE Electronics DCM8 and disable phantom power""" + for i in reversed(range(21)): + self.mixer.headamp[10].gain = i + time.sleep(0.1) + self.mixer.headamp[10].phantom = False + self.logger.info('Onyx Mic Unstaged and Phantom Power Disabled') + + def unstage_iris_mic(self): + """Unstage TLM102 and disable phantom power""" + for i in reversed(range(31)): + self.mixer.headamp[11].gain = i + time.sleep(0.1) + self.mixer.headamp[11].phantom = False + self.logger.info('Iris Mic Unstaged and Phantom Power Disabled') + def solo_onyx(self): - """placeholder method.""" + """placeholder method""" def solo_iris(self): - """placeholder method.""" + """placeholder method""" - def toggle_workstation_to_onyx(self): - self.state.ws_to_onyx = not self.state.ws_to_onyx - onyx_conn = configuration.get('vban_onyx') - if self.state.ws_to_onyx: - with vban_cmd.api('potato', outbound=True, **onyx_conn) as vban: - vban.vban.instream[0].on = True + def _fade_mixer(self, target_fader, fade_in=True): + """Fade the mixer's fader to the target level.""" + current_fader = self.mixer.lr.mix.fader + step = 1 if fade_in else -1 + while (fade_in and current_fader < target_fader) or (not fade_in and current_fader > target_fader): + current_fader += step + self.mixer.lr.mix.fader = current_fader + time.sleep(0.05) + + def _toggle_workstation_routing(self, state_attr, target_name, vban_config_key): + """Toggle routing of workstation audio to either Onyx or Iris via VBAN.""" + + current_state = getattr(self.state, state_attr) + new_state = not current_state + setattr(self.state, state_attr, new_state) + + target_conn = configuration.get(vban_config_key) + + if new_state: + with vban_cmd.api('potato', outbound=True, **target_conn) as vban: + vban.vban.instream[1].on = True self.vm.strip[5].gain = -6 self.vm.vban.outstream[2].on = True - - # Fade out the workstation - 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) + self._fade_mixer(-90, fade_in=False) + self.logger.info(f'Workstation audio routed to {target_name}') else: - with vban_cmd.api('potato', outbound=True, **onyx_conn) as vban: + with vban_cmd.api('potato', outbound=True, **target_conn) as vban: vban.vban.instream[0].on = False self.vm.strip[5].gain = 0 self.vm.vban.outstream[2].on = False + self._fade_mixer(-36, fade_in=True) + self.logger.info('Workstation audio routed back to monitor speakers') - # Fade in the workstation - current_fader = self.mixer.lr.mix.fader - while current_fader < -36: - current_fader += 1 - self.mixer.lr.mix.fader = current_fader - time.sleep(0.05) + 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[2].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[2].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') diff --git a/duckypad_twitch/duckypad.py b/duckypad_twitch/duckypad.py index ee4a9b7..ec62020 100644 --- a/duckypad_twitch/duckypad.py +++ b/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): - self.streamlabs.disconnect() + def __exit__(self, exc_value, exc_type, exc_tb): 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( - self.scene, - '_'.join([word.lower() for word in self.stream.current_scene.split()]), - ) - fn() + try: + fn = getattr( + self.scene, + to_snakecase(self.stream.current_scene), + ) + 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 diff --git a/duckypad_twitch/enums.py b/duckypad_twitch/enums.py new file mode 100644 index 0000000..2ea95f3 --- /dev/null +++ b/duckypad_twitch/enums.py @@ -0,0 +1,6 @@ +from enum import IntEnum + +Buttons = IntEnum('Buttons', 'mute_mics only_discord only_stream', start=0) + +Strips = IntEnum('Strips', 'onyx_mic iris_mic onyx_pc iris_pc', start=0) +Buses = IntEnum('Buses', 'MR18 ASIO[1,2] ASIO[3,4] ASIO[5,6] ASIO[7,8] onyx_mic iris_mic both_mics', start=5) diff --git a/duckypad_twitch/macros/duckypad.py b/duckypad_twitch/macros/duckypad.py index b96461d..4f6142a 100644 --- a/duckypad_twitch/macros/duckypad.py +++ b/duckypad_twitch/macros/duckypad.py @@ -16,33 +16,28 @@ def register_hotkeys(duckypad): keyboard.add_hotkey('F14', duckypad.audio.only_discord) keyboard.add_hotkey('F15', duckypad.audio.only_stream) keyboard.add_hotkey('F16', duckypad.audio.sound_test) - keyboard.add_hotkey('F17', duckypad.audio.solo_onyx) - keyboard.add_hotkey('F18', duckypad.audio.solo_iris) - keyboard.add_hotkey('F19', duckypad.audio.toggle_workstation_to_onyx) + 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.solo_onyx) + keyboard.add_hotkey('F20', duckypad.audio.solo_iris) + keyboard.add_hotkey('F21', duckypad.audio.toggle_workstation_to_onyx) + keyboard.add_hotkey('F22', duckypad.audio.toggle_workstation_to_iris) + keyboard.add_hotkey('F23', duckypad.audio.toggle_tv_audio_to_onyx) + keyboard.add_hotkey('F24', duckypad.audio.toggle_tv_audio_to_iris) def scene_hotkeys(): - keyboard.add_hotkey('ctrl+F13', duckypad.scene.onyx_only) - keyboard.add_hotkey('ctrl+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) + 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+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) + keyboard.add_hotkey('ctrl+alt+F13', duckypad.obsws.start_stream) + keyboard.add_hotkey('ctrl+alt+F14', duckypad.obsws.stop_stream) def duckypad_hotkeys(): keyboard.add_hotkey('ctrl+F21', duckypad.reset) @@ -51,7 +46,6 @@ def register_hotkeys(duckypad): audio_hotkeys, scene_hotkeys, obsws_hotkeys, - streamlabs_controller_hotkeys, duckypad_hotkeys, ): step() @@ -69,5 +63,5 @@ def run(): register_hotkeys(duckypad) - print('press ctrl+m to quit') - keyboard.wait('ctrl+m') + print('press ctrl+shift+F24 to quit') + keyboard.wait('ctrl+shift+F24') diff --git a/duckypad_twitch/obsws.py b/duckypad_twitch/obsws.py index c329f48..a51ef94 100644 --- a/duckypad_twitch/obsws.py +++ b/duckypad_twitch/obsws.py @@ -4,7 +4,6 @@ import obsws_python as obsws from . import configuration from .layer import ILayer -from .states import OBSWSState from .util import ensure_obsws logger = logging.getLogger(__name__) @@ -14,12 +13,13 @@ class OBSWS(ILayer): def __init__(self, duckypad): super().__init__(duckypad) self.request = self.event = None - self._state = OBSWSState() @property def identifier(self): return type(self).__name__ + ### State Management ### + @property def state(self): return self._state @@ -29,8 +29,6 @@ class OBSWS(ILayer): self._state = val def reset_states(self): - resp = self.request.get_input_mute('Mic/Aux') - self.state.mute_mic = resp.input_muted resp = self.request.get_stream_status() self._duckypad.stream.is_live = resp.output_active @@ -44,7 +42,6 @@ class OBSWS(ILayer): self.event.callback.register( [ self.on_stream_state_changed, - self.on_input_mute_state_changed, self.on_current_program_scene_changed, self.on_exit_started, ] @@ -58,47 +55,54 @@ class OBSWS(ILayer): if client: client.disconnect() - def on_current_program_scene_changed(self, data): - 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"}') + ### Event Handlers ### def on_stream_state_changed(self, data): self._duckypad.stream.is_live = data.output_active self.logger.info(f'stream is {"live" if self._duckypad.stream.is_live else "offline"}') + def on_current_program_scene_changed(self, data): + self._duckypad.stream.current_scene = data.scene_name + match data.scene_name: + case 'START': + self.logger.info('Start scene enabled.. ready to go live!') + case 'DUAL STREAM': + self.logger.info('Dual Stream Scene enabled') + case 'BRB': + self.logger.info('BRB: game pcs muted') + case 'END': + self.logger.info('End Scene enabled.') + case 'ONYX SOLO': + self.logger.info('Onyx Solo Scene enabled, Iris game pc muted') + case 'IRIS SOLO': + self.logger.info('Iris Solo Scene enabled, Onyx game pc muted') + def on_exit_started(self, _): self.event.unsubscribe() + ### OBSWS Request Wrappers ### + @ensure_obsws - def call(self, fn_name, *args): + def _call(self, fn_name, *args): fn = getattr(self.request, fn_name) resp = fn(*args) return resp - def start(self): - self.call('set_current_program_scene', 'START') + def switch_to_scene(self, scene_name): + self._call('set_current_program_scene', scene_name) - def brb(self): - self.call('set_current_program_scene', 'BRB') + def start_stream(self): + 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('set_current_program_scene', 'END') + self._call('start_stream') - def live(self): - self.call('set_current_program_scene', 'LIVE') + def stop_stream(self): + 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('set_input_mute', 'Mic/Aux', val) - - def toggle_mute_mic(self): - self.call('toggle_input_mute', 'Mic/Aux') - - def toggle_stream(self): - self.call('toggle_stream') + self._call('stop_stream') diff --git a/duckypad_twitch/scene.py b/duckypad_twitch/scene.py index 39ba229..c652a0c 100644 --- a/duckypad_twitch/scene.py +++ b/duckypad_twitch/scene.py @@ -1,5 +1,6 @@ import logging +from .enums import Strips from .layer import ILayer from .states import SceneState @@ -31,50 +32,37 @@ class Scene(ILayer): def reset_states(self): self._state = SceneState() - def onyx_only(self): - if self._duckypad.streamlabs.switch_scene('onyx_only'): - self.vm.strip[2].mute = False - self.vm.strip[3].mute = True - self.logger.info('Only Onyx Scene enabled, Iris game pc muted') - - def iris_only(self): - if self._duckypad.streamlabs.switch_scene('iris_only'): - self.vm.strip[2].mute = True - self.vm.strip[3].mute = False - self.logger.info('Only Iris Scene enabled, Onyx game pc muted') - - def dual_scene(self): - if self._duckypad.streamlabs.switch_scene('dual_scene'): - self.vm.strip[2].apply({'mute': False, 'gain': 0}) - self.vm.strip[3].apply({'A5': True, 'mute': False, 'gain': 0}) - self.logger.info('Dual Scene enabled') - - def onyx_big(self): - if self._duckypad.streamlabs.switch_scene('onyx_big'): - self.vm.strip[2].apply({'mute': False, 'gain': 0}) - self.vm.strip[3].apply({'mute': False, 'gain': -3}) - self.logger.info('Onyx Big scene enabled') - - def iris_big(self): - if self._duckypad.streamlabs.switch_scene('iris_big'): - self.vm.strip[2].apply({'mute': False, 'gain': -3}) - self.vm.strip[3].apply({'mute': False, 'gain': 0}) - self.logger.info('Iris Big enabled') - def start(self): - if self._duckypad.streamlabs.switch_scene('start'): - self.vm.strip[2].mute = True - self.vm.strip[3].mute = True - self.logger.info('Start scene enabled.. ready to go live!') + self.vm.strip[Strips.onyx_pc].mute = True + self.vm.strip[Strips.iris_pc].mute = True + self.obsws.switch_to_scene('START') + + def dual_stream(self): + ENABLE_PC = { + 'mute': False, + 'A5': True, + 'gain': 0, + } + self.vm.strip[Strips.onyx_pc].apply(ENABLE_PC) + self.vm.strip[Strips.iris_pc].apply(ENABLE_PC) + self.obsws.switch_to_scene('DUAL STREAM') def brb(self): - if self._duckypad.streamlabs.switch_scene('brb'): - self.vm.strip[2].mute = True - self.vm.strip[3].mute = True - self.logger.info('BRB: game pcs muted') + self.vm.strip[Strips.onyx_pc].mute = True + self.vm.strip[Strips.iris_pc].mute = True + self.obsws.switch_to_scene('BRB') def end(self): - if self._duckypad.streamlabs.switch_scene('end'): - self.vm.strip[2].mute = True - self.vm.strip[3].mute = True - self.logger.info('End scene enabled.') + self.vm.strip[Strips.onyx_pc].mute = True + self.vm.strip[Strips.iris_pc].mute = True + self.obsws.switch_to_scene('END') + + def onyx_solo(self): + self.vm.strip[Strips.onyx_pc].mute = False + self.vm.strip[Strips.iris_pc].mute = True + self.obsws.switch_to_scene('ONYX SOLO') + + def iris_solo(self): + self.vm.strip[Strips.onyx_pc].mute = True + self.vm.strip[Strips.iris_pc].mute = False + self.obsws.switch_to_scene('IRIS SOLO') diff --git a/duckypad_twitch/states.py b/duckypad_twitch/states.py index a5be0f0..7b20332 100644 --- a/duckypad_twitch/states.py +++ b/duckypad_twitch/states.py @@ -17,6 +17,9 @@ class AudioState: solo_iris: bool = False ws_to_onyx: bool = False + ws_to_iris: bool = False + tv_to_onyx: bool = False + tv_to_iris: bool = False @dataclass @@ -29,8 +32,3 @@ class SceneState: start: bool = False brb: bool = False end: bool = False - - -@dataclass -class OBSWSState: - mute_mic: bool = True diff --git a/duckypad_twitch/streamlabs.py b/duckypad_twitch/streamlabs.py deleted file mode 100644 index 84f842f..0000000 --- a/duckypad_twitch/streamlabs.py +++ /dev/null @@ -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 = '' diff --git a/duckypad_twitch/util.py b/duckypad_twitch/util.py index eba4792..5523945 100644 --- a/duckypad_twitch/util.py +++ b/duckypad_twitch/util.py @@ -1,24 +1,3 @@ -import slobs_websocket - - -def ensure_sl(func): - """ensure a streamlabs websocket connection has been established""" - - def wrapper(self, *args): - if self._duckypad.streamlabs.conn.ws is None: - try: - try: - self.connect() - except AttributeError: - self._duckypad.streamlabs.connect() - except slobs_websocket.exceptions.ConnectionFailure: - self._duckypad.streamlabs.conn.ws = None - return - return func(self, *args) - - return wrapper - - def ensure_obsws(func): """ensure an obs websocket connection has been established""" @@ -31,3 +10,17 @@ def ensure_obsws(func): return func(self, *args) return wrapper + +def ensure_mixer_fadeout(func): + """ensure mixer fadeout is stopped before proceeding""" + + def wrapper(self, *args): + if self.mixer.lr.mix.fader > -90: + self._fade_mixer(-90, fade_in=False) + return func(self, *args) + + return wrapper + +def to_snakecase(scene_name: str) -> str: + """Convert caplitalized words to lowercase snake_case""" + return '_'.join(word.lower() for word in scene_name.split())