Extracted App class from runner

main.py

@@ -14,18 +14,30 @@ logging.basicConfig(filename='pixelserver.log',level=config.LogLevel)
 
 running = True
 
+
+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)
+    def pushData(self, data):
+        self.data = data
+        for listener in self.listeners:
+            with listener:
+                listener.notify_all()
+
 if config.UseGui:
     import pygame
 
     class Gui(threading.Thread):
-        def __init__(self):
+        def __init__(self, datasource):
             super().__init__()
-            self.data = b"\x00"*config.ScreenX*config.ScreenY*3
+            self.datasource = datasource
-            self.lock = threading.Lock()
+            self.cv = threading.Condition()
-
+            self.datasource.addListener(self.cv)
-        def setData(self, data):
-            with self.lock:
-                self.data = data
 
         def run(self):
             global running
@@ -35,35 +47,36 @@ if config.UseGui:
             pygame.display.set_caption("Pixelserver - GUI Vis")
             while running:
                 for event in pygame.event.get():
-                    if event.type == pygame.QUIT:
+                    pass
-                        logging.debug("GUI quits app")
+                with self.cv:
-                        running = False
+                    self.cv.wait()
+                    data = self.datasource.getData()
                 screen.fill((0, 255, 0))
-                with self.lock:
                 try:
                     for x in range(config.ScreenX):
                         for y in range(config.ScreenY):
                             i = x+y*config.ScreenX
-                                r = (self.data[i*3+0])
+                            r = (data[i*3+0])
-                                g = (self.data[i*3+1])
+                            g = (data[i*3+1])
-                                b = (self.data[i*3+2])
+                            b = (data[i*3+2])
 
                             pygame.draw.rect(screen, (r, g, b), pygame.Rect(sf*x, sf*y, sf, sf))
                 except:
                     pass
-                time.sleep(1/30)
+                    raise
                 pygame.display.flip()
             logging.info("Closing GUI")
+        def join(self):
+            with self.cv:
+                self.cv.notify_all()
+            super().join()
 
 class SerialWriter(threading.Thread):
-    def __init__(self):
+    def __init__(self, datasource):
         super().__init__()
-        self.lock = threading.Lock()
+        self.cv = threading.Condition()
-        self.data = b"\x00"*config.ScreenX*config.ScreenY*3
+        self.datasource = datasource
-
+        self.datasource.addListener(self.cv)
-    def setData(self, data):
-        with self.lock:
-            self.data = data
     
     def run(self):
         should_connect = True
@@ -75,49 +88,120 @@ class SerialWriter(threading.Thread):
                     ser = serial.Serial(config.Serial)
                     should_connect = False
                     logging.info("Serial Opened")
-                with self.lock:
+                with self.cv:
+                    self.cv.wait(timeout = 1/30)
+                    data = self.datasource.getData()
                 ser.write(b"\01")
-                    ser.write(self.data)
+                ser.write(data)
-                time.sleep(1/120)
             except Exception as e:
                 if ser != None:
                     ser.close()
                     ser = None
                 logging.warning("Serial was close because: "+str(e))
                 should_connect = True
-                time.sleep(0.1)
+                time.sleep(5)
         logging.info("Closing SerialWriter")
+    def joint(self):
+        self.cv.notify_all()
+        super().join()
 
 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:
+        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:
+        while running and self.running:
             try:
                 self.log += runner.app.stderr.read(1).decode("utf-8")
             except Exception as e:
-                time.sleep(0.1) 
+                time.sleep(1) 
         logging.info("LogReader closed")
+    def stop(self):
+        self.running = False
 
+class App(threading.Thread):
+    def __init__(self, cmd, param, listener):
+        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 = running
+        self.listener = listener
+
+    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
 
 class AppRunner(threading.Thread):
     def __init__(self):
@@ -125,80 +209,58 @@ class AppRunner(threading.Thread):
         self.currentApp = -1
         self.requestedApp = 0
         self.app = None
-        self.lock = threading.Lock()
+        self.cv = threading.Condition()
         self.param = ""
-        self.serial = SerialWriter()
+        self.datasource = DataSource(b"\x00"*config.ScreenX*config.ScreenY*3)
+        self.serial = SerialWriter(self.datasource)
         self.serial.start()
-        self.last_update = time.time()
-        self.watchdog = WatchDog(lambda: self.appTimedOut(), lambda: self.terminateApp())
-        self.watchdog.start()
-        self.logreader = LogReader(self)
-        self.logreader.start()
         if config.UseGui:
-            self.gui = Gui()
+            self.gui = Gui(self.datasource)
             self.gui.start()
 
     def requestApp(self, app, param=""):
-        with self.lock:
+        with self.cv:
             self.requestedApp = app
             self.param = param
-
+            self.cv.notify_all()
-    def terminateApp(self):
+        logging.info("Requesting app: "+str(app))
-        if self.app != None:
-            logging.error("Terminate app "+config.Apps[self.currentApp]["name"])
-            # don't ask
-            self.app.terminate()
-        if self.app != None:
-            self.app.kill()
-            self.app.stdout.close()
-            self.app.stderr.close()
-            self.last_update = time.time()
-        self.requestedApp = 0
-
-    def appTimedOut(self):
-        return time.time()-self.last_update > config.NoDataTimeout
 
     def updateApp(self):
         if self.app != None:
-            self.app.terminate()
+            self.app.stop()
-            self.app.stderr.close()
+        logging.info("Starting app "+config.Apps[self.requestedApp]["name"])
         cmd = config.Apps[self.requestedApp]["cmd"]
         logging.debug(str(cmd))
-        if type(cmd) != list:
+        self.app = App(cmd, self.param, self.cv)
-            cmd = [cmd,]
+        self.app.datasource.addListener(self.cv)
-        args = cmd+[str(config.ScreenX), str(config.ScreenY), self.param]
+        self.app.start()
-        self.app = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True)
-        self.logreader.clear()
 
     def run(self):
         logging.info("Starting Apprunner")
         while running:
-            with self.lock:
+            with self.cv:
-                if self.app == None or self.app.poll() != None:
+                if self.app == None or not self.app.alive():
                     self.requestedApp = 0   
                 if self.requestedApp != None:
                     self.currentApp = self.requestedApp
                     self.updateApp()
                     self.requestedApp = None
-            d = self.app.stdout
+            
-            oshandle = d.fileno()
+                self.cv.wait()
-            try:
+            if self.app != None:
-                data = os.read(oshandle, config.ScreenX*config.ScreenY*3)
+                data = self.app.datasource.getData()
-                self.last_update = time.time()
+                self.datasource.pushData(data)
-                self.serial.setData(data)
-                if config.UseGui:
-                    self.gui.setData(data)
-            except:
-                logging.debug("Exception in Apprunner.run")
-        self.watchdog.join()
         self.serial.join()
-        self.logreader.join()
         if config.UseGui:
             self.gui.join()
         logging.info("Close Apprunner")
 
     def getLog(self):
-        return self.logreader.log
+        if self.app == None:
+            return ""
+        return self.app.getLog()
+
+
 
 runner = AppRunner()
 runner.start()