Reading Real Time Values With Pyserial While Plotting

So here is the deal, I have a module which sends out data over the serial port at 9600 baud and I am using the matplotlib to plot that data in real time. I wrote this code #! pyth

Solution 1:

I know the code below seems long and overly complex for your simple problem, but manual optimization is usually more code and more potential bugs. That's why premature optimization is almost always a mistake.

In __init__ it sets up the plot, setting references for the axis, canvas, line (it starts with a line drawn off screen), and the pre-animation background. Additionally __init__ registers callbacks to handle resizing and shutdown. The on_resize callback is needed to update the background (used for the blit) when the window is resized. The on_close callback uses a lock to update the running status. I haven't eliminated all race conditions with this, but this works to prevent a _tkinter.TclError caused by trying to blit to a terminated Tk app. I only tested with Tk, and on only one machine. YMMV, and I'm open to suggestions.

In the run method I've added a call to canvas.flush_events(). This should keep the plot window from hanging if you try to drag the window around and resize it. The while loop in this method calls self.get_new_values() to set the data in the plot. It then updates the plot using the configured method. If self.blit is true, it uses canvas.blit, else it uses pyplot.draw.

The variable spf (samples per frame) controls how many samples are plotted in a frame. You can use it in your implementation of get_new_values to determine the number of bytes to read (e.g. 2 * self.spf for 2 bytes per sample). I've set the default to 120, which is 5 frames per second if your data rate is 600 samples per second. You have to find the sweet spot that maximizes throughput vs time resolution in the graph while also keeping up with the incoming data.

Reading your data into a NumPy array instead of using a Python list will likely speed up processing. Also, it would give you easy access to tools to downsample and analyze the signal. You can read into a NumPy array directly from a byte string, but make sure you get the endianess right:

>>>data = b'\x01\xff'#big endian 256 + 255 = 511>>>np.little_endian   #my machine is little endian
True
>>>y = np.fromstring(data, dtype=np.uint16); y  #so this is wrong
array([65281], dtype=uint16)
>>>if np.little_endian: y = y.byteswap()>>>y #fixed
array([511], dtype=uint16)

Code:

from __future__ import division
from matplotlib import pyplot
import threading

classMain(object):
    def__init__(self, samples_per_frame=120, blit=True):
        self.blit = blit
        self.spf = samples_per_frame
        pyplot.ion()
        self.ax = pyplot.subplot(111)
        self.line, = self.ax.plot(range(self.spf), [-1] * self.spf)
        self.ax.axis([0, self.spf, 0, 65536])
        pyplot.draw()
        self.canvas = self.ax.figure.canvas
        self.background = self.canvas.copy_from_bbox(self.ax.bbox)

        self.canvas.mpl_connect('resize_event', self.on_resize)
        self.canvas.mpl_connect('close_event', self.on_close)
        self.lock = threading.Lock()
        self.run()

    defget_new_values(self):
        import time
        import random
        #simulate receiving data at 9600 bps (no cntrl/crc)
        time.sleep(2 * self.spf * 8 / 9600)
        y = [random.randrange(65536) for i inrange(self.spf)]
        self.line.set_ydata(y)

    defon_resize(self, event):
        self.line.set_ydata([-1] * self.spf)
        pyplot.draw()
        self.background = self.canvas.copy_from_bbox(self.ax.bbox)

    defon_close(self, event):
        with self.lock:
            self.running = Falsedefrun(self):
        with self.lock:
            self.running = Truewhile self.running:
            self.canvas.flush_events()
            with self.lock:
                self.get_new_values()
            if self.running:
                if self.blit:
                    #blit for a higher frame rate
                    self.canvas.restore_region(self.background)
                    self.ax.draw_artist(self.line)
                    self.canvas.blit(self.ax.bbox)
                else:
                    #versus a regular draw
                    pyplot.draw()

if __name__ == '__main__':
    Main(samples_per_frame=120, blit=True)