"""The pyobs exception hierarchy, used both for local errors and to carry failures across the RPC
boundary between modules.
Two independent axes live in this hierarchy:
- **Domain exceptions** (everything above `RemoteError` below) mean "the operation failed for
reason X." These deliberately multiply into fine-grained leaves (`MoveError`, `FocusError`,
`GrabImageError`, ...) so a caller can react differently to each one, rather than catching one
broad type and inspecting a message string.
- **`RemoteError`** and its small subtree mean "the call itself didn't reach/return" (a transport
failure), not "the remote operation failed for a domain reason." That distinction doesn't
benefit from the same fine-grained treatment, so it stays deliberately small.
A domain exception raised on the far side of an RPC call is looked up in the registry below by its
fully-qualified name and re-raised locally as its real type, tagged with `remote_module` (see
`comm/xmpp/rpc.py`'s fault reconstruction) -- it does not travel wrapped in a `RemoteError`. If the
type can't be resolved because its defining module was never imported in this process, it falls back
to the nearest of its base classes that is registered here (the fault carries their names, see
`comm/xmpp/rpc.py`; not in 2.14.2), e.g. `MotionError` for a driver's own `MotionError` subclass, with
the original type name preserved as `original_type`. If there is no such base class, or the exception
was never a `PyobsError` to begin with, it arrives as `UnclassifiedError`, again with `original_type`.
Driver authors should therefore subclass the closest class of this module, not `PyobsError` itself. See
the "Exceptions across modules" guide in the documentation.
Every subclass registers itself automatically via `__init_subclass__` (`PyobsError._registry`);
there's nothing to do at definition time beyond subclassing the right branch of the hierarchy.
"""
from __future__ import annotations
import logging
from collections.abc import Callable, Coroutine
from typing import Any, NamedTuple
__title__ = "Exceptions"
[docs]
class PyobsError(Exception):
"""Base class for all exceptions.
Define your own errors by subclassing the closest class of this module (e.g. `MotionError`), not
`PyobsError` itself: a caller that did not import your module then still receives the nearest of these
classes, otherwise it gets `UnclassifiedError`.
"""
_registry: dict[str, type[PyobsError]] = {}
def __init_subclass__(cls, **kwargs: Any) -> None:
super().__init_subclass__(**kwargs)
PyobsError._registry[f"{cls.__module__}.{cls.__qualname__}"] = cls
def __init__(self, message: str | None = None, **context: Any) -> None:
self.message = message
self.logged = False
self.context = context
for key, value in context.items():
setattr(self, key, value)
def __str__(self) -> str:
msg = f"<{self.__class__.__name__}>"
if self.message is not None:
msg += f" {self.message}"
return msg
[docs]
def log(self, log: logging.Logger, level: str, message: str, **kwargs: Any) -> None:
if self.logged:
return
log.log(logging.getLevelName(level), message, **kwargs)
self.logged = True
[docs]
@classmethod
def resolve(cls, qualified_name: str) -> type[PyobsError] | None:
return cls._registry.get(qualified_name)
#######################################
[docs]
class ModuleError(PyobsError):
pass
[docs]
class GeneralError(PyobsError):
pass
[docs]
class ImageError(PyobsError):
pass
[docs]
class ArchiveError(PyobsError):
"""An archive backend couldn't answer a query -- it was unreachable, or it returned a non-200
response. Distinct from an archive that answers fine but has no matching frames, which is a
normal empty result rather than an error. Deliberately one type across list_options/
list_frames and every backend, not split by failure mode: callers react the same way to all of
them (back off and retry later), and only the message distinguishes "HTTP 502 from <url>" from
"connection refused"."""
pass
[docs]
class MotionError(PyobsError):
pass
[docs]
class InitError(MotionError):
pass
[docs]
class ParkError(MotionError):
pass
[docs]
class MoveError(MotionError):
pass
[docs]
class PointingLimitError(PyobsError):
"""The requested destination is outside the configured pointing limits (today the minimum altitude, later
e.g. sun distance or dome limits). Nothing was attempted and the device is fine, so this is an expected
condition: defer or skip the target, don't alert and don't retry right away. Deliberately not a
MotionError, so that it never counts towards a module's repeated-motion-failure escalation.
Remote rejections also do not count towards broad RemoteError severity handlers; explicit
handlers for this class or its ancestors still apply."""
pass
[docs]
class GrabImageError(PyobsError):
pass
[docs]
class AbortedError(PyobsError):
pass
[docs]
class FocusError(PyobsError):
pass
[docs]
class AcquisitionError(PyobsError):
pass
[docs]
class DeviceBusyError(PyobsError):
"""The device can't service this request right now because it's already busy with another
operation (e.g. an exposure/sequence already running, or another motion in progress) -- back
off and retry, as opposed to GrabImageError/MoveError/etc., which mean the operation was
actually attempted and failed. Deliberately one type across camera/telescope/roof/focuser
modules alike, not split by device or by which specific operation was busy -- no caller reacts
differently to those variants."""
pass
[docs]
class ModuleStartingError(PyobsError):
"""The module hasn't finished its own startup yet (still inside `open()`, e.g. connecting to
hardware) and isn't safe to command -- distinct from DeviceBusyError, which means the module is
fully up but its device is occupied with another operation. Back off and retry once the module
reports ModuleState.READY, rather than treating this as a domain failure."""
pass
[docs]
class NotSupportedError(PyobsError):
"""This module doesn't support the requested operation at all -- a capability the module only
optionally implements (e.g. an alt/az-only telescope asked to move_radec) isn't available,
as opposed to a specific attempt at that operation failing. Cross-cutting: any module with an
optional-capability mixin can raise this, not just telescopes."""
pass
[docs]
class InvalidArgumentError(PyobsError):
"""The caller passed an argument this method rejects (unknown name, out-of-range value, ...).
Unlike a plain ValueError, this survives an RPC round trip as itself, so `except
exc.InvalidArgumentError:` around a proxy call actually catches it -- a bare ValueError would
silently degrade to UnclassifiedError the moment the call crosses XMPP, even though it works
fine locally (LocalComm, direct calls, tests), which is exactly the kind of inconsistency that
only shows up once code goes from a local test to a networked deployment. Deliberately one
type across every setter-shaped method, not split per method or per argument -- no caller
reacts differently to "unknown filter" vs. "invalid focus value."""
pass
[docs]
class UnclassifiedError(PyobsError):
"""Wraps an exception that isn't part of the deliberate PyobsError contract -- either it never
was a PyobsError to begin with (a builtin, a vendor SDK exception) and escaped a module's own
method body unconverted, or it crossed an RPC boundary and couldn't be reconstructed as its real
type because its defining module was never imported in this process and none of its base classes
(other than `PyobsError`) is known here either. The original type name survives as `original_type`
even when the class itself doesn't."""
pass
#######################################
[docs]
class RemoteError(PyobsError):
"""The call itself didn't reach/return -- a transport failure, not a domain exception raised
by the remote operation (see the module docstring for that distinction). Kept deliberately
small: "the call failed to even happen" doesn't need the same fine-grained treatment as a
domain failure."""
pass
[docs]
class RemoteTimeoutError(RemoteError):
pass
[docs]
class ForbiddenError(RemoteError):
"""Raised when a caller is not permitted to invoke a method under the target module's ACL policy."""
pass
#######################################
[docs]
class LoggedException(NamedTuple):
time: float
exception: PyobsError
[docs]
class ExceptionHandler(NamedTuple):
exc_type: type[PyobsError]
limit: int
timespan: float | None = None
module: str | None = None
callback: Callable[[PyobsError], Coroutine[Any, Any, None]] | None = None
#######################################