daq_recorder.py

# This script is licensed under BSD License, see file LICENSE.txt, or search for `License` in the SDK online help.
#
# (c) TASKING Germany GmbH, 2023
#
#
# This script simply connects to a most recently used instance of winIDEA
# and starts recording data using the fast data acquisition (DAQ) controller.
# In order to do this the application must be running on the target, otherwise
# no data will be received.
# 
# Specify variable names and their acquisition times in command line.
 
 
import sys
import time
import isystem.connect as ic
 
 
winidea_id = ''
 
 
def printHelp():
    print("Usage:   daqRecorder.py [--verbose] <var1 samplingRate1> <var2 samplingRate2> ... <varN samplingRateN>")
    print()
    print("         Max number of DAQ items (N) is displayed before the start of the DAQ acquisition process.")
    print()
    print("         --verbose: also prints details regarding the DAQ acquisition process.")
    print()
    print("         If 'samplingRateInSeconds' == 0, the smallest possible sampling")
    print("         time is used. Supported values: 0.001, 0.01, 0.1, 1 second.")
    print()
    print("         To read data from digital and analog channels use name")
    print("         DigitalIn.DIN# and AnalogIn.AIN# where # is the desired channel number.")
    print()
    print("         To read from a desired (32bit only) memory location use 0x#")
    print("         where # is the desired memory address.")
    print()
    print("Example: daqRecorder.py main_loop_counter 0.1 g_int 0.001")
    print("         records main_loop_counter each 100 ms, and g_int with 1 ms period.")
 
    
def initTarget(cmgr):
    """
    This function initializes the target. Customize it according to
    your needs.
    """
 
    debugCtrl = ic.CDebugFacade(cmgr)
    debugCtrl.download()
    debugCtrl.runUntilFunction('main')
    debugCtrl.waitUntilStopped()
 
    return 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
    
# Supports:
# DIN
# DIN0
# DIN27
# DIN13457
def getDinSignalMask(varName):
    prefix = 'DigitalIn.DIN'
    
    # No suffix
    if not varName.startswith(prefix):
        return 0, 0
    
    indices = varName[len(prefix):].strip()
    
    # just DIN
    if len(indices) == 0:
        return 8, 0xff
    
    mask = 0
    bitCount = 0
        
    # DIN346
    try:
        for c in indices:
            num = int(c)
            
            if (num >= 0  and  num < 8):
                bit = (1 << num)
                
                if mask & bit == 0:
                    mask |= bit
                    bitCount += 1
            else:
                sys.exit(-1)
    finally:
        return bitCount, mask
    
 
