fixed pulse2osc and added pty testing scripts in conjunction with socat

This commit is contained in:
Stefan Kögl 2014-03-16 11:20:36 +01:00
parent 35fa90f8d6
commit 7548bf3f25
5 changed files with 70 additions and 44 deletions

View File

@ -33,6 +33,7 @@ except ImportError as e:
print(e) print(e)
from chaosc.osc_lib import OSCMessage from chaosc.osc_lib import OSCMessage
class Platform(object): class Platform(object):
def __init__(self, args): def __init__(self, args):
self.args = args self.args = args
@ -79,6 +80,6 @@ def init(name):
args = create_args(name) args = create_args(name)
platform = Platform(args) platform = Platform(args)
platform.connect() platform.connect()
atexit.register(platform.close)
return platform return platform

View File

@ -20,29 +20,36 @@
from __future__ import absolute_import from __future__ import absolute_import
import atexit from sensors2osc.common import *
import socket
from sensors2osc import common
atexit.register(common.close)
def main(): def main():
args, osc_sock = common.init("ekg2osc") platform = init("ekg2osc")
actor = args.actor actor = platform.args.actor
while 1: while 1:
try: try:
t = ord(common.serial_sock.read(1)) t = platform.serial_sock.read(1)
except socket.error, msg:
# got disconnected?
print "serial socket error!!!", msg
platform.reconnect()
try:
t = ord(t)
except TypeError, e:
continue
try:
print "got value", t print "got value", t
osc_message = OSCMessage("/%s/ekg" % actor) osc_message = OSCMessage("/%s/ekg" % actor)
osc_message.appendTypedArg(t, "i") osc_message.appendTypedArg(t, "i")
osc_sock.sendall(osc_message.encode_osc()) platform.osc_sock.sendall(osc_message.encode_osc())
except socket.error, msg: except socket.error, msg:
# got disconnected? print "cannot connect to chaosc"
print "lost connection!!!" continue
common.reconnect(args)
if __name__ == '__main__': if __name__ == '__main__':

View File

@ -20,9 +20,11 @@
from __future__ import absolute_import from __future__ import absolute_import
from datetime import datetime
from sensors2osc.common import * from sensors2osc.common import *
atexit.register(close)
class RingBuffer(object): class RingBuffer(object):
def __init__(self, length): def __init__(self, length):
@ -41,7 +43,7 @@ class RingBuffer(object):
def getData(self): def getData(self):
print "getData", self.ring_buf, self.head print "getData", self.ring_buf, self.head
data = list() data = list()
for i in range(7, 1, -1): for i in range(self.length + 1, 1, -1):
value = self.ring_buf[(self.head - i) % self.length] value = self.ring_buf[(self.head - i) % self.length]
if value == -1: if value == -1:
self.reset() self.reset()
@ -59,44 +61,60 @@ class RingBuffer(object):
def main(): def main():
args, osc_sock = init("pulse2osc") platform = init("pulse2osc")
actor = platform.args.actor
buf = RingBuffer(6) buf = RingBuffer(6)
heartbeat_on = False heartbeat_on = False
while 1: while 1:
try: try:
t = ord(serial_sock.read(1)) t = platform.serial_sock.read(1)
print "got value", t
buf.append(t)
if t == 0:
try:
heart_signal, heart_rate, o2, pulse = buf.getData()
if pulse == 245 and not heartbeat_on:
osc_message = OSCMessage("/%s/heartbeat" % actor)
osc_message.appendTypedArg(1, "i")
osc_message.appendTypedArg(heart_rate, "i")
osc_message.appendTypedArg(o2, "i")
osc_sock.sendall(osc_message.encode_osc())
print "heartbeat", datetime.datetime.now(), heart_signal
heartbeat_on = True
elif pulse == 1 and heartbeat_on:
#print "off heartbeat", datetime.datetime.now(), heart_signal
heartbeat_on = False
osc_message = OSCMessage("/%s/heartbeat" % actor)
osc_message.appendTypedArg(0, "i")
osc_message.appendTypedArg(heart_rate, "i")
osc_message.appendTypedArg(o2, "i")
osc_sock.sendall(osc_message.encode_osc())
except ValueError, e:
print e
except socket.error, msg: except socket.error, msg:
# got disconnected? # got disconnected?
print "lost connection!!!" print "serial socket error!!!", msg
reconnect(args) platform.reconnect()
try:
t = ord(t)
except TypeError, e:
continue
print "got value", t
buf.append(t)
if t == 0:
try:
heart_signal, heart_rate, o2, pulse = buf.getData()
except ValueError, e:
print e
continue
if pulse == 245 and not heartbeat_on:
heartbeat_on = True
try:
osc_message = OSCMessage("/%s/heartbeat" % actor)
osc_message.appendTypedArg(1, "i")
osc_message.appendTypedArg(heart_rate, "i")
osc_message.appendTypedArg(o2, "i")
platform.osc_sock.sendall(osc_message.encode_osc())
print "on heartbeat", datetime.now(), heart_signal, heart_rate, o2, pulse
except socket.error, msg:
print "cannot connect to chaosc"
continue
elif pulse == 1 and heartbeat_on:
print "off heartbeat", datetime.now(), heart_signal, heart_rate, o2, pulse
heartbeat_on = False
try:
osc_message = OSCMessage("/%s/heartbeat" % actor)
osc_message.appendTypedArg(0, "i")
osc_message.appendTypedArg(heart_rate, "i")
osc_message.appendTypedArg(o2, "i")
platform.osc_sock.sendall(osc_message.encode_osc())
except socket.error, msg:
print "cannot connect to chaosc"
continue
if __name__ == '__main__': if __name__ == '__main__':