lj-clitools/generators/osc2redis.py

132 lines
3.1 KiB
Python
Raw Normal View History

2020-11-11 16:31:08 +00:00
#!/usr/bin/python3
# -*- coding: utf-8 -*-
# -*- mode: Python -*-
'''
Forward pointlist to redis key
END POINT Format : (x,y,color)
/pl/0/0 "[(150.0, 230.0, 255), (170.0, 170.0, 255), (230.0, 170.0, 255), (210.0, 230.0, 255), (150.0, 230.0, 255)]"
v0.1.0
LICENCE : CC
by Cocoa, Sam Neurohack
'''
from OSC3 import OSCServer, OSCClient, OSCMessage
import sys
from time import sleep
import argparse
import ast
import redis
argsparser = argparse.ArgumentParser(description="osc2redis generator")
argsparser.add_argument("-i","--ip",help="IP to bind to (0.0.0.0 by default)",default="0.0.0.0",type=str)
argsparser.add_argument("-p","--port",help="OSC port to bind to (9002 by default)",default=9002,type=str)
argsparser.add_argument("-r","--rip",help="Redis server IP (127.0.0.1 by default)",default="127.0.0.1",type=str)
argsparser.add_argument("-o","--rout",help="Redis port (6379 by default)",default=6379,type=str)
argsparser.add_argument("-v","--verbose",action="store_true",help="Verbose output")
args = argsparser.parse_args()
verbose = args.verbose
ip = args.ip
port = int(args.port)
rip = args.rip
rport = int(args.rout)
r = redis.StrictRedis(host=rip, port=rport, db=0)
def debug(msg):
if( verbose == False ):
return
print(msg)
oscserver = OSCServer( (ip, port) )
oscserver.timeout = 0
run = True
# this method of reporting timeouts only works by convention
# that before calling handle_request() field .timed_out is
# set to False
def handle_timeout(self):
self.timed_out = True
# funny python's way to add a method to an instance of a class
import types
oscserver.handle_timeout = types.MethodType(handle_timeout, oscserver)
def validate(pointlist):
state = True
if len(pointlist)<9:
state = False
try:
pl = bytes(pointlist, 'ascii')
check = ast.literal_eval(pl.decode('ascii'))
except:
state = False
return state
# RAW OSC Frame available ?
def OSC_frame():
# clear timed_out flag
oscserver.timed_out = False
# handle all pending requests then return
while not oscserver.timed_out:
oscserver.handle_request()
# default handler
def OSChandler(oscpath, tags, args, source):
oscaddress = ''.join(oscpath.split("/"))
print("fromOSC Default OSC Handler got oscpath :", oscpath, "from :" + str(source[0]), "args :", args)
print(oscpath.find("/pl/"), len(oscpath))
if oscpath.find("/pl/") ==0 and len(args)==1:
print("correct OSC type :'/pl/")
if validate(args[0]) == True and len(oscpath) == 7:
print("new pl for key ", oscpath, ":", args[0])
if r.set(oscpath,args[0])==True:
debug("exports::redis set("+str(oscpath)+") to "+args[0])
else:
print("Bad pointlist -> msg trapped.")
else:
print("BAD OSC Message :", oscpath)
oscserver.addMsgHandler( "default", OSChandler )
# simulate a "game engine"
while run:
# do the game stuff:
sleep(0.01)
# call user script
OSC_frame()
oscserver.close()