def recordVariables(cmgr,
                    debugCtrl,
                    variables,
                    isVerboseLogging):
    """
    Parameters:
 
    debugCtrl - iSYSTEM's debugCtrl controller, which provides access to target
    
    variables - array of two element arrays, where the first element contains
                variable name and the second element contains sampling interval,
                for example: [['main_loop_counter', 0.1], ['g_char', 0]].
                Sampling interval set to 0 means the smallest interval possible.
                
    isVerboseLogging - prints details regarding daq acquisition process.
    
    Returns:
        List of rows, where each row is a list containing time stamp and recorded
        values in the same order as variable names were specified. Example for
        two samples of three variables:
        [[0, 23, 45, 4.35],
        [0.1, 24, -525, 1.78]
        ]
 
    """
 
    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('System properties:')
    print('   MaxItems = ', daqInfo.getMaxItems())
    print('   MaxItemSize = ', daqInfo.getMaxItemSize())
    print()
 
    print('Configuration:')
    for varSamplData in variables:
        varName = varSamplData[0]
        samplingTime = varSamplData[1]
        print("   - '%s': %f"%(varName, samplingTime))
    print()
    
    daqVariables = ic.DAQConfigVector()
    for varSamplData in variables:
        varName = varSamplData[0]
        samplTime = realTimeToDAQTimeFlags(varSamplData[1])
        
        # Direct memory address reading
        if varName.startswith('0x'): # Sample: 0x20001000,2 for a 16bit integer @ address 20001000
            bySize = 4  # Default size of variables is 32bit.
            bitNumIdx = varName.index(',')
            if bitNumIdx > 0:
                bySizeStr = varName[bitNumIdx+1:]
                bySize = int(bySizeStr) // 8
                varName = varName[:bitNumIdx]
            print("   Size in bytes: %d"%(bySize))
 
            addr = varName[2:]
            memAddr = int(addr, 16)
            print("   addr: 0x%x"%(memAddr))
 
            daqVariables.append(ic.CDAQConfigItem(bySize, 0, memAddr, samplTime))
               
        # Reading a regular variable
        else:
            daqVariables.append(ic.CDAQConfigItem(varName, samplTime))
 
    daqCtrl.configure(daqVariables)
 
    # note the time of the DAQ system
    daqTimeStart = daqCtrl.status().getTime()
 
    # enable DAQ on the entire SoC
    daqCtrl.enableGlobal(True)
 
    numVars = len(variables)    
    varNames = []
    
    longestVarName = 0
    for varData in variables:
        varName = varData[0]
        varNames.append(varName)
        nameLen = len(varName)
        if nameLen > longestVarName:
            longestVarName = nameLen
            
    startTime = time.time()
    RECORDING_TIME_S = 5  # seconds
    RECORDING_END_TIME = startTime + RECORDING_TIME_S   
 
    while time.time() < RECORDING_END_TIME:
        daqStatus = daqCtrl.status()
        # if any sample is available, display the status and print the samples  
        if daqStatus.getNumSamplesAvailable() > 0:
            if isVerboseLogging:
                print(f"available no. of samples: {daqStatus.getNumSamplesAvailable()}")
            if daqStatus.getOverflow():
                print('SAMPLING OVERFLOW!')
 
            # read available samples into daqSamples
            daqSamples = ic.DAQSampleVector()
            
            if isVerboseLogging:
                t0 = time.time()
            daqCtrl.read(daqSamples)
            if isVerboseLogging:
                t1 = time.time()
            
            if isVerboseLogging:
                print()
                print("Sample acquisition duration: %fms"%((t1-t0)*1000))
                print("Sample count: %d"%(len(daqSamples)))
            
            # print 'Number of samples = ', daqSamples.size()
            for daqSample in daqSamples:
                
                varName = varNames[daqSample.getIndex()]
 
                sampleTimeMs = (daqSample.getTime() - daqTimeStart) / 1000
                columnIndex = daqSample.getIndex()+1
            
                var = daqCtrl.getDataValue(daqSample)
                
                if var.isTypeUnsigned()  or  var.isTypeSigned ():
                    value = var.getLong()
                elif var.isTypeFloat():
                    value = var.getDouble()
                elif var.isTypeAddress():
                    value = var.getAddress().m_aAddress
                elif var.isTypeCompound():
                    value = 'Struct'
 
                print(f"{varName:>12} @ {sampleTimeMs:9.3f} ms:  {value}")
                
        # exit the loop if 'q' was pressed
 
    daqCtrl.enableGlobal(False)
 
def main():
 
    if len(sys.argv) < 2 or '--help' in sys.argv:
        printHelp()
        return
 
    samplingInfo = []
    varNames = []
    isVerboseLogging = False
    
    if '--verbose' in sys.argv:
        isVerboseLogging = True
        sys.argv.remove('--verbose')
    
    # create (varName, samplingTime) pairs from params in command line
    # Example input parameters:
    #    a_my_signal 0.01 a_my_int 0 a_my_char 1 DigitalIn.DIN0 0.1 AnalogIn.AIN1 0.1 DigitalIn.DIN125
    # For digital mask testing: 
    #    DigitalIn.DIN  1 DigitalIn.DIN2 1 DigitalIn.DIN123 1 DigitalIn.DIN765 1 DigitalIn.DIN23456 1
    for idx in range(1, len(sys.argv), 2):
        varName = sys.argv[idx]
        varNames.append(varName)
        samplingInfo.append([varName, float(sys.argv[idx + 1])])
 
    cmgr = ic.ConnectionMgr()
    cmgr.initLogger('daq', 'daqExample.log', ic.CLogger.PYTHON)
    cmgr.connect(ic.CConnectionConfig().instanceId(winidea_id))
    print()
    if not cmgr.isAttached():
        print("The connection to winIDEA has not been established - exiting script.")
        sys.exit(-1)
    else:
        print("Established connection to winIDEA.")
        print()
    debugCtrl = initTarget(cmgr)
 
    debugCtrl.run() # run target so that variable is changing
    
    # Here you can modify recording time and file name.
    recordVariables(cmgr, debugCtrl, samplingInfo, isVerboseLogging)
 
 
if __name__ == '__main__':
    main()