SK6812 Remote über den Django Webserver steuern

Bisher haben wir den SK6812 LED Streifen direkt per Skript auf dem Raspberry angesteuert. Die erste Lösung war hierbei per Kommandozeile den Modus zu wechseln. Das ist uns aber auf Dauer zu umständlich. Deswegen wollen wir von jedem beliebigen Gerät, sei es ein Rechner, Tablet oder Smartphone den Streifen ansteuern.

Django Webserver vorbereiten

Hierzu verwenden wir unseren Django-Webserver, den wir sowieso für die Visualisierung verwenden. Mit Hilfe von HTTP Streaming werden die Befehle an die Raspberry geschickt. Dazu nutzen wir die StreamingHttpResponse von Django.

Mit dieser ist es einfach möglich, Daten regelmäßig aus einem Webserver auszulesen. Im folgenden Besipiel wird alle 5 Sekunden der neue Modus ausgegeben.

LED_MODES = {}
for i in range(1,30):
    LED_MODES[i] = -1;

class LedView(View):
    def get_data(self):
        while True:
            jdata = json.dumps({'mode': LED_MODES[self.id]})
            LED_MODES[self.id] = -1
            yield "{0:<100}".format(jdata)
            time.sleep(5)
                     
    def get(self, request, id):
        self.id = int(id)
        return StreamingHttpResponse(self.get_data())
        

Ruft man nun die zugehörige Adresse zum Beispiel mit einem Webbrowser auf, so wird alle 5 Sekunden ein neuer JSON Block geschickt. Dazu muss der view natürlich per URL aufrufbar sein, also ein entsprechender Eintrag ins der urls.py vorhanden sein.

Da es leichter ist, eine feste Länge zu parsen, wird der gesendete Block immer auf 100 Zeichen festgelegt. Dies lässt genug Platz für weitere Parameter. Später sollen hier zum Beispiel mal die Farbe übertragen werden, die die Basis des jeweiligen Modus sein soll.

Die Daten die gesendet werden stammen dabei aus dem Dictionary LED_MODES. Der Einfachheit halber wird dieses mit dem Wert -1 für den Modus initialisiert. Um diese zu ändern, brauchen wir einen anderen View, der im Dictionary den Wert ändert. Dazu dient der folgende View.

class LedConfigView(View):
    def get(self, request, id, mode=-1):
        context = {}
        room = get_object_or_404(Room, id=id)
        context['room'] = room
        LED_MODES[int(id)] = int(mode)
        return render(request, 'led.html', context)

Mit diesen beiden Views haben wir jetzt die Server Seite erledigt.

Streaming Client auf dem Raspberry

Um einen Streaming Client in Python zum Laufen zu bringen wird nicht viel benötigt. Mithilfe der Bibliothek requests können wir eine Anfrage vom Typ Stream erzeugen und mit der Funktion iter_content werden dann die einzelnen Pakete gelesen.

import requests, json
url = 'unsereURL'

r = requests.get(url, stream=True)

for line in r.iter_content(chunk_size=100):
    if line:
        try:
            data = json.loads(line.strip())
            print (data)
        except Exception as inst:
            pass

Beim Aufruf von iter_content nutzen wir die feste Blocklänge von 100 Zeichen aus. Dies führt dazu, dass immer genau ein Datenblock verarbeitet wird.

SK6812 per Thread steuern

Am Ende kommen wir nun dazu, wie man die Funktion am besten in die Steuerung einbauen kann. Das Geheimnis sind Threads, denn es muss die gesamte Zeit auf neue Signale gewartet werden, während der SK6812 LED Streifen angesteuert wird.

Dazu werden im Haupt-Thread auf neue Befehle gewartet. Der Haupt-Thread startet dann als neuen Thread jeweils die Funktion, die angezeigt werden soll.

Für die einzelnen Funktionen gibt es dafür eine Basis-Thread-Klasse, die die Arbeit für die einzelnen Modi vereinfachen soll. Die Variable strip ist das für die Ansteuerung benötige Objekt, was wir schon aus dem ersten Skript kennen. Die Variable wait_ms wird für das Timing der Updates verwendet.

import threading

class LedThread(threading.Thread):
    def __init__(self, strip, wait_ms=250):
        threading.Thread.__init__(self)
        self.strip = strip
        self.wait_ms = wait_ms
        self.intensity = 1.0
        self.stop = False

    def join(self, timeout=None):
        for i in range(1, 25):
            self.intensity = 1.0 - i * 0.04
            time.sleep(self.wait_ms/1000.0)
        self.stop = True
        threading.Thread.join(self, timeout)

Die Variable intensity soll beim berechnen dafür verwendet werden, um einen Effekt ein- oder ausfaden zu lassen. Das Ausfaden ist in der Funktion join schon implementiert. Auch die Variable stop wird im Join verwendet. Sie soll dazu verwendet werden, die eigentliche Funktion kontrolliert abzubrechen.

Eine einfache Ableitung ist die folgende Klasse, den den Streifen einfach nur Schwarz werden lässt, also ausschaltet.

class DarkStrip(LedThread):
    def run(self):
        for i in range(strip.numPixels()):
            strip.setPixelColor(i, Color(0,0,0))
        while not self.stop:
            strip.show()
            time.sleep(self.wait_ms/1000.0)

