mirror of
https://github.com/onyx-and-iris/streamlabs-socketio-py
synced 2026-04-07 13:43:30 +00:00
v2 proposal. See CHANGELOG unreleased.
This commit is contained in:
@@ -1,119 +1,44 @@
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Union
|
||||
|
||||
import socketio
|
||||
from loguru import logger
|
||||
from observable import Observable
|
||||
|
||||
from .error import SteamlabsSIOConnectionError, SteamlabsSIOError
|
||||
from .error import SteamlabsSIOConnectionError
|
||||
from .models import as_dataclass
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Client:
|
||||
def __init__(self, token=None, raw=False):
|
||||
self.logger = logger.getChild(self.__class__.__name__)
|
||||
self.token = token or self._token_from_toml()
|
||||
EVENT_TYPES: set[str] = (
|
||||
{'donation'} # streamlabs
|
||||
| {'follow', 'subscription', 'host', 'bits', 'raid'} # twitch
|
||||
| {'follow', 'subscription', 'superchat'} # youtube
|
||||
)
|
||||
|
||||
def __init__(self, *, token: str, raw: bool = False):
|
||||
self._logger = logger.bind(name=self.__class__.__name__)
|
||||
self._token = token
|
||||
self._raw = raw
|
||||
self.sio = socketio.Client()
|
||||
self.sio.on('connect', self.connect_handler)
|
||||
self.sio.on('event', self.event_handler)
|
||||
self.sio.on('disconnect', self.disconnect_handler)
|
||||
self.sio.on('connect', self._connect_handler)
|
||||
self.sio.on('event', self._event_handler)
|
||||
self.sio.on('disconnect', self._disconnect_handler)
|
||||
self.obs = Observable()
|
||||
self.event_types: set[str] = (
|
||||
{'donation'} # streamlabs
|
||||
| {'follow', 'subscription', 'host', 'bits', 'raid'} # twitch
|
||||
| {'follow', 'subscription', 'superchat'} # youtube
|
||||
)
|
||||
|
||||
def __enter__(self):
|
||||
def __enter__(self) -> 'Client':
|
||||
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:
|
||||
self.logger.exception(f'{type(e).__name__}: {e}')
|
||||
raise SteamlabsSIOConnectionError(
|
||||
'no connection could be established to the Streamlabs SIO server'
|
||||
) from e
|
||||
self.log_mode()
|
||||
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.'
|
||||
raise SteamlabsSIOConnectionError(ERR_MSG) from e
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
|
||||
self.sio.disconnect()
|
||||
|
||||
@property
|
||||
def raw(self) -> bool:
|
||||
return self._raw
|
||||
def _connect_handler(self) -> None:
|
||||
self._logger.info('Connected to Streamlabs Socket API')
|
||||
|
||||
@raw.setter
|
||||
def raw(self, val: bool) -> None:
|
||||
self._raw = val
|
||||
self.log_mode()
|
||||
|
||||
def log_mode(self):
|
||||
info = (
|
||||
'Raw mode' if self.raw else 'Normal mode',
|
||||
'activated.',
|
||||
'JSON messages' if self.raw else 'Event objects',
|
||||
'will be passed to callbacks.',
|
||||
)
|
||||
self.logger.info(' '.join(info))
|
||||
|
||||
def _token_from_toml(self) -> str:
|
||||
"""
|
||||
Retrieves the Streamlabs token from a TOML configuration file.
|
||||
This method attempts to load the token from a 'config.toml' file located
|
||||
either in the current working directory or in the user's home configuration
|
||||
directory under '.config/streamlabsio/'.
|
||||
Returns:
|
||||
str: The Streamlabs token retrieved from the TOML configuration file.
|
||||
Raises:
|
||||
SteamlabsSIOError: If no configuration file is found, if the file cannot
|
||||
be decoded, or if the required 'streamlabs' section or 'token' key is
|
||||
missing from the configuration file.
|
||||
"""
|
||||
|
||||
try:
|
||||
import tomllib
|
||||
except ModuleNotFoundError:
|
||||
import tomli as tomllib
|
||||
|
||||
def get_filepath() -> Union[Path, None]:
|
||||
filepaths = (
|
||||
Path.cwd() / 'config.toml',
|
||||
Path.home() / '.config' / 'streamlabsio' / 'config.toml',
|
||||
)
|
||||
for filepath in filepaths:
|
||||
if filepath.exists():
|
||||
return filepath
|
||||
return None
|
||||
|
||||
filepath = get_filepath()
|
||||
if not filepath:
|
||||
raise SteamlabsSIOError('no token provided and no config.toml file found')
|
||||
|
||||
try:
|
||||
with open(filepath, 'rb') as f:
|
||||
conn = tomllib.load(f)
|
||||
except tomllib.TOMLDecodeError as e:
|
||||
ERR_MSG = f'Error decoding {filepath}: {e}'
|
||||
self.logger.exception(ERR_MSG)
|
||||
raise SteamlabsSIOError(ERR_MSG) from e
|
||||
|
||||
if 'streamlabs' not in conn or 'token' not in conn['streamlabs']:
|
||||
ERR_MSG = (
|
||||
'config.toml does not contain a "streamlabs" section '
|
||||
'or the "streamlabs" section does not contain a "token" key'
|
||||
)
|
||||
self.logger.exception(ERR_MSG)
|
||||
raise SteamlabsSIOError(ERR_MSG)
|
||||
|
||||
return conn['streamlabs']['token']
|
||||
|
||||
def connect_handler(self) -> None:
|
||||
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.
|
||||
Args:
|
||||
@@ -126,19 +51,30 @@ class Client:
|
||||
None
|
||||
"""
|
||||
|
||||
if 'for' in data and data['type'] in self.event_types:
|
||||
message = data['message'][0]
|
||||
self.obs.trigger(
|
||||
data['for'],
|
||||
data['type'],
|
||||
message if self.raw else as_dataclass(data['type'], message),
|
||||
)
|
||||
self.logger.debug(data)
|
||||
match data:
|
||||
case {
|
||||
'for': str() as target,
|
||||
'type': str() as event_type,
|
||||
'message': list() as message_list,
|
||||
} if event_type in Client.EVENT_TYPES:
|
||||
# The message type is expected to be a list, however, in practice, it often contains only one item.
|
||||
# To ensure compatibility with the actual data structure,
|
||||
# we will iterate over the message list and trigger events for each message item.
|
||||
for message in message_list:
|
||||
self.obs.trigger(
|
||||
target,
|
||||
event_type,
|
||||
message if self._raw else as_dataclass(event_type, message),
|
||||
)
|
||||
self._logger.debug(
|
||||
f'Triggered event: {target} {event_type} {message}'
|
||||
)
|
||||
case _:
|
||||
self._logger.debug(f'Unexpected event data: {data}')
|
||||
|
||||
def disconnect_handler(self) -> None:
|
||||
self.logger.info('Disconnected from Streamlabs Socket API')
|
||||
def _disconnect_handler(self) -> None:
|
||||
self._logger.info('Disconnected from Streamlabs Socket API')
|
||||
|
||||
|
||||
def connect(**kwargs) -> Client:
|
||||
SIO_cls = Client
|
||||
return SIO_cls(**kwargs)
|
||||
def request_client_object(token: str, raw: bool = False) -> Client:
|
||||
return Client(token=token, raw=raw)
|
||||
|
||||
Reference in New Issue
Block a user