mirror of
https://github.com/onyx-and-iris/duckypad-twitch.git
synced 2026-01-02 17:57:49 +00:00
remove streamlabs code, only communicate with OBS
added methods to Audio class for: stage gaining microphones toggling audio to/from ws toggling audio to/from tv
This commit is contained in:
Onyx and Iris
parent
d07581593f
commit
227a973949
@ -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')
|
||||
|
||||
@ -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
|
||||
|
||||
6
duckypad_twitch/enums.py
Normal file
6
duckypad_twitch/enums.py
Normal file
@ -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)
|
||||
@ -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')
|
||||
|
||||
@ -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')
|
||||
|
||||
@ -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')
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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 = ''
|
||||
@ -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())
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user