NI-DAQmx Python: How to control same AO channel with finite and continuous sample mode?

[jhuang25]

Hello!

I am working on code to write a sinusoidal signal with frequency 71 Hz and varying amplitude from an AO channel. I would like the amplitude of the signal to ramp up at a constant rate from 0 to A. Once the amplitude reaches A, the signal should continue indefinitely with regeneration until interrupted by user input. Upon user input, the signal should ramp back down to 0. An example is illustrated below.

I believe the ramp up/down outputs should utilize FINITE sample mode (it only lasts for a finite time until the signal reaches amplitude A or 0), whereas the steady-state output should utilize CONTINUOUS sample mode (it lasts for an indefinite amount of time until user keyboard input). Furthermore, the ramp up/down should transition seamlessly with the steady-state portion, without any discontinuities in voltage. I am not sure how to achieve this.

 

I have tried writing different variations of code with the nidaqmx Python API to achieve this. A snippet from one variation is shown below:

electrode_dict = {'electrode_x+':'PXI1Slot6/ao17', 'electrode_x-':'PXI1Slot6/ao21', 'electrode_y+':'PXI1Slot6/ao0', 'electrode_y-':'PXI1Slot6/ao8', 'electrode_z+':'PXI1Slot6/ao12', 'electrode_z-':'PXI1Slot6/ao4'}
electrode_x_dict = {'electrode_x+':'PXI1Slot6/ao17', 'electrode_x-':'PXI1Slot6/ao21'}
electrode_y_dict = {'electrode_y+':'PXI1Slot6/ao0', 'electrode_y-':'PXI1Slot6/ao8'}
electrode_z_dict = {'electrode_z+':'PXI1Slot6/ao12', 'electrode_z-':'PXI1Slot6/ao4'}

def XeFlashes_new(task_spin, sampling_rate_AO, f_e=6.23e3, amp_e=2, ramp_rate_e=0.05, axis='none'):

    duration_ramp = amp_e/ramp_rate_e

    waveforms_up = []
    waveforms_down = []
    waveforms_spin = []

    if not axis=='none':
        # choose which channel(s) to output waveform(s) on
        if axis == 'X':
            spin_electrodes = [item for pair in zip(electrode_y_dict.values(), electrode_z_dict.values()) for item in pair]
        elif axis == 'Y':
            spin_electrodes = [item for pair in zip(electrode_z_dict.values(), electrode_x_dict.values()) for item in pair]
        elif axis == 'Z':
            spin_electrodes = [item for pair in zip(electrode_x_dict.values(), electrode_y_dict.values()) for item in pair]
        else:
            raise Exception("'axis' parameter is not recognized. Should be 'X', 'Y', or 'Z'.")

        for vv in enumerate(spin_electrodes):
            # add analog output channels for electrodes
            cc = task_spin.ao_channels.add_ao_voltage_chan(vv[1], min_val=-4, max_val=4)
            cc.ao_term_cfg = TerminalConfiguration.RSE

            # ramping time
            t_ramp = np.linspace(0, duration_ramp, int(duration_ramp * sampling_rate_AO), endpoint=False)

            # create waveforms for ramp up
            waveform_ramp_up = (t_ramp * ramp_rate_e) * np.sin(2 * np.pi * f_e * t_ramp + vv[0] * np.pi/2)
            waveforms_up.append(waveform_ramp_up)

            # create waveforms for ramp down
            waveform_ramp_down = (amp_e - t_ramp * ramp_rate_e) * np.sin(2 * np.pi * f_e * t_ramp + vv[0] * np.pi/2)
            waveforms_down.append(waveform_ramp_down)

            # create one period of waveforms for continuous section
            t = np.linspace(0, 1/f_e, int(1/f_e * sampling_rate_AO), endpoint=False)
            waveform = amp_e * np.sin(2 * np.pi * f_e * t + vv[0] * np.pi/2) # each adjacent electrode is phase-shifted by pi/2
            waveforms_spin.append(waveform)
    
    # allow regeneration
    task_spin.regen_mode = RegenerationMode.ALLOW_REGENERATION

    # configure timing
    task_spin.timing.samp_clk_src='OnboardClock'
    task_spin.timing.samp_clk_rate = sampling_rate_AO
    task_spin.timing.cfg_samp_clk_timing(rate=sampling_rate_AO,
                                       sample_mode=AcquisitionType.CONTINUOUS)

    # ramp up for finite amount of time
    task_spin.write(np.array(waveforms_up))
    task_spin.start()
    task_spin.wait_until_done(timeout=duration_ramp+1)
    task_spin.stop()

    # output sinusoids until user keyboard input
    task_spin.write(np.array(waveforms_spin))
    task_spin.start()
    input('Generating voltage continously with regeneration. Press Enter to stop.\n')
    task_spin.stop()

    # ramp down for finite amount of time
    task_spin.write(np.array(waveforms_down))
    task_spin.start()
    task_spin.wait_until_done(timeout=duration_ramp+1)
    task_spin.stop()

    return

