Skip to content

alicatlib.errors

Typed exception hierarchy. Every exception raised by the library is a subclass of AlicatError and carries a structured ErrorContext. See Design §5.17 and Troubleshooting for remediation guidance by error type.

alicatlib.errors

Typed error hierarchy for :mod:alicatlib.

Every exception raised by the library is a subclass of :class:AlicatError and carries a structured :class:ErrorContext. The context is deliberately a typed dataclass (not **kwargs) so IDEs and mypy --strict can reason about it, and so rendering is consistent across tracebacks.

Design reference: docs/design.md §5.17.

AlicatCapabilityError 

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

Bases: AlicatError

The device cannot perform the requested command.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatCommandRejectedError 

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

Bases: AlicatProtocolError

The device replied with its error marker (? / similar).

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatConfigurationError 

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

Bases: AlicatError

User-supplied configuration was invalid.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatConnectionError 

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

Bases: AlicatTransportError

Connection could not be established or was lost.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatDiscoveryError 

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

Bases: AlicatError

Device discovery failed.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatError 

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

Bases: Exception

Base class for every exception raised by :mod:alicatlib.

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

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    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 port or elapsed_s).

Allocates a fresh instance via cls.__new__ and copies attribute state directly. Avoids re-invoking __init__ — many subclasses (AlicatMediumMismatchError, AlicatFirmwareError, UnknownGasError and friends) have bespoke keyword-only signatures that don't accept (message, *, context=), and copy.copy would silently dispatch through them via :meth:Exception.__reduce__.

