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
|
"""OBS stream-deck: recording, scenes, audio channels, transitions and sources."""
import threading
from ..module import Module, ModuleError
from ..obsclient import ObsClient
AUDIO_KINDS = {
"pulse_input_capture", "pulse_output_capture", "alsa_input_capture", "alsa_output_capture",
"wasapi_input_capture", "wasapi_output_capture", "coreaudio_input_capture", "coreaudio_output_capture",
"sndio_input_capture", "sndio_output_capture", "jack_input_capture", "jack_output_capture",
}
def make(cfg: dict) -> Module:
o = cfg.get("obs", {})
client = cfg.get("client") or ObsClient(
host=o.get("host", "127.0.0.1"), port=o.get("port", 4455), password=o.get("password", ""))
lock = threading.Lock()
stop = threading.Event()
def req(kind, data=None):
with lock:
return client.request(kind, data)
def connect_loop():
while not stop.is_set():
if not getattr(client, "connected", False):
try:
client.connect()
except Exception:
pass
stop.wait(5)
def state():
if not getattr(client, "connected", False):
return {"connected": False}
try:
scenes = req("GetSceneList")
rec = req("GetRecordStatus")
stream = req("GetStreamStatus")
trans = req("GetCurrentSceneTransition")
kinds = req("GetTransitionKindList").get("transitionKinds", [])
current = scenes.get("currentProgramSceneName")
sources = []
if current:
for item in req("GetSceneItemList", {"sceneName": current}).get("sceneItems", []):
sources.append({"id": item["sceneItemId"], "name": item["sourceName"],
"on": item["sceneItemEnabled"]})
inputs = []
for inp in req("GetInputList").get("inputs", []):
if inp.get("inputKind") not in AUDIO_KINDS:
continue
try:
vol = req("GetInputVolume", {"inputName": inp["inputName"]})
muted = req("GetInputMute", {"inputName": inp["inputName"]})
except ModuleError:
continue
inputs.append({"name": inp["inputName"],
"volume": vol.get("inputVolumeMul", 0),
"muted": muted.get("inputMuted", False)})
except ModuleError:
return {"connected": False}
return {
"connected": True,
"scenes": [s["sceneName"] for s in scenes.get("scenes", [])],
"currentScene": current,
"recording": rec.get("outputActive", False),
"recordingTime": rec.get("outputDuration", 0) / 1000,
"streaming": stream.get("outputActive", False),
"transitions": kinds,
"transition": trans.get("transitionName"),
"sources": sources,
"inputs": inputs,
}
def scene(body):
name = body.get("scene")
if not isinstance(name, str) or not name:
raise ModuleError("scene required")
req("SetCurrentProgramScene", {"sceneName": name})
def record(body):
req("StartRecord" if body.get("on") else "StopRecord")
def stream(body):
req("StartStream" if body.get("on") else "StopStream")
def mute(body):
req("SetInputMute", {"inputName": str(body.get("input")), "inputMuted": bool(body.get("muted"))})
def volume(body):
req("SetInputVolume", {"inputName": str(body.get("input")),
"inputVolumeMul": float(body.get("level", 0))})
def transition(body):
req("SetCurrentSceneTransition", {"transitionName": str(body.get("name"))})
def trigger(body):
req("TriggerStudioModeTransition")
def source(body):
req("SetSceneItemEnabled", {"sceneName": str(body.get("scene")),
"sceneItemId": int(body.get("id")),
"sceneItemEnabled": bool(body.get("on"))})
return Module(
name="obs", title="OBS", state=state,
actions={"scene": scene, "record": record, "stream": stream, "mute": mute,
"volume": volume, "transition": transition, "trigger": trigger, "source": source},
start=lambda: threading.Thread(target=connect_loop, daemon=True).start(),
stop=lambda: (stop.set(), client.close()),
)
|
