Plugin system¶

Introduction¶

Assuming the following package structure:

your_package
├── pyproject.toml        # or setup.py
└── src
    └── your_module
        ├── __init__.py
        ├── your_plugins.py
        └── ...

where the ert plugins are defined in your_plugins.py, then discovery is done by registering your plugin via a setuptools entry point, with the namespace ert:

# setup.py
setup(
    ...
    entry_points={"ert": ["your_module_jobs = your_module.your_plugins"]},
    ...
)

# pyproject.toml
[project.entry-points.ert]
your_module_jobs = "your_module.your_plugins"

This entry point should point to the module where your ert plugin(s) exists. (Notice that the entry point expects a list, so you can register multiple modules).

Kinds of plugins¶

A plugin is created with the ert.plugin decorator and the function name describes what kind of plugin it is.

Forward models¶

To install forward model steps that you want to have available in ERT you can either use the simplified installable_jobs function name, or installable_forward_model_steps which gives a lot more control. The individual steps in a forward model were previously called “jobs” in Ert, and this is still partly present in the source code.

from typing import Optional

import ert


@ert.plugin(name="my_plugin")
def installable_jobs() -> dict[str, str]:
    return {
        "job_name": "/path/to/workflow.config"
    }


class MyForwardModel(ert.ForwardModelStepPlugin):
    def __init__(self):
        super().__init__(
            name="MY_FORWARD_MODEL",
            command=["my_executable", "<parameter1>", "<parameter2>"],
        )

    def validate_pre_realization_run(
        self, fm_step_json: ert.ForwardModelStepJSON
    ) -> ert.ForwardModelStepJSON:
        if fm_step_json["argList"][0] not in ["start", "stop"]:
            raise ert.ForwardModelStepValidationError(
                "First argument to MY_FORWARD_MODEL must be either start or stop"
            )
        return fm_step_json

    def validate_pre_experiment(self, fm_step_json: ert.ForwardModelStepJSON) -> None:
        pass

    @staticmethod
    def documentation() -> Optional[ert.ForwardModelStepDocumentation]:
        return ert.ForwardModelStepDocumentation(
            category="utility.templating",
            source_package="my_plugin",
            source_function_name="MyForwardModel",
            description="my plugin description",
        )


@ert.plugin(name="my_plugin")
def installable_forward_model_steps() -> list[ert.ForwardModelStepPlugin]:
    return [MyForwardModel]

Notice that by using installable_forward_model_steps, validation can be added where the methods validate_pre_experiment or validate_pre_realization_run can throw ForwardModelStepValidationError to indicate that the configuration of the forward model step is invalid (which ert then handles gracefully and presents nicely to the user). If you want to show a warning in cases where the configuration cannot be validated pre-experiment, you can use the ForwardModelStepWarning.warn(...) method.

To provide documentation for a forward model step given with installable_jobs, use the job_documentation name. If you are the plugin that provided the step with the name step_name, then you respond with the documentation as specified, else respond with None.

import ert

@ert.plugin(name="my_plugin")
def job_documentation(step_name: str):
    if step_name == "my_step":
         return {
             "description": "step description",
             "examples": "...",
             "category": "test.category.for.step",
         }

When creating documentation in ERT, forward model steps will be grouped by their main categories (ie. the category listed before the first dot).

Workflow job¶

There are two ways to install workflow jobs in ERT. Depending on whether you already have a configuration file or need to include additional documentation, you can choose between the installable_workflow_jobs hook or the legacy_ertscript_workflow hook.

Using the installable_workflow_jobs hook

The hook is specified as follows:

import ert

@ert.plugin(name="my_plugin")
def installable_workflow_jobs():
     return {
         "wf_job_name": "/path/to/workflow_job.config",
     }

The configuration file needed to use the installable_workflow_jobs hook must point to an executable and specify its arguments. The built-in internal CSV_EXPORT workflow job is shown as an example:

INTERNAL    True
SCRIPT      /path/to/csv_export.py
MIN_ARG     1
MAX_ARG     2
ARG_TYPE    0  STRING
ARG_TYPE    1  STRING

Implement the hook specification as follows to register the workflow job CSV_EXPORT:

import ert

@ert.plugin(name="ert")
def installable_workflow_jobs() -> Dict[str, str]:
     return {
         "CSV_EXPORT": "/path/to/csv_export"
     }

Using the legacy_ertscript_workflow hook

The second approach does not require creating a workflow job configuration file up-front, and allows adding documentation.

@no_type_check
@hook_specification
def legacy_ertscript_workflow(config: WorkflowConfigs) -> None:
    """
    This hook allows the user to register a workflow with the config object. A workflow
    must add the class inheriting from ErtScript and an optional name.

    :param config: A handle to the main workflow config.
    """

Minimal example:

import ert

class MyJob(ert.ErtScript):
    def run(self):
        print("Hello World")

@ert.plugin(name="my_plugin")
 def legacy_ertscript_workflow(config):
     config.add_workflow(MyJob, "MY_JOB")

Full example:

import ert

class MyJob(ert.ErtScript):
    def run(self):
        print("Hello World")

@ert.plugin(name="my_plugin")
 def legacy_ertscript_workflow(config: ert.WorkflowConfigs):
     workflow: ert.ErtScriptWorkflow = config.add_workflow(MyJob, "MY_JOB")
     workflow.parser = my_job_parser  # Optional
     workflow.description = "My job description"  # Optional
     workflow.examples = "example of use"  # Optional

The configuration object and properties are as follows.

ert.plugins.hook_specifications.jobs.legacy_ertscript_workflow(config: WorkflowConfigs) → None¶

This hook allows the user to register a workflow with the config object. A workflow must add the class inheriting from ErtScript and an optional name.

Parameters:: config – A handle to the main workflow config.

class ert.plugins.workflow_config.WorkflowConfigs¶

Top level workflow config object, holds all workflow configs.

__init__() → None¶

__new__(**kwargs)¶

add_workflow(ert_script: type[ErtScript], name: str = '', description: str = '', examples: str | None = None, parser: Callable[[], ArgumentParser] | None = None, category: str = 'other') → ErtScriptWorkflow¶

Parameters:

category – dot separated string
parser – will extract information to use in documentation
examples – must be valid rst, will be added to documentation
description – must be valid rst, defaults to __doc__
ert_script – class which inherits from ErtScript
name – Optional name for workflow (default is name of class)

Returns:

Instantiated workflow config.

class ert.plugins.workflow_config.ErtScriptWorkflow¶

Single workflow configuration object

__init__(name: str, min_args: int | None = None, max_args: int | None = None, arg_types: list[~ert.config.parsing.schema_item_type.SchemaItemType] = <factory>, stop_on_fail: bool | None = None, ert_script: type[~ert.plugins.ert_script.ErtScript] = None, description: str = '', examples: str | None = None, parser: ~collections.abc.Callable[[], ~argparse.ArgumentParser] | None = None, category: str = 'other') → None¶

category: str = 'other'¶

description: str = ''¶

ert_script: type[ErtScript] = None¶

examples: str | None = None¶

is_plugin() → bool¶

parser: Callable[[], ArgumentParser] | None = None¶

Logging configuration¶

The logging can be configured by plugins to add custom log handlers.

ert.plugins.hook_specifications.logging.add_log_handle_to_root() → Handler¶

Create a log handle which will be added to the root logger in the main entry point.

Returns:: A log handle that will be added to the root logger

Minimal example to log to a new file:

import ert

 @ert.plugin(name="my_plugin")
 def add_log_handle_to_root():
     import logging
     fh = logging.FileHandler('spam.log')
     formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
     fh.setFormatter(formatter)
     return fh