Measurement Computing   Easy to Use | Easy to Integrate | Easy to Support catalog banner

Python for Windows Continuous Background with Device Discovery

Expand / Collapse

Python for Windows Continuous Background with Device Discovery

The following Python for Windows example demonstrates device discovery and a continuous background acquisition. 

Once started, the acquisition is monitored using the get status function that returns total count and buffer index position. Data is retrieved from the device buffer when it is more than half full (index > half). When this happens, it transfers data from the device buffer to the user buffer. The acquisition loop continues to monitor the index and when it rolls over to zero, transfers data from the upper half of the device buffer. It continues to do this back-n-forth or ping-ponging between the low and upper halves. 

The data is automatically scaled to volts via the ScaleData feature. This requires the use of the functions scaled_win_buf_alloc to allocate memory and scaled_win_buf_to_array to transfer the data. The ScaleOption.ScaleData informs the AInScan function that the ScaleData feature is in use. For simplicity, only one row of data is printed each time the buffer is transferred.

To install the Python for Windows support, copy the following URL into your browser and follow the instructions provided.

This example was built using Python 3.7 and PyCharm. The code module is provided via a download link at the end of this article.

from __future__ import absolute_import, division, print_function

import time

from builtins import *  # @UnusedWildImport

from mcculw import ul
from mcculw.enums import ScanOptions, FunctionType, Status
from examples.console import util
from mcculw.ul import ULError, a_input_mode
from mcculw.enums import InterfaceType
from mcculw.enums import ULRange
from mcculw.enums import AnalogInputMode

use_device_detection = True

def run_example():
    board_num = 0
    board_index = 0
    find_device = "USB-1608GX-2AO"
    if use_device_detection:
        board_num = -1
        dev_list = ul.get_daq_device_inventory(InterfaceType.USB)
        if len(dev_list) > 0:
            for device in dev_list:
                if str(device) == find_device:
                    print(f"Found {find_device} board number = {board_index}")
                    print(f"Serial number: {device.unique_id}")
                    print(f"Product type: {hex(device.product_id)}")
                    board_num = board_index
                    ul.create_daq_device(board_num, device)
                board_index = board_index + 1
            if board_num == -1:
                print(f"Device {find_device} not found")
            print("No devices detected")
    # **********End of Discovery************

    rate = 1000
    points_per_channel = 2000
    low_chan = 0
    high_chan = 3
    num_chans = 4

    total_count = points_per_channel * num_chans
    half_count = int(total_count / 2)
    # If the hardware supports the SCALEDATA option, it is easiest to
    # use it.
    scan_options = ScanOptions.CONTINUOUS | ScanOptions.BACKGROUND | ScanOptions.SCALEDATA

    memhandle = ul.scaled_win_buf_alloc(total_count)

    # Convert the memhandle to a ctypes array.
    # Use the memhandle_as_ctypes_array_scaled method for scaled
    # buffers.
    ctypes_array = util.memhandle_as_ctypes_array_scaled(memhandle)

    # Note: the ctypes array will no longer be valid after win_buf_free is
    # called.
    # A copy of the buffer can be created using win_buf_to_array or
    # win_buf_to_array_32 before the memory is freed. The copy can be used
    # at any time.

    # Check if the buffer was successfully allocated
    if not memhandle:
        print("Failed to allocate memory.")

    a_input_mode(board_num, AnalogInputMode.SINGLE_ENDED)

        # Start the scan
            board_num, low_chan, high_chan, total_count,
            rate, ULRange.BIP10VOLTS, memhandle, scan_options)

        # Create a format string that aligns the data in columns
        # plus two for curr_index and curr_count
        row_format = "{:8}" * (num_chans + 3)

        # Print the channel name headers
        labels = []
        for ch_num in range(low_chan, high_chan + 1):
            labels.append("CH" + str(ch_num) + "\t")


        # boolean flag used to toggle reading upper and lower buffer
        read_lower = True
        # Start updating the displayed values
        status, curr_count, curr_index = ul.get_status(
            board_num, FunctionType.AIFUNCTION)

        buf_data = util.memhandle_as_ctypes_array_scaled(memhandle)
        last = 0
        diff = 0
        while status != Status.IDLE and curr_count < 100000:
            # Make sure a data point is available for display.
            if curr_count > 0:
                # curr_index points to the start of the last completed
                # channel scan that was transferred between the board and
                # the data buffer. Display the latest value for each
                # channel.

                # display_data = []
                if (curr_index > half_count) and (read_lower == True):
                    diff = curr_count - last
                    ul.scaled_win_buf_to_array(memhandle, buf_data, 0, int(half_count))
                        '{:.3f}\t {:.3f}\t {:.3f}\t {:.3f}\t {:d}\t {:d}\t {:d}'.format(buf_data[0], buf_data[1],
                                                                                        buf_data[3], curr_index,
                    last = curr_count
                    read_lower = False
                elif (curr_index < half_count) and (read_lower == False):
                    diff = curr_count - last
                    ul.scaled_win_buf_to_array(memhandle, buf_data, int(half_count), int(half_count))
                        '{:.3f}\t {:.3f}\t {:.3f}\t {:.3f}\t {:d}\t {:d}\t {:d}'.format(buf_data[0], buf_data[1],
                                                                                        buf_data[3], curr_index,
                    last = curr_count
                    read_lower = True

            status, curr_count, curr_index = ul.get_status(
                board_num, FunctionType.AIFUNCTION)

        # Stop the background operation (this is required even if the
        # scan completes successfully)
        ul.stop_background(board_num, FunctionType.AIFUNCTION)

        print("Scan completed successfully.")
    except ULError as e:
        # Free the buffer in a finally block to prevent errors from causing
        # a memory leak.

        if use_device_detection:

if __name__ == '__main__':

Rate this Article:


Add Your Comments

For comments email

Article ID: 50803

Last Modified:2/18/2020 9:35:19 AM

Article has been viewed 204 times.