-
Notifications
You must be signed in to change notification settings - Fork 3
/
osc.py
120 lines (99 loc) · 3.44 KB
/
osc.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
#!/usr/bin/env python3
"""
Example:
import time
from osc import *
try:
address = sys.argv[1]
except:
address = "/hello/world"
try:
data = sys.argv[2]
except:
data = None
def message_handler(location, address, data):
print(location)
print(address)
print(data)
receiver = Receiver(23232, message_handler)
Sender(23232).send(address, data)
time.sleep(0.1)
"""
import sys, os, time, json, threading, queue
from .log import log
from .lib import OSC
verbose = True
class Sender(object):
# note: OSC.OSCMultiClient() does not appear to work, so writing the equivalent in wrappers here
def __init__(self, *location):
self.targets = {}
self.add_target(*location)
def add_target(self, *location):
location = self._get_host_port(*location)
if not location:
return
if verbose:
log.info("OSC adding target %s:%s" % location)
self.targets[location] = OSC.OSCClient()
self.targets[location].connect(location)
def remove_target(self, *location):
location = self._get_host_port(*location)
if not location or not location in self.targets:
return
if verbose:
log.info("OSC removing target %s:%s" % location)
del self.targets[location]
def send(self, address, data=None):
message = OSC.OSCMessage()
message.setAddress(address)
if data is not None:
if type(data) is dict:
data = json.dumps(data)
if type(data) is tuple:
data = list(data)
message.extend(data)
if verbose:
log.info("OSC send (%s): %s" % (len(self.targets), message))
for location, target in self.targets.items():
try:
target.send(message, timeout=1)
except Exception as e:
log.error("%s (%s)" % (log.exc(e), message))
def _get_host_port(self, *target):
host, port = None, None
assert len(target) <= 2
if len(target) == 1:
host, port = '127.0.0.1', target[0]
if len(target) == 2:
host, port = target
if host is not None and port is not None:
return host, port
return None
class Receiver(threading.Thread):
def __init__(self, port, message_handler=None, blocking=False):
threading.Thread.__init__(self)
self.daemon = True
self.message_handler = message_handler
self.location = '0.0.0.0', port
self.server = OSC.OSCServer(self.location)
if verbose:
log.info("Started OSC Receiver on port %s" % port)
self.start()
if blocking:
try:
while True:
time.sleep(5)
except (KeyboardInterrupt, SystemExit):
pass
def run(self):
self.server.addMsgHandler("default", self.default_handler)
while True:
try:
self.server.handle_request()
except Exception as e:
log.error(log.exc(e))
def default_handler(self, address, tags, data, location): # tags show data type ('i': integer, 's': string)
if verbose:
log.info("%s: %s %s" % (location[0], address, data))
if self.message_handler is not None:
self.message_handler(location[0], address, data)