AquaDMX/aquadmx.py

import configparser
from enum import Enum
import time
import os

from stupidArtnet import StupidArtnetServer

from homeassistant_api import Processing, Client
from homeassistant_api.processing import process_json

# Globals
dataBuffer = []

# Set Home Assistant Light based on DMX device and data.
def setLight(device, data):
    start = device.getStartChannel()
    if device.getLightType() == LightType.DIMMER:
        light.turn_on(entity_id=device.getDeviceID(),
                      brightness=data[start],
                      transition=0)
    elif device.getLightType() == LightType.RGB:
        light.turn_on(entity_id=device.getDeviceID(),
                  brightness=data[start],
                  transition=0,
                  rgb_color=[data[start+1], data[start+2], data[start+3]])

# DMX callback for changes to the universe.
def dmx_callback(data):
    global dataBuffer
    if dataBuffer:
        diffs = [i[0]+1 for i in enumerate(data) if data[i[0]] != dataBuffer[i[0]]]
        for d in u1.getDevicesByID(diffs):
            setLight(d, data)
    else:
        # Set all lights on first
        for d in u1.getDevices():
            setLight(d, data)
    dataBuffer = data

# Get a list of items in a config key
def configList(config, key):
    return [i.strip(' ') for i in config[key].split('\n')]

# Returns a list of config items.
def configOptionalList(config, key):
    if key in config:
        return configList(config, key)
    else:
        return []

# Types of Lights Supported.
class LightType(Enum):
    DIMMER = 1
    RGB = 2

# Data structure for a DMX Device.
class DMXDevice(object):

    startChannel = 0

    # Initialize DMX Device.
    def __init__(self, deviceID, lightType):
        self.deviceID = deviceID
        self.lightType = lightType
        if lightType is LightType.DIMMER:
            self.numChannels = 1
        elif lightType is LightType.RGB:
            self.numChannels = 4

    # Returns number of channels the device uses.
    def getNumChannels(self):
        return self.numChannels

    # Set the start channel for the device.
    def setStartChannel(self, start):
        self.startChannel = start

    # Get the start channel for the device.
    def getStartChannel(self):
        return self.startChannel

    # Get the end channel for the device.
    def getEndChannel(self):
        return self.startChannel+self.numChannels-1

    # Get the type of light.
    def getLightType(self):
        return self.lightType

    # Get the device ID for Home Assistant.
    def getDeviceID(self):
        return self.deviceID

    # Print settings of the device.
    def print(self):
        print('{}: {}-{}'.format(self.deviceID, self.getStartChannel(), self.getEndChannel()))
            
# Data structure for a DMX Universe.
class DMXUniverse(object):

    maxChannels = 512

    # initialize DMX Universe.
    def __init__(self):
        self.channels = 1
        self.devices = []

    # Add a device to the DMX Universe.
    def addDevice(self, device):
        device.setStartChannel(self.channels)
        self.devices.append(device)
        self.channels += device.getNumChannels()

    # Get channels in use in the DMX Universe.
    def getChannels(self):
        return self.channels

    # Get a list of devices in the DMX Universe.
    def getDevices(self):
        return self.devices

    # Get a list of device in the DMX Universe that match the given
    # list of channels.
    def getDevicesByID(self, channels):
        l = []
        for d in self.devices:
            for c in channels:
                if d.getStartChannel() <= c and c <= d.getEndChannel():
                    l.append(d)
        return set(l)

    # Print DMX Universe settings.
    def print(self):
        for d in self.devices:
            d.print()

# Configs
config = configparser.ConfigParser()
config.sections()
programPath = os.path.dirname(os.path.realpath(__file__))
config.read('{}/settings.ini'.format(programPath))

# Home Assistant
URL = '{}/api'.format(config['HomeAssistant']['url'])
TOKEN = config['HomeAssistant']['token']
client = Client(URL, TOKEN)
light = client.get_domain('light')

# Lights
u1 = DMXUniverse()
dimmers = configOptionalList(config['Lights'], 'dimmer')
for l in dimmers:
    u1.addDevice(DMXDevice(l, LightType.DIMMER))
rgbs = configOptionalList(config['Lights'], 'rgb')
for l in rgbs:
    u1.addDevice(DMXDevice(l, LightType.RGB))

# DMX
universe = 1
server = StupidArtnetServer()
u1_listener = server.register_listener(universe,
                                       callback_function=dmx_callback)

# Start server
print('AquaDMX is listening')
print()
print('Home Assistant ID: DMX Channel Start-DMX Channel End')
u1.print()
while True:
    time.sleep(1)
    pass