Read analog input with Python nidaqmx package

[KathiSt]

Hi guys,

I am using the NI-USB 6002 for reading in an analog input. My goal is to read always the same amount of samples. If there are too less samples available, I use an await from the asyncio-package.

Here is my code, which I execute with Python 3.10.4 in Spyder:

# Function for the recording task
async def record_emg(event):

# Define the input channels
input_channels = ["Dev1/ai0"]

# Define the parameters
sample_rate= 2000 # Sample rate for the input channels
window_time = 66e-3 # Timeslot for one window
overlap = 0.5 # overlap between old an new data for one window

# Calculate window_size (samples for one window for ml-algorithm) and
# samples_per_channel (samples which are read at one timepoint)
window_size = (int)(window_time * sample_rate)
samples_per_channel = (int)(window_time * sample_rate * overlap)

# Define the queue for the window (ml-algorithm)
data_queue = deque([0]*window_size, maxlen=window_size)

# Create tasks for the analog inputs
with nidaqmx.Task() as input_task:

# Configure analog input channel
for channel in input_channels:
input_task.ai_channels.add_ai_voltage_chan(channel,
terminal_config=nidaqmx.constants.TerminalConfiguration.RSE)

# Configure the sample rate and the number of samples per channel
input_task.timing.cfg_samp_clk_timing(sample_rate,
sample_mode=nidaqmx.constants.AcquisitionType.CONTINUOUS)

input_task.in_stream.regen_mode = nidaqmx.constants.RegenerationMode.DONT_ALLOW_REGENERATION

starttask_time = time.monotonic()
input_task.start()

while True:

# Guarantees, that the number of "read samples is equal to
# the number of samples which are used for one signalpart
while True:

input_data = input_task.read(1)

# Check the available samples
available_samples = input_task.in_stream.avail_samp_per_chan
logging.debug(f"Available samples before reading: {available_samples}")

logging.debug(f"Task done: {input_task.is_task_done()}")

# If there are more samples available than needed -->
# read in the needed samples.
if available_samples >= samples_per_channel:

logging.debug(f"Read data! {samples_per_channel}")
# Read all needed values and add them to the queue
input_data = input_task.read(number_of_samples_per_channel = samples_per_channel) # [mV]
logging.debug(f"{len(input_data)} are read in!")
available_samples = input_task.in_stream.avail_samp_per_chan
logging.debug(f"Available samples after reading: {available_samples}")
logging.debug("Start calculation!!!!!!!!!!!!!!!!!!!!!!")
data_queue.extend(input_data)
break

# Otherwise, if there are not enough samples than
# needed --> wait.
else:
logging.debug("Not enough samples?????")

# Wait as long as half of the missing values are
# available. Meanwhile, execute output_task.
await asyncio.sleep((samples_per_channel - available_samples) / sample_rate / 2)

# Check the available samples and sleep afterwards
available_samples = input_task.in_stream.avail_samp_per_chan

# Wait as long as half of the missing values are available.
# Meanwhile, execute output_task.
await asyncio.sleep((samples_per_channel-available_samples) / sample_rate / 2)

# Function for the closed-loop system
async def closed_loop():

# Define the event, which makes interaction between both tasks possible
event = asyncio.Event()

# Create task
record_emg_task = asyncio.create_task(record_emg(event))
task_list = [record_emg_task]

try:
await asyncio.wait(task_list, return_when = asyncio.FIRST_EXCEPTION)
for task in task_list:
if task.done() and task.exception() is not None:
raise task.exception()
except asyncio.CancelledError as e:
logging.debug(f"One of the tasks was cancelled! CancelledError: {str(e)}")
except Exception as e:
logging.debug(f"An exception occurred: {str(e)}")
finally:
for task in task_list:
task.cancel()
await task

# Main code
if __name__ == "__main__":

logging.basicConfig(level=logging.DEBUG, format='%(asctime)-15s %(message)s')

try:
asyncio.run(closed_loop(), debug=True)
except KeyboardInterrupt:
logging.debug("KeyboardInterrupt")
except nidaqmx.DaqError as e:
logging.debug(f"Dataacquisition-Error: {str(e)}")
finally:
logging.debug("Press Enter to exit...")

But with this code I have the following problem: The code runs as expected at the beginning, means he is always waiting for enough samples with await, and if there are enough, read the samples and extend it to the queue. For debugging reasons I added the line with "read(1)".

After a while (the time how long it runs correctly is really different, from 5min to 40min), the variable "avail_samp_per_chan" shows a constant value. Means, the program isn't able to read more values cause there are too less available for reading in. Cause of "read(1)" the program reads always one value, until the buffer is empty. The program itself runs until this point but new values never come again. 

Here is a visualization of the available samples along the time:

You can see the blue dots, which are the available samples. After some time, no new samples appear and the available samples decrease cause of "read(1)".

I don't know why there are no new values, so I decided to ask the community. Anybody an idea what happens here? I am not sure if it is a software, hardware or interface problem.

BR, Katharina