LJ/clitools/filters/redilysis.py

301 line
11 KiB
Python
Executable File

#!/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 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