""" BRB Timer for OBS Studio https://github.com/beporter/brb-timer """ from __future__ import annotations import base64 import contextlib from dataclasses import dataclass import datetime import functools import hashlib import http import inspect import json import logging import os import re import socket import ssl import struct import sys from textwrap import dedent import threading import time from typing import Callable, Dict from urllib import error, parse, request # ========================================================================= SCRIPT_NAME = "BRB Timer" SCRIPT_VERSION = 1.1 LOG_LEVEL = logging.INFO # Standard python log levels. Ref: https://docs.python.org/3.12/library/logging.html#logging-levels logging.basicConfig() logging.getLogger().setLevel(LOG_LEVEL) brblog = logging.getLogger(__name__) brblog.setLevel(LOG_LEVEL) # ========================================================================= try: import obspython as S # type: ignore except ImportError: # This bit of code replaces the real obspython module with a # testing mock. The script won't "work" without obs, but it'll # let you run `python -i brb_timer.py` to access an interactive # terminal without failing on # `NameError: name 'obs' is not defined.` to let you test the # rest of it. from unittest.mock import Mock S = Mock() S.script_log = lambda _lvl, msg: print(msg) logging.warning('obspython module unavailable. Replaced with a mock.') # ========================================================================= class DEFAULTS: AUTO_HIDE_SECS = 120 TEXT_TIMER = f"{SCRIPT_NAME} - Timer" TRANSITION_SHOW = f"{SCRIPT_NAME} - Show Transition" TRANSITION_HIDE = f"{SCRIPT_NAME} - Hide Transition" ########################################################################### # Chat Commands and Messages # ========================================================================= class COMMANDS: BRB = "!brb" BACK = "!back" AT = "!at" # ========================================================================= class Message(str): def __call__(self, **kwargs): return self.format(**kwargs) # ========================================================================= class MessageMeta(type): def __getattribute__(cls, name): value = super().__getattribute__(name) if isinstance(value, str) and not isinstance(value, Message): return Message(value) return value # ========================================================================= class MESSAGES(metaclass=MessageMeta): """ This class cleans up the code that uses these format strings. Instead of: str.format(MESSAGES.BRB_STARTED, streamer='some user') You can use: MESSAGES.BRB_STARTED(streamer="some user") """ # Access control. ONLY_MOD_START = f"Only mods can start a {COMMANDS.BRB}." ONLY_MOD_END = f"Only mods can end a {COMMANDS.BRB} with {COMMANDS.BACK}." # Reporting script state. ALREADY_RUNNING = ( f"A {COMMANDS.BRB} is already running! " f"Must use {COMMANDS.BACK} first." ) NO_BRB = ( f"There's no {COMMANDS.BRB} running. " f"A mod must use {COMMANDS.BRB} first." ) BRB_STARTED = ( "{streamer} is taking a break! " "Guess how long till they're back (without going over!) " f"Type `{COMMANDS.AT} MM:SS` in chat!" ) BRB_FINISHED = ( "{streamer} is back after {time}! " "Winner is: {winner}. They were {diff} off." ) BRB_FINISHED_NO_GUESSES = ( "{streamer} is back after {time}! " "But there were no valid " f"{COMMANDS.AT} guesses so there's no winner. " "(Price is Right rules: closest without going over!)" ) # Guessing. BAD_GUESS = ( "Couldn't understand {user}'s guess. " "Format is MINS:SECS. " "For example: '1:35:42' means you think the streamer " "will return in 1 hour, 35 mins and 42 seconds." ) ALREADY_GUESSED = ( "{user} already guessed {guess}. " "You can't change your guess!" ) GUESS_ACCEPTED = "{user} guesses {guess}." ########################################################################### # Convience datetime and timedelta methods # ========================================================================= class DT: #---------------------------------------------------------------------- @classmethod def strfdelta(self, delta: int) -> str: if delta >= 3600: fmt = '%H:%M:%S' elif delta >= 60: fmt = '%M:%S' else: fmt = '%S secs' return datetime.datetime.fromtimestamp( delta, tz=datetime.timezone.utc, ).strftime(fmt).lstrip('0') #---------------------------------------------------------------------- @classmethod def strpdelta(self, time_str: str) -> int | False: """ Parse the provided `[[HH:]MM:]SS` string into a integer number of seconds. Ref: timedelta https://stackoverflow.com/a/51916936/70876 Ref: regex https://stackoverflow.com/a/8318367/70876 """ pattern = re.compile( r''' ^(?: (?: (?P[01]?\d|2[0-3]): )? (?P[0-5]?\d): )? (?P[0-5]?\d)$ ''', re.X, ) parts = pattern.match(time_str) if parts is None: return False time_params = {name: float(param) for name, param in parts.groupdict().items() if param} return int(datetime.timedelta(**time_params).total_seconds()) ########################################################################### # Twitch Chat Client # ========================================================================= @dataclass class ChatMessage(object): """ Message passing container from the IRC client to the command parser. """ username: str display_name: str message: str is_mod: bool is_broadcaster: bool # ========================================================================= type ChatMsgCallback = Callable[[ChatMessage], None] # ========================================================================= class _WebSocket: """ Minimal RFC6455 WebSocket client implemented with only the Python standard library. """ MAGIC = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" #---------------------------------------------------------------------- def __init__(self, url: str, timeout: float = 10): self.url = url self.timeout = timeout self.socket: socket.socket | ssl.SSLSocket | None = None self._recv_buffer = bytearray() self._fragment_buffer = bytearray() self._fragment_opcode: int | None = None #---------------------------------------------------------------------- def connect(self) -> None: parsed = parse.urlsplit(self.url) if parsed.scheme != "wss": raise ValueError( f"Only wss:// URLs are supported: {self.url!r}" ) host = parsed.hostname if not host: raise ValueError(f"Invalid WebSocket URL: {self.url!r}") port = parsed.port or 443 path = parsed.path or "/" if parsed.query: path += "?" + parsed.query self.socket = ssl.create_default_context().wrap_socket( socket.create_connection( (host, port), timeout=self.timeout, ), server_hostname=host, ) # RFC 6455 opening handshake. key = base64.b64encode(os.urandom(16)).decode("ascii") request = ( f"GET {path} HTTP/1.1\r\n" f"Host: {host}:{port}\r\n" f"Upgrade: websocket\r\n" f"Connection: Upgrade\r\n" f"Sec-WebSocket-Key: {key}\r\n" f"Sec-WebSocket-Version: 13\r\n" f"\r\n" ).encode("ascii") self.socket.sendall(request) status_line, headers = self._read_http_response() try: version, status, reason = status_line.split(" ", 2) except ValueError as e: raise ConnectionError( f"Invalid WebSocket HTTP response: {status_line!r}" ) from e if version != "HTTP/1.1": raise ConnectionError( f"Unexpected WebSocket HTTP version: {version!r}" ) if status != "101": raise ConnectionError( f"WebSocket handshake failed: {status} {reason}" ) if headers.get("upgrade", "").lower() != "websocket": raise ConnectionError( "WebSocket handshake missing Upgrade: websocket" ) connection_tokens = { token.strip().lower() for token in headers.get("connection", "").split(",") } if "upgrade" not in connection_tokens: raise ConnectionError( "WebSocket handshake missing Connection: Upgrade" ) accept = headers.get("sec-websocket-accept") if accept is None: raise ConnectionError( "WebSocket handshake missing Sec-WebSocket-Accept" ) expected_accept = base64.b64encode( hashlib.sha1((key + self.MAGIC).encode("ascii")).digest() ).decode("ascii") if accept.strip() != expected_accept: raise ConnectionError( "Invalid WebSocket Sec-WebSocket-Accept" ) if "sec-websocket-extensions" in headers: raise ConnectionError( "WebSocket extensions are not supported" ) self.socket.settimeout(1.0) #---------------------------------------------------------------------- def recv_message(self) -> str | bytes | None: """ Receive the next complete WebSocket message. Returns: str for text messages bytes for binary messages None for a clean WebSocket close """ while True: fin, opcode, payload = self._recv_frame() # Continuation frame. if opcode == 0x0: if self._fragment_opcode is None: raise ConnectionError( "Unexpected WebSocket continuation frame" ) self._fragment_buffer.extend(payload) if not fin: continue message_opcode = self._fragment_opcode data = bytes(self._fragment_buffer) self._fragment_buffer.clear() self._fragment_opcode = None if message_opcode == 0x1: return data.decode("utf-8") if message_opcode == 0x2: return data raise ConnectionError( f"Invalid fragmented opcode: {message_opcode}" ) # Text or binary data frame. if opcode in (0x1, 0x2): if self._fragment_opcode is not None: raise ConnectionError( "New WebSocket data frame received " "during fragmented message" ) if fin: if opcode == 0x1: return payload.decode("utf-8") return payload self._fragment_opcode = opcode self._fragment_buffer.clear() self._fragment_buffer.extend(payload) continue # Close. if opcode == 0x8: if len(payload) == 1: raise ConnectionError( "WebSocket close frame has invalid payload length" ) self._send_close(payload) if len(payload) >= 2: close_code = struct.unpack("!H", payload[:2])[0] raise ConnectionError( f"WebSocket closed by server (code {close_code})" ) raise ConnectionError("WebSocket closed by server") # Ping. if opcode == 0x9: self._send_frame(0xA, payload) continue # Pong. if opcode == 0xA: continue raise ConnectionError( f"Unsupported WebSocket opcode: {opcode}" ) #---------------------------------------------------------------------- def close(self) -> None: sock = self.socket if sock is None: return try: self._send_close(struct.pack("!H", 1000)) except (OSError, ValueError): pass finally: self.socket = None try: sock.close() except OSError: pass #---------------------------------------------------------------------- def _read_http_response(self) -> tuple[str, dict[str, str]]: data = bytearray() while b"\r\n\r\n" not in data: try: chunk = self._recv_socket(4096) except socket.timeout: continue if not chunk: raise ConnectionError( "EOF during WebSocket handshake" ) data.extend(chunk) if len(data) > 64 * 1024: raise ConnectionError( "WebSocket handshake response exceeds 64 KiB" ) header_bytes, remainder = data.split(b"\r\n\r\n", 1) # Preserve anything that happened to arrive after the HTTP headers. self._recv_buffer.extend(remainder) lines = header_bytes.decode("iso-8859-1").split("\r\n") if not lines or not lines[0]: raise ConnectionError( "Empty WebSocket handshake response" ) headers: dict[str, str] = {} for line in lines[1:]: if not line: continue if ":" not in line: raise ConnectionError( f"Malformed WebSocket HTTP header: {line!r}" ) name, value = line.split(":", 1) headers[name.strip().lower()] = value.strip() return lines[0], headers #---------------------------------------------------------------------- def _recv_frame(self) -> tuple[bool, int, bytes]: first_two = self._recv_exact(2) first = first_two[0] second = first_two[1] fin = bool(first & 0x80) if first & 0x70: raise ConnectionError( "WebSocket frame uses unsupported RSV bits" ) opcode = first & 0x0F masked = bool(second & 0x80) length = second & 0x7F if masked: raise ConnectionError( "Server sent a masked WebSocket frame" ) if length == 126: length = struct.unpack("!H", self._recv_exact(2))[0] elif length == 127: length = struct.unpack("!Q", self._recv_exact(8))[0] if length & (1 << 63): raise ConnectionError( "Invalid WebSocket 64-bit frame length" ) # Control frames must be final and <= 125 bytes. if opcode >= 0x8: if not fin: raise ConnectionError( "Fragmented WebSocket control frame" ) if length > 125: raise ConnectionError( "WebSocket control frame exceeds " "125-byte limit" ) # Continuation frames require an active fragmented message. if opcode == 0x0 and self._fragment_opcode is None: raise ConnectionError( "Unexpected WebSocket continuation frame" ) # Reject undefined control opcodes. if opcode >= 0x8 and opcode not in ( 0x8, # Close 0x9, # Ping 0xA, # Pong ): raise ConnectionError( f"Unknown WebSocket control opcode: {opcode}" ) payload = self._recv_exact(length) return fin, opcode, payload #---------------------------------------------------------------------- def _recv_exact(self, size: int) -> bytes: while len(self._recv_buffer) < size: chunk = self._recv_socket( max(4096, size - len(self._recv_buffer)) ) if not chunk: raise ConnectionError("WebSocket EOF") self._recv_buffer.extend(chunk) result = bytes(self._recv_buffer[:size]) del self._recv_buffer[:size] return result #---------------------------------------------------------------------- def _recv_socket(self, size: int) -> bytes: if self.socket is None: raise ConnectionError("WebSocket is not connected") return self.socket.recv(size) #---------------------------------------------------------------------- def _send_close(self, payload: bytes = b"") -> None: if len(payload) == 1: raise ValueError( "WebSocket close payload cannot have length 1" ) if len(payload) > 125: raise ValueError( "WebSocket close payload exceeds " "125-byte control-frame limit" ) try: self._send_frame(0x8, payload) except OSError: pass # The peer may already have closed the TCP connection. #---------------------------------------------------------------------- def _send_frame(self, opcode: int, payload: bytes = b"") -> None: if self.socket is None: raise ConnectionError("WebSocket is not connected") if opcode >= 0x8 and len(payload) > 125: raise ValueError( "WebSocket control frame payload " "cannot exceed 125 bytes" ) # RFC 6455 requires client-to-server frames to be masked. mask = os.urandom(4) length = len(payload) if length < 126: header = struct.pack("!BB", 0x80|opcode, 0x80|length) elif length <= 0xFFFF: header = struct.pack("!BBH", 0x80|opcode, 0x80|126, length) else: header = struct.pack("!BBQ", 0x80|opcode, 0x80|127, length) masked_payload = bytes( value ^ mask[index % 4] for index, value in enumerate(payload) ) self.socket.sendall(header + mask + masked_payload) # ========================================================================= class TwitchEventPubClient: """ Twitch's EventSub WebSocket is a text-data/server-to-client connection. The client only needs to: - receive text/binary/continuation frames - respond to Ping with Pong - send the WebSocket Close frame """ # BRB Timer for Chat by beporter@users.sourceforge.net # https://dev.twitch.tv/console/apps/ja5swzyzsr1euwm0e53h1sxqhk553l APP_CLIENT_ID = "ja5swzyzsr1euwm0e53h1sxqhk553l" KICKOFF_URL: str = 'https://beporter.github.io/brb-timer/start.html' ID_URL: str = 'https://id.twitch.tv' API_URL: str = 'https://api.twitch.tv' EVENTSUB_URL: str = 'wss://eventsub.wss.twitch.tv/ws' # https://dev.twitch.tv/docs/api/reference/#send-chat-message OAUTH_SCOPES: list[str] = [ 'user:read:chat', 'user:write:chat', 'user:bot', ] RECONNECT_DELAY: int = 5 CONNECT_TIMEOUT: int = 10 HTTP_TIMEOUT: int = 10 #---------------------------------------------------------------------- def __init__( self, oauth: str, callback: ChatMsgCallback, ): self.oauth: str = oauth.removeprefix("oauth:") self.callback: Callable = callback self.socket: socket.socket = None self.send_lock: threading.Lock = threading.Lock() self.running: bool = False self.thread: threading.Thread = None self.stop_event: threading.Event = threading.Event() self.client_id: str = self.APP_CLIENT_ID self.expires_at: int = 0 self.username: str = '' self.bot_user_id: int = None # ID of bot connecting to channel. self.channel_user_id: int = None # ID of channel's owner. self.keepalive_timeout: int = 10 self.last_sent: str = '' #---------------------------------------------------------------------- def start(self) -> None: if self.running: return self.running = True self.stop_event.clear() self.thread = threading.Thread( target=self._receive_loop_thread, daemon=True, name=f"{SCRIPT_NAME}-TwitchEventSub", ) self.thread.start() #---------------------------------------------------------------------- def close(self) -> None: """ Permanently close the client. The receiver thread owns the WebSocket and is responsible for closing it. """ self.stop() if self.thread is not None and self.thread is not threading.current_thread(): self.thread.join(timeout=2) if self.thread.is_alive(): OBS.error(f"Failed joining Twitch EventSub thread.") raise self.thread = None OBS.info("Twitch EventSub client shut down.") #---------------------------------------------------------------------- def stop(self) -> None: self.running = False self.stop_event.set() #---------------------------------------------------------------------- def send_chat(self, message: str) -> None: """ Send a chat message through Twitch Helix. This is the EventSub/API equivalent of IRC PRIVMSG. Ref: https://dev.twitch.tv/docs/api/reference/#send-chat-message """ with self.send_lock: if not self.client_id or not self.bot_user_id: raise ConnectionError( "Twitch client is not connected" ) if not self.channel_user_id: raise ConnectionError( "Twitch channel identity is unavailable" ) resp = self._api_request( "helix/chat/messages", body={ "broadcaster_id": self.channel_user_id, # Broadcaster's channel "sender_id": self.bot_user_id, # Who you're posting as "message": message, }, ) if not resp: raise ConnectionError( "Twitch Send Chat Message failed" ) data = resp.get("data", [])[0] if not data: raise ConnectionError( f"Twitch returned no chat result." ) if not data.get("is_sent"): raise ConnectionError( "Twitch rejected chat message: " f"{data.get('drop_reason')!r}" ) self.last_sent = message #---------------------------------------------------------------------- def _receive_loop_thread(self) -> None: while self.running: try: self._connect() self._read_loop() except ConnectionError as e: OBS.error(f"Connection error: {e!r}") pass finally: self._close_socket() if self.running: # Pause if we're looping. self._wait_for_reconnect() #---------------------------------------------------------------------- def _connect(self) -> None: """ Create a fresh EventSub WebSocket session. """ if not self._initialize_identity(): self.running = False return self.socket = _WebSocket(self.EVENTSUB_URL, timeout=self.CONNECT_TIMEOUT) self.socket.connect() raw = self.socket.recv_message() if raw is None: raise ConnectionError("EventSub closed during welcome") if not isinstance(raw, str): raise ConnectionError("EventSub welcome was not a text frame") try: welcome = json.loads(raw) except json.JSONDecodeError as e: raise ConnectionError(f"Invalid EventSub welcome: {e}") metadata = welcome.get("metadata", {}) if metadata.get("message_type") != "session_welcome": raise ConnectionError( "Expected EventSub session_welcome, got " f"{metadata.get('message_type')!r}" ) session = (welcome.get("payload", {}).get("session", {})) session_id = session.get("id") if not session_id: raise ConnectionError( "EventSub welcome did not contain a session ID" ) self.keepalive_timeout = int( session.get("keepalive_timeout_seconds", 10) ) self._create_chat_subscription(session_id) OBS.info( "Twitch EventSub connected " f"(session={session_id!r})." ) #---------------------------------------------------------------------- def _initialize_identity(self) -> None: """ Check the passed oauth token to determine if it's still valid, and who it belongs to. Ref: https://dev.twitch.tv/docs/authentication/validate-tokens#how-to-validate-a-token """ resp = self._api_request('oauth2/validate') if not resp: OBS.error("oauth2/validate failed.") return False if not resp['login']: OBS.error("OAuth token is not attached to a user.") return False if not set(self.OAUTH_SCOPES).issubset(resp['scopes']): missing_scopes = ", ".join(set(self.OAUTH_SCOPES) - set(resp['scopes'])) OBS.error( "OAuth token is lacking necessary scopes: " f"{missing_scopes!s}" ) return False self.user = resp.get('login', '') self.client_id = resp.get('client_id', self.APP_CLIENT_ID) self.expires_at = int(time.time()) + int(resp.get('expires_in', 0)) # int seconds self.username = resp.get('login', '') self.channel_user_id = resp.get('user_id', '') # Channel is always the streamer's id... self.bot_user_id = self.channel_user_id # ...posting as themselves. if ( not self.client_id or not self.bot_user_id or not self.channel_user_id ): OBS.error("Twitch OAuth validation did not return client_id/user_id") return False return True #---------------------------------------------------------------------- def _create_chat_subscription( self, session_id: str, ) -> None: """ Subscribe the EventSub WebSocket session to chat messages. Twitch requires the bot's user ID in the user_id condition. """ body = { "type": "channel.chat.message", "version": "1", "condition": { "broadcaster_user_id": self.channel_user_id, # Broadcaster's channel "user_id": self.bot_user_id, # Who you're listening as }, "transport": { "method": "websocket", "session_id": session_id, }, } resp = self._api_request('helix/eventsub/subscriptions', body=body) if not resp: raise ConnectionError( "Failed to create Twitch EventSub subscription." ) #---------------------------------------------------------------------- def _read_loop(self) -> None: if self.socket is None: raise ConnectionError("EventSub WebSocket is not connected") while self.running: try: raw = self.socket.recv_message() except socket.timeout: # The 1-second socket timeout exists solely to make stop() # responsive. It is not a connection failure. if not self.running: return continue if not self.running: return # Ends thread excution. if not isinstance(raw, str): continue try: message = json.loads(raw) except json.JSONDecodeError as e: OBS.debug(f"Ignoring invalid EventSub JSON: {e!r}") continue message_type = message.get("metadata", {}).get("message_type") if message_type == "notification": self._handle_notification(message) elif message_type == "session_keepalive": continue elif message_type == "session_reconnect": self._handle_reconnect(message) return elif message_type == "revocation": self._handle_revocation(message) return #---------------------------------------------------------------------- def _handle_notification(self, message: dict) -> None: payload = message.get('payload', {}) subscription = payload.get('subscription', {}) if subscription.get('type') != 'channel.chat.message': return chat_message = self._parse_eventsub_message(payload.get('event', {})) if chat_message is not None: self.callback(chat_message) #---------------------------------------------------------------------- def _parse_eventsub_message(self, event: dict) -> ChatMessage | None: username = event.get('chatter_user_login', '') text = event.get('message', {}).get('text') badges = event.get('badges', []) if not username or text is None: return None return ChatMessage( username, event.get('chatter_user_name', username), text, any( badge.get('set_id') == 'moderator' and badge.get('id') == '1' for badge in badges ), any( badge.get('set_id') == 'broadcaster' and badge.get('id') == '1' for badge in badges ), ) #---------------------------------------------------------------------- def _handle_reconnect(self, message: dict) -> None: """ Twitch specifically requires us to keep the old connection alive until the new connection has delivered its welcome message. """ session = message.get('payload', {}).get('session', {}) reconnect_url = session.get('reconnect_url') if not reconnect_url: raise ConnectionError( "EventSub reconnect message has no reconnect URL" ) OBS.info("Twitch requested EventSub reconnect.") old_ws = self.socket new_ws = _WebSocket(reconnect_url, timeout=self.CONNECT_TIMEOUT) try: new_ws.connect() raw = new_ws.recv_message() if not isinstance(raw, str): raise ConnectionError( "EventSub reconnect did not return " "a text welcome message" ) welcome = json.loads(raw) message_type = welcome.get('metadata', {}).get('message_type') if (message_type != 'session_welcome'): raise ConnectionError( "Invalid EventSub reconnect welcome" ) new_session = welcome.get('payload', {}).get('session', {}) self.keepalive_timeout = int( new_session.get( "keepalive_timeout_seconds", self.keepalive_timeout, ) ) # Only switch after the new socket has successfully completed # the WebSocket handshake AND received its EventSub welcome. self.socket = new_ws # Now that the new connection has inherited the subscriptions, # the old socket can be closed. if old_ws is not None: old_ws.close() OBS.debug("Twitch EventSub reconnect completed.") except ( OSError, BlockingIOError, ConnectionError, TimeoutError, ) as e: OBS.error(f"reconnect failure: {e!r}") new_ws.close() #---------------------------------------------------------------------- def _handle_revocation(self, message: dict) -> None: subscription = (message.get("payload", {}).get("subscription", {})) OBS.error( "Twitch EventSub subscription revoked: " f"type={subscription.get('type')!r}, " f"status={subscription.get('status')!r}" ) # Don't reconnect after a revoked authorization. self.running = False self.stop_event.set() #---------------------------------------------------------------------- def _wait_for_reconnect(self) -> None: self.stop_event.wait(self.RECONNECT_DELAY) #---------------------------------------------------------------------- def _close_socket(self) -> None: """ Close the network connection. Called by the receiver thread. """ ws = self.socket self.socket = None if ws is not None: try: ws.close() except OSError: pass #---------------------------------------------------------------------- def _api_request( self, path: str, body: dict|None = None, params: dict[str, str|int] = {}, extra_headers: dict[str, str|int] = {}, ) -> object|False: data = None if body is not None: data = json.dumps(body).encode("utf-8") extra_headers["Content-Type"] = "application/json" req = request.Request( f"{self._server(path)}/{path}?" + parse.urlencode(params), method='POST' if body is not None else 'GET', headers=self._headers(extra_headers), data=data, ) try: with request.urlopen(req, timeout=self.HTTP_TIMEOUT) as r: return json.load(r) except json.JSONDecodeError: detail = f"Invalid JSON payload: {r!r}" except UnicodeDecodeError: detail = f"Non-unicode payload: {r!r}" except error.HTTPError as e: match e.code: case http.HTTPStatus.UNAUTHORIZED: try: detail = json.load(e.fp) except json.JSONDecodeError: detail = "Unauthorized. (Invalid JSON payload.)" except UnicodeDecodeError: detail = "Unauthorized. (Non-unicode payload.)" # Ref: https://dev.twitch.tv/docs/api/guide/#twitch-rate-limits case http.HTTPStatus.TOO_MANY_REQUESTS: total = e.headers.get('Ratelimit-Limit') remaining = e.headers.get('Ratelimit-Remaining') reset = e.headers.get('Ratelimit-Reset') detail = ( f"Request for {e.url} was rate limited. " f"({remaining}/{total} requests remaining. " f"Reset in {reset!s} secs.)" ) case _: detail = e.reason OBS.error(f"Twitch API request failure: {detail}") return False #---------------------------------------------------------------------- def _server(self, path: str) -> str: match path.split('/', 2)[0]: case 'oauth2': return self.ID_URL case _: return self.API_URL #---------------------------------------------------------------------- def _headers( self, extra: dict[str, str|int] = {}, ) -> dict[str, str|int]: if ( 'client-id' not in map(str.lower, extra.keys()) and len(self.client_id) > 0 ): extra['Client-ID'] = self.client_id return { 'User-Agent': f"{SCRIPT_NAME} v{SCRIPT_VERSION}", 'Accept': 'application/json', 'Authorization': f"Bearer {self.oauth}", **extra, } ########################################################################### # Events class # ========================================================================= class Events: SHOW_TRANSITION_TYPE = 'slide_transition' HIDE_TRANSITION_TYPE = 'slide_transition' SHOW_TRANSITION_DIR = 'left' HIDE_TRANSITION_DIR = 'right' TRANSITION_DURATION_MS = 300 source_name: str = DEFAULTS.TEXT_TIMER running: bool = False start_time: int = 0 stop_time: int = 0 auto_hide_secs: int = DEFAULTS.AUTO_HIDE_SECS hide_time: int = 0 guesses: dict[str, int] = {} token: str = '' #---------------------------------------------------------------------- def on_event(self, event): if event != S.OBS_FRONTEND_EVENT_FINISHED_LOADING: return if not OBS.source_exists(self.source_name): self.create_source(self.source_name, '--:--') # Make sure the timer is hidden on startup. OBS.sceneitem_set_visible_by_name(self.source_name, False) OBS.debug(f"on_event complete") #---------------------------------------------------------------------- def create_source( self, source_name: str, text: str = "hello world", ) -> bool: """ If we were provided a source_name that doesn't exist in the active scene, create one and add it into the active scene using some defaults that the OBS user can subsequently tweak. Ref: https://github.com/upgradeQ/Streaming-Software-Scripting-Reference/blob/b876ee8e5a57/src/add_nested.py#L16 Ref: https://github.com/obsproject/obs-studio/blob/32.0.4/plugins/text-freetype2/text-freetype2.c#L170 """ if OBS.source_exists(source_name): OBS.info( f"Source '{source_name}' already exists in current scene." "Nothing created." ) return False # Create the text source for the timer display with OBS.source_create_text(source_name, text) as source, \ OBS.scene_current() as scene: OBS.source_save(source) if scene is None: OBS.error("Current frontend source is not an OBS scene.") return False # Add source to scene. # Ref: https://github.com/obsproject/obs-studio/blob/32.0.4/libobs/obs.h#L153 with OBS.scene_add(scene, source) as sceneitem: if sceneitem is None: OBS.error( f"Could not add source '{source_name}' to current scene.", ) return False # Set starting position. OBS.sceneitem_position_set( sceneitem, # These constants don't work for some reason. #S.OBS_ALIGN_RIGHT | S.OBS_ALIGN_TOP, ((1 << 1) | (1 << 2)) ) # Add transitions to scene item. with OBS.transition_source_create( DEFAULTS.TRANSITION_SHOW, self.SHOW_TRANSITION_TYPE, 'show', self.SHOW_TRANSITION_DIR, ) as show_trans: OBS.source_save(show_trans) OBS.sceneitem_add_transition( sceneitem, show_trans, 'show', self.TRANSITION_DURATION_MS, ) with OBS.transition_source_create( DEFAULTS.TRANSITION_HIDE, self.HIDE_TRANSITION_TYPE, 'hide', self.HIDE_TRANSITION_DIR, ) as hide_trans: OBS.source_save(hide_trans) OBS.sceneitem_add_transition( sceneitem, hide_trans, 'hide', self.TRANSITION_DURATION_MS, ) OBS.info( f"Created new text source named '{source_name}' in current scene." ) return True #---------------------------------------------------------------------- def on_chat(self, message: ChatMessage) -> None: """ Invoked by the IRC client whenever a chat message is received. This router is only responsible for determining whether to respond to an event and routing it to a handler, or ignore it. """ global chat_client # Don't respond to our own messages. if message.message == chat_client.last_sent: return match message.message.strip().split()[0]: case COMMANDS.BRB: self.command_brb(message) case COMMANDS.AT: self.command_at(message) case COMMANDS.BACK: self.command_back(message) case _: OBS.debug('No BRB commands matched. Skipping.') #---------------------------------------------------------------------- def ticker(self): """ Updates the text source contents. Scheduled as a per-second timer. Can only access running python state, not OBS settings or props for this script. Can NOT be scheduled via obs.timer_add() from a separate python thread (such as the irc client). """ if self.hide_time > 0 and int(time.time()) > self.hide_time: OBS.info('Timer hide_time reached, hiding on-screen text.') OBS.sceneitem_set_visible_by_name(self.source_name, False) self.hide_time = 0 return if not self.running: return OBS.source_set_text_by_name(self.source_name, self.ticker_text()) #---------------------------------------------------------------------- def ticker_text(self) -> str: if self.running: diff_secs: int = int(time.time()) - self.start_time else: diff_secs: int = self.stop_time - self.start_time dt: datetime.datetime = datetime.datetime.fromtimestamp( diff_secs, datetime.timezone.utc, ) return dt.strftime('%M:%S') #---------------------------------------------------------------------- def guess_start(self) -> None: if self.running: return if self.source_name: self.running = True self.start_time = int(time.time()) self.stop_time = 0 self.hide_time = 0 # (Timer is already running, so just show it.) OBS.sceneitem_set_visible_by_name(self.source_name, True) #---------------------------------------------------------------------- def has_guess(self, username: str) -> int | False: if username in self.guesses: return self.guesses[username] return False #---------------------------------------------------------------------- def add_guess(self, username: str, seconds: int) -> bool: if not self.running: return False # Can't guess when brb isn't running. if self.has_guess(username) is not False: return False # User already has a guess registered. self.guesses[username] = seconds return True # Guess added. #---------------------------------------------------------------------- def guess_end(self, auto_hide_secs: int) -> None: # Handle a stream ending with no !brb's having been run. if not self.running: return # Otherwise shut down cleanly. self.running = False self.stop_time = int(time.time()) self.hide_time = self.stop_time + auto_hide_secs #---------------------------------------------------------------------- def guess_winner(self) -> tuple[str, int] | tuple[False, None]: """ Returns false if no !brb has run, or is still running, or if nobody registered any guesses. Returns a tuple of (username, guessed_secs) when a winner is present. """ if self.running or len(self.guesses) == 0: return (False, None) # No winner when brb is still active or nobody guessed. qualified = self._qualified_guesses() if not qualified: return (False, None) winner_username, guessed_secs = next(reversed(qualified.items())) return (winner_username, guessed_secs) #---------------------------------------------------------------------- def _qualified_guesses(self) -> Dict[str, int]: # Exclude any guess larger than the actual seconds. This may # be an empty set. actual_secs = self.stop_time - self.start_time qualified = { username: seconds for username, seconds in self.guesses.items() if seconds <= actual_secs } # Sort the remaining guesses by number of seconds. The # largest (last) guess is the one closest to actual_secs # without going over. return dict( sorted( qualified.items(), key=lambda item: item[1], ) ) #---------------------------------------------------------------------- def send_chat(self, msg: str) -> None: """ Convenience/consistency wrapper for sending messages out through IRC. Ensures the client is started. """ global chat_client if chat_client is None or not chat_client.running: OBS.warn(f"Tried to send chat, but client is not connected. ({msg})") return chat_client.send_chat(msg) #---------------------------------------------------------------------- def command_brb(self, msg: ChatMessage) -> None: """ Handle a !brb command. """ global chat_client OBS.debug("Handling !brb.") # Validate command was sent by broadcaster or nod. if not (msg.is_mod or msg.is_broadcaster): OBS.info("Only mods can start brb.") self.send_chat(MESSAGES.ONLY_MOD_START) return if self.running: OBS.info("brb already running.") self.send_chat(MESSAGES.ALREADY_RUNNING) return self.guess_start() # Send the starting chat message. self.send_chat(MESSAGES.BRB_STARTED(streamer=chat_client.user)) OBS.debug("brb handled.") #---------------------------------------------------------------------- def command_at(self, msg: ChatMessage) -> None: """ Handle an !at command. """ OBS.debug("Handling !at.") if not self.running: OBS.info("No brb running.") self.send_chat(MESSAGES.NO_BRB) return [_cmd, time_str, *_rest] = msg.message.split() delta = DT.strpdelta(time_str) if not delta: OBS.debug(f"Failed to parse !at in message: {msg}") self.send_chat(MESSAGES.BAD_GUESS(user=msg.username)) return existing_guess = self.has_guess(msg.username) if existing_guess: OBS.info(f"User {msg.username} already has a guess registered: {existing_guess}") self.send_chat(MESSAGES.ALREADY_GUESSED( user=msg.username, guess=DT.strfdelta(existing_guess), )) return if self.add_guess(msg.username, delta): OBS.info(f"Guess registered for user {msg.username}: {delta}") self.send_chat(MESSAGES.GUESS_ACCEPTED( user=msg.username, guess=DT.strfdelta(delta), )) OBS.debug("!at handled.") #---------------------------------------------------------------------- def command_back(self, msg: ChatMessage) -> None: """ Handler that's called when chat_client returns a ChatMessage with a !back command. """ global chat_client OBS.debug("Handling !back.") # Validate command was sent by broadcaster or nod. if not (msg.is_mod or msg.is_broadcaster): OBS.info("Only mods can run !back.") self.send_chat(MESSAGES.ONLY_MOD_END) return if not self.running: OBS.info("No brb running.") self.send_chat(MESSAGES.NO_BRB) return # Disallow further guessing. if self.running: OBS.info("Stopping guesses.") self.guess_end(self.auto_hide_secs) # Announce the winner. (winner, guess_secs) = self.guess_winner() actual_secs = self.stop_time - self.start_time if winner: OBS.info(f"!brb winner is: {winner}.") self.send_chat(MESSAGES.BRB_FINISHED( streamer=chat_client.user, time=DT.strfdelta(actual_secs), winner=winner, diff=DT.strfdelta(actual_secs - guess_secs), )) else: OBS.info("No (valid) guesses, so no winner.") self.send_chat(MESSAGES.BRB_FINISHED_NO_GUESSES( streamer=chat_client.user, time=DT.strfdelta(actual_secs), )) OBS.info(f"Auto-hiding timer in {self.auto_hide_secs} seconds.") OBS.debug("!back handled.") ########################################################################### # OBS Wrapper # ========================================================================= type OBSTimer = Callable[[], None] class OBS: #---------------------------------------------------------------------- @classmethod def source_exists(self, source_name: str) -> bool: """ Returns True if a source with source_name is already present in the currently active scene. Ref: https://github.com/upgradeQ/Streaming-Software-Scripting-Reference/blob/b876ee8e5/src/toggle_sceneitem_vis.py#L9-L13 """ with self.scene_current() as scene: si = S.obs_scene_find_source(scene, source_name) # Don't release unless an explicit ref was saved. return (si is not None) #---------------------------------------------------------------------- @classmethod def source_set_text_by_name(self, source_name: str, text: str) -> None: source = S.obs_get_source_by_name(source_name) # Release! if source is not None: self.debug(f"updating text contents to {text}") settings = S.obs_data_create() # Release! S.obs_data_set_string(settings, "text", text) S.obs_source_update(source, settings) S.obs_source_release(source) S.obs_data_release(settings) #---------------------------------------------------------------------- @classmethod def source_save(self, source) -> bool: # source: obs_source_t """ Save the provided source to OBS's persistent storage. """ return S.obs_save_source(source) #---------------------------------------------------------------------- @classmethod @contextlib.contextmanager def source_create( self, source_type: str, source_name: str, settings = None, # obs_data_t ): # yield obs_source_t """ Yield a brand new source. """ with self.data_yield(settings) if settings is not None else self.data() as settings: try: source = S.obs_source_create( source_type, source_name, settings, None, ) if source is None: raise ValueError(f"Unable to create a new Source object.") yield source finally: if source is not None: S.obs_source_release(source) #---------------------------------------------------------------------- @classmethod @contextlib.contextmanager def source_create_text( self, source_name: str, text: str = "", source_type_id: str = None, ): # yield obs_source_t """ Create and yield the first text source type available in the current OBS build. """ # OBS uses different built-in text source IDs on different platforms. if source_type_id is None: if sys.platform.startswith("win"): source_types = [ "text_gdiplus_v2", "text_gdiplus", ] else: source_types = [ "text_ft2_source_v2", "text_ft2_source", ] else: source_types = [source_type_id] # Import default text settings. with self.text_settings(text, font_size_px=144) as settings: # Check if the source type exists before trying to create the source. for source_id in source_types: display_name = S.obs_source_get_display_name(source_id) if display_name is None: continue # Return the first successfully created text source. with self.source_create(source_id, source_name, settings) as source: if source is not None: self.debug( f"Using text source type '%s' (%s)." % (display_name, source_id), ) yield source return raise ValueError(f"Couldn't find an available text source type.") #---------------------------------------------------------------------- @classmethod def sceneitem_set_visible_by_name(self, source_name: str, visible: bool) -> bool: with self.scene_current() as scene: si = S.obs_scene_find_source(scene, source_name) # Don't release unless an explicit ref was saved. if si is not None: if S.obs_sceneitem_visible(si) is not visible: self.debug(f"{'showing' if visible else 'hiding'} sceneitem {source_name}.") S.obs_sceneitem_set_visible(si, visible) #S.obs_sceneitem_release(si) # DON'T release. #---------------------------------------------------------------------- @classmethod def sceneitem_add_transition( self, sceneitem, # obs_sceneitem_t transition_source, # obs_source_t visibility: str, duration: int = 300, ) -> None: # Input validation. if visibility not in ("show", "hide"): raise ValueError("visibility must be 'show' or 'hide'") # Attach the transition to the scene item. S.obs_sceneitem_set_transition( sceneitem, visibility == "show", transition_source, ) # Set transition duration. S.obs_sceneitem_set_transition_duration( sceneitem, visibility == "show", duration, ) #---------------------------------------------------------------------- @classmethod @contextlib.contextmanager def scene_current(self): # yield obs_scene_t """ Yields an obs_scene object and auto-releases afterward. Usage example: with OBS.scene_current() as scene: # do something with `scene`. """ try: # Must be released with obs_source_release() scene_source = S.obs_frontend_get_current_scene() # Does not need to be released! scene = S.obs_scene_from_source(scene_source) yield scene finally: if scene_source is not None: S.obs_source_release(scene_source) #---------------------------------------------------------------------- @classmethod @contextlib.contextmanager def scene_add( self, scene, # obs_scene_t source, # obs_source_t ): # yield obs_sceneitem_t """ Yields a sceneitem object resulting from adding the provided source to the provided scene. Uses the currently active scene if none is provided. """ try: sceneitem = S.obs_scene_add(scene, source) if sceneitem is None: raise ValueError( "Unable to create a sceneitem for source." ) yield sceneitem finally: if sceneitem is not None: pass #S.obs_sceneitem_release(sceneitem) #---------------------------------------------------------------------- @classmethod def sceneitem_position_set( self, sceneitem, # obs_sceneitem_t alignment: int, # OBS_ALIGN_* pos_x: int = None, pos_y: int = 0, z_index: int = S.OBS_ORDER_MOVE_TOP, # OBS_ORDER_* ) -> None: # Set on-screen alignment. S.obs_sceneitem_set_alignment(sceneitem, alignment) # Set on-screen position. if pos_x is None: vi = S.obs_video_info() S.obs_get_video_info(vi) pos_x = vi.base_width S.obs_sceneitem_set_pos(sceneitem, self.vec2(pos_x, pos_y)) S.obs_sceneitem_set_order(sceneitem, z_index) # Set item bounds. # S.obs_sceneitem_set_bounds_type(sceneitem, S.OBS_BOUNDS_SCALE_TO_WIDTH) # S.obs_sceneitem_set_bounds_alignment(sceneitem, S.OBS_ALIGN_CENTER) # S.obs_sceneitem_set_bounds(sceneitem, S.vec2(400, 100)) #---------------------------------------------------------------------- @classmethod @contextlib.contextmanager def transition_source_create( self, source_name: str, transition_type: str, visibility: str, direction: str, ): """ Creates a new transition Source object with the provided settings. Returns False if any part of the process fails. """ # Input validation. if visibility not in ("show", "hide"): raise ValueError("visibility must be 'show' or 'hide'") if direction not in ("left", "right"): raise ValueError("direction must be 'left' or 'right'") try: with self.data() as settings: S.obs_data_set_string(settings, "direction", direction) transition = S.obs_source_create( transition_type, source_name, settings, None, ) if transition is None: raise ValueError( f"Failed to create {visibility} {transition_type} source." ) yield transition finally: if transition is not None: S.obs_source_release(transition) #---------------------------------------------------------------------- @classmethod @contextlib.contextmanager def text_settings( self, text: str, width: int = None, wrap: bool = False, outline: bool = True, drop_shadow: bool = False, color_top: int = 0xffffffff, # Format: 0xrrggbbaa color_bottom: int = 0xffffffff, font_face: str = "Monaco", font_style: str = "Regular", font_size_px: int = 96, ): # yield obs_data_t """ Helper that yields an obs_data object pre-configured with our stock text Source settings. """ with self.data() as settings: S.obs_data_set_string(settings, 'text', text) S.obs_data_set_bool(settings, 'word_wrap', wrap) if width is not None: S.obs_data_set_int(settings, 'custom_width', width) S.obs_data_set_bool(settings, 'outline', outline) S.obs_data_set_bool(settings, 'drop_shadow', drop_shadow) # Ref: https://docs.obsproject.com/reference-properties#c.obs_properties_add_color_alpha S.obs_data_set_int(settings, 'color1', color_top) S.obs_data_set_int(settings, 'color2', color_bottom) with self.data() as font: S.obs_data_set_string(font, 'face', font_face) S.obs_data_set_string(font, 'style', font_style) S.obs_data_set_int(font, 'size', font_size_px) S.obs_data_set_int(font, 'flags', 0) S.obs_data_set_obj(settings, 'font', font) yield settings #---------------------------------------------------------------------- @classmethod @contextlib.contextmanager def data( self, source_settings = None, # obs_data_t ): # yield obs_data_t """ Helper that yields an obs_data object either from the passed settings, or from scratch. Auto-releases after use. Usage example: with OBS.data() as d: # do something with `d`. """ try: if source_settings: data = S.obs_source_get_settings(source_settings) else: data = S.obs_data_create() if data is None: raise ValueError( f"Could not create a new data object." ) yield data finally: if data is not None: S.obs_data_release(data) #---------------------------------------------------------------------- @classmethod @contextlib.contextmanager def data_yield( self, data, # obs_data_t ): # yield obs_data_t """ Yield the passed data object. Does NOT release it. """ try: yield data finally: pass #---------------------------------------------------------------------- @classmethod def vec2(self, x: int, y: int): # -> obs_vec2_t v = S.vec2() v.x = x v.y = y return v #---------------------------------------------------------------------- @classmethod def error(self, msg: str) -> None: self._log(msg, S.LOG_ERROR) #---------------------------------------------------------------------- @classmethod def warn(self, msg: str) -> None: self._log(msg, S.LOG_WARNING) #---------------------------------------------------------------------- @classmethod def info(self, msg: str) -> None: self._log(msg, S.LOG_INFO) #---------------------------------------------------------------------- @classmethod def debug(self, msg: str) -> None: self._log(msg, S.LOG_DEBUG) #---------------------------------------------------------------------- @classmethod def _log(self, msg: str, level: int) -> None: # Grab our frame and all parents we don't want to exclude. stack = [ f for f in inspect.stack(0) if f.frame.f_code.co_qualname not in [ 'dispatch..handle', ] ] # The last element in the pre-filtered slice of (up to) 3 is the # farthest-most caller. This handles the case where there's less # than 2 stacks above us. callers_caller_frameinfo = stack[0:3][-1] # Dig into the frame details to get the fully qualified method name. calling_method = callers_caller_frameinfo.frame.f_code.co_qualname # Only log if the message is at or below the configured limit. if self._should_log(level): S.script_log(level, f"[{calling_method}] {msg}") #---------------------------------------------------------------------- @classmethod def _log_level_map(self, level: int) -> int: """ Converts a standard python log level to an OBS logging level. Also accepts OBS levels and passes them through. """ mappings = { logging.CRITICAL: S.LOG_ERROR, logging.ERROR: S.LOG_ERROR, logging.WARNING: S.LOG_WARNING, logging.INFO: S.LOG_INFO, logging.DEBUG: S.LOG_DEBUG, logging.NOTSET: S.LOG_DEBUG, S.LOG_ERROR: S.LOG_ERROR, S.LOG_WARNING: S.LOG_WARNING, S.LOG_INFO: S.LOG_INFO, S.LOG_DEBUG: S.LOG_DEBUG, } return mappings[level] #---------------------------------------------------------------------- @classmethod @functools.cache def _should_log(self, level: int) -> bool: global LOG_LEVEL # Order matters! (least chatty --> most chatty) ordered_levels = (S.LOG_ERROR, S.LOG_WARNING, S.LOG_INFO, S.LOG_DEBUG) max_level = self._log_level_map(LOG_LEVEL) allowed_levels = ordered_levels[0:ordered_levels.index(max_level)+1] return level in allowed_levels ########################################################################### # Global State e: Events = Events() chat_client: TwitchEventPubClient = None #-------------------------------------------------------------------------- def token_expired() -> bool: global chat_client token_expires_at = getattr(chat_client, 'expires_at', 0) return token_expires_at > 0 and time.time() >= token_expires_at ########################################################################### # OBS Scripting API #-------------------------------------------------------------------------- def script_description() -> str: """ Describes the script in the OBS GUI. Uses Qt formatting with a subset of HTML. Ref: https://doc.qt.io/archives/qt-5.15/richtext-html-subset.html Ref: (for raw svg) https://dashboard.twitch.tv/ > (Nav Menu) > Moderation > Inspect icon Ref: (svg to data uri) https://codeshack.io/svg-to-data-uri-converter/ Ref: (color changer) https://www.svggenie.com/tools/svg-color-changer """ mod_svg = dedent(""" Broadcaster and mods only """.strip()) cell_style = 'style="background-color: #444444;"' return dedent(f"""

