simon/led-server
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

renaming folders and files to python format

Browse Source
This commit is contained in:
dezeyer
2019-05-11 14:28:42 +00:00
parent ade378d128
commit 798775805c
19 changed files with 509 additions and 268 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/Utils/BaseEffect.py
+100
View File
@@ -0,0 +1,100 @@
import time
import threading
import typing
import copy
class BaseEffect(threading.Thread):
# The Name and the Description of the Effect,
# should be overwritten by the inheritancing Effect
name: str = "Undefined"
desc: str = "No Description"
# Something that will be used to show descriptions and value options
# of the parameters the effect will accept, in a way, that eg the webclient can decide,
# if the parameters can be toggeled by a button/checkbox/slider/whatever
effectParameters: list = []
# musiceffect things
# TODO we need a recorderController where the effect binds to and gets its values, this is shit...
hasNewFttValues = False
ftt: typing.Tuple[typing.List[float],typing.List[float]] = ([],[])
stop:bool = False
def __init__(self):
threading.Thread.__init__(self)
self.effectRGBStripList = []
# when a strip is added or removed or the options change, the thread will not restart,
# the changes are pushed to the thread at livetime. so avoid log running loops without
# accessing getRGBStrips and getEffectParams. It could happen, that a new effect for example already uses
# a rgbStip while the loop of the old effect is running and has not got the changes.
def run(self):
self.effectParameterValues = []
for effectParameterIndex, effectParameter in enumerate(self.effectParameters):
self.effectParameterValues.append([])
for option in effectParameter.options:
self.effectParameterValues[effectParameterIndex].append(option[2])
print(self.effectParameterValues)
self.init()
while not self.stop:
# run the effect in endless while
self.effect()
self.end()
# Init is called bevor the loop, use setEffectParams for init values
# and access them in the effect with getEffectParams
def init(self):
return
# see run(): never save effectRGBStrips() as a variable and access it as often as possible
# to avoid two effects accessing the rgbStrip
def effect(self):
while 1:
print("ET "+self.name+" effect() function needs to be replaced in inheritancing Effect")
# called when the effect is stopped
def end(self):
return
# when called the effect loop will stop and the thread can be terminated
def stopEffect(self):
self.stop = True
# the effect itself will know its own strips. i don't know if it would be better if the effectController self
# should have this list, but then i must do something like nested arrays, naah.
def addRGBStrip(self,rgbStrip):
self.effectRGBStripList.append(rgbStrip)
self.onRGBStripAdded(rgbStrip)
# remove a strip, if the effect has no more strips, the effect thread guardian will kill it
def removeRGBStrip(self,rgbStrip):
self.effectRGBStripList.remove(rgbStrip)
# for overriding by the effect, when a strip is added
def onRGBStripAdded(self,rgbStrip):
return
# for overriding by the effect, when a strip is added
def onEffectParameterValuesUpdated(self):
return
def updateFttValues(self,ftt: typing.Tuple[typing.List[float],typing.List[float]]):
self.hasNewFttValues = True
self.ftt = ftt
# returns a list of the RGBStrips used by this effect
def effectRGBStrips(self):
return self.effectRGBStripList
def addEffectParameter(self,effectParameter):
self.effectParameters.append(effectParameter)
# set Params as descriped in getParamsDescription()
def updateEffectParameterValues(self,effectParameterIndex, effectParameterValues):
print("updateEffectParameterValues",effectParameterIndex,effectParameterValues)
for effectParameterValue in effectParameterValues:
if self.effectParameters[int(effectParameterIndex)].testValue(int(effectParameterValue),int(effectParameterValues[effectParameterValue])):
self.effectParameterValues[int(effectParameterIndex)][int(effectParameterValue)] = int(effectParameterValues[effectParameterValue])
self.onEffectParameterValuesUpdated()
server/Utils/EffectController.py
+245
View File
@@ -0,0 +1,245 @@
# import offEffect
from Effects.OffEffect import OffEffect
from Utils.BaseEffect import BaseEffect
import time
import threading
import os
import sys
#TypeCheck
from typing import Type, List
from Utils.RGBStrip import RGBStrip
from Utils.RGBStripController import RGBStripController
class EffectController:
'''
the effectController handles the effects and pushes the values to the rgbStripContoller
it also calls the backendProvider's onChange function when there are changes made on the effects:
- start/stop effects
- changes parameters of specific effects
- moves the rgbStrips to the effects
- runs a guardian that detects dead effectThreads and moves Strips to offEffect
'''
rememberRGBStripEffectThreads: dict = {}
# list of current running effects
effectThreads: List[BaseEffect] = []
# the always running base effect
offEffectThreadObject: BaseEffect
# maybe i will seperate this later
onControllerChangeHandler: List[Type[object]] = []
#rgbStripController
rgbStripController: RGBStripController
def __init__(self,rgbStripController: RGBStripController) -> None:
'''Init the EffectController with the RGBStripController'''
self.rgbStripController = rgbStripController
# load the effects
self.effectsList: List[Type[BaseEffect]] = self.getEffectsListFromDir()
# start the offEffect by default
self.offEffectThreadObject = self.startEffect(OffEffect,[])
# - a bit of failover handling, remove dead threads from effectThread array
# - move strips without an effect to the offEffect
self.effectGuardian = self.EffectGuardian(self)
self.effectGuardian.start()
def startEffect(self, effectClass: Type[BaseEffect], rgbStrips: list, params: list = []) -> BaseEffect: #TODO params as a object -> EffectParams
'''starts a effect by given class, rgbStrips and params ([[index,[param,param,param]],[index,[param,param,param]]])'''
newEffect = effectClass()
if len(self.effectThreads) > 0 and len(rgbStrips) > 0 or len(self.effectThreads) == 0:
newEffect.start()
self.updateEffectParameters(newEffect, params)
self.effectThreads.append(newEffect)
for rgbStrip in rgbStrips:
self.moveRGBStripToEffectThread(rgbStrip, newEffect)
self.noticeControllerChange()
return newEffect
def getEffects(self) -> List[Type[BaseEffect]]:
'''
returns all effectClasses but offEffect, since offEffect will never be killed
and should not be running twice
'''
# alle außer offEffect
return self.effectsList
def getEffectThreads(self) -> List[BaseEffect]:
'''returns all running effectThreads'''
return self.effectThreads
def getEffectRGBStrips(self, effectThreadObject: BaseEffect) -> List[RGBStrip]:
'''returns a list of the RGBStrips used by this effect'''
return effectThreadObject.effectRGBStrips()
def moveRGBStripToEffectThread(self, rgbStrip: RGBStrip, effectThreadObject: BaseEffect) -> None:
'''move a rgbStip to a running effectThread'''
# cycle throught all effects and
# remove Strip from effect if added
for et in self.effectThreads:
if rgbStrip in et.effectRGBStrips():
et.removeRGBStrip(rgbStrip)
if effectThreadObject.isAlive():
self.rememberRGBStripEffectThreads[rgbStrip.STRIP_NAME] = et
effectThreadObject.addRGBStrip(rgbStrip)
# check if any effectThread has no more rgbStrips and if so, stop it
# if the effectThread has no more strips, we stop it and remove it.
for x, effectThread in enumerate(self.effectThreads):
# but ignore the first effectthread, since that is our offeffect that should never be killed
if len(effectThread.effectRGBStrips()) == 0 and x is not 0:
effectThread.stopEffect()
self.effectThreads.remove(effectThread)
self.noticeControllerChange()
def updateEffectParameters(self, effectThreadObject: BaseEffect, effectParameters: list):
'''updates parameter of a running effectThread ([[index,[param,param,param]],[index,[param,param,param]]])'''
for effectParameter in effectParameters:
effectThreadObject.updateEffectParameterValues(
effectParameter[0], effectParameter[1])
self.noticeControllerChange()
def stopAll(self):
'''stops all effectThreads and set the rgbStrips to off'''
print("effectController stopAll()")
for effectThread in self.effectThreads:
print("effectController killing... "+str(effectThread))
effectThread.stopEffect()
print("effectController killed "+str(effectThread))
print("effectController stopping Guardian... ")
self.effectGuardian.stop()
print("effectController stopped Guardian")
time.sleep(0.5)
def noticeControllerChange(self) -> None:
'''inform the controllerChangeHandler to update the client'''
for controllerChangeHandler in self.onControllerChangeHandler:
controllerChangeHandler()
def addOnControllerChangeHandler(self, hander: Type[object]) -> None: #TODO class instead of function for OnControllerChangeHandler?
'''add onControllerChangeHandler'''
print("addOnControllerChangeHandler", str(hander))
self.onControllerChangeHandler.append(hander)
# send data to this client
hander()
def removeOnControllerChangeHandler(self, hander: Type[object]): #TODO class instead of function for OnControllerChangeHandler?
'''remove onControllerChangeHandler'''
print("removeOnControllerChangeHandler", str(hander))
if hander in self.onControllerChangeHandler:
self.onControllerChangeHandler.remove(hander)
else:
print('\n\n -> client was never registered!')
def getEffectsListFromDir(self) -> List[Type[BaseEffect]]:
'''automaticly loads all modules from effects subdir and adds them to the list of effects if they have the BaseEffect as subclass'''
effectsList: List[Type[BaseEffect]] = []
for raw_module in os.listdir(os.path.abspath(os.path.join(os.path.join(os.path.dirname(__file__), os.pardir),"Effects"))):
if raw_module == '__init__.py' or raw_module == 'OffEffect.py' or raw_module[-3:] != '.py':
continue
effectModule = __import__("Effects."+raw_module[:-3], fromlist=[raw_module[:-3]])
effectClass = getattr(effectModule, raw_module[:-3])
if issubclass(effectClass,BaseEffect):
effectsList.append(effectClass)
print("Loaded effects: ",effectsList)
return effectsList
def onRGBStripRegistered(self,rgbStrip: RGBStrip):
''' moves new rgbStrip to the offEffectThread and notice all Controller Clients'''
if rgbStrip.STRIP_NAME in self.rememberRGBStripEffectThreads:
restoredEffect: BaseEffect = self.rememberRGBStripEffectThreads[rgbStrip.STRIP_NAME]
if restoredEffect.isAlive():
print("stripname in rememberRGBStripEffectThreads -> same effect was never stopped, move to")
self.moveRGBStripToEffectThread(rgbStrip,restoredEffect)
else:
print("stripname in rememberRGBStripEffectThreads -> create new effect")
effectParamValueList: list = []
for x in range(len(restoredEffect.effectParameterValues)):
valdict: dict = {}
for y in range(len(restoredEffect.effectParameterValues[x])):
valdict[y] = restoredEffect.effectParameterValues[x][y]
effectParamValueList.append([x,valdict])
self.startEffect(restoredEffect.__class__,[rgbStrip],effectParamValueList)
self.rememberRGBStripEffectThreads.pop(rgbStrip.STRIP_NAME)
else:
print("stripname was not in rememberRGBStripEffectThreads")
self.offEffectThreadObject.addRGBStrip(rgbStrip)
self.noticeControllerChange()
def onRGBStripUnRegistered(self,rgbStrip: RGBStrip):
'''
cycle throught all effectThreadss and
remove Strip from effect if added
'''
et: BaseEffect
for et in self.effectThreads:
if rgbStrip in et.effectRGBStrips():
et.removeRGBStrip(rgbStrip)
# if the effectThread has no more strips, we stop it and remove it.
for x, effectThread in enumerate(self.effectThreads):
# but ignore the first effectthread, since that is our offeffect that should never be killed
if len(effectThread.effectRGBStrips()) == 0 and x is not 0:
effectThread.stopEffect()
self.effectThreads.remove(effectThread)
self.noticeControllerChange()
def updateFttValues(self,ftt: tuple):
''' TODO this is a themorary funtion util a recorderController is build'''
effectThread: BaseEffect
for effectThread in self.effectThreads:
effectThread.updateFttValues(ftt)
class EffectGuardian(threading.Thread):
'''
a bit of failover handling, remove dead threads from effectThread array
move strips without an effect to the offEffect
This Guardian detects Effect Threads that are killed in some unnormal way
This is not the procedere how effects are stoped
Something went wrong when the guarian is removing dead threads!
'''
def __init__(self, effectController):
threading.Thread.__init__(self, name='EffectGuardian')
self.effectController = effectController
self.stopped = False
def run(self):
while not self.stopped:
for effectThread in self.effectController.effectThreads:
# if Thread was killed by something else, we remove it from the list
if not effectThread.isAlive():
for rgbStrip in effectThread.effectRGBStrips():
self.effectController.moveRGBStripToEffectThread(rgbStrip,self.effectController.offEffectThreadObject)
self.effectController.effectThreads.remove(effectThread)
print("effectController:effectGuardian removed dead Thread " +
str(effectThread) + ". There must be an error in code!")
self.effectController.noticeControllerChange()
time.sleep(1)
def stop(self):
self.stopped = True
server/Utils/EffectParameter.py
+42
View File
@@ -0,0 +1,42 @@
import threading
class EffectParameter(object):
# The Name and the Description of the EffectParameter,
# should be overwritten by the inheritancing EffectParameter
name="Undefined"
desc = "No Description"
# In the order you expect the options to be set
# [
# [min/off,max/on,current,"description"],
# [min/off,max/on,current,"description"],
# [min/off,max/on,current,"description"],
# ]
options: list = []
def __init__(self,name,desc,initOptions = []):
self.name = name
self.desc = desc
self.options = initOptions
# check if the given values are plausible
def testValue(self,index,value):
if value >= self.options[index][0] \
and value <= self.options[index][1]:
return True
else:
return False
class colorpicker(EffectParameter):
name="UndefinedColorpicker"
desc="No Description"
type="colorpicker"
# check if the given values are plausible
def testValue(self,index,value):
return True
class slider(EffectParameter):
name="UndefinedSlider"
desc="No Description"
type="slider"
server/Utils/PyAudioRecorder.py
+139
View File
@@ -0,0 +1,139 @@
#
# The idea is to have only one thread accessing the audio input source instead
# of every music enabled thread itself. also, different fuctions calculating
# frequencys and so on shoud run in own threads to the leds get more updates
#
# in history one thread was calculating bpm, freqence average and max values in one thread
# before updating the leds, what the pi was a bit slow for and you cloud count every update
# of the leds
# i think most of it is from https://github.com/shunfu/python-beat-detector/
import numpy
import pyaudio
import threading
import time
class PyAudioRecorder:
"""Simple, cross-platform class to record from the default input device."""
def __init__(self):
self.RATE = 44100
self.BUFFERSIZE = 2**12
self.secToRecord = .1
self.kill_threads = False
self.has_new_audio = False
self.setup()
# since the server only can handly one input (for now) this thread will calculate
# some basic things for the musicEffectsThreads, so they can update the leds more often.
# it is always posible to catch the fft() function and do your own thing in your musikEffect.
# calculate bpm, this is the same for all clientThreads
self.beats_idx = 0
self.bpm_list = []
self.prev_beat = time.perf_counter()
self.low_freq_avg_list = []
self.lastTime = time.time()
self.recorderClients = []
def setup(self):
self.buffers_to_record = int(
self.RATE * self.secToRecord / self.BUFFERSIZE)
if self.buffers_to_record == 0:
self.buffers_to_record = 1
self.samples_to_record = int(self.BUFFERSIZE * self.buffers_to_record)
self.chunks_to_record = int(self.samples_to_record / self.BUFFERSIZE)
self.sec_per_point = 1. / self.RATE
self.p = pyaudio.PyAudio()
# start the PyAudio class
for i in range(self.p.get_device_count()):
devinfo = self.p.get_device_info_by_index(i)
print(i, devinfo["name"])
# make sure the default input device is broadcasting the speaker output
# there are a few ways to do this
# e.g., stereo mix, VB audio cable for windows, soundflower for mac
self.in_stream = self.p.open(format=pyaudio.paInt16,
channels=1,
rate=self.RATE,
input=True,
frames_per_buffer=self.BUFFERSIZE)
print("Using default input device: {:s}".format(
self.p.get_default_input_device_info()['name']))
self.audio = numpy.empty(
(self.chunks_to_record * self.BUFFERSIZE), dtype=numpy.int16)
def close(self):
print("pyAudioRecorder closed")
self.kill_threads = True
self.p.close(self.in_stream)
### RECORDING AUDIO ###
def get_audio(self):
"""get a single buffer size worth of audio."""
audio_string = self.in_stream.read(self.BUFFERSIZE)
return numpy.fromstring(audio_string, dtype=numpy.int16)
def record(self):
while not self.kill_threads:
for i in range(self.chunks_to_record):
self.audio[i*self.BUFFERSIZE:(i+1)
* self.BUFFERSIZE] = self.get_audio()
self.has_new_audio = True
def start(self):
print("pyAudioRecorder started")
self.t = threading.Thread(target=self.record)
self.t.start()
### MATH ###
def downsample(self, data, mult):
"""Given 1D data, return the binned average."""
overhang = len(data) % mult
if overhang:
data = data[:-overhang]
data = numpy.reshape(data, (len(data) / mult, mult))
data = numpy.average(data, 1)
return data
def fft(self, data=None, trim_by=10, log_scale=False, div_by=100):
if not data:
data = self.audio.flatten()
left, right = numpy.split(numpy.abs(numpy.fft.fft(data)), 2)
ys = numpy.add(left, right[::-1])
if log_scale:
ys = numpy.multiply(20, numpy.log10(ys))
xs = numpy.arange(self.BUFFERSIZE/2, dtype=float)
if trim_by:
i = int((self.BUFFERSIZE/2) / trim_by)
ys = ys[:i]
xs = xs[:i] * self.RATE / self.BUFFERSIZE
if div_by:
ys = ys / float(div_by)
return xs, ys
### multithreading things ###
def registerRecorderClient(self,recorderClient):
self.recorderClients.append(recorderClient)
def unregisterRecorderClient(self,recorderClient):
self.recorderClients.remove(recorderClient)
class recorderClient(threading.Thread):
def __init__():
# toggle to true on beat, false when client got the value
self.onBeat = False
# when registering the client i want to be able to define how long the avg list should be
self.y_max_freq_avg_list = []
server/Utils/RGBStrip.py
+98
View File
@@ -0,0 +1,98 @@
import uuid
import typing
class RGBStrip:
# name = the name off the the strip, defined by the client connecting to the server
# uid = unique id, if the strip sends one, use this (later maybe, or never, whatever)
# lenght = the lenght off the strip, for future use of eg WS2812b strips, will be 1 by default
STRIP_NAME: str
STRIP_LENGHT: int
def __init__(self,name: str,onValuesUpdateHandler: object,lenght: int=1,address: typing.Tuple[str,int] = ("0.0.0.0",0000)):
# UID should be updateable later, or not?
# when updating, be sure it does not exist
self.STRIP_UID = str(uuid.uuid4())
self.STRIP_NAME = name
self.STRIP_LENGHT = lenght
self.address = address
self.onValuesUpdateHandler = onValuesUpdateHandler
self.red = [0]*self.STRIP_LENGHT
self.green = [0]*self.STRIP_LENGHT
self.blue = [0]*self.STRIP_LENGHT
def RGB(self,red: int,green: int,blue: int,brightness: int = 100):
if(red < 0):
red = 0
if(red > 255):
red = 255
if(green < 0):
green = 0
if(green > 255):
green = 255
if(blue < 0):
blue = 0
if(blue > 255):
blue = 255
if(brightness < 0):
brightness = 0
if(brightness > 100):
brightness = 100
for x in range(self.STRIP_LENGHT):
self.red[x] = int(red/100*brightness)
self.green[x] = int(green/100*brightness)
self.blue[x] = int(blue/100*brightness)
self.show()
def WS2812b(self,id,red,green,blue,brightness=100):
if id < 0 and id > self.STRIP_LENGHT:
print(self.STRIP_NAME," is max ",self.STRIP_LENGHT," Pixels long!")
return
else:
if(red < 0):
red = 0
if(red > 255):
red = 255
if(green < 0):
green = 0
if(green > 255):
green = 255
if(blue < 0):
blue = 0
if(blue > 255):
blue = 255
if(brightness < 0):
brightness = 0
if(brightness > 100):
brightness = 100
self.red[id] = int(red/100*brightness)
self.green[id] = int(green/100*brightness)
self.blue[id] = int(blue/100*brightness)
def show(self):
self.onValuesUpdateHandler(self)
def off(self):
for x in range(self.STRIP_LENGHT):
self.red[x] = 0
self.green[x] = 0
self.blue[x] = 0
self.onValuesUpdateHandler(self)
def getData(self):
return [self.red,self.green,self.blue]
server/Utils/RGBStripController.py
+59
View File
@@ -0,0 +1,59 @@
from Utils.RGBStrip import RGBStrip
import time
import threading
import json
class RGBStripController:
'''
rgbStrips register themselves at the rgbStripContoller
the rgbStripController calls the backend Provider's onChange function
when there are new values for the strip
'''
def __init__(self):
#threading.Thread.__init__(self)
self.rgbStrips = []
self.onRGBStripRegisteredHandler = []
self.onRGBStripUnRegisteredHandler = []
def registerRGBStrip(self,rgbStripName: str,onValuesUpdateHandler: object,ledcount=1):
# maybe we can use an unique id if the strip reconnects later, eg push the uid
# to the client on first connect and if he reconnects he sould send it back again.
# the wmos could use the mac adress, if there is a python script it can save the uid
# in a file or so.
strip = RGBStrip(rgbStripName,onValuesUpdateHandler,ledcount)
self.rgbStrips.append(strip)
self.noticeRGBStripRegisteredHandler(strip)
return strip
def unregisterRGBStrip(self,strip):
self.rgbStrips.remove(strip)
self.noticeRGBStripUnRegisteredHandler(strip)
# returns all registered rgbStips
def getRGBStrips(self):
return self.rgbStrips
# inform all onRGBStripRegisteredHandler about the new RGBStrip
def noticeRGBStripRegisteredHandler(self,rgbStrip):
for hander in self.onRGBStripRegisteredHandler:
hander(rgbStrip)
# add onRGBStripRegisteredHandler
def addOnRGBStripRegisteredHandler(self, function):
self.onRGBStripRegisteredHandler.append(function)
# remove onRGBStripRegisteredHandler
def removeOnRGBStripRegisteredHandler(self, function):
self.onRGBStripRegisteredHandler.remove(function)
# inform all onRGBStripUnRegisteredHandder about the removed RGBStrip
def noticeRGBStripUnRegisteredHandler(self,rgbStrip):
for hander in self.onRGBStripUnRegisteredHandler:
hander(rgbStrip)
# add onRGBStripUnRegisteredHandler
def addOnRGBStripUnRegisteredHandler(self, function):
self.onRGBStripUnRegisteredHandler.append(function)
# remove onRGBStripUnRegisteredHandler
def removeOnRGBStripUnRegisteredHandler(self, function):
self.onRGBStripUnRegisteredHandler.remove(function)
server/Utils/__init__.py
View File