"""The AutoWISP exception hierarchy.
Every exception AutoWISP raises on purpose derives from
:class:`AutoWISPError`, which carries enough context (the affected
artifact(s), the pipeline run, the worker that raised it, a short
user-facing message, and an arbitrary ``details`` dict) to power CLI
messages, BUI surfacing, and post-mortem debugging from a single source
of truth. See ``error_handling_plan.md`` for the full design.
This module imports only :class:`FrozenRow` from the database package
(itself dependency-free), so the hierarchy stays free of a hard
SQLAlchemy dependency.
"""
import importlib.metadata
import os
import platform
import socket
import traceback as traceback_module
from datetime import datetime, timezone
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Optional, Sequence
import numpy
from autowisp.database.frozen_row import FrozenRow
git_id = "$Id: cbd3289b0eb32e12d1bdf639e748bcddb5a8819c $"
# A type-dispatch sanitizer: one return per handled kind reads clearer
# than nesting.
# pylint: disable=too-many-return-statements
[docs]
def sanitize_for_json(obj, max_inline_array_size=64):
"""Coerce an otherwise-unserializable object to a JSON-friendly form.
Intended as the ``default=`` argument to :func:`json.dump`/`json.dumps`
when writing the error sidecar: ``details`` may carry numpy
scalars/arrays, :class:`~pathlib.Path`, :class:`~datetime.datetime`,
sets, and arbitrary objects. This is *total* -- it never raises, so a
pathological value can never turn recording an error into a second
error; the last resort is ``repr()``.
Use directly (``default=sanitize_for_json``) for the default
threshold, or ``default=functools.partial(sanitize_for_json,
max_inline_array_size=N)`` to override it.
Args:
obj: The value ``json`` could not serialize natively.
max_inline_array_size(int): ndarrays with at most this many
elements are dumped in full; larger ones are summarized so a
stray full-frame array cannot write hundreds of MB.
Returns:
A JSON-serializable stand-in for ``obj``.
"""
if isinstance(obj, numpy.ndarray):
if obj.size <= max_inline_array_size:
return obj.tolist()
return {
"__ndarray__": {
"shape": list(obj.shape),
"dtype": str(obj.dtype),
"head": obj.ravel()[:max_inline_array_size].tolist(),
}
}
if isinstance(obj, numpy.generic):
return obj.item()
if isinstance(obj, Path):
# as_posix() so the serialized path is stable across platforms
# (str() would emit backslashes on Windows).
return obj.as_posix()
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, (set, frozenset)):
return list(obj)
try:
return repr(obj)
except Exception: # pylint: disable=broad-except
return "<unrepresentable>"
# --- Environment / resource provenance. -------------------------------
# These live here (the leaf error module, alongside ``sanitize_for_json``)
# because every layer that records or bundles an error needs them --
# ``error_context`` (crashed-worker resources), ``error_persistence``
# (crash-time environment in the sidecar), ``crash_report`` (report-time
# provenance), and ``run_pipeline`` (the run host) -- and all already
# import this module, so this is the one cycle-free home.
[docs]
def get_hostname():
"""This machine's name, recorded consistently across the pipeline.
A single source so the *run* host (``PipelineRun.host``) and the
*report* host (:func:`autowisp.crash_report.collect_provenance`) agree
-- otherwise ``socket.getfqdn()`` in one place and
``socket.gethostname()`` in another make one box look like two.
``gethostname`` is preferred: it is fast and returns the clean local
name rather than a reverse-DNS ``*.in-addr.arpa`` form on a
loopback-only host.
"""
return socket.gethostname()
[docs]
def collect_environment(
packages=(
"autowisp",
"astrowisp",
"numpy",
"scipy",
"pandas",
"sqlalchemy",
"astropy",
),
):
"""Platform + key package versions of the *current* process.
Recorded into the error sidecar at crash time (by
``error_persistence``) so a report reflects the environment that
actually produced the failure -- immune to the report being built
later, after packages were upgraded. Report-time provenance records the
same shape for the machine building the report, so comparing the two
reveals drift. Never raises.
Args:
packages(iterable): Distribution names whose versions to record
(the runtime stack whose combination determines whether a crash
reproduces); those not installed are omitted.
Returns:
dict: ``platform`` / ``python_version`` / ``packages``.
"""
versions = {}
for name in packages:
try:
versions[name] = importlib.metadata.version(name)
except importlib.metadata.PackageNotFoundError:
continue
except Exception: # pylint: disable=broad-except
continue
try:
plat = platform.platform()
python_version = platform.python_version()
except Exception: # pylint: disable=broad-except
plat = python_version = None
return {
"platform": plat,
"python_version": python_version,
"packages": versions,
}
[docs]
def collect_resource_snapshot():
"""Best-effort machine-memory snapshot (bytes), for diagnosing OOM.
System memory pressure is the tell for an OOM / macOS-jetsam kill: a
``SIGKILL`` with no native traceback plus a nearly-full machine points
at memory, not a crash. The dead worker's own peak RSS is gone by the
time the parent looks, but the machine's RAM ceiling and the parent's
RSS are strong signal. Cross-OS via ``psutil`` (a hard dependency).
Never raises -- a failure yields a partial or empty dict rather than
turning the recording of one error into a second error.
Returns:
dict: Any of ``ram_total`` / ``ram_available`` (bytes),
``ram_percent_used`` (percent), ``process_rss`` (bytes).
"""
snapshot = {}
try:
import psutil # pylint: disable=import-outside-toplevel
except Exception: # pylint: disable=broad-except
return snapshot
try:
virtual_memory = psutil.virtual_memory()
snapshot["ram_total"] = int(virtual_memory.total)
snapshot["ram_available"] = int(virtual_memory.available)
snapshot["ram_percent_used"] = float(virtual_memory.percent)
except Exception: # pylint: disable=broad-except
pass
try:
snapshot["process_rss"] = int(psutil.Process().memory_info().rss)
except Exception: # pylint: disable=broad-except
pass
return snapshot
[docs]
class Component(str, Enum):
"""Which broad part of AutoWISP an error belongs to."""
STEP = "step"
PIPELINE = "pipeline"
BUI = "bui"
[docs]
class FileKind(str, Enum):
"""The sort of file a :class:`RelatedFile` points at."""
RAW_IMAGE = "raw_image"
CALIBRATED_IMAGE = "calibrated_image"
MASTER_BIAS = "master_bias"
MASTER_DARK = "master_dark"
MASTER_FLAT = "master_flat"
MASTER_PHOTREF = "master_photref"
DR_FILE = "dr_file"
LIGHTCURVE = "lightcurve"
CATALOG = "catalog"
CONFIG = "config"
OUTPUT = "output"
OTHER = "other"
[docs]
def _rebuild_autowisp_error(cls, args, state):
"""Reconstruct an :class:`AutoWISPError` subclass for unpickling.
Bypasses ``__init__`` (so keyword-only arguments on subclasses do not
block unpickling) and restores both ``BaseException.args`` (where the
message lives, in C-level storage outside ``__dict__``) and the
instance ``__dict__`` directly. See :meth:`AutoWISPError.__reduce__`.
Args:
cls(type): The concrete exception class to rebuild.
args(tuple): The original ``self.args`` (the message).
state(dict): The instance ``__dict__`` captured at pickle
time.
Returns:
AutoWISPError: The reconstructed exception.
"""
obj = cls.__new__(cls)
obj.args = args
obj.__dict__.update(state)
return obj
[docs]
class AutoWISPError(Exception):
"""Base class for every AutoWISP-raised exception.
Every concrete subclass selects a :class:`Component`.
Attributes:
component(Component): Set on each subclass; verified by tests.
related_files(tuple): The :class:`RelatedFile` entries this
error is about.
pipeline_run(FrozenRow or None): Snapshot of the
``PipelineRun`` row (see :class:`FrozenRow`), set by the
pipeline driver when it wraps a step's exception, or by
``wisp-*`` entry points. ``None`` for runs with no DB row.
crashed(datetime or None): When the failure surfaced, filled
in by the top-level handler.
subprocess_id(int or None): PID of the multiprocessing worker
that raised, when the exception travelled out of a Pool;
``None`` for errors raised in the main process.
user_message(str): Short, free of jargon, suitable for the
BUI.
details(dict): Arbitrary key/value pairs giving extra context
about the failure (e.g. shape mismatches, expected/actual
values, parsed config). Useful to both users and developers.
"""
component: Component # set on each subclass; verified by tests
# The many keyword-only arguments are the whole point: this is the
# single context-carrying constructor for every AutoWISP error.
# pylint: disable=too-many-arguments
[docs]
def __init__(
self,
message: str,
*,
related_files: Sequence[RelatedFile] = (),
pipeline_run: Optional[FrozenRow] = None,
crashed: Optional[datetime] = None,
subprocess_id: Optional[int] = None,
user_message: Optional[str] = None,
details: Optional[dict] = None,
):
"""Store the context attributes (see class ``Attributes``)."""
super().__init__(message)
self.related_files = tuple(related_files)
self.pipeline_run = pipeline_run
self.crashed = crashed
self.subprocess_id = subprocess_id
self.user_message = user_message or message
self.details = dict(details or {})
[docs]
def __reduce__(self):
"""Pickle by restoring ``__dict__`` rather than re-running
``__init__``.
Subclasses (e.g. :class:`StepError`) accept keyword-only
arguments and carry context attributes that are not part of
``self.args``, so the default exception unpickler -- which calls
``cls(*self.args)`` -- would both drop those fields and (for
required kwargs) raise ``TypeError``. Reconstructing through
``__new__`` + ``__dict__`` keeps every field intact and lets the
exception travel back out of a multiprocessing worker faithfully.
Returns:
tuple: ``(callable, args)`` per the pickle protocol.
"""
return (
_rebuild_autowisp_error,
(type(self), self.args, self.__dict__.copy()),
)
[docs]
def stamp_subprocess(self) -> None:
"""Record the current PID as the raising sub-process.
Called by the worker before re-raising out of a multiprocessing
Pool. Idempotent: a value already set in a deeper worker wins.
Returns:
None
"""
if self.subprocess_id is None:
self.subprocess_id = os.getpid()
[docs]
def with_pipeline_run(
self, run: Optional[FrozenRow], *, crashed: Optional[datetime] = None
) -> "AutoWISPError":
"""Attach a :class:`FrozenRow` snapshot of the ``PipelineRun``.
Args:
run(FrozenRow or None): Row snapshot to attach. Built by
``snapshot_row`` (parent) or from config primitives
(worker), so host/started are already populated; nothing
is reconstructed here.
crashed(datetime or None): Failure time; defaults to now
if not already set on the exception.
Returns:
AutoWISPError: ``self``, so it can be used inline before
re-raising.
"""
self.pipeline_run = run
self.crashed = self.crashed or crashed or datetime.now(timezone.utc)
return self
[docs]
def to_detail_dict(self) -> dict:
"""Return the heavy, non-column fields for the error sidecar.
Complements the queryable columns of the ``Error`` row:
everything here is what does *not* live inline on the row -- the
full technical message, the complete related-file list (a superset
of the artifact FKs), the arbitrary ``details`` dict, and the
formatted traceback (``__cause__`` chain included). The result may
still contain numpy/`Path`/etc. inside ``details``; serialize it
with ``json.dump(..., default=sanitize_for_json)``.
Returns:
dict: The sidecar payload (see ``error_handling_plan.md``).
"""
return {
"schema_version": 1,
"message": str(self),
"related_files": [
{
"kind": related.kind.value,
"path": related.path.as_posix(),
"role": related.role,
}
for related in self.related_files
],
"details": self.details,
"traceback": "".join(
traceback_module.format_exception(
type(self), self, self.__traceback__
)
),
}
[docs]
class StepError(AutoWISPError):
"""Failure inside a processing step.
Attributes:
step_name(str or None): Name of the step that failed. May be
``None`` at raise time and filled in from the ambient
context by the capture layer.
"""
component = Component.STEP
def __init__(
self, message: str, *, step_name: Optional[str] = None, **kwargs
):
super().__init__(message, **kwargs)
self.step_name = step_name
[docs]
class PipelineError(AutoWISPError):
"""Failure in the orchestration layer (not "the algorithm")."""
component = Component.PIPELINE
[docs]
class BUIError(AutoWISPError):
"""Failure in the Django views/forms/templates of the BUI."""
component = Component.BUI
# --- Concrete step-level exceptions, one per pipeline stage. ----------
[docs]
class CalibrationError(StepError):
"""Failure in the calibrate step."""
[docs]
class OutsideImageError(CalibrationError):
"""Attempt to access image data outside the bounds of the image.
Raised only in the calibration path, so it specializes
:class:`CalibrationError` rather than the generic :class:`StepError`.
"""
[docs]
class StackToMasterError(StepError):
"""Failure stacking calibration frames into a master."""
[docs]
class FindStarsError(StepError):
"""Failure in the find_stars step."""
[docs]
class NoSourcesFoundError(FindStarsError):
"""Source extraction completed but found no sources in a frame.
Distinct from a generic :class:`FindStarsError` so the step can mark
just that frame as failed and carry on (a clouded or badly defocused
frame is not a reason to abort the run), while every *other*
find_stars failure still propagates.
"""
[docs]
class SolveAstrometryError(StepError):
"""Failure in the solve_astrometry step."""
[docs]
class FitStarShapeError(StepError):
"""Failure in the fit_star_shape step."""
[docs]
class MeasurePhotometryError(StepError):
"""Failure in the measure_aperture_photometry step."""
[docs]
class FitPSFMapError(StepError):
"""Failure in the fit_source_extracted_psf_map step."""
[docs]
class FitMagnitudesError(StepError):
"""Failure in the fit_magnitudes step."""
[docs]
class CreateLightCurvesError(StepError):
"""Failure in the create_lightcurves step."""
[docs]
class EPDError(StepError):
"""Failure in the EPD detrending step."""
[docs]
class TFAError(StepError):
"""Failure in the TFA detrending step."""
[docs]
class DetrendingStatError(StepError):
"""Failure computing detrending statistics."""
# --- Cross-cutting step "reason" exceptions. --------------------------
#
# These name *why* a step failed (a bad image, incompatible images, a
# non-converging iteration) rather than *which* step. They are raised
# from shared low-level modules (fits_utilities, image_utilities,
# iterative_rejection_util, ...) reached by many steps, so they specialize
# the generic StepError; the step identity is supplied separately by the
# ``step_name`` the capture layer stamps from the ambient context.
[docs]
class ImageMismatchError(StepError):
"""Attempt to combine incompatible images in some way."""
[docs]
class BadImageError(StepError):
"""An image does not look like it is expected to."""
[docs]
class ConvergenceError(StepError):
"""Some iterative procedure failed to converge."""
[docs]
class WorkerCrashedError(StepError):
"""A multiprocessing worker died without preserving its exception.
Re-raise wrapper used when a worker dies in a way that does not
propagate the original exception (segfault, OOM-killer, ``os._exit``).
A :class:`StepError` (component ``step``): although the *parent*
synthesises it -- the worker cannot describe its own death -- the
failure is in the algorithm running inside a step, and the error
belongs to that step. It is a single generic class rather than one of
the per-stage subclasses because the parent has only the ambient step
*name*, not the failing step's exception type. That ``step_name`` --
stamped from the ambient context like any other ``StepError`` -- is
what lets crash-report log-collection resolve the run/step whose logs
to gather (see :func:`autowisp.error_context._worker_crashed`).
"""
[docs]
class CatalogError(StepError):
"""A problem with the reference catalog (query or coverage/consistency).
A cross-cutting :class:`StepError` -- catalog trouble is not specific to
one stage: a live Gaia query can exhaust its retries, and a cached
catalog can fail to cover the frames or mismatch the required epoch /
magnitude range / field of view during solve_astrometry, find_stars,
fit_star_shape, etc. Raised as one catchable type across all of them
(``step_name`` is stamped from the ambient context), so callers can
``except CatalogError`` regardless of which step triggered it.
"""
# --- Pipeline-level exceptions. ---------------------------------------
[docs]
class ConfigurationError(PipelineError):
"""Invalid or inconsistent pipeline configuration."""
[docs]
class DatabaseError(PipelineError):
"""Failure interacting with the pipeline database."""
[docs]
class HDF5LayoutError(PipelineError):
"""Error caused by invalid specification of HDF5 layout."""
[docs]
class ResourceError(PipelineError):
"""Insufficient disk / memory / CPU to continue."""
[docs]
class MasterSelectionError(PipelineError):
"""Failure selecting the master frame(s) for an image."""
[docs]
class PhotrefBindingError(PipelineError):
"""Failure binding a photometric reference."""
[docs]
class DependencyResolutionError(PipelineError):
"""Failure resolving processing-step dependencies."""
# --- BUI-level exceptions. --------------------------------------------
[docs]
class ViewError(BUIError):
"""Failure rendering or handling a BUI view."""
[docs]
class ProjectStateError(BUIError):
"""The BUI project is in a state that blocks the requested action."""