#!/usr/bin/python3 # -*- coding: utf-8 -*- # -*- mode: Python -*- ''' redilysis v0.1.0 A complex effect that depends on redis keys for audio analysis see https://git.interhacker.space/teamlase/redilysis for more informations about the redilysis project LICENCE : CC by cocoa ''' from __future__ import print_function import argparse import ast import os import math import random import redis import sys import time name = "filters::redilysis" def debug(*args, **kwargs): if( verbose == False ): return print(*args, file=sys.stderr, **kwargs) def now(): return time.time() * 1000 # The list of available modes and the redis keys they need oModeList = { "rms_noise": ["rms"], "rms_size": ["rms"], "bpm_size": ["bpm"] } CHAOS = 1 REDIS_FREQ = 300 # General Args argsparser = argparse.ArgumentParser(description="Redilysis filter") argsparser.add_argument("-v","--verbose",action="store_true",help="Verbose") # Redis Args argsparser.add_argument("-i","--ip",help="IP address of the Redis server ",default="127.0.0.1",type=str) argsparser.add_argument("-p","--port",help="Port of the Redis server ",default="6379",type=str) argsparser.add_argument("-s","--redis-freq",help="Query Redis every x (in milliseconds). Default:{}".format(REDIS_FREQ),default=REDIS_FREQ,type=int) # General args argsparser.add_argument("-x","--centerX",help="geometrical center X position",default=400,type=int) argsparser.add_argument("-y","--centerY",help="geometrical center Y position",default=400,type=int) argsparser.add_argument("-f","--fps",help="Frame Per Second",default=30,type=int) # Modes And Common Modes Parameters argsparser.add_argument("-m","--modelist",required=True,help="Comma separated list of modes to use from: {}".format("i, ".join(oModeList.keys())),type=str) argsparser.add_argument("--chaos",help="How much disorder to bring. High value = More chaos. Default {}".format(CHAOS), default=CHAOS, type=str) args = argsparser.parse_args() ip = args.ip port = args.port redisFreq = args.redis_freq verbose = args.verbose fps = args.fps centerX = args.centerX centerY = args.centerY chaos = float(args.chaos) optimal_looptime = 1 / fps modeList = args.modelist.split(",") redisKeys = [] for mode in modeList: if not mode in oModeList: print("Mode '{}' is invalid. Exiting.".format(mode)) sys.exit(2) redisKeys += oModeList[mode] redisKeys = list(set(redisKeys)) debug(name,"Redis Keys:{}".format(redisKeys)) redisData = {} redisLastHit = now() - redisFreq r = redis.Redis( host=ip, port=port) # Records the last bpm last_bpm = time.time() def gauss(x, mu, sigma): return( math.exp(-math.pow((x-mu),2)/(2*math.pow(sigma,2))/math.sqrt(2*math.pi*math.pow(sigma,2)))) def bpm_size( pl ): global last_bpm bpm = float(redisData["bpm"]) # Milliseconds ber beat milliSecondsPerBeat = int(60 / bpm * 1000) # Calculate the intensity based on bpm coming/leaving # The curb is a gaussian mu = math.sqrt(milliSecondsPerBeat) milliTimeToLastBeat = (time.time() - last_bpm) * 1000 milliTimeToNextBeat = (milliSecondsPerBeat - milliTimeToLastBeat) intensity = gauss( milliTimeToNextBeat, 0 , mu) debug(name,"bpm_size","milliSecondsPerBeat:{}\tmu:{}".format(milliSecondsPerBeat, mu)) debug(name,"bpm_size","milliTimeToLastBeat:{}\tmilliTimeToNextBeat:{}\tintensity:{}".format(milliTimeToLastBeat, milliTimeToNextBeat, intensity)) if milliTimeToNextBeat <= 0 : last_bpm = time.time() for i, point in enumerate(pl): ref_x = point[0]-centerX ref_y = point[1]-centerY #debug(name,"In new ref x:{} y:{}".format(point[0]-centerX,point[1]-centerY)) angle=math.atan2( point[0] - centerX , point[1] - centerY ) l = ref_y / math.cos(angle) new_l = l * intensity #debug(name,"bpm_size","angle:{} l:{} new_l:{}".format(angle,l,new_l)) new_x = math.sin(angle) * new_l + centerX new_y = math.cos(angle) * new_l + centerY #debug(name,"x,y:({},{}) x',y':({},{})".format(point[0],point[1],new_x,new_y)) pl[i][0] = new_x pl[i][1] = new_y #debug( name,"bpm_noise output:{}".format(pl)) return pl def rms_size( pl ): rms = float(redisData["rms"]) for i, point in enumerate(pl): ref_x = point[0]-centerX ref_y = point[1]-centerY debug(name,"In new ref x:{} y:{}".format(point[0]-centerX,point[1]-centerY)) angle=math.atan2( point[0] - centerX , point[1] - centerY ) l = ref_y / math.cos(angle) debug(name,"angle:{} l:{}".format(angle,l)) new_l = l + rms * chaos new_x = math.sin(angle) * new_l + centerX new_y = math.cos(angle) * new_l + centerY debug(name,"x,y:({},{}) x',y':({},{})".format(point[0],point[1],new_x,new_y)) pl[i][0] = new_x pl[i][1] = new_y #debug( name,"rms_noise output:{}".format(pl)) return pl def rms_noise( pl ): rms = float(redisData["rms"]) debug(name, "pl:{}".format(pl)) for i, point in enumerate(pl): #debug(name,"rms_noise chaos:{} rms:{}".format(chaos, rms)) xRandom = random.uniform(-1,1) * rms * chaos yRandom = random.uniform(-1,1) * rms * chaos #debug(name,"rms_noise xRandom:{} yRandom:{}".format(xRandom, yRandom)) pl[i][0] += xRandom pl[i][1] += yRandom #debug( name,"rms_noise output:{}".format(pl)) return pl def updateRedis(): global redisLastHit global redisData for key in redisKeys: redisData[key] = r.get(key).decode('ascii') debug("name","updateRedis key:{} value:{}".format(key,redisData[key])) if key == 'bpm': redisData['bpm_ttl'] = r.pttl(key) debug(name,"redisData:{}".format(redisData)) try: while True: # it is time to query redis if now() - redisLastHit > redisFreq: updateRedis() start = time.time() line = sys.stdin.readline() if line == "": time.sleep(0.01) line = line.rstrip('\n') pointsList = ast.literal_eval(line) # Do the filter for mode in modeList: pointsList = locals()[mode](pointsList) print( pointsList, flush=True ) looptime = time.time() - start # debug(name+" looptime:"+str(looptime)) if( looptime < optimal_looptime ): time.sleep( optimal_looptime - looptime) # debug(name+" micro sleep:"+str( optimal_looptime - looptime)) except EOFError: debug(name+" break")# no more information