OlaLED
Introduction
In this topic, you will learn how to install a python script to control a LED / RGB system via webcontrol (or other python script).
Prerequisites:
You need to have a fully functional ola system.
Install script
create a directory to store all these new functionnality
mkdir -p /srv/dmx/
write a this new python script in this folder :
nano /srv/dmx/server.py
#!/usr/bin/python
import SocketServer
import threading
import time
from ola_send_dmx import *
my_thread = 0
my_command = 'start'
actual_program = 'color' ## program wanted
param1 = 'red' ## first param
intensity = 100 ## current intensity
actual_fixed_color = 0 ## if fixed color choice, put args here to remember (if having to change intensity)
R_before = 0 # current color state
G_before = 0
B_before = 0
pitch = 1
class MyTCPHandler(SocketServer.BaseRequestHandler):
"""
The RequestHandler class for our server.
It is instantiated once per connection to the server, and must
override the handle() method to implement communication to the
client.
"""
def handle(self):
# self.request is the TCP socket connected to the client
global my_thread
self.data = self.request.recv(1024).strip()
# was a test
#if self.data == 'reset' :
# global my_command
# my_command = 'reset'
#else :
command = self.data.split()
command = command[0]
if command == 'intensity': # We separe intensity because should lauch in parallel
another_thread = Thread_dmx(self.data) # thread named different for able controlling mythread
another_thread.start()
print "thread intensite lance"
# just send back the same data, but upper-cased
self.request.send(self.data.upper())
else:
if my_thread != 0 : #if thread is already launched and is not about intensity
try:
my_thread.stop() #stop thread
my_thread.join() #stop program to be sure thread is stopped
except NameError:
print "An error occured when trying to stop previous thread."
print "launching thread " + self.data
my_thread = Thread_dmx(self.data)
my_thread.start()
# just send back the same data to the client, but upper-cased
self.request.send(self.data.upper())
class Thread_dmx(threading.Thread):
def __init__(self, nom = ''):
threading.Thread.__init__(self)
self.nom = nom
self.Terminated = False
def run(self):
global actual_program
global intensity
global fixed_color
## separate command and arguments ##
words = self.nom.split()
if len(words) == 0:
print('Error: empty request\n')
return
command = words[0]
args = words[1:]
methodname = 'do_' + command
## relauch sending DMX for fixed color, if intensity changed
if command=='intensity':
intensity = abs(int(args[0]))
if intensity > 100:
return
if actual_program == 'fade':
method = getattr(self, 'do_fade')
method(*fixed_color)
else:
actual_program = command
## take care of inexistent function
if not hasattr(self, methodname):
print('Error: ',command,' is not a valid command')
return
## save color if fixed color
if command=='fade':
fixed_color = args
## launch corresponding function
method = getattr(self, methodname)
method(*args)
## Stop the thread if called
def stop(self):
self.Terminated = True
## Go to specified DMX adresses from previous adresses in fade
def do_fade(self,R_after,G_after,B_after,speed):
global R_before
global G_before
global B_before
global intensity
R_after=int(round((int(R_after)*intensity)/100))
G_after=int(round((int(G_after)*intensity)/100))
B_after=int(round((int(B_after)*intensity)/100))
if(R_after==R_before and G_after==G_before and B_after==B_before):
print('Etat deja atteint')
return
speed=float(speed)
actual_intensity = intensity
time_to_sleep=(speed/max(abs(R_before-R_after),abs(G_before-G_after),abs(B_before-B_after)))
# count to know wich pitch to apply (processor free)
speed_reference=float(0.01)
print(time_to_sleep)
pitch=int(speed_reference/time_to_sleep)
print(pitch)
if pitch == 0:
pitch = int(1)
print(pitch)
## while ending color are not reached or no order for stopping, increment data
while (R_before-R_after!=0 or G_before-G_after!=0 or B_before-B_after!=0) and (not self.Terminated) :
## If intensity was changed, take care of this
if actual_intensity != intensity:
R_after=int(round((int(R_after)*intensity)/100))
G_after=int(round((int(G_after)*intensity)/100))
B_after=int(round((int(B_after)*intensity)/100))
actual_intensity = intensity
R_diff=R_before-R_after
G_diff=G_before-G_after
B_diff=B_before-B_after
if R_diff <= -pitch :
R_before += pitch
elif R_diff >= pitch :
R_before-= pitch
else :
R_before = R_after
if G_diff <= -pitch :
G_before += pitch
elif G_diff >= pitch :
G_before -= pitch
else :
G_before=G_after
if B_diff <= -pitch :
B_before += pitch
elif B_diff >=pitch :
B_before -= pitch
else :
B_before=B_after
#print('AFTER : ',R_after,G_after,B_after)
print('values =',R_before,G_before,B_before)
print('sleep =',time_to_sleep*pitch)
print('pitch =',pitch)
dmx_send(R_before,G_before,B_before)
time.sleep(time_to_sleep*pitch)
## make a chase from red to blue, indefinitely
def do_chase(self,speed):
intensity = 255
speed = float(speed)
while not self.Terminated :
self.do_fade(intensity,0,intensity,speed)
print("7")
print("1")
self.do_fade(intensity,0,0,speed)
print("2")
self.do_fade(intensity,intensity,0,speed)
print("3")
self.do_fade(0,intensity,0,speed)
print("4")
self.do_fade(0,intensity,intensity,speed)
print("5")
self.do_fade(0,0,intensity,speed)
print("6")
#self.do_fade(intensity,0,intensity,speed)
#print("7")
if __name__ == "__main__":
HOST, PORT = "localhost", 9999
# Create the server, binding to localhost on port 9999
server = SocketServer.TCPServer((HOST, PORT), MyTCPHandler)
# Activate the server; this will keep running until you
# interrupt the program with Ctrl-C
server.serve_forever()
copy client_wrapper.py in this folder :
cp /usr/src/ola-0.7.4/python/examples/client_wrapper.py /srv/dmx/
create a new python file function : ola_send_dmx.py
nano /srv/dmx/ola_send_dmx.py
Writted by David Magniez pubmag at free dot fr
