onyx_online/streamlabs-socketio-py
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/streamlabs-socketio-py synced 2026-04-01 11:19:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

{Client}.sio is now a private attribute

Browse Source 
{Client}.wait() added to expose {Client}._sio.wait() and {Client}._sio.sleep()
This commit is contained in:
onyx-and-iris 2026-03-25 05:40:20 +00:00
parent 77d252ebe2
commit e1fa0eea79
1 changed files with 24 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
streamlabsio/client.py
View File

@ -17,15 +17,15 @@ class Client:
self._logger = logger.bind(name=self.__class__.__name__) self._logger = logger.bind(name=self.__class__.__name__)
self._token = token self._token = token
self._raw = raw self._raw = raw
self.sio = socketio.Client() self._sio = socketio.Client()
self.sio.on('connect', self._connect_handler) self._sio.on('connect', self._connect_handler)
self.sio.on('event', self._event_handler) self._sio.on('event', self._event_handler)
self.sio.on('disconnect', self._disconnect_handler) self._sio.on('disconnect', self._disconnect_handler)
self.obs = Observable() self.obs = Observable()
def __enter__(self) -> 'Client': def __enter__(self) -> 'Client':
try: try:
self.sio.connect(f'https://sockets.streamlabs.com?token={self._token}') self._sio.connect(f'https://sockets.streamlabs.com?token={self._token}')
except socketio.exceptions.ConnectionError as e: except socketio.exceptions.ConnectionError as e:
self._logger.exception('Connection to Streamlabs Socket API failed') self._logger.exception('Connection to Streamlabs Socket API failed')
ERR_MSG = 'Failed to connect to Streamlabs Socket API. Please check your token and network connection.' ERR_MSG = 'Failed to connect to Streamlabs Socket API. Please check your token and network connection.'
@ -33,14 +33,31 @@ class Client:
return self return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None: def __exit__(self, exc_type, exc_val, exc_tb) -> None:
self.sio.disconnect() self._sio.disconnect()
def wait(self, seconds: float | None = None) -> None:
"""Exposes the wait and sleep methods of the socketio client to allow the user to keep the connection alive.
Args:
seconds (float | None): If None, the method will block indefinitely.
Returns:
None
"""
if seconds is None:
self._sio.wait()
else:
self._sio.sleep(seconds)
def _connect_handler(self) -> None: def _connect_handler(self) -> None:
self._logger.info('Connected to Streamlabs Socket API') self._logger.info('Connected to Streamlabs Socket API')
def _event_handler(self, data: dict) -> None: def _event_handler(self, data: dict) -> None:
""" """
Handles incoming events and triggers corresponding OBS actions. Handles incoming events and triggers the callback functions.
Args: Args:
data (dict): The event data containing information about the event. data (dict): The event data containing information about the event.
Expected keys: Expected keys: