pixelserver2/main.py

#!/usr/bin/env python3
import config
import subprocess
import os
import os.path
import serial
import threading
import json
import bottle
from bottle import route, run, request, post
import time
import sys
import logging
import math

logging.basicConfig(filename='pixelserver.log', level=config.LogLevel)
running = True

########################################################################
#                                Utils                                 #
########################################################################
class DataSource:
    def __init__(self, initial):
        self.data = initial
        self.listeners = []
    def getData(self):
        return self.data
    def addListener(self, listener):
        self.listeners.append(listener)
        return self
    def pushData(self, data):
        self.data = data
        for listener in self.listeners:
            with listener:
                listener.notify_all()

class WatchDog(threading.Thread):
    def __init__(self, check, action):
        super().__init__()
        self.check = check
        self.action = action
        self.running = True
    def run(self):
        while running and self.running:
            if self.check():
                logging.error("Watchdog: Executed")
                self.action()
            time.sleep(1)
    def stop(self):
        self.running = False

class LogReader(threading.Thread):
    def __init__(self, runner):
        super().__init__()
        self.runner = runner
        self.log = ""
        self.running = True
    def clear(self):
        self.log = ""
    def getLog(self):
        return self.log
    def run(self):
        logging.info("LogReader started")
        while running and self.running:
            try:
                self.log += self.runner.app.stderr.read(1).decode("utf-8")
            except Exception as e:
                print(e)
                logging.error(str(e))
                time.sleep(1) 
        logging.info("LogReader closed")
    def stop(self):
        self.running = False

########################################################################
#                                  GUI                                 #
########################################################################
if config.UseGui:
    import pygame

    class Gui(threading.Thread):
        def __init__(self, datasource):
            super().__init__()
            self.cv = threading.Condition()
            self.datasource = datasource.addListener(self.cv)
        def run(self):
            logging.info("Starting GUI")
            sf = config.GuiScaleFactor
            screen = pygame.display.set_mode((sf*config.ScreenX, sf*config.ScreenY))
            pygame.display.set_caption("Pixelserver - GUI Vis")
            while running:
                for event in pygame.event.get():
                    pass
                with self.cv:
                    self.cv.wait()
                    data = self.datasource.getData()
                screen.fill((0, 255, 0))
                for x in range(config.ScreenX):
                    for y in range(config.ScreenY):
                        i = x+y*config.ScreenX
                        color = (data[i*3+0], data[i*3+1], data[i*3+2])
                        pygame.draw.rect(screen, color, pygame.Rect(sf*x, sf*y, sf, sf))
                pygame.display.flip()
            logging.info("Closing GUI")
        def join(self):
            with self.cv:
                self.cv.notify_all()
            super().join()

########################################################################
#                                 Serial                               #
########################################################################
class SerialWriter(threading.Thread):
    def __init__(self, datasource):
        super().__init__()
        self.cv = threading.Condition()
        self.datasource = datasource.addListener(self.cv)
        self.updateGamma = False
    def run(self):
        should_connect = True
        ser = None
        logging.info("Starting SerialWriter")
        while running:
            try:
                if should_connect:
                    ser = serial.Serial(config.Serial)
                    should_connect = False
                    logging.info("Serial Opened")
                with self.cv:
                    self.cv.wait(timeout = 1/30)
                    data = self.datasource.getData()
                    if self.updateGamma:
                        buf = bytearray(b"\x00")*3*256
                        for i in range(256):
                            apply = lambda x, g: max(0, min(255, int(math.pow(x/255, g)*255)))
                            buf[i]     = apply(i, self.r)
                            buf[i+256] = apply(i, self.g)
                            buf[i+512] = apply(i, self.b)
                        ser.write(b"\x02")
                        ser.write(buf)
                        self.updateGamma = False
                ser.write(b"\01")
                ser.write(data)
                ser.flush()
            except Exception as e:
                if ser != None:
                    ser.close()
                    ser = None
                logging.warning("Serial was close because: "+str(e))
                should_connect = True
                time.sleep(5)
        logging.info("Closing SerialWriter")
    def join(self):
        with self.cv:
            self.cv.notify_all()
        super().join()
    def setGamma(self, r, g, b):
        with self.cv:
            self.r, self.g, self.b = r, g, b
            self.updateGamma = True
            self.cv.notify_all()

########################################################################
#                                   App                                #
########################################################################
class App(threading.Thread):
    def __init__(self, cmd, param, listener, is_persistent):
        super().__init__()
        #start app
        if type(cmd) != list:
            cmd = [cmd,]
        args = cmd+[str(config.ScreenX), str(config.ScreenY), param]
        self.app = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True)
        self.last_update = time.time()
        self.cv = threading.Condition()
        self.watchdog = WatchDog(lambda: self.isAppTimedOut(), lambda: self.terminateApp())
        self.watchdog.start()
        self.logreader = LogReader(self)
        self.logreader.start()
        self.datasource = DataSource(b"\x00"*config.ScreenX*config.ScreenY*3)
        self.running = True
        self.listener = listener
        self.is_persistent = is_persistent
    def run(self):
        while running and self.running and self.alive():
            oshandle = self.app.stdout.fileno()
            try:
                data = os.read(oshandle, config.ScreenX*config.ScreenY*3)
                assert len(data) == config.ScreenX*config.ScreenY*3
                self.last_update = time.time()
                self.datasource.pushData(data)
            except Exception as e:
                logging.debug("Exception in App.run")
        with self.listener:
            self.listener.notify_all()
        self.watchdog.stop()
        self.logreader.stop()
        self.watchdog.join()
        self.logreader.join()
    def alive(self):
        return self.app.poll() == None
    def stop(self):
        self.running = False
        self.app.kill()
        self.app.stdout.close()
        self.app.stderr.close()
        self.watchdog.stop()
        self.logreader.stop()
    def getLog(self):
        return self.logreader.getLog()
    def terminateApp(self):
        logging.error("Terminate app!")
        self.stop()
    def isAppTimedOut(self):
        return time.time()-self.last_update > config.NoDataTimeout

########################################################################
#                                  Main                                #
########################################################################
class AppRunner(threading.Thread):
    def __init__(self):
        super().__init__()
        self.currentApp = -1
        self.requestedApp = 0
        self.app = None
        self.cv = threading.Condition()
        self.param = ""
        self.datasource = DataSource(b"\x00"*config.ScreenX*config.ScreenY*3)
        self.serial = SerialWriter(self.datasource)
        self.serial.start()
        self.persistent_apps = {}
        #start persistent apps
        for app, i in zip(config.Apps, range(len(config.Apps))):
            if app["persistent"]:
                self.startApp(i)
        if config.UseGui:
            self.gui = Gui(self.datasource)
            self.gui.start()
    def requestApp(self, app, param=""):
        with self.cv:
            self.requestedApp = app
            self.param = param
            self.cv.notify_all()
        logging.info("Requesting app: "+str(app))
    def startApp(self, i, param=""):
        app = config.Apps[i]
        newapp = App(app["cmd"], param, self.cv, is_persistent=app["persistent"])
        newapp.datasource.addListener(self.cv)
        newapp.start()
        if app["persistent"]:
            self.persistent_apps[self.currentApp] = newapp
        return newapp
    def updateApp(self):
        if self.app != None and not self.app.is_persistent:
            self.app.stop()
        self.currentApp = self.requestedApp
        if (self.currentApp in self.persistent_apps.keys() 
            and self.persistent_apps[self.currentApp].alive()):
            self.app = self.persistent_apps[self.currentApp]
        else:
            self.app = self.startApp(self.requestedApp, self.param)
    def run(self):
        logging.info("Starting Apprunner")
        while running:
            with self.cv:
                if self.app == None or not self.app.alive():
                    self.requestedApp = 0   
                if self.requestedApp != None:
                    self.updateApp()
                    self.requestedApp = None
                data = self.app.datasource.getData()
                self.datasource.pushData(data)
                self.cv.wait()
        self.serial.join()
        if config.UseGui:
            self.gui.join()
        logging.info("Close Apprunner")
    def getLog(self):
        if self.app == None:
            return ""
        return self.app.getLog()
    def setGamma(self, r, g, b):
        self.serial.setGamma(r, g, b)

########################################################################
#                              Web Api                                 #
########################################################################
@bottle.route('/<:re:.*>', method='OPTIONS')
def enable_cors_generic_route():
    add_cors_headers()

@bottle.hook('after_request')
def enable_cors_after_request_hook():
    add_cors_headers()

def add_cors_headers():
    bottle.response.headers['Access-Control-Allow-Origin'] = '*'
    bottle.response.headers['Access-Control-Allow-Methods'] = \
        'GET, POST, PUT, OPTIONS'
    bottle.response.headers['Access-Control-Allow-Headers'] = \
        'Origin, Accept, Content-Type, X-Requested-With, X-CSRF-Token'

def startApp(name, param=""):
    for i in range(len(config.Apps)):
        if config.Apps[i]["name"] == name:
            runner.requestApp(i, param)
            return "ok"
    return "not_found"

@route("/apps/list")
def apps_list():
    s = []
    for app in config.Apps:
        s.append({"name":       app["name"],
                  "guiname":    app["guiname"],
                  "persistent": app["persistent"]})
    return json.dumps(s)
    
@route("/apps/start/<name>")
def apps_start_param(name):
    return startApp(name)

@post("/apps/start/<name>")
def apps_start_post(name):
    param = request.forms.get('param')
    return startApp(name, param)
    
@route("/apps/start/<name>/<param>")
def apps_start(name, param):
    return startApp(name, param)

@route("/apps/log")
def apps_log():
    return runner.getLog()

@route("/apps/running") 
def  apps_running():
    return config.Apps[runner.currentApp]["name"]

@route("/")
def index():
    return bottle.static_file("index.html", root='html')

@route("/setgamma/<r>/<g>/<b>")
def setGamma(r, g, b):
    r = float(r)
    g = float(g)
    b = float(b)
    runner.setGamma(r, g, b)
    return "ok"

########################################################################
#                              Startup                                 #
########################################################################
#normalize config
for app in config.Apps:
    if "persistent" not in app.keys():
        app["persistent"] = False
    if "guiname" not in app.keys():
        app["guiname"] = app["name"]
    cmd = app["cmd"]
    #remove non existing apps
    if type(cmd) == str and not os.path.isfile(cmd):
        config.Apps.remove(app)
        logging.warning("Removed app "+app["name"])

runner = AppRunner()
runner.start()

run(host=config.WebHost, port=config.WebPort)

########################################################################
#                              Shutdown                                #
########################################################################
running = False
runner.join()