Skip to content

dtollib.errors

The DtolError hierarchy and ErrorContext. See Troubleshooting for decoding common SDK error codes.

dtollib.errors

Typed exception hierarchy for :mod:dtollib.

Every library exception inherits from :class:DtolError and carries a structured :class:ErrorContext describing the failing operation.

The full hierarchy from docs/design.md §17.3 is declared here up front even though not all of these are raised yet. Forward-defining the hierarchy now means later work doesn't need to add error classes that ecosystem consumers depend on subclass-testing against.

DtolBackendError 

DtolBackendError(message='', *, context=None)

Bases: DtolError

The backend rejected an operation or surfaced a generic SDK failure.

Used when the failure is not a clean fit for the more specific subclasses. Wraps the originating ctypes exception via __cause__ when available.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolBufferOverrunError 

DtolBufferOverrunError(message='', *, context=None)

Bases: DtolCapiError

OLDA_WM_OVERRUN_ERROR — continuous AI driver fell behind.

Signals that the consumer + buffer pool can't keep up with the configured sample rate. AcquisitionSummary.overruns_observed > 0 after a run is the soft signal; this exception is the hard one. Increase BufferPlan.buffers or BufferPlan.samples_per_buffer.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolBufferUnderrunError 

DtolBufferUnderrunError(message='', *, context=None)

Bases: DtolCapiError

OLDA_WM_UNDERRUN_ERROR — continuous AO consumer ran dry.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolCapabilityError 

DtolCapabilityError(message='', *, context=None)

Bases: DtolError

Requested feature not supported per the live CapabilitySet.

Examples: configuring continuous mode on a subsystem that reports only OLSSC_SUP_SINGLEVALUE; requesting read_inprocess on a board without the inprocess-flush capability.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolCapiError 

DtolCapiError(message='', *, context=None)

Bases: DtolError

Raw ECODE from oldaapi / olmem failed.

Parent for the buffer / trigger flavours below. Specific OLSTATUS ranges and codes map into the more precise subclasses; everything else lands here.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolConfigurationError 

DtolConfigurationError(message='', *, context=None)

Bases: DtolError

Configuration-level error (bad spec, missing required field, ...).

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolConfirmationRequiredError 

DtolConfirmationRequiredError(message='', *, context=None)

Bases: DtolError

A safety-gated operation was attempted without confirm=True.

Raised by AO / DO / CO writes outside their safe_min/safe_max band, by pulse-train start, by auto-calibration, and by any operation that mutates persistent device state. Matches the ecosystem ConfirmationRequiredError convention shared with :mod:watlowlib and :mod:sartoriuslib.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolConnectionError 

DtolConnectionError(message='', *, context=None)

Bases: DtolError

Communication with the DT-Open Layers board was lost or unavailable.

Aligns with the ecosystem ConnectionError convention (matching :class:watlowlib.WatlowConnectionError, :class:alicatlib.AlicatConnectionError, :class:sartoriuslib.SartoriusConnectionError, and :class:nidaqlib.NIDaqConnectionError). Raised when a USB DT module is unplugged or powered down mid-session, or when the driver handle is invalidated — the board is gone, as opposed to a transient, retry-safe blip (:class:DtolTransientError).

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolDependencyError 

DtolDependencyError(message='', *, context=None)

Bases: DtolError

A required dependency is unavailable.

Most common cases: the DT-Open Layers SDK is not installed (no oldaapi*.dll / olmem*.dll on the resolution chain), bitness mismatch (32-bit Python attempting to load oldaapi64.dll), or running on a non-Windows platform.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolError 

DtolError(message='', *, context=None)

Bases: Exception

Base class for every :mod:dtollib exception.

Carries a typed :class:ErrorContext. The message is the human-readable summary; the context is the machine-readable detail.

Initialise with a human-readable message and optional context.

Parameters:

Name Type Description Default
message str

Short, human-readable summary suitable for logs.

 ''
context ErrorContext | None

Structured fields about the failing operation. None yields an empty :class:ErrorContext.

 None
Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

with_context 

with_context(**updates)

Return a copy of this error with its context updated.

Useful when an inner layer raises and an outer layer wants to enrich the context (for instance adding task_name or operation).

Source code in src/dtollib/errors.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def with_context(self, **updates: Any) -> Self:
    """Return a copy of this error with its context updated.

    Useful when an inner layer raises and an outer layer wants to enrich
    the context (for instance adding ``task_name`` or ``operation``).
    """
    cls = type(self)
    new = cls.__new__(cls)
    new.args = self.args
    try:
        new.__dict__.update(self.__dict__)
    except AttributeError:  # pragma: no cover — no slotted subclass today
        for slot in getattr(cls, "__slots__", ()):
            if hasattr(self, slot):
                object.__setattr__(new, slot, getattr(self, slot))
    new.context = self.context.merged(**updates)
    new.__cause__ = self.__cause__
    new.__context__ = self.__context__
    new.__traceback__ = self.__traceback__
    return new

DtolReadError 

DtolReadError(message='', *, context=None)

Bases: DtolError

A read against the underlying SDK failed.

Wraps olDaGetSingleValue*, olDaGetBuffer, olDaCopyFromBuffer failures that are not classified into a more specific subclass.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolResourceError 

DtolResourceError(message='', *, context=None)

Bases: DtolError

A resource conflict was detected.

