read_functions.py 文件源码

def _read_raw(task_handle, read_array, num_samps_per_chan, timeout):
    samples_read = ctypes.c_int()
    number_of_bytes_per_sample = ctypes.c_int()

    cfunc = lib_importer.windll.DAQmxReadRaw
    if cfunc.argtypes is None:
        with cfunc.arglock:
            if cfunc.argtypes is None:
                cfunc.argtypes = [
                    lib_importer.task_handle, ctypes.c_int, ctypes.c_double,
                    wrapped_ndpointer(dtype=read_array.dtype, flags=('C', 'W')),
                    ctypes.c_uint, ctypes.POINTER(ctypes.c_int),
                    ctypes.POINTER(ctypes.c_int), ctypes.POINTER(c_bool32)]

    error_code = cfunc(
        task_handle, num_samps_per_chan, timeout, read_array,
        read_array.nbytes, ctypes.byref(samples_read),
        ctypes.byref(number_of_bytes_per_sample), None)
    check_for_error(error_code)

    return samples_read.value, number_of_bytes_per_sample.value