This repository has been archived by the owner on Apr 5, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 36
/
main.py
268 lines (208 loc) · 9.8 KB
/
main.py
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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
from __future__ import division
from websocket import WebSocketApp
from tinydb import TinyDB, Query
from sys import exit, stdout
from os import path
import logging
import json
import mido
TEMPLATES = {
"ToggleSourceVisibility": """{
"request-type": "GetSceneItemProperties",
"message-id": "%d",
"item": "%s"
}""",
"ReloadBrowserSource": """{
"request-type": "GetSourceSettings",
"message-id": "%d",
"sourceName": "%s"
}"""
}
SCRIPT_DIR = path.dirname(path.realpath(__file__))
def map_scale(inp, ista, isto, osta, osto):
return osta + (osto - osta) * ((inp - ista) / (isto - ista))
def get_logger(name, level=logging.INFO):
log_format = logging.Formatter('[%(asctime)s] (%(levelname)s) %(message)s')
std_output = logging.StreamHandler(stdout)
std_output.setFormatter(log_format)
std_output.setLevel(level)
file_output = logging.FileHandler(path.join(SCRIPT_DIR, "debug.log"))
file_output.setFormatter(log_format)
file_output.setLevel(level)
logger = logging.getLogger(name)
logger.setLevel(logging.DEBUG)
logger.addHandler(file_output)
logger.addHandler(std_output)
return logger
class MidiHandler:
# Initializes the handler class
def __init__(self, config_path="config.json", ws_server="localhost", ws_port=4444):
# Setting up logging first and foremost
self.log = get_logger("midi_to_obs")
# Internal service variables
self._action_buffer = []
self._action_counter = 2
self.log.debug("Trying to load config file from %s" % config_path)
self.db = TinyDB(config_path, indent=4)
result = self.db.search(Query().type.exists())
if not result:
self.log.critical("Config file %s doesn't exist or is damaged" % config_path)
# ENOENT (No such file or directory)
exit(2)
self.log.info("Successfully parsed config file")
port_name = str(result[0]["value"])
self.log.debug("Retrieved MIDI port name `%s`" % port_name)
del result
try:
self.log.debug("Attempting to open midi port")
self.port = mido.open_input(name=port_name, callback=self.handle_midi_input)
except:
self.log.critical("The midi device %s is not connected or has a different name" % port_name)
self.log.critical("Please plug the device in or run setup.py again and restart this script")
# EIO 5 (Input/output error)
exit(5)
self.log.info("Successfully initialized midi port `%s`" % port_name)
del port_name
# Properly setting up a Websocket client
self.log.debug("Attempting to connect to OBS using websocket protocol")
self.obs_socket = WebSocketApp("ws://%s:%d" % (ws_server, ws_port))
self.obs_socket.on_message = self.handle_obs_message
self.obs_socket.on_error = self.handle_obs_error
self.obs_socket.on_close = self.handle_obs_close
self.obs_socket.on_open = self.handle_obs_open
def handle_midi_input(self, message):
self.log.debug("Received %s message from midi: %s" % (message.type, message))
if message.type == "note_on":
return self.handle_midi_button(message.type, message.note)
# `program_change` messages can be only used as regular buttons since
# they have no extra value, unlike faders (`control_change`)
if message.type == "program_change":
return self.handle_midi_button(message.type, message.program)
if message.type == "control_change":
return self.handle_midi_fader(message.control, message.value)
def handle_midi_button(self, type, note):
query = Query()
results = self.db.search((query.msg_type == type) & (query.msgNoC == note))
if not results:
self.log.debug("Cound not find action for note %s", note)
return
for result in results:
if self.send_action(result):
break
def handle_midi_fader(self, control, value):
query = Query()
results = self.db.search((query.msg_type == "control_change") & (query.msgNoC == control))
if not results:
self.log.debug("Cound not find action for fader %s", control)
return
for result in results:
input_type = result["input_type"]
action = result["action"]
if input_type == "button":
if value == 127 and not self.send_action(result):
continue
break
if input_type == "fader":
command = result["cmd"]
scaled = map_scale(value, 0, 127, result["scale_low"], result["scale_high"])
if command == "SetSourcePosition" or command == "SetSourceScale":
self.obs_socket.send(action % scaled)
break
# Super dirty hack but @AlexDash says that it works
# @TODO: find an explanation _why_ it works
if command == "SetVolume":
# Yes, this literally raises a float to a third degree
self.obs_socket.send(action % scaled**3)
break
if command == "SetSourceRotation" or command == "SetTransitionDuration" or command == "SetSyncOffset":
self.obs_socket.send(action % int(scaled))
break
def handle_obs_message(self, message):
self.log.debug("Received new message from OBS")
payload = json.loads(message)
self.log.debug("Successfully parsed new message from OBS")
if "error" in payload:
self.log.error("OBS returned error: %s" % payload["error"])
return
message_id = payload["message-id"]
self.log.debug("Looking for action with message id `%s`" % message_id)
for action in self._action_buffer:
(buffered_id, template, kind) = action
if buffered_id != int(payload["message-id"]):
continue
del buffered_id
self.log.info("Action `%s` was requested by OBS" % kind)
if kind == "ToggleSourceVisibility":
# Dear lain, I so miss decent ternary operators...
invisible = "false" if payload["visible"] else "true"
self.obs_socket.send(template % invisible)
elif kind == "ReloadBrowserSource":
source = payload["sourceSettings"]["url"]
target = source[0:-1] if source[-1] == '#' else source + '#'
self.obs_socket.send(template % target)
self.log.debug("Removing action with message id %s from buffer" % message_id)
self._action_buffer.remove(action)
break
def handle_obs_error(self, ws, error=None):
# Protection against potential inconsistencies in `inspect.ismethod`
if error is None and isinstance(ws, BaseException):
error = ws
if isinstance(error, (KeyboardInterrupt, SystemExit)):
self.log.info("Keyboard interrupt received, gracefully exiting...")
self.close(teardown=True)
else:
self.log.error("Websocket error: %" % str(error))
def handle_obs_close(self, ws):
self.log.error("OBS has disconnected, timed out or isn't running")
self.log.error("Please reopen OBS and restart the script")
def handle_obs_open(self, ws):
self.log.info("Successfully connected to OBS")
def send_action(self, action_request):
action = action_request.get("action")
if not action:
# @NOTE: this potentionally should never happen but you never know
self.log.error("No action supplied in current request")
return False
request = action_request.get("request")
if not request:
self.log.debug("No request body for action %s, sending action" % action)
self.obs_socket.send(action)
# Success, breaking the loop
return True
template = TEMPLATES.get(request)
if not template:
self.log.error("Missing template for request %s" % request)
# Keep searching
return False
target = action_request.get("target")
if not target:
self.log.error("Missing target in %s request for %s action" % (request, action))
# Keep searching
return False
self._action_buffer.append([self._action_counter, action, request])
self.obs_socket.send(template % (self._action_counter, target))
self._action_counter += 1
# Explicit return is necessary here to avoid extra searching
return True
def start(self):
self.log.info("Connecting to OBS...")
self.obs_socket.run_forever()
def close(self, teardown=False):
self.log.debug("Attempting to close midi port")
self.port.close()
self.log.info("Midi connection has been closed successfully")
# If close is requested during keyboard interrupt, let the websocket
# client tear itself down and make a clean exit
if not teardown:
self.log.debug("Attempting to close OBS connection")
self.obs_socket.close()
self.log.info("OBS connection has been closed successfully")
self.log.debug("Attempting to close TinyDB instance on config file")
self.db.close()
self.log.info("Config file has been successfully released")
def __end__(self):
self.log.info("Exiting script...")
self.close()
if __name__ == "__main__":
handler = MidiHandler()
handler.start()