Live Plot of CV Data



  • Hi

    I was wondering if there is an easy way I can plot the measurement data of the Rodeostat live while running e.g. a CV measurement? It should be noted that I am not in any way experienced when it comes to Python programming and I am also only a beginner when it comes to Arduino programming.
    I have managed to create a data collection system with live data plotting for our old manual potentiostat/galavanostat using Arduino and C# but I chose to order the Rodeostat instead of trying to develop a similar microcontroller based system myself.

    Best Regards
    Steffen Vindt



  • @Steffen

    Yes. The easiest way is probably to create a sub-class of the Potentiostat class with it own run_test method which creates the live plot instead of the progress bar. I give a simple example below.

    from __future__ import print_function
    import json
    import threading
    import Queue
    import matplotlib.pyplot as plt
    import potentiostat as ps
    
    
    class PotentiostatLivePlot(ps.Potentiostat):
        """
        Custom version of potentiostat with displays a live plot when running a test.
        """
    
        def __init__(self, port, timeout=10.0, debug=False):
            super(PotentiostatLivePlot,self).__init__(port,timeout=timeout,debug=debug)
            self.data_queue = Queue.Queue()
    
    
        def get_default_volt_lims(self):
            volt_range = self.get_volt_range()
            volt_range_value = float(volt_range[:-1])
            volt_lims = -volt_range_value, volt_range_value
            return volt_lims
    
    
        def get_default_curr_lims(self):
            curr_range = self.get_curr_range()
            curr_range_value = float(curr_range[:-2])
            curr_lims = -curr_range_value, curr_range_value
            return curr_lims
    
    
        def get_curr_unit(self):
            curr_range = self.get_curr_range()
            curr_unit = curr_range[-2:]
            return curr_unit
    
    
        def receive_data(self,timeunit):
            while True:
                # Get json data from the device
                sample_json = self.readline()
                sample_json = sample_json.strip()
                try:
                    sample_dict = json.loads(sample_json.decode())
                except ValueError:
                    continue
                # Put new values in data queue
                if len(sample_dict) > 0:
                    tval = sample_dict[ps.TimeKey]*ps.TimeUnitToScale[timeunit]
                    volt = sample_dict[ps.VoltKey]
                    curr = sample_dict[ps.CurrKey]
                    self.data_queue.put({'tval': tval, 'volt': volt, 'curr': curr})
                else:
                    self.data_queue.put({})
    
    
    
        def run_test(self, testname, param=None, filename=None, volt_lims=None, curr_lims=None, timeunit='s'):
    
            if timeunit not in ps.TimeUnitToScale:
                raise RuntimeError('uknown timeunit option {0}'.format(timeunit))
    
            if volt_lims is None:
                volt_lims = self.get_default_volt_lims()
    
            if curr_lims is None:
                curr_lims = self.get_default_curr_lims()
    
            curr_unit = self.get_curr_unit() 
    
            if param is not None:
                self.set_param(testname,param)
    
            if filename is not None:
                fid = open(filename,'w')
    
            t_done = self.get_test_done_time(testname, timeunit=timeunit)
    
            # Create figure and live plot
            fig = plt.figure()
            plt.ion()
    
            volt_axes = plt.subplot(211)
            volt_line, = plt.plot([0],[0],'b')
            plt.xlim(0,t_done)
            plt.ylim(*volt_lims)
            plt.grid('on')
            plt.ylabel('(V)')
            plt.title('Potentiostat Live Plot')
    
            curr_axes = plt.subplot(212,sharex=volt_axes)
            curr_line, = plt.plot([0],[0],'b')
            plt.xlim(0,t_done)
            plt.ylim(*curr_lims)
            plt.grid('on')
            plt.ylabel('({0})'.format(curr_unit))
            plt.xlabel('({0})'.format(timeunit))
    
            fig.canvas.flush_events()
            plt.pause(0.001)
    
            # Create lists for storing incoming data
            tval_list = [] 
            volt_list = [] 
            curr_list = []
    
            # Setup up working thread to collect data from device
            data_worker = threading.Thread(target=self.receive_data,args=(timeunit,))
            data_worker.daemon = True
    
            # Send command to run the test
            cmd_dict = {
                    ps.CommandKey: ps.RunTestCmd, 
                    ps.TestKey: testname
                    }
            msg_dict = self.send_cmd(cmd_dict)
            self.test_running = True
    
            data_worker.start()
            done = False
    
            while not done:
    
                # Get available data from data queue and add to list
                while self.data_queue.qsize() > 0:
                    data = self.data_queue.get(True)
                    if not data:
                        done = True # Empty data means run is complete
                        break
    
                    # Get data items and append to lists
                    tval = data['tval']
                    volt = data['volt']
                    curr = data['curr']
                    tval_list.append(tval)
                    volt_list.append(volt)
                    curr_list.append(curr)
    
                    # Write data to file
                    if filename is not None:
                        fid.write('{0:1.3f}, {1:1.4f}, {2:1.4f}\n'.format(tval,volt,curr))
    
                    print('{0:1.3f}, {1:1.4f}, {2:1.4f}'.format(tval,volt,curr))
    
                # Update live plot
                volt_line.set_xdata(tval_list)
                volt_line.set_ydata(volt_list)
                curr_line.set_xdata(tval_list)
                curr_line.set_ydata(curr_list)
                fig.canvas.flush_events()
                plt.pause(0.001)
    
            self.test_running = False
    
            if filename is not None:
                fid.close()
    
            return tval_list, volt_list, curr_list 
    

     

    Assuming you have saved this to a new module called potentiostat_live_plot.py then you could call it as follows.

     

    from potentiostat_live_plot import PotentiostatLivePlot
    
    dev = PotentiostatLivePlot(port='/dev/ttyACM0')
    
    dev.set_curr_range('100uA')
    dev.set_sample_period(10)
    
    name = 'cyclic'
    param = {
            'quietValue' : 0.0,
            'quietTime'  : 1000,
            'amplitude'  : 2.0,
            'offset'     : 0.0,
            'period'     : 1000,
            'numCycles'  : 15,
            'shift'      : 0.0,
            }
    
    t, volt, curr = dev.run_test(name,param,filename='data.txt', curr_lims=(-50,50))
    
    raw_input('Press enter to quit')
    

     

    The live plot should look something like the show image below.

     

    live plot image

     

    In this example there are two plots which are updated in realtime - one showing the output voltage vs time and the other showing the current vs time.



  • Hi Will

    Thanks a lot :)
    How should I change "potentiostat_live_plot.py" to plot a classical CV with the potential on the x-axis and the current on the y-axis?

    Best Regards
    Steffen



  • @Steffen

    Here is a modified version which plots the classical CV type plot

    from __future__ import print_function
    import json
    import threading
    import Queue
    import matplotlib.pyplot as plt
    import potentiostat as ps
    
    
    class PotentiostatLivePlot(ps.Potentiostat):
        """
        Custom version of potentiostat with displays a live plot when running a test.
        """
    
        def __init__(self, port, timeout=10.0, debug=False):
            super(PotentiostatLivePlot,self).__init__(port,timeout=timeout,debug=debug)
            self.data_queue = Queue.Queue()
    
    
        def get_default_volt_lims(self):
            volt_range = self.get_volt_range()
            volt_range_value = float(volt_range[:-1])
            volt_lims = -volt_range_value, volt_range_value
            return volt_lims
    
    
        def get_default_curr_lims(self):
            curr_range = self.get_curr_range()
            curr_range_value = float(curr_range[:-2])
            curr_lims = -curr_range_value, curr_range_value
            return curr_lims
    
    
        def get_curr_unit(self):
            curr_range = self.get_curr_range()
            curr_unit = curr_range[-2:]
            return curr_unit
    
    
        def receive_data(self,timeunit):
            while True:
                # Get json data from the device
                sample_json = self.readline()
                sample_json = sample_json.strip()
                try:
                    sample_dict = json.loads(sample_json.decode())
                except ValueError:
                    continue
                # Put new values in data queue
                if len(sample_dict) > 0:
                    tval = sample_dict[ps.TimeKey]*ps.TimeUnitToScale[timeunit]
                    volt = sample_dict[ps.VoltKey]
                    curr = sample_dict[ps.CurrKey]
                    self.data_queue.put({'tval': tval, 'volt': volt, 'curr': curr})
                else:
                    self.data_queue.put({})
    
    
    
        def run_test(self, testname, param=None, filename=None, volt_lims=None, curr_lims=None, timeunit='s'):
    
            if timeunit not in ps.TimeUnitToScale:
                raise RuntimeError('uknown timeunit option {0}'.format(timeunit))
    
            if volt_lims is None:
                volt_lims = self.get_default_volt_lims()
    
            if curr_lims is None:
                curr_lims = self.get_default_curr_lims()
    
            curr_unit = self.get_curr_unit() 
    
            if param is not None:
                self.set_param(testname,param)
    
            if filename is not None:
                fid = open(filename,'w')
    
            t_done = self.get_test_done_time(testname, timeunit=timeunit)
    
            # Create figure and live plot
            fig = plt.figure()
            plt.ion()
    
            ax = plt.subplot(111)
            line, = plt.plot([0],[0],'b')
            plt.xlim(*volt_lims)
            plt.ylim(*curr_lims)
            plt.grid('on')
            plt.xlabel('(V)')
            plt.ylabel('({0})'.format(curr_unit))
            plt.title('Potentiostat Live Plot')
    
            fig.canvas.flush_events()
            plt.pause(0.001)
    
            # Create lists for storing incoming data
            tval_list = [] 
            volt_list = [] 
            curr_list = []
    
            # Setup up working thread to collect data from device
            data_worker = threading.Thread(target=self.receive_data,args=(timeunit,))
            data_worker.daemon = True
    
            # Send command to run the test
            cmd_dict = {
                    ps.CommandKey: ps.RunTestCmd, 
                    ps.TestKey: testname
                    }
            msg_dict = self.send_cmd(cmd_dict)
            self.test_running = True
    
            data_worker.start()
            done = False
    
            while not done:
    
                # Get available data from data queue and add to list
                while self.data_queue.qsize() > 0:
                    data = self.data_queue.get(True)
                    if not data:
                        done = True # Empty data means run is complete
                        break
    
                    # Get data items and append to lists
                    tval = data['tval']
                    volt = data['volt']
                    curr = data['curr']
                    tval_list.append(tval)
                    volt_list.append(volt)
                    curr_list.append(curr)
    
                    # Write data to file
                    if filename is not None:
                        fid.write('{0:1.3f}, {1:1.4f}, {2:1.4f}\n'.format(tval,volt,curr))
    
                    print('{0:1.3f}, {1:1.4f}, {2:1.4f}'.format(tval,volt,curr))
    
                # Update live plot
                line.set_xdata(volt_list)
                line.set_ydata(curr_list)
                fig.canvas.flush_events()
                plt.pause(0.001)
    
            self.test_running = False
    
            if filename is not None:
                fid.close()
    
            return tval_list, volt_list, curr_list 
    

     
    Note, this version plots the potential and current during the quiet period as well as during the test. However, it should be pretty easy to modify if don't want to show the data during the quiet period. Maybe just discard the incoming data (e.g. don't append it to the lists) until the quiet period is complete.



  • Another way would be to raise an async event each time a new value is obtained from the serial?. (it would be triggered "sample_rate" times per second, just as each val is received from the teensy).

    That approach could enable any component to register for the event and update itself with each new value, without making the class responisble for plotting. Just thinking out loud.

    Right around here:

            if len(sample_dict) > 0:
                tval = sample_dict[ps.TimeKey]*ps.TimeUnitToScale[timeunit]
                volt = sample_dict[ps.VoltKey]
                curr = sample_dict[ps.CurrKey]
                self.data_queue.put({'tval': tval, 'volt': volt, 'curr': curr})
    

    I'll work some code over march and commit to the blog. I´ll also work on some bindings to store long series on DB tables instead of datafile.txt.

    Question... can the iorodeo repos be forked with GIT? cheers.



  • @pguillem

    I think separating the plotting from class is a nice idea. And I like the idea of using an event. You might even get away with just the data queue (and no event) ... I hacked together a simple example below

    from __future__ import print_function
    import json
    import threading
    import Queue
    import potentiostat as ps
    
    
    class PotentiostatWithQueue(ps.Potentiostat):
    
        def __init__(self, port, timeout=10.0, debug=False):
            super(PotentiostatWithQueue,self).__init__(port,timeout=timeout,debug=debug)
            self.data_queue = Queue.Queue()
    
        def receive_data(self,timeunit):
    
            while self.test_running:
                # Get json data from the device
                sample_json = self.readline()
                sample_json = sample_json.strip()
                try:
                    sample_dict = json.loads(sample_json.decode())
                except ValueError:
                    continue
                # Put new values in data queue
                if len(sample_dict) > 0:
                    tval = sample_dict[ps.TimeKey]*ps.TimeUnitToScale[timeunit]
                    volt = sample_dict[ps.VoltKey]
                    curr = sample_dict[ps.CurrKey]
                    self.data_queue.put({'tval': tval, 'volt': volt, 'curr': curr})
                else:
                    self.data_queue.put({})
                    self.test_running = False
    
        def run_test(self, testname, param=None, timeunit='s'):
    
            if timeunit not in ps.TimeUnitToScale:
                raise RuntimeError('uknown timeunit option {0}'.format(timeunit))
            if param is not None:
                self.set_param(testname,param)
    
            # Setup up working thread to collect data from device
            data_worker = threading.Thread(target=self.receive_data,args=(timeunit,))
            data_worker.daemon = True
            self.data_queue = Queue.Queue()
    
            # Send command to run the test
            cmd_dict = {
                    ps.CommandKey: ps.RunTestCmd, 
                    ps.TestKey: testname
                    }
            msg_dict = self.send_cmd(cmd_dict)
            self.test_running = True
            data_worker.start()
    

     
    Then some other part of your program could pull data from the queue as it comes it and you could plot it as you want. Here is a quick and dirty example. I'm assuming the the above snippet is saved as potentiostat_with_queue.py.

    from __future__ import print_function
    import Queue
    from potentiostat_with_queue import PotentiostatWithQueue
    import matplotlib.pyplot as plt
    
    volt_lims = (-2.1,2.0)
    curr_lims = (-50,50)
    
    dev = PotentiostatWithQueue(port='/dev/ttyACM0')
    
    dev.set_curr_range('100uA')
    dev.set_sample_period(10)
    
    name = 'cyclic'
    param = {
            'quietValue' : 0.0,
            'quietTime'  : 1000,
            'amplitude'  : 2.0,
            'offset'     : 0.0,
            'period'     : 2000,
            'numCycles'  : 2,
            'shift'      : 0.0,
            }
    
    dev.set_param(name,param)
    t_done = dev.get_test_done_time(name)
    
    # Create figure and live plot
    fig = plt.figure()
    plt.ion()
    
    ax = plt.subplot(111)
    line, = plt.plot([0],[0],'b')
    plt.xlim(*volt_lims)
    plt.ylim(*curr_lims)
    plt.grid('on')
    plt.xlabel('(V)')
    plt.ylabel('(uA)')
    plt.title('Potentiostat Live Plot')
    
    fig.canvas.flush_events()
    plt.pause(0.001)
    
    # Lists for incoming data
    tval_list = [] 
    volt_list = [] 
    curr_list = []
    
    # Start test
    done = False
    dev.run_test(name)
    
    while not done:
    
        # Get data from queue and add store for plotting
        have_new_data = False
        while True:
    
            try:
                data = dev.data_queue.get(False)
            except Queue.Empty:
                break
    
            if data:
                tval = data['tval']
                volt = data['volt']
                curr = data['curr']
                print('{0:1.3f}, {1:1.4f}, {2:1.4f}'.format(tval,volt,curr))
                tval_list.append(tval)
                volt_list.append(volt)
                curr_list.append(curr)
                have_new_data = True
            else:
                done = True
                break
    
        # Update live plot
        if have_new_data:
            line.set_xdata(volt_list)
            line.set_ydata(curr_list)
            fig.canvas.flush_events()
            plt.pause(0.001)
    
    raw_input('done')
    

     
    In a GUI you might use something a bit different than the while loop shown in the simple example above - maybe a timer callback e.g., something which sets the rate at which you want to update your plot. I'm guessing you probably won't always want to update the plot as frequently as data comes in from the device - it would depend on the sample frequency (or period). The sample rate can be anything from very slow up to 1000Hz which might be a bit fast for updating the plot in the GUI (or at least there is not reason to update the plot that fast). So the idea would be to empty the data queue (which could have multiple samples in it) and then update plot in each timer callback.

    The repository on bitbucket uses mercurial - so you can't directly fork with git. You might be able to use something like git-remote-hg https://github.com/felipec/git-remote-hg

    In the past I've had OK luck with hg-git for mercurial/git for keeping hg and git repos in sync. http://hg-git.github.io/ It can be a little tricky.

    You could also use this https://help.github.com/articles/about-github-importer/ However, this is probably only one way trip e.g no going back to hg. So maybe not the best for sharing.

    Of course you could also just use mercurial (hg). If you know git then it would probably only take a few minutes to orient yourself with mercurial https://github.com/sympy/sympy/wiki/Git-hg-rosetta-stone



  • @pguillem
    Good point. I believe it will also be a more logical structure when it comes to GUI development.

    @Will-Dickson
    Thanks a lot Will. Both scripts works well.



  • @Will-Dickson

    Nice code. Do you think this would work for chronoamperometry if I change the parameters in the second script?



  • @Rodeo_First

    Yes. However for chronoamperometry you will probably want to change the plot so that it displays current vs time rather than current vs voltage as the voltages are just stepped in chronoamperometry. Even better you could have two real-time plots - one showing the voltage vs time and the other showing the current vs time. This way you could see when the voltages are stepped.



  • @Will-Dickson

    Thanks again, it was really helpful.


Log in to reply