data_recorder_with_live_plot.py

# This script is licensed under BSD License, see file LICENSE.txt.
# (c) TASKING Germany GmbH, 2023
#
#
# This script opens a window with chart and reads and draws values of target
# variable given in command line in real-time for 5 seconds. Since
# data acquisition is used, sampling times can be around 1ms (target dependent).
 
# This script requires 'matplotlib', 'numpy', and 'scipy' to be
# installed. They are available at:
#  http://matplotlib.sourceforge.net/
#  http://www.scipy.org/Download
 
import matplotlib
matplotlib.use('TkAgg')
 
import sys
import pylab as p
import time
import pylab
import numpy
import matplotlib.animation as animation
 
import isystem.connect as ic
 
 
winidea_id = ''
 
 
def printHelp():
    print('Usage:')
    print('  dataRecorderWithLivePlot.py <recordingTime> <viewTime> <miny> <maxy> <varName> <samplingTime>')
    print('    recordingTime - how long to record data in seconds')
    print('    viewTime     - time interval shown on X axis in seconds')
    print('    minY         - lower limit on Y axis')
    print('    maxY         - upper limit on Y axis')
    print('    varName      - name of the variable to record')
    print('    samplingTime - sampling time. It is rounded to the nearest ')
    print('                   lower available number (1s, 100 ms,')
    print('                   10 ms, 1 ms). If specified as 0, the lowest available ')
    print('                   sampling time is used.')
    print('Example:')
    print('  dataRecorderWithLivePlot.py 10 3 -300 20000 main_loop_counter 0.1 g_baseStruct.i_base 0')
 
 
def parseCmdLine():
 
    if len(sys.argv) < 7:
        printHelp()
        print()
        raise Exception('Not enough command line parameters!')
    
    samplingInfo = []
 
    recordingTime = float(sys.argv[1])
    viewTime = float(sys.argv[2])
    minY = float(sys.argv[3])
    maxY = float(sys.argv[4])
    
    # create (varName, samplingTime) pairs from params in command line
    for idx in range(5, len(sys.argv), 2):
        samplingInfo.append([sys.argv[idx], float(sys.argv[idx + 1])])
 
    return [recordingTime, viewTime, minY, maxY], samplingInfo
 
 
def initTarget():
    cmgr = ic.ConnectionMgr()
    cmgr.connect(ic.CConnectionConfig().instanceId(winidea_id))
 
    debugCtrl = ic.CDebugFacade(cmgr)
    print('Downloading ... ', end=' ')
    debugCtrl.download()
    print('finished!')
    debugCtrl.runUntilFunction('main')
    debugCtrl.waitUntilStopped()
    debugCtrl.run()
 
    return cmgr, debugCtrl
 
 
def realTimeToDAQTimeFlags(samplTime):
    """
    Converts the given sampling time in seconds to EDAQSamplingFlags.
    """
    if samplTime < 0.001:
        return ic.CDAQController.daqSampleMax
    elif samplTime < 0.01:
        return ic.CDAQController.daqSample1ms
    elif samplTime < 0.1:
        return ic.CDAQController.daqSample10ms
    elif samplTime < 1:
        return ic.CDAQController.daqSample100ms
    else:
        return ic.CDAQController.daqSample1s
    
    
def initDAQ(cmgr, samplingInfo):
    daqCtrl = ic.CDAQController(cmgr)
 
    # check if DAQ system is available
    daqInfo = daqCtrl.info()
    if daqInfo.getMaxItems() == 0:
        raise Exception("Data Acquisition (DAQ) system is not available.")
        
    print('MaxItems = ', daqInfo.getMaxItems())
 
    daqVariables = ic.DAQConfigVector()
    for varSamplData in samplingInfo:
        varName = varSamplData[0]
        if varSamplData[1] == 0:
            daqVariables.append(ic.CDAQConfigItem(varName))
        else:
            samplTime = realTimeToDAQTimeFlags(varSamplData[1])
            daqVariables.append(ic.CDAQConfigItem(varName, samplTime))
            
    # note the time of the DAQ system
    daqTimeStart = daqCtrl.status().getTime()
 
    daqCtrl.configure(daqVariables)
 
    # enable DAQ on the entire SoC
    daqCtrl.enableGlobal(True)
 
    return [daqCtrl, daqTimeStart]
 
 
def readValuesFromTarget():
    timeStamps = pylab.array([])
    values = []
    daqStatus = daqCtrl.status()
    # if any sample is available, display the status and print the samples  
    if daqStatus.getNumSamplesAvailable() > 0:
        if daqStatus.getOverflow():
            print('SAMPLING OVERFLOW!')
 
        # read available samples into daqSamples
        daqSamples = ic.DAQSampleVector()
        daqCtrl.read(daqSamples)
 
        for daqSample in daqSamples:
            timeStamps = pylab.append(timeStamps, (daqSample.getTime() - daqTimeStart) / 1e6)
            values.append(daqCtrl.getDataValue(daqSample).getLong())
 
    return timeStamps, values
 
 
def animate(i):
    global timeOffset, xData, yData, endTime, line
    
    timeStamps, values = readValuesFromTarget()
    
    if time.time() < endTime and values:
 
        print(values, i)
 
        toff = timeStamps[-1] - viewTime
        if toff > 0:
            timeStamps -= toff
            xData -= toff - timeOffset
            timeOffset = toff
 
        xData = pylab.append(xData, timeStamps)
        yData = pylab.append(yData, values)
        
        xData = xData[xData > 0]
        yData = yData[-len(xData):]
            
        line.set_xdata(xData) 
        line.set_ydata(yData)
        
    return line,
 
 
#Init only required for blitting to give a clean slate.
def init():
    global xData, yData
    
    line.set_xdata(xData)
    line.set_ydata(yData)
    return line,
 
 
def main():
    global daqCtrl, yData, xData, line, daqTimeStart, endTime, viewTime, timeOffset
 
    plotParams, samplingInfo = parseCmdLine()
    cmgr, debugCtrl = initTarget()
    daqCtrl, daqTimeStart = initDAQ(cmgr, samplingInfo)
    xData = pylab.array([])
    yData = pylab.array([])
    timeOffset = 0
    
    fig = pylab.figure()
    ax = fig.add_subplot(111)
    line, = ax.plot([0], [0])
 
    # define ranges on X and Y axis
    recordingTime, viewTime, minY, maxY = plotParams
    pylab.axis([0, viewTime, minY, maxY])
 
    startTime = time.time()
    endTime = startTime + recordingTime
    
    ani = animation.FuncAnimation(fig, animate, numpy.arange(0, 10), init_func=init,
                                  interval=25, blit=True)
 
    pylab.show()
 
 
if __name__ == '__main__':
    main()