from __future__ import annotations
from nidaqmx.constants import READ_ALL_AVAILABLE, FillMode
from nidaqmx.stream_readers._channel_reader_base import ChannelReaderBase
from nidaqmx.types import CtrFreq, CtrTick, CtrTime
[docs]
class CounterReader(ChannelReaderBase):
"""Reads samples from a counter input channel in an NI-DAQmx task."""
[docs]
def read_many_sample_double(
self, data, number_of_samples_per_channel=READ_ALL_AVAILABLE, timeout=10.0
):
"""Reads one or more floating-point samples from a single counter input channel in a task.
This read method accepts a preallocated NumPy array to hold the
samples requested, which can be advantageous for performance and
interoperability with NumPy and SciPy.
Passing in a preallocated array is valuable in continuous
acquisition scenarios, where the same array can be used
repeatedly in each call to the method.
Args:
data (numpy.ndarray): Specifies a preallocated 1D NumPy
array of floating-point values to hold the samples
requested.
Each element in the array corresponds to a sample from
the channel. The size of the array must be large enough
to hold all requested samples from the channel in the
task; otherwise, an error is thrown.
number_of_samples_per_channel (Optional[int]): Specifies the
number of samples to read.
If you set this input to nidaqmx.constants.
READ_ALL_AVAILABLE, NI-DAQmx determines how many samples
to read based on if the task acquires samples
continuously or acquires a finite number of samples.
If the task acquires samples continuously and you set
this input to nidaqmx.constants.READ_ALL_AVAILABLE, this
method reads all the samples currently available in the
buffer.
If the task acquires a finite number of samples and you
set this input to nidaqmx.constants.READ_ALL_AVAILABLE,
the method waits for the task to acquire all requested
samples, then reads those samples. If you set the
"read_all_avail_samp" property to True, the method reads
the samples currently available in the buffer and does
not wait for the task to acquire all requested samples.
timeout (Optional[float]): Specifies the amount of time in
seconds to wait for samples to become available. If the
time elapses, the method returns an error and any
samples read before the timeout elapsed. The default
timeout is 10 seconds. If you set timeout to
nidaqmx.constants.WAIT_INFINITELY, the method waits
indefinitely. If you set timeout to 0, the method tries
once to read the requested samples and returns an error
if it is unable to.
Returns:
int:
Indicates the number of samples acquired by each channel.
NI-DAQmx returns a single value because this value is the
same for all channels.
"""
number_of_samples_per_channel = self._task._calculate_num_samps_per_chan(
number_of_samples_per_channel
)
self._verify_array(data, number_of_samples_per_channel, False, True)
_, samps_per_chan_read = self._interpreter.read_counter_f64_ex(
self._handle,
number_of_samples_per_channel,
timeout,
FillMode.GROUP_BY_CHANNEL.value,
data,
)
return samps_per_chan_read
[docs]
def read_many_sample_pulse_frequency(
self,
frequencies,
duty_cycles,
number_of_samples_per_channel=READ_ALL_AVAILABLE,
timeout=10.0,
):
"""Reads one or more pulse samples in terms of frequency from a single counter input channel in a task.
This read method accepts preallocated NumPy arrays to hold the
samples requested, which can be advantageous for performance and
interoperability with NumPy and SciPy.
Passing in preallocated arrays is valuable in continuous
acquisition scenarios, where the same array can be used
repeatedly in each call to the method.
Args:
frequencies (numpy.ndarray): Specifies a preallocated 1D
NumPy array of floating-point values to hold the frequency
portion of the pulse samples requested.
Each element in the array corresponds to a sample from
the channel. The size of the array must be large enough
to hold all requested samples from the channel in the
task; otherwise, an error is thrown.
duty_cycles (numpy.ndarray): Specifies a preallocated 1D
NumPy array of floating-point values to hold the duty
cycle portion of the pulse samples requested.
Each element in the array corresponds to a sample from
the channel. The size of the array must be large enough
to hold all requested samples from the channel in the
task; otherwise, an error is thrown.
number_of_samples_per_channel (Optional[int]): Specifies the
number of samples to read.
If you set this input to nidaqmx.constants.
READ_ALL_AVAILABLE, NI-DAQmx determines how many samples
to read based on if the task acquires samples
continuously or acquires a finite number of samples.
If the task acquires samples continuously and you set
this input to nidaqmx.constants.READ_ALL_AVAILABLE, this
method reads all the samples currently available in the
buffer.
If the task acquires a finite number of samples and you
set this input to nidaqmx.constants.READ_ALL_AVAILABLE,
the method waits for the task to acquire all requested
samples, then reads those samples. If you set the
"read_all_avail_samp" property to True, the method reads
the samples currently available in the buffer and does
not wait for the task to acquire all requested samples.
timeout (Optional[float]): Specifies the amount of time in
seconds to wait for samples to become available. If the
time elapses, the method returns an error and any
samples read before the timeout elapsed. The default
timeout is 10 seconds. If you set timeout to
nidaqmx.constants.WAIT_INFINITELY, the method waits
indefinitely. If you set timeout to 0, the method tries
once to read the requested samples and returns an error
if it is unable to.
Returns:
int:
Indicates the number of samples acquired by each channel.
NI-DAQmx returns a single value because this value is the
same for all channels.
""" # noqa: W505 - doc line too long (111 > 100 characters) (auto-generated noqa)
number_of_samples_per_channel = self._task._calculate_num_samps_per_chan(
number_of_samples_per_channel
)
self._verify_array(frequencies, number_of_samples_per_channel, False, True)
self._verify_array(duty_cycles, number_of_samples_per_channel, False, True)
_, _, samps_per_chan_read = self._interpreter.read_ctr_freq(
self._handle,
number_of_samples_per_channel,
timeout,
FillMode.GROUP_BY_CHANNEL.value,
frequencies,
duty_cycles,
)
return samps_per_chan_read
[docs]
def read_many_sample_pulse_ticks(
self, high_ticks, low_ticks, number_of_samples_per_channel=READ_ALL_AVAILABLE, timeout=10.0
):
"""Reads one or more pulse samples in terms of ticks from a single counter input channel in a task.
This read method accepts preallocated NumPy arrays to hold the
samples requested, which can be advantageous for performance and
interoperability with NumPy and SciPy.
Passing in preallocated arrays is valuable in continuous
acquisition scenarios, where the same array can be used
repeatedly in each call to the method.
Args:
high_ticks (numpy.ndarray): Specifies a preallocated 1D
NumPy array of 32-bit unsigned integer values to hold
the high ticks portion of the pulse samples requested.
Each element in the array corresponds to a sample from
the channel. The size of the array must be large enough
to hold all requested samples from the channel in the
task; otherwise, an error is thrown.
low_ticks (numpy.ndarray): Specifies a preallocated 1D NumPy
array of 32-bit unsigned integer values to hold the low
ticks portion of the pulse samples requested.
Each element in the array corresponds to a sample from
the channel. The size of the array must be large enough
to hold all requested samples from the channel in the
task; otherwise, an error is thrown.
number_of_samples_per_channel (Optional[int]): Specifies the
number of samples to read.
If you set this input to nidaqmx.constants.
READ_ALL_AVAILABLE, NI-DAQmx determines how many samples
to read based on if the task acquires samples
continuously or acquires a finite number of samples.
If the task acquires samples continuously and you set
this input to nidaqmx.constants.READ_ALL_AVAILABLE, this
method reads all the samples currently available in the
buffer.
If the task acquires a finite number of samples and you
set this input to nidaqmx.constants.READ_ALL_AVAILABLE,
the method waits for the task to acquire all requested
samples, then reads those samples. If you set the
"read_all_avail_samp" property to True, the method reads
the samples currently available in the buffer and does
not wait for the task to acquire all requested samples.
timeout (Optional[float]): Specifies the amount of time in
seconds to wait for samples to become available. If the
time elapses, the method returns an error and any
samples read before the timeout elapsed. The default
timeout is 10 seconds. If you set timeout to
nidaqmx.constants.WAIT_INFINITELY, the method waits
indefinitely. If you set timeout to 0, the method tries
once to read the requested samples and returns an error
if it is unable to.
Returns:
int:
Indicates the number of samples acquired by each channel.
NI-DAQmx returns a single value because this value is the
same for all channels.
""" # noqa: W505 - doc line too long (107 > 100 characters) (auto-generated noqa)
number_of_samples_per_channel = self._task._calculate_num_samps_per_chan(
number_of_samples_per_channel
)
self._verify_array(high_ticks, number_of_samples_per_channel, False, True)
self._verify_array(low_ticks, number_of_samples_per_channel, False, True)
_, _, samps_per_chan_read = self._interpreter.read_ctr_ticks(
self._handle,
number_of_samples_per_channel,
timeout,
FillMode.GROUP_BY_CHANNEL.value,
high_ticks,
low_ticks,
)
return samps_per_chan_read
[docs]
def read_many_sample_pulse_time(
self, high_times, low_times, number_of_samples_per_channel=READ_ALL_AVAILABLE, timeout=10.0
):
"""Reads one or more pulse samples in terms of time from a single counter input channel in a task.
This read method accepts preallocated NumPy arrays to hold the
samples requested, which can be advantageous for performance and
interoperability with NumPy and SciPy.
Passing in preallocated arrays is valuable in continuous
acquisition scenarios, where the same array can be used
repeatedly in each call to the method.
Args:
high_times (numpy.ndarray): Specifies a preallocated 1D
NumPy array of floating-point values to hold the high
time portion of the pulse samples requested.
Each element in the array corresponds to a sample from
the channel. The size of the array must be large enough
to hold all requested samples from the channel in the
task; otherwise, an error is thrown.
low_times (numpy.ndarray): Specifies a preallocated 1D
NumPy array of floating-point values to hold the low
time portion of the pulse samples requested.
Each element in the array corresponds to a sample from
the channel. The size of the array must be large enough
to hold all requested samples from the channel in the
task; otherwise, an error is thrown.
number_of_samples_per_channel (Optional[int]): Specifies the
number of samples to read.
If you set this input to nidaqmx.constants.
READ_ALL_AVAILABLE, NI-DAQmx determines how many samples
to read based on if the task acquires samples
continuously or acquires a finite number of samples.
If the task acquires samples continuously and you set
this input to nidaqmx.constants.READ_ALL_AVAILABLE, this
method reads all the samples currently available in the
buffer.
If the task acquires a finite number of samples and you
set this input to nidaqmx.constants.READ_ALL_AVAILABLE,
the method waits for the task to acquire all requested
samples, then reads those samples. If you set the
"read_all_avail_samp" property to True, the method reads
the samples currently available in the buffer and does
not wait for the task to acquire all requested samples.
timeout (Optional[float]): Specifies the amount of time in
seconds to wait for samples to become available. If the
time elapses, the method returns an error and any
samples read before the timeout elapsed. The default
timeout is 10 seconds. If you set timeout to
nidaqmx.constants.WAIT_INFINITELY, the method waits
indefinitely. If you set timeout to 0, the method tries
once to read the requested samples and returns an error
if it is unable to.
Returns:
int:
Indicates the number of samples acquired by each channel.
NI-DAQmx returns a single value because this value is the
same for all channels.
""" # noqa: W505 - doc line too long (106 > 100 characters) (auto-generated noqa)
number_of_samples_per_channel = self._task._calculate_num_samps_per_chan(
number_of_samples_per_channel
)
self._verify_array(high_times, number_of_samples_per_channel, False, True)
self._verify_array(low_times, number_of_samples_per_channel, False, True)
_, _, samps_per_chan_read = self._interpreter.read_ctr_time(
self._handle,
number_of_samples_per_channel,
timeout,
FillMode.GROUP_BY_CHANNEL.value,
high_times,
low_times,
)
return samps_per_chan_read
[docs]
def read_many_sample_uint32(
self, data, number_of_samples_per_channel=READ_ALL_AVAILABLE, timeout=10.0
):
"""Reads one or more 32-bit unsigned integer samples from a single counter input channel in a task.
This read method accepts a preallocated NumPy array to hold the
samples requested, which can be advantageous for performance and
interoperability with NumPy and SciPy.
Passing in a preallocated array is valuable in continuous
acquisition scenarios, where the same array can be used
repeatedly in each call to the method.
Args:
data (numpy.ndarray): Specifies a preallocated 1D NumPy
array of 32-bit unsigned integer values to hold the
samples requested.
Each element in the array corresponds to a sample from
the channel. The size of the array must be large enough
to hold all requested samples from the channel in the
task; otherwise, an error is thrown.
number_of_samples_per_channel (Optional[int]): Specifies the
number of samples to read.
If you set this input to nidaqmx.constants.
READ_ALL_AVAILABLE, NI-DAQmx determines how many samples
to read based on if the task acquires samples
continuously or acquires a finite number of samples.
If the task acquires samples continuously and you set
this input to nidaqmx.constants.READ_ALL_AVAILABLE, this
method reads all the samples currently available in the
buffer.
If the task acquires a finite number of samples and you
set this input to nidaqmx.constants.READ_ALL_AVAILABLE,
the method waits for the task to acquire all requested
samples, then reads those samples. If you set the
"read_all_avail_samp" property to True, the method reads
the samples currently available in the buffer and does
not wait for the task to acquire all requested samples.
timeout (Optional[float]): Specifies the amount of time in
seconds to wait for samples to become available. If the
time elapses, the method returns an error and any
samples read before the timeout elapsed. The default
timeout is 10 seconds. If you set timeout to
nidaqmx.constants.WAIT_INFINITELY, the method waits
indefinitely. If you set timeout to 0, the method tries
once to read the requested samples and returns an error
if it is unable to.
Returns:
int:
Indicates the number of samples acquired by each channel.
NI-DAQmx returns a single value because this value is the
same for all channels.
""" # noqa: W505 - doc line too long (107 > 100 characters) (auto-generated noqa)
number_of_samples_per_channel = self._task._calculate_num_samps_per_chan(
number_of_samples_per_channel
)
self._verify_array(data, number_of_samples_per_channel, False, True)
_, samps_per_chan_read = self._interpreter.read_counter_u32_ex(
self._handle,
number_of_samples_per_channel,
timeout,
FillMode.GROUP_BY_CHANNEL.value,
data,
)
return samps_per_chan_read
[docs]
def read_one_sample_double(self, timeout=10):
"""Reads a single floating-point sample from a single counter input channel in a task.
Args:
timeout (Optional[float]): Specifies the amount of time in
seconds to wait for samples to become available. If the
time elapses, the method returns an error and any
samples read before the timeout elapsed. The default
timeout is 10 seconds. If you set timeout to
nidaqmx.constants.WAIT_INFINITELY, the method waits
indefinitely. If you set timeout to 0, the method tries
once to read the requested samples and returns an error
if it is unable to.
Returns:
float: Indicates a single floating-point sample from the
task.
"""
return self._interpreter.read_counter_scalar_f64(self._handle, timeout)
[docs]
def read_one_sample_pulse_frequency(self, timeout=10):
"""Reads a pulse sample in terms of frequency from a single counter input channel in a task.
Args:
timeout (Optional[float]): Specifies the amount of time in
seconds to wait for samples to become available. If the
time elapses, the method returns an error and any
samples read before the timeout elapsed. The default
timeout is 10 seconds. If you set timeout to
nidaqmx.constants.WAIT_INFINITELY, the method waits
indefinitely. If you set timeout to 0, the method tries
once to read the requested samples and returns an error
if it is unable to.
Returns:
nidaqmx.types.CtrFreq:
Indicates a pulse sample in terms of frequency from the task.
"""
freq, duty_cycle = self._interpreter.read_ctr_freq_scalar(self._handle, timeout)
return CtrFreq(freq, duty_cycle)
[docs]
def read_one_sample_pulse_ticks(self, timeout=10):
"""Reads a pulse sample in terms of ticks from a single counter input channel in a task.
Args:
timeout (Optional[float]): Specifies the amount of time in
seconds to wait for samples to become available. If the
time elapses, the method returns an error and any
samples read before the timeout elapsed. The default
timeout is 10 seconds. If you set timeout to
nidaqmx.constants.WAIT_INFINITELY, the method waits
indefinitely. If you set timeout to 0, the method tries
once to read the requested samples and returns an error
if it is unable to.
Returns:
nidaqmx.types.CtrTick:
Indicates a pulse sample in terms of ticks from the task.
"""
high_ticks, low_ticks = self._interpreter.read_ctr_ticks_scalar(self._handle, timeout)
return CtrTick(high_ticks, low_ticks)
[docs]
def read_one_sample_pulse_time(self, timeout=10):
"""Reads a pulse sample in terms of time from a single counter input channel in a task.
Args:
timeout (Optional[float]): Specifies the amount of time in
seconds to wait for samples to become available. If the
time elapses, the method returns an error and any
samples read before the timeout elapsed. The default
timeout is 10 seconds. If you set timeout to
nidaqmx.constants.WAIT_INFINITELY, the method waits
indefinitely. If you set timeout to 0, the method tries
once to read the requested samples and returns an error
if it is unable to.
Returns:
nidaqmx.types.CtrTime:
Indicates a pulse sample in terms of time from the task.
"""
high_time, low_time = self._interpreter.read_ctr_time_scalar(self._handle, timeout)
return CtrTime(high_time, low_time)
[docs]
def read_one_sample_uint32(self, timeout=10):
"""Reads a single 32-bit unsigned integer sample from a single counter input channel in a task.
Args:
timeout (Optional[float]): Specifies the amount of time in
seconds to wait for samples to become available. If the
time elapses, the method returns an error and any
samples read before the timeout elapsed. The default
timeout is 10 seconds. If you set timeout to
nidaqmx.constants.WAIT_INFINITELY, the method waits
indefinitely. If you set timeout to 0, the method tries
once to read the requested samples and returns an error
if it is unable to.
Returns:
int:
Indicates a single 32-bit unsigned integer sample from the
task.
""" # noqa: W505 - doc line too long (103 > 100 characters) (auto-generated noqa)
return self._interpreter.read_counter_scalar_u32(self._handle, timeout)