Currently I encounter an issue where wait_until_done() cannot be used in conjunction with AcquisitionType.CONTINUOUS, as it is intended for finite acquisition. I have also tried controlling the same channel with two separate tasks (one task created with finite sample mode, the other with continuous sample mode), but this results in a resource error, as nidaqmx does not allow the same channel to be controlled by two tasks simultaneously.

A similar functionality has been successfully implemented in LabVIEW, so I believe it is achievable in Python as well. I'm open to any suggestions on how to fix/rewrite my code.

Thank you!

[dsb@NI · Accepted solution]

You want Continuous mode from the first sample to the last sample. A task is a grouping of channel and timing information - you should not change the timing mode while the task is running.

If you are tolerant of glitches, allow regeneration. The LabVIEW Sound and Vibration Toolkit supports output with ramp up and ramp down. SVT went with design decisions a long time ago to prefer DSA hardware and to never allow regeneration. It was the only way to guarantee smooth, deterministic test signals. While you are using the DAQmx Python API, the DAQmx concepts should translate.

Sample Mode = Continuous

Allow Regeneration = False

[Kevin_Price]

I second the motion -- you need to be in continuous sampling mode throughout.  It's not possible to change the timing mode between continuous and finite without adding some kind of anomaly to the waveform (and it would be difficult to control or know exactly what kind of anomaly you'd be getting.  It would depend on several things, some of them out of your control.)

Following are some tips and things to consider.  I'll assume NON-regeneration throughout because it lets you service your task in a consistent way throughout the whole generation process.

1. There will always be *some* latency in a buffered output task between the time when you calculate and write new data to the task and when that data becomes a signal in the real world. 

2. There are DAQmx properties and techniques that can help reduce that latency, but you need to be careful.  At the extreme of reducing it *too* much, you risk a task-killing buffer underflow error that stops your waveform suddenly.  You need to find an appropriate risk/reward tradeoff for your specific app.

    Specifics of how to reduce latency can vary by device and are a bigger topic than I can address right now.  Some searches here about latency may yield further help in the meantime.

3. There's a very general rule of thumb around here for servicing buffered tasks.  It says that most tasks perform reliably for the long term if you interact with them at ~10 Hz, i.e., 100 msec of data at a time.

    Note that in the case of output tasks, this does NOT limit your total latency to 100 msec.  Some devices have large onboard FIFO buffers that could give you *seconds* worth of latency unless you explicitly configure them otherwise.

4. You'll likely make your life easier if each chunk of data represents an integer number of 71 Hz sine cycles.  You can create this array one time only and use it throughout.  You'd just need to know when you're ramping (and where you are within the ramp) so you can multiply the array elements by the appropriate portion of a 0->1 or 1->0 ramp.

5. 7 full cycles at 71 Hz would get you pretty close to the 10 Hz servicing rule of thumb target.  10 full cycles might be easier to think about and gets you in the same ballpark at 7.1 Hz.

6. You can afford to define the sine waves with pretty high resolution, say 100 samples per cycle or more.  That would be a data acq rate of 7100 Hz, which is not generally difficult to achieve.

7. Here's a very picky detail that you can *probably* ignore.  It's probably not possible to set up a sample rate that's *exactly* 7100 Hz.  You can only do integer divisors of your device's internal timebase clock.  My quick calc suggests you may get an actual sample rate that misses your 7100 Hz target by ~0.25 Hz.   Again, my guess is that you can probably decide to just live with this.

Summary: there's kind of a lot to have to deal with, but you very likely *can* get there from here.  The more latency you can accommodate from user keystroke to real world signal change, the easier your coding job gets.

-Kevin P

[dsb@NI]

Kevin's point about latency is real, especially for platforms such as ethernet cDAQ where the DAQmx driver doesn't limit the latency. SVT compares generated waveform time to system time (and waits if necessary) to avoid building a long latency. Here is a shipping example with the maximum latency property added:

Here is what the generated output looks like:

I know you are using Python, so I throw this example your way to let you know that your task is achievable. Since you posted in the LabVIEW forum, I'm hoping there might also be a chance you are considering LabVIEW 😉. All kidding aside, I think it is worth you considering the value of your time.

