osc_led/osc_server.py

#!/usr/bin/python3
# Import needed modules from osc4py3
from osc4py3.as_eventloop import *
from osc4py3 import oscbuildparse
from osc4py3.oscmethod import *
import logging
from pprint import pprint
import argparse
import colorsys
import gpiozero
import osc_client
import json

# Set up basic vars and logging
logging.basicConfig(format='%(asctime)s - %(threadName)s ø %(name)s - '
            '%(levelname)s - %(message)s')
log = logging.getLogger("ledserver")
log.setLevel(logging.INFO)
osc_log=logging.getLogger("osc")
osc_log.setLevel(logging.WARNING)
clientname="ledserver"

# Get the normalised RGB values for the HSV (i.e, the colour float designating hue from OSC)
def hsv2rgb(h,s,v): 
    return tuple(round(i*255) for i in colorsys.hsv_to_rgb(h,s,v))

# Some little helper methods to set colours in the terminal output
def termrgb(rgb):
        return "\033[38;2;{};{};{}m".format(rgb[0],rgb[1],rgb[2])
def termendc():
        return '\033[0m'
def output_led_colour(rgb):
        return termrgb(rgb)+"█"+termendc()


# Our main class, the LED

class LED:
        # These should be largely self-explanatory
        # Float, from OSC FX param 2
        brightness=0
        # RGB array, currently as int 0-255, but will probably end up float 0-1.0, converted from hue
        colour=[0,0,0]
        #Float, from OSC FX param 1
        hue=0
        # Bool, Intial state, off. Should always align to whether the LED is on or off, brightness being >0
        state=False
        # The pin number of the LED
        pin=None
        # The OSC target FX param
        target=0
        name=""

        # List for logging state
        mode=["off","on"]

        # Initialize the object, set up GPIO
        def __init__(self,pin,target): 
            self.pin=pin
            self.target=target
            self.led=gpiozero.PWMLED(pin)
        # Set the brightness, this only triggers on a delta threshold so that we don't go crazy
        def set_brightness(self,value):
            if abs(self.brightness-value) > .005:
                self.brightness=value
                self.led.value=self.state*self.brightness
                log.info("Setting brightness for LED on pin {} to {}".format(self.pin, self.brightness))
                
        # Set the colour, only on a delta threshold        
        def set_colour(self,value):
            if abs(self.hue-value) > .01:
                self.colour=hsv2rgb(value,1,1)
                self.hue=value
                log.info("Setting colour for LED on pin {} to {} ({})".format(self.pin,self.colour,output_led_colour(self.colour)))

        # Set an absolute state, on or off. Brightness is set regardless, so we can turn it on with 40% brightness if desired
        def set_state(self,value):  
            self.state=value
            self.led.value=self.state*self.brightness
            log.info("Turning LED {} {}".format(self.pin, self.mode[self.state]))

        # Toggle on or off, edge detecting. I.e, send True to turn on, send True again to turn off. False is ignored
        def toggle_state(self,value):
            if (value):
                self.state=not self.state
                self.led.value=self.state*self.brightness
                log.info("Turning LED {} {}".format(self.pin, self.mode[self.state]))


# Set the colour of the LEDs. This only does the first LED, but when the REAPER controls are finalised, it'll be more controllable 
def led_colour(value):
    for name,led in leds.items():
        led.set_colour(value)

# Set the brightness of all LEDs. As above, when REAPER is finalised it'll be controllable (or everything, as required)
def led_brightness(value):
    for name,led in leds.items():
            led.set_brightness(value)

# Turn an individual LED on or off based on FX param 3+
def led_state(addr,value):
        target=addr.split("/")[6]
        for led in [led for name,led in leds.items() if led.target==target]:
            led.toggle_state(bool(value))

def btn_restart():
    initialise()
    osc_client.cmd_restart()

def btn_play():
    osc_client.cmd_play()

def btn_stop():
    osc_client.cmd_stop()

def initialise(filename='config.json'):
    global config
    global leds
    global buttons
    config=json.load(open(filename))

    log.info("Creating LEDs...")

    # Create LEDs
    if 'leds' in globals():
        del leds
    leds={led['name']:LED(led['pin'],led['target']) for led in config['outputs']['leds']}
    for led in config['outputs']['leds']:
            if "defaults" not in led:
                led['defaults']={}
                for i in config['outputs']['defaults']['led'].keys():
                        led['defaults'][i]=config['outputs']['defaults']['led'][i]

            state=led['defaults']["state"] if "state" in led["defaults"] else config['outputs']["defaults"]['led']["state"]
            colour=led['defaults']['colour'] if 'colour' in led['defaults'] else config['outputs']['defaults']['led']['colour']
            brightness=led['defaults']['brightness'] if 'brightness' in led['defaults'] else config['outputs']['defaults']['led']['brightness']


            leds[led['name']].set_state(bool(state)) 
            leds[led['name']].set_colour(colour)
            leds[led['name']].set_brightness(brightness) 
    # Don't re-initialise buttons, specifically because we expect this to be called by a button
    # And unassigning the calling button will probably break things
    if 'buttons' not in globals():
        buttons={}    
        for name,button in config['inputs']['buttons'].items():
              log.info('Binding button {} to pin {}'.format(name,button['pin']))
              buttons[name]=gpiozero.Button(button['pin'])
              log.info('Binding button {} when_pressed to {}'.format(name,button['when_pressed']))
              buttons[name].when_pressed=globals()['btn_'+button['when_pressed']]
              pprint(buttons[name].when_pressed)

    
# Set up command line arguments
parser = argparse.ArgumentParser(description='OSC listener for LEDs')
parser.add_argument('--listen',action='store',default='127.0.0.1:9000',help="Listen on a specific IP/port. Defaults to 127.0.0,.1:9000")
parser.add_argument('--debug-osc',action='store_true',default=False,help="Debug OSC messages. Very verbose")
args=parser.parse_args()

# Start the system.
log.info("Starting OSC...")



# Log OSC stuff if requested
if args.debug_osc:
    osc_startup(logger=osc_log)
else: 
    osc_startup()


# Split up port/address from command line
listen_address=args.listen.split(':')[0]
listen_port=args.listen.split(':')[1]


# Start up the OSC server
osc_udp_server(listen_address,listen_port, clientname)

# Set up osc paths
# Each FX param has a callback function that is triggered whenever a message for that path is received
log.info("Binding targets...")
osc_method("/track/2/fx/1/fxparam/1/value",led_colour)
osc_method("/track/2/fx/1/fxparam/2/value",led_brightness)
osc_method("/track/2/fx/1/fxparam/[!12]/value",led_state,argscheme=OSCARG_ADDRESS+OSCARG_DATAUNPACK)



initialise()
#restart_button=gpiozero.Button(2)
#restart_button.when_pressed=osc_client.cmd_restart


# Do a massive loop listening for messages. This should really be cleaner...
finished=False
log.info("Starting loop")
while not finished:
        osc_process()