forked from ideoforms/AbletonOSC
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmanager.py
125 lines (105 loc) · 4.54 KB
/
manager.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
from ableton.v2.control_surface import ControlSurface
from . import abletonosc
import importlib
import traceback
import logging
import os
logger = logging.getLogger("abletonosc")
class Manager(ControlSurface):
def __init__(self, c_instance):
ControlSurface.__init__(self, c_instance)
self.log_level = "info"
self.start_logging()
self.handlers = []
self.show_message("AbletonOSC: Listening for OSC on port %d" % abletonosc.OSC_LISTEN_PORT)
self.osc_server = abletonosc.OSCServer()
self.schedule_message(0, self.tick)
self.init_api()
class LiveOSCErrorLogHandler(logging.StreamHandler):
def emit(handler, record):
message = record.getMessage()
message = message[message.index(":") + 2:]
try:
self.osc_server.send("/live/error", (message,))
except OSError:
# If the connection is dead, silently ignore errors as there's not much more we can do
pass
self.live_osc_error_handler = LiveOSCErrorLogHandler()
self.live_osc_error_handler.setLevel(logging.ERROR)
logger.addHandler(self.live_osc_error_handler)
def start_logging(self):
module_path = os.path.dirname(os.path.realpath(__file__))
log_dir = os.path.join(module_path, "logs")
if not os.path.exists(log_dir):
os.mkdir(log_dir, 0o755)
log_path = os.path.join(log_dir, "abletonosc.log")
self.log_file_handler = logging.FileHandler(log_path)
self.log_file_handler.setLevel(self.log_level.upper())
formatter = logging.Formatter('(%(asctime)s) [%(levelname)s] %(message)s')
self.log_file_handler.setFormatter(formatter)
logger.addHandler(self.log_file_handler)
def init_api(self):
def test_callback(params):
self.show_message("Received OSC OK")
self.osc_server.send("/live/test", ("ok",))
def reload_callback(params):
self.reload_imports()
def get_log_level_callback(params):
return (self.log_level,)
def set_log_level_callback(params):
log_level = params[0]
assert log_level in ("debug", "info", "warning", "error", "critical")
self.log_level = log_level
self.log_file_handler.setLevel(self.log_level.upper())
self.osc_server.add_handler("/live/test", test_callback)
self.osc_server.add_handler("/live/api/reload", reload_callback)
self.osc_server.add_handler("/live/api/get/log_level", get_log_level_callback)
self.osc_server.add_handler("/live/api/set/log_level", set_log_level_callback)
with self.component_guard():
self.handlers = [
abletonosc.SongHandler(self),
abletonosc.ApplicationHandler(self),
abletonosc.ClipHandler(self),
abletonosc.ClipSlotHandler(self),
abletonosc.TrackHandler(self),
abletonosc.DeviceHandler(self),
abletonosc.ViewHandler(self)
]
def clear_api(self):
self.osc_server.clear_handlers()
for handler in self.handlers:
handler.clear_api()
def tick(self):
"""
Called once per 100ms "tick".
Live's embedded Python implementation does not appear to support threading,
and beachballs when a thread is started. Instead, this approach allows long-running
processes such as the OSC server to perform operations.
"""
logger.debug("Tick...")
self.osc_server.process()
self.schedule_message(1, self.tick)
def reload_imports(self):
try:
importlib.reload(abletonosc.application)
importlib.reload(abletonosc.clip)
importlib.reload(abletonosc.clip_slot)
importlib.reload(abletonosc.device)
importlib.reload(abletonosc.handler)
importlib.reload(abletonosc.osc_server)
importlib.reload(abletonosc.song)
importlib.reload(abletonosc.track)
importlib.reload(abletonosc.view)
importlib.reload(abletonosc)
except Exception as e:
exc = traceback.format_exc()
logging.warning(exc)
if self.handlers:
self.clear_api()
self.init_api()
logger.info("Reloaded code")
def disconnect(self):
self.show_message("Disconnecting...")
logger.info("Disconnecting...")
self.osc_server.shutdown()
super().disconnect()