openavmkit.progress

Progress reporting and cooperative cancellation for long-running work.

Notebook users see OpenAVMKit's printed output. Other front ends (scripts, CI, desktop apps) need structured progress instead. Code that does long-running work calls :func:report at meaningful points; a caller that wants the events wraps the work in :func:reporting. With no listener installed, :func:report is a no-op, so notebooks behave exactly as before.

Cancellation is cooperative: a caller passes a cancel callable to :func:reporting, and every :func:report call checks it and raises :class:Cancelled if it returns True.

Skipping is the same, for one optional part of the work (an enrichment, say): a :func:report that names its part raises :class:Skipped when the caller's skip callable says that part is to be skipped. The code running the part catches it, leaves the part's results out, and carries on.

Cancelled

Bases: Exception

Raised by :func:report when the caller has requested cancellation.

ProgressEvent dataclass

ProgressEvent(stage, step, fraction, message, sub=None, part=None)

One progress update.

Attributes:

Name	Type	Description
stage	str or None	The pipeline stage being run (e.g. "assemble"), if known.
step	str	Short machine-friendly name of the current step (e.g. "load_dataframes").
fraction	float or None	Overall completion of the stage in [0, 1], if known.
message	str	Human-readable description of what is happening.
sub	float or None	Completion of the current step itself in [0, 1], if known.
part	str or None	The optional part this is in ("distances"), which can be skipped.

Skipped

Skipped(part)

Bases: BaseException

Raised by :func:report when the caller has asked to skip the part it's in.

A BaseException, like KeyboardInterrupt: the part's own except Exception handlers (a failed download, a bad batch) must not swallow it and carry on.

Source code in openavmkit/progress.py

def __init__(self, part: str):
    super().__init__(f"{part} skipped")
    self.part = part

check_cancelled

check_cancelled()

Raise :class:Cancelled if the active cancel check returns True.

Source code in openavmkit/progress.py

def check_cancelled() -> None:
    """Raise :class:`Cancelled` if the active cancel check returns True."""
    cancel = _cancel.get()
    if cancel is not None and cancel():
        raise Cancelled()

check_skipped

check_skipped(part)

Raise :class:Skipped if the active skip check says to skip part.

Source code in openavmkit/progress.py

def check_skipped(part: str) -> None:
    """Raise :class:`Skipped` if the active skip check says to skip ``part``."""
    skip = _skip.get()
    if skip is not None and skip(part):
        raise Skipped(part)

detail

detail(text, sub=None)

Say what the current step is doing now (which model group, say), without starting a new step.

Reports the last step again, its message followed by text, with sub as the step's own completion. Generic helpers that don't know which step they're in (do_per_model_group) call this. A no-op with no listener, or before any step has been reported.

Source code in openavmkit/progress.py

def detail(text: str, sub: Optional[float] = None) -> None:
    """Say what the current step is doing now (which model group, say), without starting a new step.

    Reports the last step again, its message followed by ``text``, with ``sub`` as the step's own
    completion. Generic helpers that don't know which step they're in (``do_per_model_group``) call
    this. A no-op with no listener, or before any step has been reported.
    """
    listener = _listener.get()
    current = _current.get()
    if listener is not None and current is not None:
        step, fraction, message = current
        listener(ProgressEvent(_stage.get(), step, fraction, f"{message}: {text}", sub))
    check_cancelled()

report

report(step, fraction=None, message='', sub=None, part=None)

Report progress, then raise :class:Cancelled if cancellation was requested, or :class:Skipped if the part it names is to be skipped.

Parameters:

Name	Type	Description	Default
step	str	Short machine-friendly name of the current step.	required
fraction	float	Overall completion of the current stage in [0, 1].	None
message	str	Human-readable description. Defaults to step.	''
sub	float	Completion of this step itself in [0, 1].	None
part	str	The optional part this is in: the caller may skip it here.	None

Source code in openavmkit/progress.py

def report(
    step: str,
    fraction: Optional[float] = None,
    message: str = "",
    sub: Optional[float] = None,
    part: Optional[str] = None,
) -> None:
    """Report progress, then raise :class:`Cancelled` if cancellation was requested, or
    :class:`Skipped` if the part it names is to be skipped.

    Parameters
    ----------
    step : str
        Short machine-friendly name of the current step.
    fraction : float, optional
        Overall completion of the current stage in ``[0, 1]``.
    message : str, optional
        Human-readable description. Defaults to ``step``.
    sub : float, optional
        Completion of this step itself in ``[0, 1]``.
    part : str, optional
        The optional part this is in: the caller may skip it here.
    """
    listener = _listener.get()
    if listener is not None:
        _current.set((step, fraction, message or step))
        listener(ProgressEvent(_stage.get(), step, fraction, message or step, sub, part))
    check_cancelled()
    if part is not None:
        check_skipped(part)

reporting

reporting(on_progress=None, cancel=None, stage=None, skip=None)

Install a progress listener and/or cancel check for the enclosed block.

Parameters:

Name	Type	Description	Default
on_progress	callable	Called with a :class:ProgressEvent for every :func:report.	None
cancel	callable	Returns True when the work should stop; checked at every :func:report.	None
stage	str	Stage name attached to events reported inside the block.	None
skip	callable	Given a part's name, returns True when that part should be skipped; checked at every :func:report that names a part.	None

Source code in openavmkit/progress.py

@contextmanager
def reporting(
    on_progress: Optional[ProgressCallback] = None,
    cancel: Optional[CancelCheck] = None,
    stage: Optional[str] = None,
    skip: Optional[SkipCheck] = None,
) -> Iterator[None]:
    """Install a progress listener and/or cancel check for the enclosed block.

    Parameters
    ----------
    on_progress : callable, optional
        Called with a :class:`ProgressEvent` for every :func:`report`.
    cancel : callable, optional
        Returns True when the work should stop; checked at every :func:`report`.
    stage : str, optional
        Stage name attached to events reported inside the block.
    skip : callable, optional
        Given a part's name, returns True when that part should be skipped;
        checked at every :func:`report` that names a part.
    """
    tokens = [
        _listener.set(on_progress),
        _cancel.set(cancel),
        _stage.set(stage),
    ]
    skip_token = _skip.set(skip)
    current_token = _current.set(None)
    try:
        yield
    finally:
        _current.reset(current_token)
        _skip.reset(skip_token)
        _stage.reset(tokens[2])
        _cancel.reset(tokens[1])
        _listener.reset(tokens[0])