#!/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 Licensed under GNU GPLv3 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 msNow(): return time.time() # The list of available modes => redis keys each requires to run oModeList = { "rms_noise": ["rms"], "rms_size": ["rms"], "bpm_size": ["bpm"], "bpm_detect_size": ["bpm","bpm_delay","bpm_sample_interval","beats"] } CHAOS = 1 REDISLATENCY = 30 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("-l","--redisLatency",help="Latency in ms to substract. Default:{}".format(REDISLATENCY),default=REDISLATENCY,type=float) 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 / 1000 verbose = args.verbose fps = args.fps centerX = args.centerX centerY = args.centerY redisLatency = args.redisLatency 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 = msNow() - 99999 r = redis.Redis( host=ip, port=port) # Records the last bpm tsLastBeat = 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)))) previousPTTL = 0 tsNextBeatsList = [] def bpmDetect( ): """ An helper to compute the next beat time in milliseconds Returns True if the cache was updated """ global tsNextBeatsList global previousPTTL global redisLastHit global redisLatency # Get the redis PTTL value for bpm PTTL = redisData["bpm_pttl"] # Skip early if PTTL < 0 if PTTL < 0 : debug(name,"bpmDetect skip detection : PTTL expired for 'bpm' key") return False # Skip early if the record hasn't been rewritten if PTTL <= previousPTTL : previousPTTL = PTTL #debug(name,"bpmDetect skip detection : {} <= {}".format(PTTL, previousPTTL)) return False debug(name,"bpmDetect running detection : {} > {}".format(PTTL, previousPTTL)) previousPTTL = PTTL # Skip early if beat list is empty beatsList = ast.literal_eval(redisData["beats"]) tsNextBeatsList = [] if( len(beatsList) == 0 ): return True # Read from redis bpm = float(redisData["bpm"]) msBpmDelay = float(redisData["bpm_delay"]) samplingInterval = float(redisData["bpm_sample_interval"]) # Calculate some interpolations lastBeatTiming = float(beatsList[len(beatsList) - 1]) msPTTLDelta = 2 * samplingInterval - float(PTTL) sPerBeat = 60 / bpm lastBeatDelay = msBpmDelay - lastBeatTiming*1000 + msPTTLDelta countBeatsPast = math.floor( (lastBeatDelay / 1000) / sPerBeat) #debug(name,"bpmDetect lastBeatTiming:{}\tmsPTTLDelta:{}\tsPerBeat:{}".format(lastBeatTiming,msPTTLDelta,sPerBeat)) #debug(name,"lastBeatDelay:{}\t countBeatsPast:{}".format(lastBeatDelay, countBeatsPast)) for i in range( countBeatsPast, 1000): beatTime = i * sPerBeat - lastBeatTiming if beatTime < 0: continue if beatTime * 1000 > 2 * samplingInterval : break #debug(name, "bpmDetect beat add beatTime:{} redisLastHit:{}".format(beatTime, redisLastHit)) tsNextBeatsList.append( redisLastHit + beatTime - redisLatency/1000) debug(name, "bpmDetect new tsNextBeatsList:{}".format(tsNextBeatsList)) return True def bpm_detect_size( pl ): bpmDetect() # Find the next beat in the list tsNextBeat = 0 now = time.time() msNearestBeat = None msRelativeNextBTList = list(map( lambda a: abs(now - a) * 1000, tsNextBeatsList)) msToBeat = min( msRelativeNextBTList) #debug(name,"bpm_detect_size msRelativeNextBTList:{} msToBeat:{}".format(msRelativeNextBTList,msToBeat)) # Calculate the intensity based on bpm coming/leaving # The curb is a gaussian mu = 15 intensity = gauss( msToBeat, 0 , mu) #debug(name,"bpm_size","mu:{}\t msToBeat:{}\tintensity:{}".format(mu, msToBeat, intensity)) if msToBeat < 20: debug(name,"bpm_detect_size kick:{}".format(msToBeat)) pass 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_detect_size output:{}".format(pl)) return( pl ); def bpm_size( pl ): global tsLastBeat bpm = float(redisData["bpm"]) # msseconds ber beat msPerBeat = int(60 / bpm * 1000) # Calculate the intensity based on bpm coming/leaving # The curb is a gaussian mu = math.sqrt(msPerBeat) msTimeToLastBeat = (time.time() - tsLastBeat) * 1000 msTimeToNextBeat = (msPerBeat - msTimeToLastBeat) intensity = gauss( msTimeToNextBeat, 0 , mu) debug(name,"bpm_size","msPerBeat:{}\tmu:{}".format(msPerBeat, mu)) debug(name,"bpm_size","msTimeToLastBeat:{}\tmsTimeToNextBeat:{}\tintensity:{}".format(msTimeToLastBeat, msTimeToNextBeat, intensity)) if msTimeToNextBeat <= 0 : tsLastBeat = 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 refreshRedis(): global redisLastHit global redisData # Skip if cache is sufficent diff = msNow() - redisLastHit if diff < redisFreq : #debug(name, "refreshRedis not updating redis, {} < {}".format(diff, redisFreq)) pass else: #debug(name, "refreshRedis updating redis, {} > {}".format(diff, redisFreq)) redisLastHit = msNow() for key in redisKeys: redisData[key] = r.get(key).decode('ascii') #debug(name,"refreshRedis key:{} value:{}".format(key,redisData[key])) # Only update the TTLs if 'bpm' in redisKeys: redisData['bpm_pttl'] = r.pttl('bpm') #debug(name,"refreshRedis key:bpm_ttl value:{}".format(redisData["bpm_pttl"])) #debug(name,"redisData:{}".format(redisData)) return True try: while True: refreshRedis() 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