1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
# deck-it - a minimal obs-websocket v5 client
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""A minimal obs-websocket v5 client.
OBS speaks WebSocket + JSON. On connect it sends Hello; if auth is required we
answer with Identify carrying the protocol's two-step SHA-256 proof, then wait
for Identified. Requests are JSON op 6 and their replies are matched by id.
"""
import base64
import hashlib
import json
import threading
from .module import ModuleError
def auth_string(password: str, salt: str, challenge: str) -> str:
secret = base64.b64encode(hashlib.sha256((password + salt).encode()).digest()).decode()
return base64.b64encode(hashlib.sha256((secret + challenge).encode()).digest()).decode()
class ObsClient:
def __init__(self, host="127.0.0.1", port=4455, password="", on_event=None):
self.host, self.port, self.password = host, port, password
self.on_event = on_event
self.ws = None
self.connected = False
self._lock = threading.Lock()
self._pending = {}
self._seq = 0
self._stop = threading.Event()
self._thread = None
def _recv_json(self) -> dict:
raw = self.ws.recv()
if not raw:
raise ModuleError("OBS closed the connection")
try:
return json.loads(raw)
except ValueError:
raise ModuleError("OBS sent a non-JSON reply")
def connect(self):
import websocket # only needed when actually talking to OBS
self.ws = websocket.create_connection(f"ws://{self.host}:{self.port}",
timeout=5, suppress_origin=True)
hello = self._recv_json()
auth = {}
if hello.get("d", {}).get("authentication"):
a = hello["d"]["authentication"]
auth["authentication"] = auth_string(self.password, a["salt"], a["challenge"])
self.ws.send(json.dumps({"op": 1, "d": {"rpcVersion": 1, **auth, "eventSubscriptions": 0xFFFFFFFF}}))
try:
identified = self._recv_json()
except ModuleError:
raise ModuleError("OBS rejected the password" if auth else "OBS did not identify")
if identified.get("op") != 2:
raise ModuleError("OBS did not identify")
self.connected = True
self._stop.clear()
self._thread = threading.Thread(target=self._recv_loop, daemon=True)
self._thread.start()
def _recv_loop(self):
import websocket
while not self._stop.is_set():
try:
msg = json.loads(self.ws.recv())
except websocket.WebSocketTimeoutException:
# An idle OBS is not a closed one; keep reading.
continue
except Exception:
break
op, d = msg.get("op"), msg.get("d", {})
if op == 5:
if self.on_event:
self.on_event(d.get("eventType"), d.get("eventData", {}))
elif op == 7:
req = self._pending.get(d.get("requestId"))
if req:
req[1] = d
req[0].set()
self.connected = False
def request(self, request_type: str, data: dict | None = None, timeout=5) -> dict:
if not self.ws or not self.connected:
raise ModuleError("OBS is not connected")
with self._lock:
self._seq += 1
rid = f"deck-{self._seq}"
event = threading.Event()
self._pending[rid] = [event, None]
try:
self.ws.send(json.dumps({"op": 6, "d": {"requestType": request_type, "requestId": rid,
"requestData": data or {}}}))
except Exception as e:
self._pending.pop(rid, None)
raise ModuleError(f"OBS send failed: {e}")
if not event.wait(timeout):
self._pending.pop(rid, None)
raise ModuleError("OBS did not answer")
entry = self._pending.pop(rid, None)
resp = (entry[1] if entry else None) or {}
if not resp.get("requestStatus", {}).get("result"):
raise ModuleError(f"OBS refused: {resp.get('requestStatus', {}).get('code')}")
return resp.get("responseData", {})
def close(self):
self._stop.set()
if self.ws:
try:
self.ws.close()
except Exception:
pass
self.ws = None
self.connected = False
|
