daq_recorder.py

# This script is licensed under BSD License, see file LICENSE.txt.
#
# (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()