Warning filter

amltk.scheduling.plugins.warning_filter

The WarningFilter if used to automatically filter out warnings from a Task as it runs.

This wraps your function in context manager warnings.catch_warnings() and applies your arguments to warnings.filterwarnings(), as you would normally filter warnings in Python.

Usage

import warnings
from amltk.scheduling import Scheduler
from amltk.scheduling.plugins import WarningFilter

def f() -> None:
    warnings.warn("This is a warning")

scheduler = Scheduler.with_processes(1)
task = scheduler.task(f, plugins=WarningFilter("ignore"))

╭─ Task f() -> None ───────────────────────────────────────────────────────────╮
│ ╭───────────────────────── Plugin warning-filter ──────────────────────────╮ │
│ │  Args         Kwargs                                                     │ │
│ │  ('ignore',)  {}                                                         │ │
│ ╰──────────────────────────────────────────────────────────────────────────╯ │
╰──────────────────────────── Ref: Task-f-ycUB67MH ────────────────────────────╯

WarningFilter

WarningFilter(*args: Any, **kwargs: Any)

Bases: Plugin

A plugin that disables warnings emitted from tasks.

PARAMETER	DESCRIPTION
*args	arguments to pass to warnings.filterwarnings. TYPE: Any DEFAULT: ()
**kwargs	keyword arguments to pass to warnings.filterwarnings. TYPE: Any DEFAULT: {}

Source code in src/amltk/scheduling/plugins/warning_filter.py

def __init__(self, *args: Any, **kwargs: Any):
    """Initialize the plugin.

    Args:
        *args: arguments to pass to
            [`warnings.filterwarnings`][warnings.filterwarnings].
        **kwargs: keyword arguments to pass to
            [`warnings.filterwarnings`][warnings.filterwarnings].
    """
    super().__init__()
    self.task: Task | None = None
    self.warning_args = args
    self.warning_kwargs = kwargs

name class-attribute instance-attribute

name: ClassVar = 'warning-filter'

The name of the plugin.

attach_task

attach_task(task: Task) -> None

Attach the plugin to a task.

Source code in src/amltk/scheduling/plugins/warning_filter.py

@override
def attach_task(self, task: Task) -> None:
    """Attach the plugin to a task."""
    self.task = task

events

events() -> list[Event]

Return a list of events that this plugin emits.

Likely no need to override this method, as it will automatically return all events defined on the plugin.

Source code in src/amltk/scheduling/plugins/plugin.py

def events(self) -> list[Event]:
    """Return a list of events that this plugin emits.

    Likely no need to override this method, as it will automatically
    return all events defined on the plugin.
    """
    inherited_attrs = chain.from_iterable(
        vars(cls).values() for cls in self.__class__.__mro__
    )
    return [attr for attr in inherited_attrs if isinstance(attr, Event)]

pre_submit

pre_submit(
    fn: Callable[P, R], *args: args, **kwargs: kwargs
) -> tuple[Callable[P, R], tuple, dict]

Pre-submit hook.

Wraps the function to ignore warnings.

Source code in src/amltk/scheduling/plugins/warning_filter.py

@override
def pre_submit(
    self,
    fn: Callable[P, R],
    *args: P.args,
    **kwargs: P.kwargs,
) -> tuple[Callable[P, R], tuple, dict]:
    """Pre-submit hook.

    Wraps the function to ignore warnings.
    """
    wrapped_f = _IgnoreWarningWrapper(fn, *self.warning_args, **self.warning_kwargs)
    return wrapped_f, args, kwargs