mafw.runner

Provides a container to run configurable and modular analytical tasks.

Classes

MAFwApplication([steering_file, ...])

The MAFw Application.

Bases: object

The MAFw Application.

This class takes care of reading a steering file and from the information retrieved construct a ProcessorList from the processor listed there and execute it.

It is very practical because any combination of processors can be run without having to write dedicated scripts but simply modifying a steering file.

The application will search for processors not only among the ones available in the MAFw library, but also in all other packages exposing processors via the plugin mechanism.

All parameters in the constructor are optional.

An instance can be created also without the steering_file, but such an instance cannot be executed. The steering file can be provided in a later stage via the init() method or directly to the run() method.

The user interface can be either provided directly in the constructor, or it will be taken from the steering file. In the worst case, the fallback ConsoleInterface will be used.

The plugin manager, if not provided, the global plugin manager will be retrieved from the get_plugin_manager().

A simple example is provided here below:

Listing 28 Creation and execution of a MAFwApplication

import logging
from pathlib import Path

from mafw.runner import MAFwApplication

log = logging.getLogger(__name__)

# put here your steering file
steering_file = Path('path_to_my_steering_file.toml')

try:
    # create the app
    app = MAFwApplication(steering_file)

    # run it!
    app.run()

except Exception as e:
    log.error('An error occurred!')
    log.exception(e)

Constructor parameters:

Parameters:

steering_file (Path | str, Optional) – The path to the steering file.
user_interface (UserInterfaceBase | type[UserInterfaceBase] | str, Optional) – The user interface to be used by the application.
plugin_manager (PluginManager, Optional) – The plugin manager.

_expand_processors_to_run(items: list[str], config: dict[str, Any], plugins: LoadedPlugins, seen: set[str] | None = None) → ProcessorList[source]

Constructs a ProcessorList from a list of processor names or groups.

This method recursively expands a list of processor names or group definitions into a ProcessorList object, which can be executed. It handles three cases: direct processor class matches, group definitions in the configuration, and unknown entries.

Added in version v2.0.0.

Parameters:

items (list[str]) – A list of processor names or group identifiers to be expanded.
config (dict[str, Any]) – The configuration dictionary containing processor and group definitions.
plugins (LoadedPlugins) – The loaded plugins containing available processors and groups.
seen (set[str] | None) – A set of group names that have already been processed to detect cyclic definitions.

Returns:

A ProcessorList containing the expanded processors ready for execution.

Return type:

ProcessorList

Raises:

UnknownProcessorGroup – If a group is defined without processors or if a cyclic group definition is detected.
UnknownProcessor – If a processor name is not recognized in the plugin library.

get_user_interface(user_interface: str) → UserInterfaceBase[source]

Retrieves the user interface from the plugin managers.

User interfaces are exposed via the plugin manager.

If the requested user_interface is not available, then the fallback console interface is used.

Parameters:: user_interface (str) – The name of the user interface to be used. Normally rich or console.

init(steering_file: Path | str) → None[source]

Initializes the application.

This method is normally automatically invoked by the class constructor. It can be called in a later moment to force the parsing of the provided steering file.

Parameters:: steering_file (Path | str) – The path to the steering file.

run(steering_file: Path | str | None = None) → ProcessorExitStatus[source]

Runs the application.

This method builds the ProcessorList with the processors listed in the steering file and launches its execution.

A steering file can be provided at this stage if it was not done before.

Changed in version v2.0.0: Refactor to accept replica names in the processor to run list.

Parameters:

steering_file (Path | str, Optional) – The steering file. Defaults to None.

Raises:

RunnerNotInitialized – if the application has not been initialized. Very likely a steering file was never provided.
UnknownProcessor – if a processor listed in the steering file is not available in the plugin library.

exit_status: the exit status of the application

name: the name of the application instance

plugin_manager: the plugin manager of the application instance