onyx_online/duckypad-twitch
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/duckypad-twitch.git synced 2026-04-08 12:13:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

re-run through ruff formatter

Browse Source 
add hatch+ruff badges
This commit is contained in:
Onyx and Iris
2025-01-17 20:43:34 +00:00
parent 3c979b8391
commit 5b4f3753db
11 changed files with 230 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
duckypad_twitch/__about__.py
View File

@@ -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.2"
__version__ = '1.0.3'

60
duckypad_twitch/audio.py
View File

@@ -10,7 +10,7 @@ from .states import AudioState
logger = logging.getLogger(__name__)
Buttons = IntEnum("Buttons", "mute_mics only_discord only_stream", start=0)
Buttons = IntEnum('Buttons', 'mute_mics only_discord only_stream', start=0)
class Audio(ILayer):
@@ -46,12 +46,12 @@ class Audio(ILayer):
self.vm.strip[0].mute = True
self.vm.strip[1].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[1].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):
@@ -59,11 +59,11 @@ class Audio(ILayer):
if self.state.only_discord:
self.mixer.dca[0].on = False
self.vm.strip[4].mute = True
self.logger.info("Only Discord Enabled")
self.logger.info('Only Discord Enabled')
else:
self.vm.strip[4].mute = False
self.mixer.dca[0].on = True
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):
@@ -74,59 +74,57 @@ class Audio(ILayer):
self.vm.strip[2].gain = -3
self.vm.strip[3].gain = -3
self.vm.strip[6].gain = -3
self.logger.info("Only Stream Enabled")
self.logger.info('Only Stream Enabled')
else:
self.vm.strip[2].gain = 0
self.vm.strip[3].gain = 0
self.vm.strip[6].gain = 0
self.vm.bus[5].mute = False
self.vm.bus[6].mute = False
self.logger.info("Only Stream Disabled")
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")
iris_conn = configuration.get("vban_iris")
assert all(
[onyx_conn, iris_conn]
), "expected configurations for onyx_conn, iris_conn"
onyx_conn = configuration.get('vban_onyx')
iris_conn = configuration.get('vban_iris')
assert all([onyx_conn, iris_conn]), 'expected configurations for onyx_conn, iris_conn'
with vban_cmd.api("potato", outbound=True, **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", outbound=True, **iris_conn) as vban:
with vban_cmd.api('potato', outbound=True, **iris_conn) as vban:
vban.strip[0].apply(params)
ENABLE_SOUNDTEST = {
"A1": True,
"A2": True,
"B1": False,
"B2": False,
"mono": True,
'A1': True,
'A2': True,
'B1': False,
'B2': False,
'mono': True,
}
DISABLE_SOUNDTEST = {
"A1": False,
"A2": False,
"B1": True,
"B2": True,
"mono": False,
'A1': False,
'A2': False,
'B1': True,
'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[4].apply({'B3': False, 'A1': True, 'mute': False})
self.vm.vban.outstream[0].on = True
self.vm.vban.outstream[1].on = True
self.vm.vban.outstream[0].route = 0
self.vm.vban.outstream[1].route = 0
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[1].route = 6
self.vm.strip[4].apply({"B3": True, "A1": False, "mute": True})
self.logger.info("Sound Test Disabled")
self.vm.strip[4].apply({'B3': True, 'A1': False, 'mute': True})
self.logger.info('Sound Test Disabled')
def solo_onyx(self):
"""placeholder method."""
@@ -136,14 +134,14 @@ class Audio(ILayer):
def toggle_workstation_to_onyx(self):
self.state.ws_to_onyx = not self.state.ws_to_onyx
onyx_conn = configuration.get("vban_onyx")
onyx_conn = configuration.get('vban_onyx')
if self.state.ws_to_onyx:
with vban_cmd.api("potato", **onyx_conn) as vban:
with vban_cmd.api('potato', **onyx_conn) as vban:
vban.vban.instream[0].on = True
self.vm.strip[5].gain = -6
self.vm.vban.outstream[2].on = True
else:
with vban_cmd.api("potato", **onyx_conn) as vban:
with vban_cmd.api('potato', **onyx_conn) as vban:
vban.vban.instream[0].on = False
self.vm.strip[5].gain = 0
self.vm.vban.outstream[2].on = False

6
duckypad_twitch/configuration.py
View File

@@ -7,11 +7,11 @@ 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)

19
duckypad_twitch/duckypad.py
View File

@@ -28,32 +28,29 @@ class DuckyPad:
def __exit__(self, exc_value, exc_type, traceback):
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(
f"Running function for current scene {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()]),
'_'.join([word.lower() for word in self.stream.current_scene.split()]),
)
fn()
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(
"stream is not live. Leaving both vban outstreams disabled"
)
self.logger.debug('stream is not live. Leaving both vban outstreams disabled')
def connect(*args, **kwargs):

39
duckypad_twitch/obsws.py
View File

@@ -29,15 +29,15 @@ class OBSWS(ILayer):
self._state = val
def reset_states(self):
resp = self.request.get_input_mute("Mic/Aux")
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")
assert conn is not None, "expected configuration for obs"
conn = configuration.get('obsws')
assert conn is not None, 'expected configuration for obs'
self.request = obsws.ReqClient(**conn)
self.reset_states()
self.event = obsws.EventClient(**conn)
@@ -50,25 +50,28 @@ class OBSWS(ILayer):
]
)
except (ConnectionRefusedError, TimeoutError) as e:
self.logger.error(f"{type(e).__name__}: {e}")
self.logger.error(f'{type(e).__name__}: {e}')
raise
def obs_disconnect(self):
for client in (self.request, self.event):
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.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":
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'}")
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'}"
)
self.logger.info(f'stream is {"live" if self._duckypad.stream.is_live else "offline"}')
def on_exit_started(self, _):
self.event.unsubscribe()
@@ -80,22 +83,22 @@ class OBSWS(ILayer):
return resp
def start(self):
self.call("set_current_program_scene", "START")
self.call('set_current_program_scene', 'START')
def brb(self):
self.call("set_current_program_scene", "BRB")
self.call('set_current_program_scene', 'BRB')
def end(self):
self.call("set_current_program_scene", "END")
self.call('set_current_program_scene', 'END')
def live(self):
self.call("set_current_program_scene", "LIVE")
self.call('set_current_program_scene', 'LIVE')
def mute_mic_state(self, val):
self.call("set_input_mute", "Mic/Aux", val)
self.call('set_input_mute', 'Mic/Aux', val)
def toggle_mute_mic(self):
self.call("toggle_input_mute", "Mic/Aux")
self.call('toggle_input_mute', 'Mic/Aux')
def toggle_stream(self):
self.call("toggle_stream")
self.call('toggle_stream')