Examples: HDASS already reserved, simultaneous-start pool members mismatch. Best-effort signal — the SDK is the final authority. Raised by :meth:DtolManager.add when the new task's board + element overlap with one already managed.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolSinkDependencyError 

DtolSinkDependencyError(message='', *, context=None)

Bases: DtolSinkError

A sink's optional dependency (pyarrow, asyncpg, ...) is missing.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolSinkError 

DtolSinkError(message='', *, context=None)

Bases: DtolError

Base class for sink-layer failures.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolSinkSchemaError 

DtolSinkSchemaError(message='', *, context=None)

Bases: DtolSinkError

A sink rejected an input record's shape.

Most commonly raised by row-oriented sinks (CsvSink, JsonlSink) when handed a :class:~dtollib.tasks.DaqBlock without accept_blocks=True — silently scalarising would surprise users with 1-GB CSV files at 10 kHz × 8 channels.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolSinkWriteError 

DtolSinkWriteError(message='', *, context=None)

Bases: DtolSinkError

A sink failed while writing a batch (file I/O, DB error, ...).

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolTaskStateError 

DtolTaskStateError(message='', *, context=None)

Bases: DtolError

Operation invalid for the task's current lifecycle state.

Raised, for example, by :meth:DtolSession.poll when the task is CONFIGURED_FOR_CONTINUOUS and RUNNING — two consumers on the same SDK buffer queue would race. Or by register_notification after olDaConfig (the §12.3.2 invariant).

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolTimeoutError 

DtolTimeoutError(message='', *, context=None)

Bases: DtolError

An SDK read or write exceeded its configured timeout.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolTransientError 

DtolTransientError(message='', *, context=None)

Bases: DtolError

A driver-layer error that is safe to retry without rebuilding the task.

Surfaced when an oldaapi / olmem call fails with a code in the retry-safe set (for example a buffer-done window that slid just ahead of the drainer). Consumers under :attr:ErrorPolicy.RETURN may re-attempt the operation. Distinct from :class:DtolBufferOverrunError, which signals the consumer cannot keep up at the configured sample rate.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolTriggerError 

DtolTriggerError(message='', *, context=None)

Bases: DtolCapiError

OLDA_WM_TRIGGER_ERROR — trigger condition could not be configured.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolValidationError 

DtolValidationError(message='', *, context=None)

Bases: DtolError

Client-side validation failed before any SDK call.

Raised by __post_init__ on spec dataclasses (e.g. TC range outside the type's operating envelope, MULTI_SENSOR ordering violation in the fake backend, mixing channel kinds in one TaskSpec.channels).

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

DtolWriteError 

DtolWriteError(message='', *, context=None)

Bases: DtolError

A write against the underlying SDK failed.

Raised by :meth:DtolSession.write when the backend rejects the write. Out-of-range values fail earlier as :class:DtolValidationError.

Source code in src/dtollib/errors.py 
145
146
147
148
149
150
151
152
153
154
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    """Initialise with a human-readable message and optional context.

    Args:
        message: Short, human-readable summary suitable for logs.
        context: Structured fields about the failing operation. ``None``
            yields an empty :class:`ErrorContext`.
    """
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

ErrorContext dataclass 

ErrorContext(
    task_name=None,
    board=None,
    subsystem_type=None,
    element=None,
    channel_name=None,
    channel=None,
    operation=None,
    ecode=None,
    ecode_source=None,
    ecode_message=None,
    sdk_event_kind=None,
    extra=_empty_extra(),
)

Structured context attached to every :class:DtolError.

Field shape matches docs/design.md §17.2. Cross-instrument log readers join these on (task_name, board, channel_name, ...) against the sibling libraries' equivalents.

Attributes:

Name Type Description
task_name str | None

TaskSpec.name of the task at fault.
board str | None

DT-Open Layers board name (e.g. "DT9805(00)").
subsystem_type SubsystemType | None

Failing subsystem kind, if known.
element int | None

Subsystem element index, if relevant.
channel_name str | None

Display name of the at-fault channel.
channel int | None

Physical-channel number (the SDK uses 0-based int).
operation str | None

Logical operation name ("olDaConfig", "read_block", "poll", ...).
ecode int | None

Raw ECODE / OLSTATUS value from the SDK call.
ecode_source Literal['oldaapi', 'olmem'] | None

Which DLL emitted the code ("oldaapi" vs "olmem") — controls which error-string function decodes ecode.
ecode_message str | None

Decoded SDK error string.
sdk_event_kind _SdkEventKindHolder | None

SDK notification message kind on async-error wraps (typed as object | None until the enum lands; see module docstring).
extra Mapping[str, Any]

Free-form additional context.

merged 

merged(**updates)

Return a new context with updates overlaid. Unknown keys go to extra.

Source code in src/dtollib/errors.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def merged(self, **updates: Any) -> Self:
    """Return a new context with ``updates`` overlaid. Unknown keys go to ``extra``."""
    known: dict[str, Any] = {}
    extra_updates: dict[str, Any] = {}
    for key, value in updates.items():
        if key in _CONTEXT_KNOWN_FIELDS:
            known[key] = value
        else:
            extra_updates[key] = value

    new_extra: Mapping[str, Any] = (
        MappingProxyType({**self.extra, **extra_updates}) if extra_updates else self.extra
    )
    return replace(self, **known, extra=new_extra)