Source code for coseda.centerrefinement.cancellation

"""Shared helpers for cooperative cancellation of long-running refinement jobs."""

from typing import Optional
import threading




class RefinementCancelled(Exception):
    """Raised when a user requests to stop an in-flight refinement."""





def check_cancel(stop_event: Optional[threading.Event], where: str = ""):
    """
    Raise RefinementCancelled if the provided stop_event is set.

    Parameters
    ----------
    stop_event : threading.Event or None
        Event used to request cancellation. If None, no check is performed.
    where : str
        Optional string describing the current phase, helpful for logging.
    """
    if stop_event is not None and stop_event.is_set():
        msg = "Refinement cancelled by user"
        if where:
            msg = f"{msg} during {where}"
        raise RefinementCancelled(msg)