_in_stream.py

# Do not edit this file; it was automatically generated.

from typing import Type, Optional, Union

import numpy
import numpy.typing
import deprecation
import pathlib

from nidaqmx.task.channels import Channel
from nidaqmx.utils import unflatten_channel_string
from nidaqmx.constants import (
    AcquisitionType, LoggingMode, LoggingOperation, OverwriteMode,
    READ_ALL_AVAILABLE, ReadRelativeTo, WaitMode)


class InStream:
    """
    Exposes an input data stream on a DAQmx task.

    The input data stream be used to control reading behavior and can be
    used in conjunction with reader classes to read samples from an
    NI-DAQmx task.
    """
    __slots__ = ('_task', '_handle', '_interpreter', '_timeout')

    def __init__(self, task, interpreter):
        self._task = task
        self._handle = task._handle
        self._interpreter = interpreter
        self._timeout = 10.0

        super().__init__()

    def __eq__(self, other):
        if isinstance(other, self.__class__):
            return (self._handle == other._handle and
                    self._timeout == other._timeout)
        return False

    def __hash__(self):
        return self._interpreter.hash_task_handle(self._handle) ^ hash(self._timeout)

    def __ne__(self, other):
        return not self.__eq__(other)

    def __repr__(self):
        return f'InStream(task={self._task.name})'

    @property
    def timeout(self):
        """
        float: Specifies the amount of time in seconds to wait for
            samples to become available. If the time elapses, the read
            method returns an error and any samples read before the
            timeout elapsed. The default timeout is 10 seconds. If you
            set timeout to nidaqmx.WAIT_INFINITELY, the read method
            waits indefinitely. If you set timeout to 0, the read method
            tries once to read the requested samples and returns an error
            if it is unable to.
        """
        return self._timeout

    @timeout.setter
    def timeout(self, val):
        self._timeout = val

    @timeout.deleter
    def timeout(self):
        self._timeout = 10.0

    @property
    def accessory_insertion_or_removal_detected(self):
        """
        bool: Indicates if any device(s) in the task detected the
            insertion or removal of an accessory since the task started.
            Reading this property clears the accessory change status for
            all channels in the task. You must read this property before
            you read **devs_with_inserted_or_removed_accessories**.
            Otherwise, you will receive an error.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x2f70)
        return val

    @property
    def auto_start(self):
        """
        bool: Specifies if DAQmx Read automatically starts the task  if
            you did not start the task explicitly by using DAQmx Start.
            The default value is True. When  DAQmx Read starts a finite
            acquisition task, it also stops the task after reading the
            last sample.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x1826)
        return val

    @auto_start.setter
    def auto_start(self, val):
        self._interpreter.set_read_attribute_bool(self._handle, 0x1826, val)

    @auto_start.deleter
    def auto_start(self):
        self._interpreter.reset_read_attribute(self._handle, 0x1826)

    @property
    def aux_power_error_chans(self):
        """
        List[str]: Indicates a list of names of any virtual channels in
            the task for which an auxiliary power supply error condition
            has been detected. You must read the Aux Power Error
            Channels Exist property before you read this property.
            Otherwise, you will receive an error.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x31e0, buffer_size)
        return unflatten_channel_string(val)

    @property
    def aux_power_error_chans_exist(self):
        """
        bool: Indicates if the device(s) detected an auxiliary power
            supply error condition for any channel in the task. Reading
            this property clears the error condition status for all
            channels in the task. You must read this property before you
            read the Aux Power Error Channels property. Otherwise, you
            will receive an error.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x31df)
        return val

    @property
    def avail_samp_per_chan(self):
        """
        int: Indicates the number of samples available to read per
            channel. This value is the same for all channels in the
            task.
        """

        val = self._interpreter.get_read_attribute_uint32(self._handle, 0x1223)
        return val

    @property
    def change_detect_overflowed(self):
        """
        bool: Indicates if samples were missed because change detection
            events occurred faster than the device could handle them.
            Some devices detect overflows differently than others.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x2194)
        return val

    @property
    def channels_to_read(self):
        """
        :class:`nidaqmx.task.channels.Channel`: Specifies a subset of
            channels in the task from which to read.
        """

        val = self._interpreter.get_read_attribute_string(self._handle, 0x1823)
        return Channel._factory(self._handle, val, self._interpreter)

    @channels_to_read.setter
    def channels_to_read(self, val):
        val = val.name
        self._interpreter.set_read_attribute_string(self._handle, 0x1823, val)

    @channels_to_read.deleter
    def channels_to_read(self):
        self._interpreter.reset_read_attribute(self._handle, 0x1823)

    @property
    def common_mode_range_error_chans(self):
        """
        List[str]: Indicates a list of names of any virtual channels in
            the task for which the device(s) detected a common mode
            range violation. You must read
            **common_mode_range_error_chans_exist** before you read this
            property. Otherwise, you will receive an error.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x2a99, buffer_size)
        return unflatten_channel_string(val)

    @property
    def common_mode_range_error_chans_exist(self):
        """
        bool: Indicates if the device(s) detected a common mode range
            violation for any virtual channel in the task. Common mode
            range violation occurs when the voltage of either the
            positive terminal or negative terminal to ground are out of
            range. Reading this property clears the common mode range
            violation status for all channels in the task. You must read
            this property before you read
            **common_mode_range_error_chans**. Otherwise, you will
            receive an error.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x2a98)
        return val

    @property
    def curr_read_pos(self):
        """
        int: Indicates in samples per channel the current position in
            the buffer.
        """

        val = self._interpreter.get_read_attribute_uint64(self._handle, 0x1221)
        return val

    @property
    def devs_with_inserted_or_removed_accessories(self):
        """
        List[str]: Indicates the names of any devices that detected the
            insertion or removal of an accessory since the task started.
            You must read **accessory_insertion_or_removal_detected**
            before you read this property. Otherwise, you will receive
            an error.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x2f71, buffer_size)
        return unflatten_channel_string(val)

    @property
    def di_num_booleans_per_chan(self):
        """
        int: Indicates the number of booleans per channel that NI-DAQmx
            returns in a sample for line-based reads. If a channel has
            fewer lines than this number, the extra booleans are False.
        """

        val = self._interpreter.get_read_attribute_uint32(self._handle, 0x217c)
        return val

    @property
    def excit_fault_chans(self):
        """
        List[str]: Indicates a list of names of any virtual channels in
            the task for which the device(s) detected an excitation
            fault condition. You must read **excit_fault_chans_exist**
            before you read this property. Otherwise, you will receive
            an error.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x3089, buffer_size)
        return unflatten_channel_string(val)

    @property
    def excit_fault_chans_exist(self):
        """
        bool: Indicates if the device(s) detected an excitation fault
            condition for any virtual channel in the task. Reading this
            property clears the excitation fault status for all channels
            in the task. You must read this property before you read
            **excit_fault_chans**. Otherwise, you will receive an error.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x3088)
        return val

    @property
    def input_buf_size(self):
        """
        int: Specifies the number of samples the input buffer can hold
            for each channel in the task. Zero indicates to allocate no
            buffer. Use a buffer size of 0 to perform a hardware-timed
            operation without using a buffer. Setting this property
            overrides the automatic input buffer allocation that NI-
            DAQmx performs.
        """

        val = self._interpreter.get_buffer_attribute_uint32(self._handle, 0x186c)
        return val

    @input_buf_size.setter
    def input_buf_size(self, val):
        self._interpreter.set_buffer_attribute_uint32(self._handle, 0x186c, val)

    @input_buf_size.deleter
    def input_buf_size(self):
        self._interpreter.reset_buffer_attribute(self._handle, 0x186c)

    @property
    def input_limits_fault_chans(self):
        """
        List[str]: Indicates the virtual channels that have detected
            samples outside the upper or lower limits configured for
            each channel in the task. You must read
            **input_limits_fault_chans_exist** before you read this
            property. Otherwise, you will receive an error.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x3190, buffer_size)
        return unflatten_channel_string(val)

    @property
    def input_limits_fault_chans_exist(self):
        """
        bool: Indicates if the device or devices detected a sample that
            was outside the upper or lower limits configured for each
            channel in the task. Reading this property clears the input
            limits fault channel status for all channels in the task.
            You must read this property before you read
            **input_limits_fault_chans**. Otherwise, you will receive an
            error. Note: Fault detection applies to both positive and
            negative inputs. For instance, if you specify a lower limit
            of 2 mA and an upper limit of 12 mA, NI-DAQmx detects a
            fault at 15 mA and -15 mA, but not at -6 mA because it is in
            the range of -12 mA to -2 mA.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x318f)
        return val

    @property
    def input_onbrd_buf_size(self):
        """
        int: Indicates in samples per channel the size of the onboard
            input buffer of the device.
        """

        val = self._interpreter.get_buffer_attribute_uint32(self._handle, 0x230a)
        return val

    @property
    def logging_file_path(self) -> Optional[pathlib.Path]:
        """
        pathlib.Path: Specifies the path to the TDMS file to which you
            want to log data.  If the file path is changed while the
            task is running, this takes effect on the next sample
            interval (if Logging.SampsPerFile has been set) or when
            DAQmx Start New File is called. New file paths can be
            specified by ending with "\\" or "/". Files created after
            specifying a new file path retain the same name and
            numbering sequence.
        """

        val = self._interpreter.get_read_attribute_string(self._handle, 0x2ec4)
        return pathlib.Path(val) if val else None

    @logging_file_path.setter
    def logging_file_path(self, val: Optional[Union[str, pathlib.PurePath]]):
        if val is None:
            val = ""
        val = str(val)
        self._interpreter.set_read_attribute_string(self._handle, 0x2ec4, val)

    @logging_file_path.deleter
    def logging_file_path(self):
        self._interpreter.reset_read_attribute(self._handle, 0x2ec4)

    @property
    def logging_file_preallocation_size(self):
        """
        int: Specifies a size in samples to be used to pre-allocate
            space on disk.  Pre-allocation can improve file I/O
            performance, especially in situations where multiple files
            are being written to disk.  For finite tasks, the default
            behavior is to pre-allocate the file based on the number of
            samples you configure the task to acquire.
        """

        val = self._interpreter.get_read_attribute_uint64(self._handle, 0x2fc6)
        return val

    @logging_file_preallocation_size.setter
    def logging_file_preallocation_size(self, val):
        self._interpreter.set_read_attribute_uint64(self._handle, 0x2fc6, val)

    @logging_file_preallocation_size.deleter
    def logging_file_preallocation_size(self):
        self._interpreter.reset_read_attribute(self._handle, 0x2fc6)

    @property
    def logging_file_write_size(self):
        """
        int: Specifies the size, in samples, in which data will be
            written to disk.  The size must be evenly divisible by the
            volume sector size, in bytes.
        """

        val = self._interpreter.get_read_attribute_uint32(self._handle, 0x2fc3)
        return val

    @logging_file_write_size.setter
    def logging_file_write_size(self, val):
        self._interpreter.set_read_attribute_uint32(self._handle, 0x2fc3, val)

    @logging_file_write_size.deleter
    def logging_file_write_size(self):
        self._interpreter.reset_read_attribute(self._handle, 0x2fc3)

    @property
    def logging_mode(self):
        """
        :class:`nidaqmx.constants.LoggingMode`: Specifies whether to
            enable logging and whether to allow reading data while
            logging. Log mode allows for the best performance. However,
            you cannot read data while logging if you specify this mode.
            If you want to read data while logging, specify Log and Read
            mode.
        """

        val = self._interpreter.get_read_attribute_int32(self._handle, 0x2ec5)
        return LoggingMode(val)

    @logging_mode.setter
    def logging_mode(self, val):
        val = val.value
        self._interpreter.set_read_attribute_int32(self._handle, 0x2ec5, val)

    @logging_mode.deleter
    def logging_mode(self):
        self._interpreter.reset_read_attribute(self._handle, 0x2ec5)

    @property
    def logging_pause(self):
        """
        bool: Specifies whether logging is paused while a task is
            executing. If **logging_mode** is set to Log and Read mode,
            this value is taken into consideration on the next call to
            DAQmx Read, where data is written to disk. If
            **logging_mode** is set to Log Only mode, this value is
            taken into consideration the next time that data is written
            to disk. A new TDMS group is written when logging is resumed
            from a paused state.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x2fe3)
        return val

    @logging_pause.setter
    def logging_pause(self, val):
        self._interpreter.set_read_attribute_bool(self._handle, 0x2fe3, val)

    @logging_pause.deleter
    def logging_pause(self):
        self._interpreter.reset_read_attribute(self._handle, 0x2fe3)

    @property
    def logging_samps_per_file(self):
        """
        int: Specifies how many samples to write to each file. When the
            file reaches the number of samples specified, a new file is
            created with the naming convention of <filename>_####.tdms,
            where #### starts at 0001 and increments automatically with
            each new file. For example, if the file specified is
            C:\\data.tdms, the next file name used is
            C:\\data_0001.tdms. To disable file spanning behavior, set
            this attribute to 0. If **logging_file_path** is changed
            while this attribute is set, the new file path takes effect
            on the next file created.
        """

        val = self._interpreter.get_read_attribute_uint64(self._handle, 0x2fe4)
        return val

    @logging_samps_per_file.setter
    def logging_samps_per_file(self, val):
        self._interpreter.set_read_attribute_uint64(self._handle, 0x2fe4, val)

    @logging_samps_per_file.deleter
    def logging_samps_per_file(self):
        self._interpreter.reset_read_attribute(self._handle, 0x2fe4)

    @property
    def logging_tdms_group_name(self):
        """
        str: Specifies the name of the group to create within the TDMS
            file for data from this task. If you append data to an
            existing file and the specified group already exists, NI-
            DAQmx appends a number symbol and a number to the group
            name, incrementing that number until finding a group name
            that does not exist. For example, if you specify a group
            name of Voltage Task, and that group already exists, NI-
            DAQmx assigns the group name Voltage Task #1, then Voltage
            Task #2.
        """

        val = self._interpreter.get_read_attribute_string(self._handle, 0x2ec6)
        return val

    @logging_tdms_group_name.setter
    def logging_tdms_group_name(self, val):
        self._interpreter.set_read_attribute_string(self._handle, 0x2ec6, val)

    @logging_tdms_group_name.deleter
    def logging_tdms_group_name(self):
        self._interpreter.reset_read_attribute(self._handle, 0x2ec6)

    @property
    def logging_tdms_operation(self):
        """
        :class:`nidaqmx.constants.LoggingOperation`: Specifies how to
            open the TDMS file.
        """

        val = self._interpreter.get_read_attribute_int32(self._handle, 0x2ec7)
        return LoggingOperation(val)

    @logging_tdms_operation.setter
    def logging_tdms_operation(self, val):
        val = val.value
        self._interpreter.set_read_attribute_int32(self._handle, 0x2ec7, val)

    @logging_tdms_operation.deleter
    def logging_tdms_operation(self):
        self._interpreter.reset_read_attribute(self._handle, 0x2ec7)

    @property
    def num_chans(self):
        """
        int: Indicates the number of channels that DAQmx Read reads from
            the task. This value is the number of channels in the task
            or the number of channels you specify with
            **channels_to_read**.
        """

        val = self._interpreter.get_read_attribute_uint32(self._handle, 0x217b)
        return val

    @property
    def offset(self):
        """
        int: Specifies an offset in samples per channel at which to
            begin a read operation. This offset is relative to the
            location you specify with **relative_to**.
        """

        val = self._interpreter.get_read_attribute_int32(self._handle, 0x190b)
        return val

    @offset.setter
    def offset(self, val):
        self._interpreter.set_read_attribute_int32(self._handle, 0x190b, val)

    @offset.deleter
    def offset(self):
        self._interpreter.reset_read_attribute(self._handle, 0x190b)

    @property
    def open_chans(self):
        """
        List[str]: Indicates a list of names of any open virtual
            channels. You must read **open_chans_exist** before you read
            this property. Otherwise you will receive an error.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x3101, buffer_size)
        return unflatten_channel_string(val)

    @property
    def open_chans_details(self):
        """
        List[str]: Indicates a list of details of any open virtual
            channels. You must read **open_chans_exist** before you read
            this property. Otherwise you will receive an error.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x3102, buffer_size)
        return unflatten_channel_string(val)

    @property
    def open_chans_exist(self):
        """
        bool: Indicates if the device or devices detected an open
            channel condition in any virtual channel in the task.
            Reading this property clears the open channel status for all
            channels in this task. You must read this property before
            you read **open_chans**. Otherwise, you will receive an
            error.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x3100)
        return val

    @property
    def open_current_loop_chans(self):
        """
        List[str]: Indicates a list of names of any virtual channels in
            the task for which the device(s) detected an open current
            loop. You must read **open_current_loop_chans_exist** before
            you read this property. Otherwise, you will receive an
            error.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x2a0a, buffer_size)
        return unflatten_channel_string(val)

    @property
    def open_current_loop_chans_exist(self):
        """
        bool: Indicates if the device(s) detected an open current loop
            for any virtual channel in the task. Reading this property
            clears the open current loop status for all channels in the
            task. You must read this property before you read
            **open_current_loop_chans**. Otherwise, you will receive an
            error.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x2a09)
        return val

    @property
    def open_thrmcpl_chans(self):
        """
        List[str]: Indicates a list of names of any virtual channels in
            the task for which the device(s) detected an open
            thermcouple. You must read **open_thrmcpl_chans_exist**
            before you read this property. Otherwise, you will receive
            an error.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x2a97, buffer_size)
        return unflatten_channel_string(val)

    @property
    def open_thrmcpl_chans_exist(self):
        """
        bool: Indicates if the device(s) detected an open thermocouple
            connected to any virtual channel in the task. Reading this
            property clears the open thermocouple status for all
            channels in the task. You must read this property before you
            read **open_thrmcpl_chans**. Otherwise, you will receive an
            error.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x2a96)
        return val

    @property
    def overcurrent_chans(self):
        """
        List[str]: Indicates a list of names of any virtual channels in
            the task for which the device(s) detected an overcurrent
            condition. You must read **overcurrent_chans_exist** before
            you read this property. Otherwise, you will receive an
            error. On some devices, you must restart the task for all
            overcurrent channels to recover.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x29e7, buffer_size)
        return unflatten_channel_string(val)

    @property
    def overcurrent_chans_exist(self):
        """
        bool: Indicates if the device(s) detected an overcurrent
            condition for any virtual channel in the task. Reading this
            property clears the overcurrent status for all channels in
            the task. You must read this property before you read
            **overcurrent_chans**. Otherwise, you will receive an error.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x29e6)
        return val

    @property
    def overloaded_chans(self):
        """
        List[str]: Indicates a list of names of any overloaded virtual
            channels in the task. You must read
            **overloaded_chans_exist** before you read this property.
            Otherwise, you will receive an error.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x2175, buffer_size)
        return unflatten_channel_string(val)

    @property
    def overloaded_chans_exist(self):
        """
        bool: Indicates if the device(s) detected an overload in any
            virtual channel in the task. Reading this property clears
            the overload status for all channels in the task. You must
            read this property before you read **overloaded_chans**.
            Otherwise, you will receive an error.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x2174)
        return val

    @property
    def overtemperature_chans(self):
        """
        List[str]: Indicates a list of names of any overtemperature
            virtual channels. You must read
            **overtemperature_chans_exist** before you read this
            property. Otherwise, you will receive an error.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x3082, buffer_size)
        return unflatten_channel_string(val)

    @property
    def overtemperature_chans_exist(self):
        """
        bool: Indicates if the device(s) detected an overtemperature
            condition in any virtual channel in the task. Reading this
            property clears the overtemperature status for all channels
            in the task. You must read this property before you read
            **overtemperature_chans**. Otherwise, you will receive an
            error.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x3081)
        return val

    @property
    def overwrite(self):
        """
        :class:`nidaqmx.constants.OverwriteMode`: Specifies whether to
            overwrite samples in the buffer that you have not yet read.
        """

        val = self._interpreter.get_read_attribute_int32(self._handle, 0x1211)
        return OverwriteMode(val)

    @overwrite.setter
    def overwrite(self, val):
        val = val.value
        self._interpreter.set_read_attribute_int32(self._handle, 0x1211, val)

    @overwrite.deleter
    def overwrite(self):
        self._interpreter.reset_read_attribute(self._handle, 0x1211)

    @property
    def pll_unlocked_chans(self):
        """
        List[str]: Indicates the channels that had their PLLs unlock.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x3119, buffer_size)
        return unflatten_channel_string(val)

    @property
    def pll_unlocked_chans_exist(self):
        """
        bool: Indicates whether the PLL is currently locked, or whether
            it became unlocked during the previous acquisition. Devices
            may report PLL Unlock either during acquisition or after
            acquisition.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x3118)
        return val

    @property
    def power_supply_fault_chans(self):
        """
        List[str]: Indicates the virtual channels that have detected a
            power supply fault. You must read
            **power_supply_fault_chans_exist** before you read this
            property. Otherwise, you will receive an error.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x3193, buffer_size)
        return unflatten_channel_string(val)

    @property
    def power_supply_fault_chans_exist(self):
        """
        bool: Indicates if the device or devices detected a power supply
            fault condition in any virtual channel in the task. Reading
            this property clears the power supply fault status for all
            channels in this task. You must read this property before
            you read **power_supply_fault_chans**. Otherwise, you will
            receive an error.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x3192)
        return val

    @property
    def raw_data_width(self):
        """
        int: Indicates in bytes the size of a raw sample from the task.
        """

        val = self._interpreter.get_read_attribute_uint32(self._handle, 0x217a)
        return val

    @property
    def read_all_avail_samp(self):
        """
        bool: Specifies whether subsequent read operations read all
            samples currently available in the buffer or wait for the
            buffer to become full before reading. NI-DAQmx uses this
            setting for finite acquisitions and only when the number of
            samples to read is -1. For continuous acquisitions when the
            number of samples to read is -1, a read operation always
            reads all samples currently available in the buffer.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x1215)
        return val

    @read_all_avail_samp.setter
    def read_all_avail_samp(self, val):
        self._interpreter.set_read_attribute_bool(self._handle, 0x1215, val)

    @read_all_avail_samp.deleter
    def read_all_avail_samp(self):
        self._interpreter.reset_read_attribute(self._handle, 0x1215)

    @property
    def relative_to(self):
        """
        :class:`nidaqmx.constants.ReadRelativeTo`: Specifies the point
            in the buffer at which to begin a read operation. If you
            also specify an offset with **offset**, the read operation
            begins at that offset relative to the point you select with
            this property. The default value is
            **ReadRelativeTo.CURRENT_READ_POSITION** unless you
            configure a Reference Trigger for the task. If you configure
            a Reference Trigger, the default value is
            **ReadRelativeTo.FIRST_PRETRIGGER_SAMPLE**.
        """

        val = self._interpreter.get_read_attribute_int32(self._handle, 0x190a)
        return ReadRelativeTo(val)

    @relative_to.setter
    def relative_to(self, val):
        val = val.value
        self._interpreter.set_read_attribute_int32(self._handle, 0x190a, val)

    @relative_to.deleter
    def relative_to(self):
        self._interpreter.reset_read_attribute(self._handle, 0x190a)

    @property
    def remote_sense_error_chans(self):
        """
        List[str]: Indicates a list of names of any virtual channels in
            the task for which a remote sense connection error condition
            has been detected. You must read Remote Sense Error Channels
            Exist before you read this property. Otherwise, you will
            receive an error.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x31de, buffer_size)
        return unflatten_channel_string(val)

    @property
    def remote_sense_error_chans_exist(self):
        """
        bool: Indicates if the device(s) detected an error condition of
            the remote sense connection for any channel in the task. You
            must disable the output and resolve the hardware connection
            issue to clear the error condition. You must read this
            property before you read the Remote Sense Error Channels
            property. Otherwise, you will receive an error.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x31dd)
        return val

    @property
    def reverse_voltage_error_chans(self):
        """
        List[str]: Indicates a list of names of all virtual channels in
            the task for which reverse voltage error condition has been
            detected. You must read the Reverse Voltage Error Channels
            Exist property before you read this property. Otherwise, you
            will receive an error.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x31e7, buffer_size)
        return unflatten_channel_string(val)

    @property
    def reverse_voltage_error_chans_exist(self):
        """
        bool: Indicates if the device(s) detected reverse voltage error
            for any of the channels in the task. Reverse voltage error
            occurs if the local voltage is equal to the negative
            saturated voltage. Reading this property clears the error
            condition status for all channels in the task. You must read
            this property before you read the Reverse Voltage Error
            Channels property. Otherwise, you will receive an error.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x31e6)
        return val

    @property
    def sleep_time(self):
        """
        float: Specifies in seconds the amount of time to sleep after
            checking for available samples if **wait_mode** is
            **WaitMode.SLEEP**.
        """

        val = self._interpreter.get_read_attribute_double(self._handle, 0x22b0)
        return val

    @sleep_time.setter
    def sleep_time(self, val):
        self._interpreter.set_read_attribute_double(self._handle, 0x22b0, val)

    @sleep_time.deleter
    def sleep_time(self):
        self._interpreter.reset_read_attribute(self._handle, 0x22b0)

    @property
    def sync_unlocked_chans(self):
        """
        List[str]: Indicates the channels from devices in an unlocked
            target.
        """

        buffer_size = self.get_channels_buffer_size()
        val = self._interpreter.get_read_attribute_string(self._handle, 0x313e, buffer_size)
        return unflatten_channel_string(val)

    @property
    def sync_unlocked_chans_exist(self):
        """
        bool: Indicates whether the target is currently locked to the
            grand master. Devices may report PLL Unlock either during
            acquisition or after acquisition.
        """

        val = self._interpreter.get_read_attribute_bool(self._handle, 0x313d)
        return val

    @property
    def total_samp_per_chan_acquired(self):
        """
        int: Indicates the total number of samples acquired by each
            channel. NI-DAQmx returns a single value because this value
            is the same for all channels. For retriggered acquisitions,
            this value is the cumulative number of samples across all
            retriggered acquisitions.
        """

        val = self._interpreter.get_read_attribute_uint64(self._handle, 0x192a)
        return val

    @property
    def wait_mode(self):
        """
        :class:`nidaqmx.constants.WaitMode`: Specifies how DAQmx Read
            waits for samples to become available.
        """

        val = self._interpreter.get_read_attribute_int32(self._handle, 0x2232)
        return WaitMode(val)

    @wait_mode.setter
    def wait_mode(self, val):
        val = val.value
        self._interpreter.set_read_attribute_int32(self._handle, 0x2232, val)

    @wait_mode.deleter
    def wait_mode(self):
        self._interpreter.reset_read_attribute(self._handle, 0x2232)

    def get_channels_buffer_size(self):
        channel_names = self._task.channel_names
        total_size = sum(len(name) + 2 for name in channel_names) + 1
        return total_size

    def _calculate_num_samps_per_chan(self, num_samps_per_chan):
        if num_samps_per_chan == -1:
            acq_type = self._task.timing.samp_quant_samp_mode

            if (acq_type == AcquisitionType.FINITE and
                    not self.read_all_avail_samp):
                return self._task.timing.samp_quant_samp_per_chan
            else:
                return self.avail_samp_per_chan
        else:
            return num_samps_per_chan

    def configure_logging(
            self, file_path: Union[str, pathlib.PurePath], logging_mode=LoggingMode.LOG_AND_READ,
            group_name="", operation=LoggingOperation.OPEN_OR_CREATE):
        """
        Configures TDMS file logging for the task.

        Args:
            file_path: Specifies the path to the TDMS file to
                which you want to log data.
            logging_mode (Optional[nidaqmx.constants.LoggingMode]):
                Specifies whether to enable logging and whether to allow
                reading data while logging. "log" mode allows for the
                best performance. However, you cannot read data while
                logging if you specify this mode. If you want to read
                data while logging, specify "LOG_AND_READ" mode.
            group_name (Optional[str]): Specifies the name of the group
                to create within the TDMS file for data from this task.
                If you append data to an existing file and the specified
                group already exists, NI-DAQmx appends a number symbol
                and a number to the group name, incrementing that number
                until finding a group name that does not exist. For
                example, if you specify a group name of Voltage Task,
                and that group already exists, NI-DAQmx assigns the
                group name Voltage Task #1, then Voltage Task #2. If you
                do not specify a group name, NI-DAQmx uses the name of
                the task.
            operation (Optional[nidaqmx.constants.LoggingOperation]):
                Specifies how to open the TDMS file.
        """

        self._interpreter.configure_logging(
            self._handle, str(file_path), logging_mode.value, group_name, operation.value)

    def read(self, number_of_samples_per_channel=READ_ALL_AVAILABLE):
        """
        Reads raw samples from the task or virtual channels you specify.

        Raw samples constitute the internal representation of samples in a
        device, read directly from the device or buffer without scaling or
        reordering. The native format of a device can be an 8-, 16-, or
        32-bit integer, signed or unsigned.

        NI-DAQmx does not separate raw data into channels. It returns data
        in an interleaved or non-interleaved 1D array, depending on the
        raw ordering of the device. Refer to your device documentation for
        more information.

        This method determines a NumPy array of appropriate size and data
        type to create and return based on your device specifications.

        Use the "timeout" property on the stream to specify 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.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.

        Args:
            number_of_samples_per_channel (int): Specifies the number of
                samples to read.

                If you set this input to nidaqmx.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.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.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.
        Returns:
            numpy.ndarray:

            The samples requested in the form of a 1D NumPy array. This
            method determines a NumPy array of appropriate size and data
            type to create and return based on your device specifications.
        """
        channels_to_read = self.channels_to_read
        number_of_channels = len(channels_to_read.channel_names)

        samp_size_in_bits = channels_to_read.ai_raw_samp_size
        has_negative_range = channels_to_read.ai_rng_low < 0

        if samp_size_in_bits == 32:
            if has_negative_range:
                dtype: Type[numpy.generic] = numpy.int32
            else:
                dtype = numpy.uint32
        elif samp_size_in_bits == 16:
            if has_negative_range:
                dtype = numpy.int16
            else:
                dtype = numpy.uint16
        else:
            if has_negative_range:
                dtype = numpy.int8
            else:
                dtype = numpy.uint8

        num_samps_per_chan = self._calculate_num_samps_per_chan(
            number_of_samples_per_channel)

        number_of_samples = number_of_channels * num_samps_per_chan

        numpy_array: numpy.typing.NDArray = numpy.zeros(number_of_samples, dtype=dtype)

        _, samples_read, _ = self._interpreter.read_raw(
            self._handle, num_samps_per_chan,
            self.timeout, numpy_array)

        if number_of_channels * samples_read != number_of_samples:
            return numpy_array[:number_of_channels * samples_read]
        return numpy_array

    @deprecation.deprecated(deprecated_in="1.0.0", removed_in="1.2.0", details="Use read_all instead.")
    def readall(self):
        return self.read_all()

    def read_all(self):
        """
        Reads all available raw samples from the task or virtual channels
        you specify.

        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, this method reads all
        the samples currently available in the buffer.

        If the task acquires a finite number of samples, 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.

        Raw samples constitute the internal representation of samples in a
        device, read directly from the device or buffer without scaling or
        reordering. The native format of a device can be an 8-, 16-, or
        32-bit integer, signed or unsigned.

        NI-DAQmx does not separate raw data into channels. It returns data
        in an interleaved or non-interleaved 1D array, depending on the
        raw ordering of the device. Refer to your device documentation for
        more information.

        This method determines a NumPy array of appropriate size and data
        type to create and return based on your device specifications.

        Use the "timeout" property on the stream to specify 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.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:
            numpy.ndarray:

            The samples requested in the form of a 1D NumPy array. This
            method determines a NumPy array of appropriate size and data
            type to create and return based on your device specifications.
        """
        return self.read(number_of_samples_per_channel=READ_ALL_AVAILABLE)

    @deprecation.deprecated(deprecated_in="1.0.0", removed_in="1.2.0", details="Use read_into instead.")
    def readinto(self, numpy_array):
        return self.read_into(numpy_array)

    def read_into(self, numpy_array):
        """
        Reads raw samples from the task or virtual channels you specify
        into numpy_array.

        The object numpy_array should be a pre-allocated, writable 1D
        numpy array.

        The number of samples per channel to read is determined using
        the following equation:

        number_of_samples_per_channel = math.floor(
            numpy_array_size_in_bytes / (
                number_of_channels_to_read * raw_sample_size_in_bytes))

        Raw samples constitute the internal representation of samples in a
        device, read directly from the device or buffer without scaling or
        reordering. The native format of a device can be an 8-, 16-, or
        32-bit integer, signed or unsigned.

        If you use a different integer size than the native format of the
        device, one integer can contain multiple samples or one sample can
        stretch across multiple integers. For example, if you use 32-bit
        integers, but the device uses 8-bit samples, one integer contains
        up to four samples. If you use 8-bit integers, but the device uses
        16-bit samples, a sample might require two integers. This behavior
        varies from device to device. Refer to your device documentation
        for more information.

        NI-DAQmx does not separate raw data into channels. It returns data
        in an interleaved or non-interleaved 1D array, depending on the
        raw ordering of the device. Refer to your device documentation for
        more information.

        Use the "timeout" property on the stream to specify 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 -1, 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.

        Args:
            numpy_array: Specifies the 1D NumPy array object into which
                the samples requested are read.
        Returns:
            int: Indicates the total number of samples read.
        """
        channels_to_read = self.channels_to_read
        number_of_channels = len(channels_to_read.channel_names)

        number_of_samples_per_channel, _ = divmod(
            numpy_array.nbytes, (
                number_of_channels * channels_to_read.ai_raw_samp_size // 8))

        _, samples_read, _ = self._interpreter.read_raw(
            self._handle, number_of_samples_per_channel,
            self.timeout, numpy_array)

        return samples_read

    def start_new_file(self, file_path: Union[str, pathlib.PurePath]):
        """
        Starts a new TDMS file the next time data is written to disk.

        Args:
            file_path: Specifies the path to the TDMS file to
                which you want to log data.
        """

        self._interpreter.start_new_file(self._handle, str(file_path))

    @property
    @deprecation.deprecated(deprecated_in="0.7.0", details="Use overwrite instead.")
    def over_write(self):
        return self.overwrite

    @over_write.setter
    @deprecation.deprecated(deprecated_in="0.7.0", details="Use overwrite instead.")
    def over_write(self, val):
        self.overwrite = val

    @over_write.deleter
    @deprecation.deprecated(deprecated_in="0.7.0", details="Use overwrite instead.")
    def over_write(self):
        del self.overwrite