"""NI-DAQmx error classes."""
import warnings
import deprecation
from nidaqmx.error_codes import DAQmxErrors, DAQmxWarnings
__all__ = ["DaqError", "DaqReadError", "DaqWriteError", "DaqWarning", "DaqResourceWarning"]
class Error(Exception):
"""Base error class for module."""
pass
class DaqNotFoundError(Error):
"""Error raised when NI-DAQmx driver is not installed."""
pass
class DaqNotSupportedError(Error):
"""Error raised when DAQmx is not supported on this platform."""
pass
class DaqFunctionNotSupportedError(Error):
"""Error raised when a specific function isn't supported by the installed NI-DAQmx driver."""
pass
[docs]
class DaqError(Error):
"""Error raised by any DAQmx method."""
def __init__(self, message, error_code, task_name=""):
"""Initialize a new DaqError.
Args:
message (string): Specifies the error message.
error_code (int): Specifies the NI-DAQmx error code.
""" # noqa: D417 - Missing argument descriptions in the docstring (auto-generated noqa)
self._error_code = int(error_code)
try:
self._error_type = DAQmxErrors(self._error_code)
except ValueError:
self._error_type = DAQmxErrors.UNKNOWN
# If message is empty, we try to put at least some information in it
if not message:
message = f"Description could not be found for the status code.\n\nStatus Code: {self._error_code}"
if task_name:
message = f"{message}\n\nTask Name: {task_name}"
# We do not know where the error description came from, so we add the status code if it is not already in the message # noqa: W505 - doc line too long (125 > 100 characters) (auto-generated noqa)
if str(self._error_code) not in message:
message = f"{message}\n\nStatus Code: {self._error_code}"
super().__init__(message)
@property
def error_code(self):
"""int: Specifies the NI-DAQmx error code."""
return self._error_code
@property
def error_type(self):
""":class:`nidaqmx.error_codes.DAQmxErrors`: Specifies the NI-DAQmx error type."""
return self._error_type
[docs]
class DaqReadError(DaqError):
"""Error raised by DAQmx write method that includes the amount of data that was read."""
def __init__(self, message, error_code, samps_per_chan_read, task_name=""):
"""Initialize a new DaqReadError.
Args:
message (string): Specifies the error message.
error_code (int): Specifies the NI-DAQmx error code.
""" # noqa: D417 - Missing argument descriptions in the docstring (auto-generated noqa)
super().__init__(message, error_code, task_name)
self._samps_per_chan_read = samps_per_chan_read
@property
def samps_per_chan_read(self):
"""int: Indicates the number of samples successfully read."""
return self._samps_per_chan_read
[docs]
class DaqWriteError(DaqError):
"""Error raised by DAQmx write method that includes the amount of data that was written."""
def __init__(self, message, error_code, samps_per_chan_written, task_name=""):
"""Initialize a new DaqWriteError.
Args:
message (string): Specifies the error message.
error_code (int): Specifies the NI-DAQmx error code.
samps_per_chan_written (int): Specifies the number of samples written.
""" # noqa: D417 - Missing argument descriptions in the docstring (auto-generated noqa)
super().__init__(message, error_code, task_name)
self._samps_per_chan_written = samps_per_chan_written
@property
def samps_per_chan_written(self):
"""int: Indicates the number of samples successfully written."""
return self._samps_per_chan_written
[docs]
class DaqWarning(Warning):
"""Warning raised by any NI-DAQmx method."""
def __init__(self, message, error_code):
"""Initialize a new DaqWarning.
Args:
message (string): Specifies the warning message.
error_code (int): Specifies the NI-DAQmx error code.
"""
super().__init__(f"\nWarning {error_code} occurred.\n\n{message}")
self._error_code = int(error_code)
try:
self._error_type = DAQmxWarnings(self._error_code)
except ValueError:
self._error_type = DAQmxWarnings.UNKNOWN
@property
def error_code(self):
"""int: Specifies the NI-DAQmx error code."""
return self._error_code
@property
def error_type(self):
""":class:`nidaqmx.error_codes.DAQmxWarnings`: Specifies the NI-DAQmx error type."""
return self._error_type
[docs]
class DaqResourceWarning(ResourceWarning):
"""Warning about DAQ resource usage, such as a leaking an NI-DAQmx task."""
pass
warnings.filterwarnings("always", category=DaqWarning)
warnings.filterwarnings("always", category=DaqResourceWarning)
@deprecation.deprecated(
deprecated_in="0.8.0", details="This function will be removed in a future update."
)
def check_for_error(error_code, samps_per_chan_written=None, samps_per_chan_read=None):
from nidaqmx._library_interpreter import LibraryInterpreter
return LibraryInterpreter().check_for_error(
error_code, samps_per_chan_written, samps_per_chan_read
)
@deprecation.deprecated(
deprecated_in="0.8.0", details="This function will be removed in a future update."
)
def is_string_buffer_too_small(error_code):
import nidaqmx._library_interpreter
return nidaqmx._library_interpreter.is_string_buffer_too_small(error_code)
@deprecation.deprecated(
deprecated_in="0.8.0", details="This function will be removed in a future update."
)
def is_array_buffer_too_small(error_code):
import nidaqmx._library_interpreter
return nidaqmx._library_interpreter.is_array_buffer_too_small(error_code)
class RpcError(Error):
"""An error specific to sessions to the NI gRPC Device Server."""
def __init__(self, rpc_code, description):
"""Initialize a new RpcError."""
self.rpc_code = rpc_code
self.description = description
try:
import grpc
rpc_error = str(grpc.StatusCode(self.rpc_code))
except Exception:
rpc_error = str(self.rpc_code)
super().__init__(rpc_error + ": " + self.description)
class FeatureNotSupportedError(Exception):
"""The feature is not supported at the current code readiness level."""