If you are outputting sine waves on one or two ao channels, use the free Dynamic Signal Generator (DSG) (https://www.ni.com/en/support/downloads/drivers/download.dsa-soft-front-panels.html#554445).

[jhuang25]

Thank you both so much for your quick and detailed responses!

I will use continuous sampling and no regeneration, as advised.

It turns out that I need to use a frequency of 6,230Hz, much greater than 71 Hz. My device, the PXIe-6738, has a maximum update rate of 1 MS/s, so I have chosen a sampling rate of 250 kHz.

To achieve the target 10Hz, I will output 623 full cycles per array chunk (i.e. 25,000 samples per chunk).

Below is the code I have written after reading your responses. I have yet to test it out fully—could you please let me know if you foresee any potential errors or issues? Thank you!

In particular, I am wondering:

1) I would like the device to output each "chunk" of data one right after the other (i.e. the time between the device outputting the last sample of one chunk should occur exactly 1/250,000 sec before the device outputs the first sample of the next chunk). I am particularly concerned about whether this holds during the transitions—both from ramp-up to steady-state and from steady-state back to ramp-down. Even a slight delay, such as one-tenth of a period (1/62300 sec) might significantly impact my application by causing a phase shift.

2) When you mention minimizing latency, are you referring to the time lag between user keyboard input interrupt and the start of the ramp-down? If so, I am ok with that time lag being a few seconds long if necessary.

# channel names
xe_dict = {'Xe':'PXI1Slot2/ao39'}
electrode_dict = {'electrode_x+':'PXI1Slot6/ao17', 'electrode_x-':'PXI1Slot6/ao21', 'electrode_y+':'PXI1Slot6/ao0', 'electrode_y-':'PXI1Slot6/ao8', 'electrode_z+':'PXI1Slot6/ao12', 'electrode_z-':'PXI1Slot6/ao4'}
electrode_x_dict = {'electrode_x+':'PXI1Slot6/ao17', 'electrode_x-':'PXI1Slot6/ao21'}
electrode_y_dict = {'electrode_y+':'PXI1Slot6/ao0', 'electrode_y-':'PXI1Slot6/ao8'}
electrode_z_dict = {'electrode_z+':'PXI1Slot6/ao12', 'electrode_z-':'PXI1Slot6/ao4'}
ai_dict = {'DCQPDX':'PXI1Slot4/ai12', 'electrode_x+':'PXI1Slot4/ai2', 'electrode_x-':'PXI1Slot4/ai3', 'electrode_y+':'PXI1Slot4/ai10', 'electrode_y-':'PXI1Slot4/ai11', 'electrode_z+':'PXI1Slot4/18', 'electrode_z-':'PXI1Slot4/19'}


def generate_sine_wave(frequency, amplitude, sampling_rate=250e3, number_of_samples=25e3):
    '''
    Generates a sine wave with a specified phase and amplitude
    '''

    duration_time = number_of_samples / sampling_rate # duration time should be one-tenth of a second
    t = np.linspace(0, duration_time, int(duration_time * sampling_rate), endpoint=False)

    waveforms = []

    for cc in range(4): # create array of four waveforms for each of four waveforms
        phase_shift = cc * np.pi/2 # each adjacent electrode is phase-shifted by pi/2
        waveform = amplitude * np.sin(2 * np.pi * frequency * t + phase_shift)

        waveforms.append(waveform)

    return np.array(waveforms)

def XeFlashes_28(task_spin, sampling_rate_AO=250e3, f_e=6.23e3, amp_e=2, ramp_rate_e=0.05, axis='none'):

    if not axis=='none':
        # choose which channel(s) to output waveform(s) on
        if axis == 'X':
            spin_electrodes = [item for pair in zip(electrode_y_dict.values(), electrode_z_dict.values()) for item in pair]
        elif axis == 'Y':
            spin_electrodes = [item for pair in zip(electrode_z_dict.values(), electrode_x_dict.values()) for item in pair]
        elif axis == 'Z':
            spin_electrodes = [item for pair in zip(electrode_x_dict.values(), electrode_y_dict.values()) for item in pair]
        else:
            raise Exception("'axis' parameter is not recognized. Should be 'X', 'Y', or 'Z'.")

        for vv in enumerate(spin_electrodes):
            # add analog output channels for electrodes
            cc = task_spin.ao_channels.add_ao_voltage_chan(vv[1], min_val=-4, max_val=4)
            cc.ao_term_cfg = TerminalConfiguration.RSE

    # don't allow regeneration
    task_spin.regen_mode = RegenerationMode.DONT_ALLOW_REGENERATION

    # configure timing
    task_spin.timing.samp_clk_src='OnboardClock'
    task_spin.timing.samp_clk_rate = sampling_rate_AO
    task_spin.timing.cfg_samp_clk_timing(rate=sampling_rate_AO,
                                       sample_mode=AcquisitionType.CONTINUOUS)
    
    is_first_run = True
    amp = 0

    try:
        print('Ramping up voltage continuously. Press Enter to stop and start ramp down')
        while True:
            if amp < amp_e:
                print(f'Amp changed from {amp} to {amp+ramp_rate_e}')
                amp += ramp_rate_e
                if amp == amp_e:
                    print(f'Reached amplitude of {amp_e}')
            data = generate_sine_wave(f_e, amp)
            task_spin.write(data)
            if is_first_run:
                is_first_run = False
                task_spin.start()
    except KeyboardInterrupt:
        print('Starting ramp down')
        while True:
            if amp > 0:
                amp -= ramp_rate_e
            else:
                break
            data = generate_sine_wave(f_e, amp)
            task_spin.write(data)
    finally:
        task_spin.stop()

    return