Etwas komplexer wird das ganze, wenn wir uns die Klasse für den ColorRun ansehen. Wie im vorherigen Beitrag beschrieben, ist der ColorRun ein Regenbogen der sich langsam weiter bewegt, die Farben rotieren also langsam über den gesamten Streifen.

class ColorRun(LedThread):

    def __init__(self, strip, wait_ms=50):
        LedThread.__init__(self, strip, wait_ms)

    def run(self):
        l = self.strip.numPixels()
        lc = len(RUN_COLORS) - 1
        bl = l / lc
        self.intensity = 0.0
        while True and not self.stop:
            for startPos in range(l):
            self.intensity = min(self.intensity + 0.04, 1.0)
            for i in range(l):
                relPos = (i - startPos) % bl
                block = (i - startPos) // bl %lc 
                factors = (relPos * 1.0 / bl , (bl - relPos)*1.0 / bl)

                red = RUN_COLORS[block+1][0] * factors[0] + RUN_COLORS[block][0] * factors[1]
                green = RUN_COLORS[block+1][1] * factors[0] + RUN_COLORS[block][1] * factors[1]
                blue = RUN_COLORS[block+1][2] * factors[0] + RUN_COLORS[block][2] * factors[1]
                color = Color(int(red * self.intensity), 
                int(green * self.intensity),
                int(blue * self.intensity))
                self.strip.setPixelColor(i, color)

            self.strip.show()
            time.sleep(self.wait_ms/1000.0)
            if self.stop:
                break

Wie man sehen kann, wir die Intensity für die Berechnung der Farben verwendet. Damit funktioniert dann automatisch der FadeOut. Für den FadeIn wird die Intensity berechnet und auf maximal 1 begrenzt.

Threads und WebClient verbinden

Jetzt müssen die einzelnen Threads nur noch aufgerufen werden. Diese Aufrufe sind an den übermittelten Mode verknüpft, der per json geparsed wird.

def stopActProgram(actThread):
    if actThread:
        actThread.join()
        
if __name__ == '__main__':
    strip = Adafruit_NeoPixel(LED_COUNT, LED_PIN, LED_FREQ_HZ, LED_DMA, LED_INVERT, LED_BRIGHTNESS, LED_CHANNEL, LED_STRIP)
    strip.begin()

    actThread = None
    data = -1

    while True:
        try:
            url = 'http://gemini:8000/room/led/7/'
            r = requests.get(url, stream=True)
    
            for line in r.iter_content(chunk_size=100):
                if line:
                    data =json.loads(line.strip())
                    val = int(data['mode'])
                    if val == 1:
                        stopActProgram(actThread)
                        actThread = ColorWipe3(strip, Color(255,0,0))
                        actThread.start()
                    elif val == 5:
                        stopActProgram(actThread)
                        actThread = Fire(strip)
                        actThread.start()
                    elif val == 0:
                        stopActProgram(actThread)
                        actThread = DarkStrip(strip)
                        actThread.start()
        except:
            time.sleep(20)

Sollte es irgendein Problem geben, zum Beispiel der Webserver weg sein, so wird das Skript 20 Sekunden warten, bevor es wieder eine Anfrage an den Server schickt. Dies schafft etwas mehr Robustheit.

Das fertige Skript kann jetzt einfach gestartet werden. Auf dem Webserver können jetzt die Modi gewechselt werden.

Steuerung als Service starten

Am Ende soll das Python Skript (in unserem Fall „SK6812.py“) aber nicht nur laufen, wenn es manuell gestartet wird. Deshalb soll es als Service auf dem Raspberry Pi eingetragen werden. Damit wird es bei jedem Neustart automatisch geladen.

Hierzu melden wir das Script als (Systemd) Service an. Wichtig ist, dass es dabei als root ausgeführt wird, da wir dies ja auch für die normale Ausführung benötigen. Dazu legen wir unter /etc/systemd/system eine Datei an:

sudo touch /etc/systemd/system/ledstripe.service

und fügen folgenden Inhalt ein:

[Unit]
Description=LED Stripe Service

[Service]
Type=simple
ExecStart=/usr/bin/python SK6812.py
WorkingDirectory=/home/pi/rpi_ws281x/python/examples/

[Install]
WantedBy=multi-user.target

Die angelegte Servicedatei kann mit folgenden Befehl überprüft werden:

sudo systemd-analyze verify ledstripe.service

Wenn alles in Ordnung ist, so wird keine Meldung auf der Konsole ausgegeben. Der Service kann nun also aktiviert und gestartet werden:

sudo systemctl enable ledstripe.service
sudo systemctl start ledstripe.service

Um sehen zu können, ob der Service läuft hilft folgender Befehl:

sudo systemctl status ledstripe.service

Läuft der Service, so lässt sich ab jetzt nur über das Web-Interface unser SK6812 LED Streifen steuern. Erweiterungen sind jetzt natürlich denkbar, seien es weitere Modi oder eine Farbwahl.

 

Schreibe einen Kommentar

Deine E-Mail-Adresse wird nicht veröffentlicht. Erforderliche Felder sind mit * markiert

Diese Website verwendet Akismet, um Spam zu reduzieren. Erfahre mehr darüber, wie deine Kommentardaten verarbeitet werden.