{SCRIPT_NAME} v{SCRIPT_VERSION}

Let chatters guess when the streamer will return from being AFK.

{mod_svg} {COMMANDS.BRB} Start on-screen timer and allow guessing.
  {COMMANDS.AT} MM:SS   Chatter registers their guess.
{mod_svg} {COMMANDS.BACK} Stop the timer and show the winner.

Originally written exclusively for Enns by beporter in August 2026.

""".strip()) #-------------------------------------------------------------------------- def script_defaults( settings, # obs_data_t ): # -> obs_data_t """ This lifecycle methods is called EARLY in the script's startup process. Before `script_properties()` is even called for the first time. """ S.obs_data_set_default_int( settings, 'auto_hide_secs', DEFAULTS.AUTO_HIDE_SECS, ) #-------------------------------------------------------------------------- def script_properties(): """ GUI controls to show in OBS. This is called pretty late in the startup process, so don't "modify" the controls here. (That has to happen in an `obs_property_set_modified_callback()` handler.) But the displayed controls should model the "first run" state. So if a button should be hidden until a dropdown entry is selected, hide the button in this method and let an eventual modified callback unhide it when conditions are met. """ OBS.debug('script_properties starting.') props = S.obs_properties_create() # Connect Twitch button b = S.obs_properties_add_button( props, 'twitch_connect_button', 'Connect Twitch', lambda: None, ) S.obs_property_button_set_type(b, S.OBS_BUTTON_URL) S.obs_property_button_set_url(b, TwitchEventPubClient.KICKOFF_URL) S.obs_property_set_long_description( b, ( 'Open a browser window to obtain ' 'a Twitch API token for this script to use.' ), ) # Twitch OAuth text input. t = S.obs_properties_add_text( props, 'twitch_token', 'Twitch OAuth Token', S.OBS_TEXT_PASSWORD, ) S.obs_property_set_long_description( t, 'Paste the Twitch auth token from the clicking the button above.', ) # Token expiration warning. i = S.obs_properties_add_text( props, 'twitch_creds_expired', 'Twitch token has expired. Please obtain a fresh token.', S.OBS_TEXT_INFO_WARNING, ) S.obs_property_text_set_info_word_wrap(i, True) S.obs_property_text_set_info_type(i, S.OBS_TEXT_INFO_NORMAL) S.obs_property_set_visible(i, token_expired()) # Auto-hide seconds. s = S.obs_properties_add_int( props, "auto_hide_secs", "Auto-hide Delay", 0, # min 60 * 30, # max (30 mins) 1, # step ) S.obs_property_int_set_suffix(s, ' secs') S.obs_property_set_long_description( s, ( 'Number of seconds until the script hides the stopped ' 'timer after the streamer returns from being AFK.' ), ) OBS.debug('script_properties complete.') return props #-------------------------------------------------------------------------- def script_load(settings): """ Run once when OBS first initializes this script. This is basically the __init__ for this script as it exists in OBS. Happens before the GUI is ready, so you can't query scenes or sources here. """ OBS.debug(f"script_load setting running = False.") e.running = False S.obs_frontend_add_event_callback(lambda ev: e.on_event(ev)) # Start the ticker from the main python thread. # (It will no-op unless a !brb is running.) S.timer_add(e.ticker, 1 * 1000) OBS.debug(f"script_load complete.") #-------------------------------------------------------------------------- def script_update(settings): """ Responsible for communicating the DATA behind the GUI props from obs to this script. OBS will already preserve these settings, but we need to get them into the script's runtime state for actual usage because the `settings` object isn't directly available anywhere except here and a few other places. The thing getting updated here is this very script, not the settings and not OBS. """ global chat_client OBS.debug(f"script_update starting.") new_token = S.obs_data_get_string(settings, 'twitch_token') if ( new_token is not None and len(new_token) > 0 and new_token != e.token ): e.token = new_token if chat_client is not None: OBS.debug('Closing old chat client.') chat_client.close() chat_client = None if ( len(e.token) > 0 and not token_expired() and chat_client is None ): OBS.debug('Starting new chat client.') chat_client = TwitchEventPubClient(e.token, e.on_chat) chat_client.start() new_auto_hide = S.obs_data_get_int(settings, 'auto_hide_secs') if new_auto_hide is not None: e.auto_hide_secs = new_auto_hide OBS.debug('script_update complete.') #-------------------------------------------------------------------------- def script_unload(): global chat_client OBS.debug('script_unload starting.') S.timer_remove(e.ticker) if chat_client is not None: chat_client.close() chat_client = None OBS.debug('script_unload complete.')