Merge pull request #17 from Overdrivr/graph-cleanup#3

plotting-improvements

Author: Overdrivr

pytelemetrycli/cli.py

@@ -8,6 +8,7 @@
 from pytelemetrycli.runner import Runner
 from serial.tools import list_ports
 from pytelemetrycli.ui.superplot import Superplot, PlotType
+from threading import Lock
 
 def docopt_cmd(func):
     def fn(self, arg):
@@ -44,7 +45,13 @@ def __init__(self):
         self.transport = transports.SerialTransport()
         self.telemetry = Pytelemetry(self.transport)
         self.topics = Topics()
-        self.runner = Runner(self.transport,self.telemetry)
+        self.plots = []
+        self.plotsLock = Lock()
+        self.runner = Runner(self.transport,
+                             self.telemetry,
+                             self.plots,
+                             self.plotsLock,
+                             self.topics)
         self.telemetry.subscribe(None,self.topics.process)
 
         self.types_lookup = {'--s'    :  'string',
@@ -127,18 +134,41 @@ def do_plot(self, arg):
 
 Usage: plot <topic>
         """
-        if not self.topics.exists(arg['<topic>']):
-            print("Topic ",arg['<topic>']," unknown.")
 
-        t = self.topics.xytype(arg['<topic>'])
-        if t == 'indexed':
-            self.myplot = Superplot(arg['<topic>'],PlotType.indexed)
-        else:
-            self.myplot = Superplot(arg['<topic>'])
-        q = self.myplot.start()
-        self.topics.transfer(arg['<topic>'],q)
+        topic = arg['<topic>']
+
+        if not self.topics.exists(topic):
+            print("Topic ",topic," unknown.")
+            return
+
+        if self.topics.intransfer(topic):
+            print("Topic already plotted.")
+            return
+
+        plotTypeFlag = self.topics.xytype(arg['<topic>'])
+        plotType = PlotType.linear
+
+        if plotTypeFlag == 'indexed':
+            plotType = PlotType.indexed
+
+        p = Superplot(topic,plotType)
+        q, ctrl = p.start()
+
+        # Protect self.plots from modifications from the runner thread
+        self.plotsLock.acquire()
+
+        self.plots.append({
+            'topic': topic,
+            'plot': p,     # Plot handler
+            'queue': q,    # Data queue
+            'ctrl': ctrl   # Plot control pipe
+        })
+
+        self.plotsLock.release()
+
+        self.topics.transfer(topic,q)
 
-        print("Plotting:", arg['<topic>'],' in mode [',t,']')
+        print("Plotting:", topic,' in mode [',plotTypeFlag,']')
 
     @docopt_cmd
     def do_pub(self, arg):

pytelemetrycli/runner.py

@@ -3,10 +3,13 @@
 
 # Main class
 class Runner:
-    def __init__(self, transport, telemetry):
+    def __init__(self, transport, telemetry, plots, plotsLock, topics):
 
         self.transport = transport
         self.telemetryWrapper = telemetry
+        self.plots = plots
+        self.plotsLock = plotsLock
+        self.topics = topics
 
         self.thread = None
         self.running = threading.Event()
@@ -40,6 +43,28 @@ def terminate(self):
     def run(self):
         while self.running.is_set():
             if self.connected.is_set():
+                # Update protocol decoding
                 self.telemetryWrapper.update()
+
+                # Protect the self.plots data structure from
+                # being modified from the main thread
+                self.plotsLock.acquire()
+                
+                # Poll each poll pipe to see if user closed them
+                plotToDelete = None
+                for p, i in zip(self.plots,range(len(self.plots))):
+                    if p['ctrl'].poll():
+                        if p['ctrl'].recv() == "closing":
+                            plotToDelete = i
+                            break
+
+                # Delete a plot if needed
+                if plotToDelete is not None:
+                    self.plots[plotToDelete]['ctrl'].close()
+                    topic = self.plots[plotToDelete]['topic']
+                    self.topics.untransfer(topic)
+                    self.plots.pop(plotToDelete)
+
+                self.plotsLock.release()
             else:
                 time.sleep(0.5)

pytelemetrycli/test/test_topics.py

@@ -69,6 +69,8 @@ def test_transfert_queue_indexed_data():
 
     topic.transfer(t1,q)
 
+    assert topic.intransfer(t1) == True
+
     assert q.qsize() > 0
 
     assert q.get() == [5, 123]
@@ -86,3 +88,14 @@ def test_transfert_queue_indexed_data():
     assert q.get() == [6, 222]
     assert q.get() == [7, 333]
     assert q.get() == [8, 333]
+
+    topic.untransfer(t1)
+
+    assert topic.intransfer(t1) == False
+
+    topic.process(t1,111, {'index': 5})
+    topic.process(t1,222, {'index': 6})
+    topic.process(t1,333, {'index': 7})
+    topic.process(t1,333, {'index': 8})
+
+    assert q.qsize() == 0

pytelemetrycli/topics.py

@@ -68,5 +68,14 @@ def transfer(self,topic,queue):
                         queue.put([self.transfers[topic]['lastindex'], item])
                         self.transfers[topic]['lastindex'] += 1
 
+    def untransfer(self,topic):
+        # If the topic data is already transfered to some queue
+        if topic in self.transfers:
+            # Remove it from the transfer list
+            del self.transfers[topic]
+
+    def intransfer(self,topic):
+        return topic in self.transfers
+
     def xytype(self,topic):
         return self.topics[topic]['type']

pytelemetrycli/ui/superplot.py

@@ -2,8 +2,8 @@
 from pyqtgraph.Qt import QtGui, QtCore
 import numpy as np
 import pyqtgraph as pg
-from multiprocessing import Process, Manager, Queue
-import sched, time, threading
+from multiprocessing import Process, Queue, Pipe
+import time, threading
 from sortedcontainers import SortedDict
 from enum import Enum
 
@@ -12,12 +12,20 @@ class PlotType(Enum):
     indexed = 1
 
 class Superplot():
+    """
+Self-contained plotting class that runs in its own process.
+Plotting functionality (reset the graph, .. ?) can be controlled
+by issuing message-based commands using a multiprocessing Pipe
+
+    """
     def __init__(self,name,plottype=PlotType.indexed):
         self.name = name
         self.plottype = plottype
+        self._clear()
 
+    def _clear(self):
         # Process-local buffers used to host the displayed data
-        if plottype == PlotType.linear:
+        if self.plottype == PlotType.linear:
             self.set = True
             self.x = []
             self.y = []
@@ -30,15 +38,20 @@ def __init__(self,name,plottype=PlotType.indexed):
             self.set = False
 
     def start(self):
+        # The queue that will be used to transfer data from the main process
+        # to the plot
         self.q = Queue()
+        main_pipe, self.in_process_pipe = Pipe()
         self.p = Process(target=self.run)
         self.p.start()
-        return self.q
+        # Return a handle to the data queue and the control pipe
+        return self.q, main_pipe
 
     def join(self):
         self.p.join()
 
     def _update(self):
+        # Empty data queue and process received data
         while not self.q.empty():
             item = self.q.get()
             if self.plottype == PlotType.linear:
@@ -50,16 +63,34 @@ def _update(self):
                 # TODO : Eventually, need to find high performance alternative. Maybe numpy based
                 self.xy[item[0]] = item[1]
 
+        # Initialize view on data dictionnary only once for increased performance
         if not self.set:
             self.set = True
             self.x = self.xy.keys()
             self.y = self.xy.values()
 
+        # Refresh plot data
         self.curve.setData(self.x,self.y)
 
+        try:
+            if self.in_process_pipe.poll():
+                msg = self.in_process_pipe.recv()
+                self._process_msg(msg)
+        except:
+            # If the polling failed, then the application most likely shut down
+            # So close the window and terminate as well
+            self.app.quit()
+
+    def _process_msg(self, msg):
+        if msg == "exit":
+            # TODO : Remove this line ? Redundant with send after app.exec_() ?
+            self.in_process_pipe.send("closing")
+            self.app.quit()
+        elif msg == "clear":
+            self._clear()
 
     def run(self):
-        app = QtGui.QApplication([])
+        self.app = QtGui.QApplication([])
         win = pg.GraphicsWindow(title="Basic plotting examples")
         win.resize(1000,600)
         win.setWindowTitle('pyqtgraph example: Plotting')
@@ -70,9 +101,12 @@ def run(self):
         timer.timeout.connect(self._update)
         timer.start(50)
 
-        app.exec_()
-
-
+        self.app.exec_()
+        try:
+            self.in_process_pipe.send("closing")
+        except:
+            pass
+            # Process was done, no need to process exception
 
 if __name__ == '__main__':
     # This is function is responsible for faking some data (IO, serial port, etc)
@@ -105,19 +139,26 @@ def io_indexed(running,q):
     run.set()
 
     # create the plot
-    #s = Superplot("somePlot",PlotType.linear)
-    s = Superplot("somePlot",PlotType.indexed)
+    s = Superplot("somePlot",PlotType.linear)
+    #s = Superplot("somePlot",PlotType.indexed)
 
     # get the queue used to exchange data
-    q = s.start()
+    q, ctrlPipe = s.start()
 
     # start IO thread
-    #t = threading.Thread(target=io_linear, args=(run,q))
-    t = threading.Thread(target=io_indexed, args=(run,q))
+    t = threading.Thread(target=io_linear, args=(run,q))
+    #t = threading.Thread(target=io_indexed, args=(run,q))
     t.start()
 
+    while True:
+        action = input("Type 'q' to quit. Type 'clear' to reset the graph. Type 'exit' to close the graph but stay on main thread.")
+        if action == 'clear':
+            ctrlPipe.send('clear')
+        elif action == 'exit':
+            ctrlPipe.send('exit')
+        elif action == 'q':
+            break
 
-    input("Type Enter to quit.")
     run.clear()
     print("Waiting for IO thread to join...")
     t.join()