44
duckypad_twitch/scene.py
View File

@@ -32,49 +32,49 @@ class Scene(ILayer):
self._state = SceneState()
def onyx_only(self):
if self._duckypad.streamlabs_controller.switch_scene("onyx_only"):
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")
self.logger.info('Only Onyx Scene enabled, Iris game pc muted')
def iris_only(self):
if self._duckypad.streamlabs_controller.switch_scene("iris_only"):
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")
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")
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")
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")
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"):
if self._duckypad.streamlabs_controller.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.logger.info('Start scene enabled.. ready to go live!')
def brb(self):
if self._duckypad.streamlabs_controller.switch_scene("brb"):
if self._duckypad.streamlabs_controller.switch_scene('brb'):
self.vm.strip[2].mute = True
self.vm.strip[3].mute = True
self.logger.info("BRB: game pcs muted")
self.logger.info('BRB: game pcs muted')
def end(self):
if self._duckypad.streamlabs_controller.switch_scene("end"):
if self._duckypad.streamlabs_controller.switch_scene('end'):
self.vm.strip[2].mute = True
self.vm.strip[3].mute = True
self.logger.info("End scene enabled.")
self.logger.info('End scene enabled.')

2
duckypad_twitch/states.py
View File

@@ -4,7 +4,7 @@ from dataclasses import dataclass
@dataclass
class StreamState:
is_live: bool = False
current_scene: str = ""
current_scene: str = ''
@dataclass

48
duckypad_twitch/streamlabs.py
View File

@@ -31,21 +31,17 @@ class StreamlabsController:
def connect(self):
try:
conn = configuration.get("streamlabs")
assert conn is not None, "expected configuration for streamlabs"
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}")
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._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
)
self.conn.StreamingService.streamingStatusChange += self.on_streaming_status_change
def disconnect(self):
self.conn.disconnect()
@@ -55,17 +51,15 @@ class StreamlabsController:
####################################################################################
def on_streaming_status_change(self, data):
self.logger.debug(f"streaming status changed, now: {data}")
if data in ("live", "starting"):
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}"
)
self.logger.debug(f'stream.current_scene updated to {self._duckypad.stream.current_scene}')
####################################################################################
# START/STOP the stream
@@ -74,14 +68,14 @@ class StreamlabsController:
@ensure_sl
def begin_stream(self):
if self._duckypad.stream.is_live:
self.logger.info("Stream is already online")
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")
self.logger.info('Stream is already offline')
return
self.conn.StreamingService.toggleStreaming()
@@ -91,9 +85,7 @@ class StreamlabsController:
@ensure_sl
def switch_scene(self, name):
return self.conn.ScenesService.makeSceneActive(
self._duckypad.scene.scenes[name.upper()]
)
return self.conn.ScenesService.makeSceneActive(self._duckypad.scene.scenes[name.upper()])
####################################################################################
# LAUNCH/SHUTDOWN the streamlabs process
@@ -102,16 +94,14 @@ class StreamlabsController:
@cached_property
def sl_fullpath(self) -> Path:
try:
self.logger.debug("fetching sl_fullpath from the registry")
SL_KEY = "029c4619-0385-5543-9426-46f9987161d9"
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"
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}")
self.logger.exception(f'{type(e).__name__}: {e}')
raise
def launch(self, delay=5):
@@ -126,4 +116,4 @@ class StreamlabsController:
if self.proc is not None:
self.proc.terminate()
self.proc = None
self._duckypad.stream.current_scene = ""
self._duckypad.stream.current_scene = ''