Elegato Streamdeck für Heimautomatisierung

Mittlerweile besitze ich 3 Streamdecks, die alle an einem Raspberry per USB hängen. 2 davon sind per 10m aktiven USB Verlängerungskabeln angeschlossen. Dazu habe ich ein Python Skript mit dieser Lib gebaut die man per pip installieren kann: https://pypi.org/project/streamdeck/

Die Übergabe der Tastendrücke erfolgt an einen MQTT Broker. Die Konfiguration der Tasten ebenfalls (Icons müssen allerdings noch manuell hochgeladen werden, in das gleiche Verzeichnis wo das Skript liegt).

Hier das Listing des Skripts (Es gibt einen Workaround über die USB-ID, da es nicht möglich ist die Seriennummer im key_change_callback abzufragen. Ebenfalls ist Plug and Play nicht verfügbar 😉 ):

import threading
from PIL import Image, ImageDraw, ImageFont #, ImageSequence
from StreamDeck.DeviceManager import DeviceManager
from StreamDeck.ImageHelpers import PILHelper

import paho.mqtt.client as mqtt
import json


# Generates a custom tile with run-time generated text and custom image via the
# PIL module.
def render_key_image(d, icon_filename, font_filename, label_text):
    # Create new key image of the correct dimensions, black background.
    image = PILHelper.create_image(d)

    # Resize the source image asset to best-fit the dimensions of a single key,
    # and paste it onto our blank frame centered as closely as possible.
    icon = Image.open(icon_filename).convert("RGBA")
    icon.thumbnail((image.width, image.height - 20), Image.LANCZOS)
    icon_pos = ((image.width - icon.width) // 2, 0)
    image.paste(icon, icon_pos, icon)

    # Load a custom TrueType font and use it to overlay the key index, draw key
    # label onto the image.
    draw = ImageDraw.Draw(image)
    font = ImageFont.truetype(font_filename, 14)
    label_w, label_h = draw.textsize(label_text, font=font)
    label_pos = ((image.width - label_w) // 2, image.height - 20)
    draw.text(label_pos, text=label_text, font=font, fill="white")

    return PILHelper.to_native_format(d, image)


# Prints key state change information, updates rhe key image and performs any
# associated actions when a key is pressed.
def key_change_callback(d, key, state):
    # Print new key state
    print("Deck {} Key {} = {}".format(d.id(), key, state), flush=True)
    try:
        outobj["data"]["deckid"] = dnr[d.id()]
        outobj["data"]["key"] = key
        outobj["data"]["pressed"] = state
        jsonstring = json.dumps(outobj)
        client.publish("homeauto/streamdeck", jsonstring.encode())
    except:
        print("Error sending data")


def on_connect(client, userdata, flags, rc):
    print("Connected with result code " + str(rc))
    client.subscribe("homeauto/streamdeck/control/#")


def on_message(client, userdata, msg):
    # Status Data
    if msg.topic == "homeauto/streamdeck/control/status":
        try:
            data = json.loads(msg.payload.decode())
            if data["brightness"] is not None:
                if data["deckid"] is not None:
                    for index, d in enumerate(streamdecks):
                        if data["deckid"] == dnr[d.id()]:
                            d.set_brightness(data["brightness"])
        except:
            print("Wrong Status data")
    # Config Data
    if msg.topic == "homeauto/streamdeck/control/config":
        try:
            cfgfile = open("streamdeck.cfg", "w")
            cfgfile.write(msg.payload.decode())
            cfgfile.close()
            loadcfg()
        except:
            print("Wrong Config data")


def loadcfg():
    try:
        cfgfile = open("streamdeck.cfg", "r")
        data = cfgfile.read()
        data = json.loads(data)
        font = "Roboto-Regular.ttf"
        for index, d in enumerate(streamdecks):
            for i in range(0, len(data[d.get_serial_number()])):
                imgfile = data[d.get_serial_number()][i]["image"]
                txt = data[d.get_serial_number()][i]["text"]
                image = render_key_image(d, imgfile, font, txt)
                d.set_key_image(i, image)
                cfgfile.close()
    except:
        print("Error loading config")

if __name__ == "__main__":

    client = mqtt.Client()
    #client.username_pw_set(username='mqttuser',password='hierpasswort')
    client.connect_async("localhost", 1883, 60)
    client.on_connect = on_connect
    client.on_message = on_message
    client.loop_start()
    outobj = {}
    dnr = {}
    outobj.update({"data": { "deckid": "" , "key": 0, "pressed": False}})

    streamdecks = DeviceManager().enumerate()

    print("Found {} Stream Deck(s).\n".format(len(streamdecks)))

    for index, deck in enumerate(streamdecks):
        deck.open()
        deck.reset()

        print("Opened '{}' device (serial number: '{}')".format(
            deck.deck_type(), deck.get_serial_number()))

        # Set initial screen brightness to 30%.
        deck.set_brightness(30)
        
        dnr.update({ deck.id():  deck.get_serial_number() })
        
        # Register callback function for when a key state changes.
        deck.set_key_callback(key_change_callback)

        # Wait until all application threads have terminated (for this example,
        # this is when all deck handles are closed).
        
    loadcfg()
    
    for t in threading.enumerate():
        if t is threading.currentThread():
            continue

        if t.is_alive():
            t.join()

Die Konfiguration geht per topic „homeauto/streamdeck/control/config“ und ist json formatiert:

{
    "seriennummer_deck": [
        {
            "image": "lamp.png",
            "text": "Licht"
        },
        {
            "image": "user_green.png",
            "text": "Auto L."
        },
        {
            "image": "lamp3.png",
            "text": "Nachtt."
        },
        {
            "image": "aurora1.png",
            "text": "Aurora"
        },
        {
            "image": "aurora2.png",
            "text": "Aurora"
        },
        {
            "image": "mycomputer.png",
            "text": "PC"
        },
        {
            "image": "beamer.png",
            "text": "Beamer"
        },
        {
            "image": "temperature.png",
            "text": "Heizung"
        },
        {
            "image": "close.png",
            "text": "-"
        },
        {
            "image": "led.png",
            "text": "Display"
        },
        {
            "image": "display.png",
            "text": "Monitor"
        },
        {
            "image": "display.png",
            "text": "Steckd. 3"
        },
        {
            "image": "display.png",
            "text": "Steckd. 4"
        },
        {
            "image": "lamp4.png",
            "text": "Deko 1"
        },
        {
            "image": "close.png",
            "text": "-"
        }
    ], nächster datensatz...
	
}