Source code in src/alicatlib/errors.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
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 ``port`` or ``elapsed_s``).

    Allocates a fresh instance via ``cls.__new__`` and copies attribute
    state directly. Avoids re-invoking ``__init__`` — many subclasses
    (``AlicatMediumMismatchError``, ``AlicatFirmwareError``,
    ``UnknownGasError`` and friends) have bespoke keyword-only
    signatures that don't accept ``(message, *, context=)``, and
    ``copy.copy`` would silently dispatch through them via
    :meth:`Exception.__reduce__`.
    """
    cls = type(self)
    new = cls.__new__(cls)
    # ``Exception`` slot state lives in ``self.args``; reuse it so
    # ``str(err)`` keeps the original message.
    new.args = self.args
    # Copy subclass-specific attributes (value, field_name, required_min, ...).
    # Use ``__dict__`` directly when present (most subclasses), and fall back
    # to slot iteration if a frozen variant ever appears.
    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

AlicatFirmwareError 

AlicatFirmwareError(
    *,
    command,
    reason,
    actual=None,
    required_min=None,
    required_max=None,
    required_families=None,
    context=None,
)

Bases: AlicatCapabilityError

The device's firmware version is outside the command's supported range.

Source code in src/alicatlib/errors.py 
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
def __init__(
    self,
    *,
    command: str,
    reason: str,
    actual: FirmwareVersion | None = None,
    required_min: FirmwareVersion | None = None,
    required_max: FirmwareVersion | None = None,
    required_families: frozenset[FirmwareFamily] | None = None,
    context: ErrorContext | None = None,
) -> None:
    self.command = command
    self.reason = reason
    self.actual = actual
    self.required_min = required_min
    self.required_max = required_max
    self.required_families = required_families
    required = ""
    if required_min is not None or required_max is not None:
        lo = str(required_min) if required_min is not None else "*"
        hi = str(required_max) if required_max is not None else "*"
        required = f" (requires {lo}..{hi}, have {actual})"
    elif required_families:
        fams = ", ".join(f.value for f in sorted(required_families, key=lambda x: x.value))
        required = f" (requires family in {{{fams}}}, have {actual})"
    super().__init__(
        f"Firmware check failed for {command}: {reason}{required}",
        context=context,
    )

AlicatMediumMismatchError 

AlicatMediumMismatchError(
    *,
    command,
    device_media,
    command_media,
    hint=None,
    context=None,
)

Bases: AlicatConfigurationError

A command's declared medium doesn't intersect the device's configured medium.

Raised pre-I/O from :class:alicatlib.devices.session.Session at the media gate (design §5.4, §5.9a). The typical shape: calling :meth:Device.gas on a liquid-only device, or :meth:Device.fluid on a gas-only device. The error carries the mismatch in :attr:ErrorContext.device_media and :attr:ErrorContext.command_media and points at the remediation API in its message.

Source code in src/alicatlib/errors.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
def __init__(
    self,
    *,
    command: str,
    device_media: Medium,
    command_media: Medium,
    hint: str | None = None,
    context: ErrorContext | None = None,
) -> None:
    self.command = command
    self.device_media = device_media
    self.command_media = command_media
    suffix = f" — {hint}" if hint else ""
    super().__init__(
        (
            f"{command} requires medium {command_media.name or command_media!r} but "
            f"device is configured as {device_media.name or device_media!r}{suffix}"
        ),
        context=context,
    )

AlicatMissingHardwareError 

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

Bases: AlicatCapabilityError

The device lacks hardware the command requires.

Raised from :class:alicatlib.devices.session.Session before any I/O, using the :class:alicatlib.commands.base.Capability bits declared on the :class:alicatlib.commands.base.Command spec. More useful than letting the device silently respond ? — tells the caller exactly which capability is missing (BAROMETER, MULTI_VALVE, ANALOG_INPUT, ...). See design §5.17.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatParseError 

AlicatParseError(
    message,
    *,
    field_name=None,
    expected=None,
    actual=None,
    context=None,
)

Bases: AlicatProtocolError

A response could not be parsed into its typed model.

Source code in src/alicatlib/errors.py 
341
342
343
344
345
346
347
348
349
350
351
352
353
def __init__(
    self,
    message: str,
    *,
    field_name: str | None = None,
    expected: object = None,
    actual: object = None,
    context: ErrorContext | None = None,
) -> None:
    self.field_name = field_name
    self.expected = expected
    self.actual = actual
    super().__init__(message, context=context)

AlicatProtocolError 

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

Bases: AlicatError

The bytes arrived but did not parse as a valid Alicat response.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatSinkDependencyError 

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

Bases: AlicatSinkError, AlicatConfigurationError

A sink's optional backing library is not installed.

Raised when the user instantiates (or calls open() on) a sink whose extras have not been installed — e.g. ParquetSink without alicatlib[parquet] or PostgresSink without alicatlib[postgres]. The message always names the exact extra to install so the remediation is copy-pasteable.

Multi-inherits :class:AlicatConfigurationError because callers that already branch on configuration errors (missing extras being a configuration problem from their perspective) keep working without changes.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatSinkError 

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

Bases: AlicatError

Base class for errors raised by sinks (CSV, JSONL, SQLite, Parquet, Postgres).

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatSinkSchemaError 

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

Bases: AlicatSinkError

A batch's shape is incompatible with the sink's locked schema.

Raised when a sink has locked its schema on the first batch (or validated against an existing table) and a subsequent batch carries rows whose shape can't be reconciled — for example, a Postgres target table that's missing a required column, or a Parquet writer that would need a type change mid-file.

Dropping unknown optional columns is handled by a per-sink WARN log and does not raise.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatSinkWriteError 

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

Bases: AlicatSinkError

The backing store rejected a write.

Wraps the underlying driver exception (sqlite3, asyncpg, pyarrow) so downstream error handlers don't need to import optional dependencies. The original exception is preserved via raise ... from original so tracebacks remain intact.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatStreamingModeError 

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

Bases: AlicatProtocolError

A request/response command was attempted while the client was in streaming mode.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatTimeoutError 

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

Bases: AlicatTransportError

An I/O timeout expired.

A timeout is never represented as an empty successful response.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatTransportError 

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

Bases: AlicatError

Serial/TCP transport failed to move bytes.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatUnitIdMismatchError 

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

Bases: AlicatProtocolError

The response's unit ID did not match the request's.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatUnsupportedCommandError 

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

Bases: AlicatCapabilityError

The command is not supported on this device kind.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

AlicatValidationError 

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

Bases: AlicatConfigurationError

Arguments failed validation before any I/O (range checks, missing confirm).

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

ErrorContext dataclass 

ErrorContext(
    command_name=None,
    command_bytes=None,
    raw_response=None,
    unit_id=None,
    port=None,
    firmware=None,
    device_kind=None,
    device_media=None,
    command_media=None,
    elapsed_s=None,
    extra=_empty_extra(),
)

Structured context attached to every :class:AlicatError.

Every field is optional so callers can build a context progressively as a command flows through layers (transport → protocol → session → command).

extra accepts any Mapping and is always frozen into a read-only :class:types.MappingProxyType at construction. The shared empty sentinel can therefore never be mutated through error.context.extra[k] = v.

merged 

merged(**updates)

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

Source code in src/alicatlib/errors.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
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)

InvalidUnitIdError 

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

Bases: AlicatConfigurationError

A unit ID was not a single letter A — Z.

Source code in src/alicatlib/errors.py 
127
128
129
def __init__(self, message: str = "", *, context: ErrorContext | None = None) -> None:
    super().__init__(message)
    self.context = context if context is not None else _EMPTY_CONTEXT

UnknownFluidError 

UnknownFluidError(value, *, suggestions=(), context=None)

Bases: AlicatConfigurationError

A fluid (working-liquid) name or code did not resolve against the registry.

Source code in src/alicatlib/errors.py 
222
223
224
225
226
227
228
229
230
231
232
def __init__(
    self,
    value: str | int,
    *,
    suggestions: tuple[str, ...] = (),
    context: ErrorContext | None = None,
) -> None:
    self.value = value
    self.suggestions = suggestions
    hint = f" (did you mean: {', '.join(suggestions)}?)" if suggestions else ""
    super().__init__(f"Unknown fluid: {value!r}{hint}", context=context)

UnknownGasError 

UnknownGasError(value, *, suggestions=(), context=None)

Bases: AlicatConfigurationError

A gas name or code did not resolve against the registry.

Source code in src/alicatlib/errors.py 
206
207
208
209
210
211
212
213
214
215
216
def __init__(
    self,
    value: str | int,
    *,
    suggestions: tuple[str, ...] = (),
    context: ErrorContext | None = None,
) -> None:
    self.value = value
    self.suggestions = suggestions
    hint = f" (did you mean: {', '.join(suggestions)}?)" if suggestions else ""
    super().__init__(f"Unknown gas: {value!r}{hint}", context=context)

UnknownStatisticError 

UnknownStatisticError(
    value, *, suggestions=(), context=None
)

Bases: AlicatConfigurationError

A statistic name or code did not resolve against the registry.

Source code in src/alicatlib/errors.py 
254
255
256
257
258
259
260
261
262
263
264
def __init__(
    self,
    value: str | int,
    *,
    suggestions: tuple[str, ...] = (),
    context: ErrorContext | None = None,
) -> None:
    self.value = value
    self.suggestions = suggestions
    hint = f" (did you mean: {', '.join(suggestions)}?)" if suggestions else ""
    super().__init__(f"Unknown statistic: {value!r}{hint}", context=context)

UnknownUnitError 

UnknownUnitError(value, *, suggestions=(), context=None)

Bases: AlicatConfigurationError

A unit name or code did not resolve against the registry.

Source code in src/alicatlib/errors.py 
238
239
240
241
242
243
244
245
246
247
248
def __init__(
    self,
    value: str | int,
    *,
    suggestions: tuple[str, ...] = (),
    context: ErrorContext | None = None,
) -> None:
    self.value = value
    self.suggestions = suggestions
    hint = f" (did you mean: {', '.join(suggestions)}?)" if suggestions else ""
    super().__init__(f"Unknown unit: {value!r}{hint}", context=context)