onyx_online / Streamlabs to Voicemeeter Sync

import asyncio import logging from collections import namedtuple import voicemeeterlib from pyslobs import ScenesService, config_from_ini_else_stdin, connection class Observer: """ Observer class to handle scene switch events and perform corresponding actions. Attributes: _vm: An instance of the virtual mixer. conn: Connection object to communicate with the Streamlabs ScenesService. Methods: __init__(vm, conn): Initializes the Observer with a virtual mixer and streamlabs connection. subscribe(): Subscribes to the scene switch event from the ScenesService. on_switch_scene(_, message): Handles the scene switch event and calls the appropriate method based on the scene name. on_start(): Handles actions to be performed when the scene switches to 'START_TEST'. on_brb(): Handles actions to be performed when the scene switches to 'BRB_TEST'. on_end(): Handles actions to be performed when the scene switches to 'END_TEST'. on_live(): Handles actions to be performed when the scene switches to 'LIVE_TEST'. """ def __init__(self, vm, conn): self._vm = vm self.conn = conn async def subscribe(self): await ScenesService(self.conn).scene_switched.subscribe(self.on_switch_scene) async def on_switch_scene(self, _, message): scene = message.get('name') print(f'Switching to {scene}') match scene: case 'START_TEST': await self.on_start() case 'BRB_TEST': await self.on_brb() case 'END_TEST': await self.on_end() case 'LIVE_TEST': await self.on_live() case _: pass async def on_start(self): self._vm.strip[0].mute = True self._vm.strip[1].B1 = True self._vm.strip[2].B2 = True async def on_brb(self): self._vm.strip[7].fadeto(0, 500) self._vm.bus[0].mute = True async def on_end(self): self._vm.apply( { 'strip-0': {'mute': True}, 'strip-1': {'mute': True, 'B1': False}, 'strip-2': {'mute': True, 'B1': False}, 'vban-in-0': {'on': False}, } ) async def on_live(self): self._vm.strip[0].mute = False self._vm.strip[7].fadeto(-6, 500) self._vm.strip[7].A3 = True self._vm.vban.instream[0].on = True async def rotate_scenes(conn): """ Rotates through specific scenes and makes each one active for a short duration. Args: conn: The connection object to the ScenesService. This function retrieves all scenes using the ScenesService, then iterates through a predefined list of scene names ('START_TEST', 'BRB_TEST', 'END_TEST', 'LIVE_TEST'). For each matching scene, it makes the scene active and waits for 1 second before moving to the next scene. Finally, it closes the connection. """ Scene = namedtuple('Scene', 'name id') ss = ScenesService(conn) scenes = await ss.get_scenes() for scene in (Scene(s.name, s.id) for s in scenes): if scene.name in ('START_TEST', 'BRB_TEST', 'END_TEST', 'LIVE_TEST'): await ss.make_scene_active(scene.id) await asyncio.sleep(1) conn.close() async def main(): """ Main asynchronous function to initialize and run the application. This function performs the following tasks: 1. Establishes a connection to the Voicemeeter API using the 'potato' version. 2. Creates a connection to Streamlabs using configuration from an INI file or standard input. 3. Initializes an observer to monitor and react to events from Voicemeeter and Streamlabs. 4. Runs background processing tasks concurrently, including: - Streamlabs connection background processing - Observer subscription to events - Streamlabs scene rotation The function uses asyncio.gather to run these tasks concurrently. Note: This function should be run within an asyncio event loop. """ with voicemeeterlib.api('potato') as vm: conn = connection.SlobsConnection(config_from_ini_else_stdin()) await asyncio.gather( conn.background_processing(), Observer(vm, conn).subscribe(), rotate_scenes(conn), ) if __name__ == '__main__': logging.basicConfig(level=logging.INFO) asyncio.run(main())