onyx_online revised this gist . Go to revision
1 file changed, 130 insertions
vm_streamlabs_sync.py(file created)
| @@ -0,0 +1,130 @@ | |||
| 1 | + | import asyncio | |
| 2 | + | import logging | |
| 3 | + | from collections import namedtuple | |
| 4 | + | ||
| 5 | + | import voicemeeterlib | |
| 6 | + | from pyslobs import ScenesService, config_from_ini_else_stdin, connection | |
| 7 | + | ||
| 8 | + | ||
| 9 | + | class Observer: | |
| 10 | + | """ | |
| 11 | + | Observer class to handle scene switch events and perform corresponding actions. | |
| 12 | + | Attributes: | |
| 13 | + | _vm: An instance of the virtual mixer. | |
| 14 | + | conn: Connection object to communicate with the Streamlabs ScenesService. | |
| 15 | + | Methods: | |
| 16 | + | __init__(vm, conn): | |
| 17 | + | Initializes the Observer with a virtual mixer and streamlabs connection. | |
| 18 | + | subscribe(): | |
| 19 | + | Subscribes to the scene switch event from the ScenesService. | |
| 20 | + | on_switch_scene(_, message): | |
| 21 | + | Handles the scene switch event and calls the appropriate method based on the scene name. | |
| 22 | + | on_start(): | |
| 23 | + | Handles actions to be performed when the scene switches to 'START_TEST'. | |
| 24 | + | on_brb(): | |
| 25 | + | Handles actions to be performed when the scene switches to 'BRB_TEST'. | |
| 26 | + | on_end(): | |
| 27 | + | Handles actions to be performed when the scene switches to 'END_TEST'. | |
| 28 | + | on_live(): | |
| 29 | + | Handles actions to be performed when the scene switches to 'LIVE_TEST'. | |
| 30 | + | """ | |
| 31 | + | ||
| 32 | + | def __init__(self, vm, conn): | |
| 33 | + | self._vm = vm | |
| 34 | + | self.conn = conn | |
| 35 | + | ||
| 36 | + | async def subscribe(self): | |
| 37 | + | await ScenesService(self.conn).scene_switched.subscribe(self.on_switch_scene) | |
| 38 | + | ||
| 39 | + | async def on_switch_scene(self, _, message): | |
| 40 | + | scene = message.get('name') | |
| 41 | + | print(f'Switching to {scene}') | |
| 42 | + | match scene: | |
| 43 | + | case 'START_TEST': | |
| 44 | + | await self.on_start() | |
| 45 | + | case 'BRB_TEST': | |
| 46 | + | await self.on_brb() | |
| 47 | + | case 'END_TEST': | |
| 48 | + | await self.on_end() | |
| 49 | + | case 'LIVE_TEST': | |
| 50 | + | await self.on_live() | |
| 51 | + | case _: | |
| 52 | + | pass | |
| 53 | + | ||
| 54 | + | async def on_start(self): | |
| 55 | + | self._vm.strip[0].mute = True | |
| 56 | + | self._vm.strip[1].B1 = True | |
| 57 | + | self._vm.strip[2].B2 = True | |
| 58 | + | ||
| 59 | + | async def on_brb(self): | |
| 60 | + | self._vm.strip[7].fadeto(0, 500) | |
| 61 | + | self._vm.bus[0].mute = True | |
| 62 | + | ||
| 63 | + | async def on_end(self): | |
| 64 | + | self._vm.apply( | |
| 65 | + | { | |
| 66 | + | 'strip-0': {'mute': True}, | |
| 67 | + | 'strip-1': {'mute': True, 'B1': False}, | |
| 68 | + | 'strip-2': {'mute': True, 'B1': False}, | |
| 69 | + | 'vban-in-0': {'on': False}, | |
| 70 | + | } | |
| 71 | + | ) | |
| 72 | + | ||
| 73 | + | async def on_live(self): | |
| 74 | + | self._vm.strip[0].mute = False | |
| 75 | + | self._vm.strip[7].fadeto(-6, 500) | |
| 76 | + | self._vm.strip[7].A3 = True | |
| 77 | + | self._vm.vban.instream[0].on = True | |
| 78 | + | ||
| 79 | + | ||
| 80 | + | async def rotate_scenes(conn): | |
| 81 | + | """ | |
| 82 | + | Rotates through specific scenes and makes each one active for a short duration. | |
| 83 | + | Args: | |
| 84 | + | conn: The connection object to the ScenesService. | |
| 85 | + | This function retrieves all scenes using the ScenesService, then iterates through | |
| 86 | + | a predefined list of scene names ('START_TEST', 'BRB_TEST', 'END_TEST', 'LIVE_TEST'). | |
| 87 | + | For each matching scene, it makes the scene active and waits for 1 second before | |
| 88 | + | moving to the next scene. Finally, it closes the connection. | |
| 89 | + | """ | |
| 90 | + | ||
| 91 | + | Scene = namedtuple('Scene', 'name id') | |
| 92 | + | ||
| 93 | + | ss = ScenesService(conn) | |
| 94 | + | scenes = await ss.get_scenes() | |
| 95 | + | for scene in (Scene(s.name, s.id) for s in scenes): | |
| 96 | + | if scene.name in ('START_TEST', 'BRB_TEST', 'END_TEST', 'LIVE_TEST'): | |
| 97 | + | await ss.make_scene_active(scene.id) | |
| 98 | + | await asyncio.sleep(1) | |
| 99 | + | conn.close() | |
| 100 | + | ||
| 101 | + | ||
| 102 | + | async def main(): | |
| 103 | + | """ | |
| 104 | + | Main asynchronous function to initialize and run the application. | |
| 105 | + | This function performs the following tasks: | |
| 106 | + | 1. Establishes a connection to the Voicemeeter API using the 'potato' version. | |
| 107 | + | 2. Creates a connection to Streamlabs using configuration from an INI file or standard input. | |
| 108 | + | 3. Initializes an observer to monitor and react to events from Voicemeeter and Streamlabs. | |
| 109 | + | 4. Runs background processing tasks concurrently, including: | |
| 110 | + | - Streamlabs connection background processing | |
| 111 | + | - Observer subscription to events | |
| 112 | + | - Streamlabs scene rotation | |
| 113 | + | The function uses asyncio.gather to run these tasks concurrently. | |
| 114 | + | Note: | |
| 115 | + | This function should be run within an asyncio event loop. | |
| 116 | + | """ | |
| 117 | + | ||
| 118 | + | with voicemeeterlib.api('potato') as vm: | |
| 119 | + | conn = connection.SlobsConnection(config_from_ini_else_stdin()) | |
| 120 | + | await asyncio.gather( | |
| 121 | + | conn.background_processing(), | |
| 122 | + | Observer(vm, conn).subscribe(), | |
| 123 | + | rotate_scenes(conn), | |
| 124 | + | ) | |
| 125 | + | ||
| 126 | + | ||
| 127 | + | if __name__ == '__main__': | |
| 128 | + | logging.basicConfig(level=logging.INFO) | |
| 129 | + | ||
| 130 | + | asyncio.run(main()) | |
Newer
Older