Thank you again for your help!

[Kevin_Price · Accepted solution]

I don't really know Python and can't speak to any specific details.  But the broad strokes of what you posted look like you're heading in the right direction.

One possible exception is that it appears to me that you'd be changing the sine wave amplitude in discrete steps rather than a really smooth ramp, where the amplitude would be changing slightly from one sample to the next.  It looks like you have one constant amplitude for each 0.1 second chunk of sine wave to be written.  When in the ramping zone, you should have an array representing 0.1 second worth of your amplitude ramp that you multiply by the 0.1 second worth of sine wave, element-by-element.

---

2) When you mention minimizing latency, are you referring to the time lag between user keyboard input interrupt and the start of the ramp-down? If so, I am ok with that time lag being a few seconds long if necessary.

Yes that's the latency I meant and that's great news, it helps a *lot* to have a wide timing tolerance for it.

---

1) I would like the device to output each "chunk" of data one right after the other (i.e. the time between the device outputting the last sample of one chunk should occur exactly 1/250,000 sec before the device outputs the first sample of the next chunk). I am particularly concerned about whether this holds during the transitions—both from ramp-up to steady-state and from steady-state back to ramp-down. Even a slight delay, such as one-tenth of a period (1/62300 sec) might significantly impact my application by causing a phase shift.

DAQmx does a great job managing exactly this kind of seamlessness during a continuous task.  There are tradeoffs however.  Some things to know about DAQmx and buffers:

1. The first stage of buffering is the task buffer.  When you call a DAQmx Write function, the data heads toward this buffer.  But DAQmx manages the process pretty carefully to prevent you from stomping on top of data you wrote previously but which hasn't yet been sent down to the device.

2. The call to do this write will block and not return until it has waited (if necessary) for space to free up so it can write all the new data immediately after the end of the previously-written data.  DAQmx manages this process for you and this is what helps make those transitions seamless.

3. The size of the task buffer is set by the # samples you write to the task prior to starting the task.  I would generally recommend that you make this at least 2x the size that you'll keep writing incrementally.  In your case with a fairly high sample rate and a high tolerance for latency, I'd go with 10x, i.e., one second worth of task buffer.

4. As a general rule, a bigger task buffer makes your app more resilient but at the expense of greater latency for output changes.

5. In the background, DAQmx will be transferring data from the task buffer down to the device where it will enter the onboard FIFO.  Some devices allow you to manually set the size of the FIFO.  Other devices require you to use an indirect method where you set the "data transfer condition" to one of the following choices:

- Less Than Full: DAQmx tries to keep the FIFO full which increases both resilience and latency.  This is often the default.

- Half Full: a compromise that is very likely to be resilient enough with only half as much FIFO-induced latency

- Empty: DAQmx tries to keep transferring data just barely in time.  In reality, it only waits for FIFO to be "nearly empty".  This minimizes latency but puts you at some risk for a task-killing underflow error.

    At your high sample rate, the default size or "Less Than Full" are likely just fine.

-Kevin P

[jhuang25]

---

One possible exception is that it appears to me that you'd be changing the sine wave amplitude in discrete steps rather than a really smooth ramp, where the amplitude would be changing slightly from one sample to the next.  It looks like you have one constant amplitude for each 0.1 second chunk of sine wave to be written.  When in the ramping zone, you should have an array representing 0.1 second worth of your amplitude ramp that you multiply by the 0.1 second worth of sine wave, element-by-element.

Okay, thanks for pointing this out. I think the discrete steps are okay for my application (I realize that in the example I initially gave, the ramp was smooth rather than discrete 😁).

---

DAQmx does a great job managing exactly this kind of seamlessness during a continuous task.  There are tradeoffs however.  Some things to know about DAQmx and buffers

---

Great, thank you so much for your help! I have tested out my code fully and it seems to all work now.