Replace interface towards eclrun and flow using fmstep_config

src/ert/config/ert_config.py

@@ -439,6 +439,17 @@ def apply_config_content_defaults(content_dict: dict, config_dir: str):
                 path.join(config_dir, content_dict[ConfigKeys.RUNPATH_FILE])
             )
 
+    @staticmethod
+    def _uppercase_subkeys_and_stringify_subvalues(
+        nested_dict: Dict[str, Dict[str, Any]],
+    ) -> Dict[str, Dict[str, str]]:
+        fixed_dict: dict[str, dict[str, str]] = {}
+        for key, value in nested_dict.items():
+            fixed_dict[key] = {
+                subkey.upper(): str(subvalue) for subkey, subvalue in value.items()
+            }
+        return fixed_dict
+
     @classmethod
     def read_site_config(cls) -> ConfigDict:
         site_config_file = site_config_location()
@@ -748,6 +759,9 @@ def handle_default(fm_step: ForwardModelStep, arg: str) -> str:
 
         job_list_errors = []
         job_list: List[ForwardModelStepJSON] = []
+        env_pr_fm_step = cls._uppercase_subkeys_and_stringify_subvalues(
+            ErtPluginManager().get_forward_model_configuration()
+        )
         for idx, fm_step in enumerate(forward_model_steps):
             substituter = Substituter(fm_step)
             fm_step_json = {
@@ -771,7 +785,9 @@ def handle_default(fm_step: ForwardModelStep, arg: str) -> str:
                     handle_default(fm_step, substituter.substitute(arg))
                     for arg in fm_step.arglist
                 ],
-                "environment": substituter.filter_env_dict(fm_step.environment),
+                "environment": substituter.filter_env_dict(
+                    dict(env_pr_fm_step.get(fm_step.name, {}), **fm_step.environment)
+                ),
                 "exec_env": substituter.filter_env_dict(fm_step.exec_env),
                 "max_running_minutes": fm_step.max_running_minutes,
             }

src/ert/plugins/__init__.py

@@ -27,6 +27,7 @@ def inner(*args: P.args, **kwargs: P.kwargs) -> Any:
                     "installable_workflow_jobs",
                     "help_links",
                     "installable_forward_model_steps",
+                    "forward_model_configuration",
                     "ecl100_config_path",
                     "ecl300_config_path",
                     "flow_config_path",

src/ert/plugins/hook_implementations/forward_model_steps.py

@@ -1,12 +1,9 @@
 import os
-import shutil
 import subprocess
 from pathlib import Path
 from textwrap import dedent
 from typing import List, Literal, Optional, Type
 
-import yaml
-
 from ert import (
     ForwardModelStepDocumentation,
     ForwardModelStepJSON,
@@ -207,12 +204,12 @@ def __init__(self) -> None:
                 str(
                     (
                         Path(__file__)
-                        / "../../../resources/forward_models/res/script/ecl100.py"
+                        / "../../../resources/forward_models/run_reservoirsimulator.py"
                     ).resolve()
                 ),
-                "<ECLBASE>",
-                "-v",
+                "eclipse",
                 "<VERSION>",
+                "<ECLBASE>",
                 "-n",
                 "<NUM_CPU>",
                 "<OPTS>",
@@ -265,12 +262,12 @@ def __init__(self) -> None:
                 str(
                     (
                         Path(__file__)
-                        / "../../../resources/forward_models/res/script/ecl300.py"
+                        / "../../../resources/forward_models/run_reservoirsimulator.py"
                     ).resolve()
                 ),
-                "<ECLBASE>",
-                "-v",
+                "e300",
                 "<VERSION>",
+                "<ECLBASE>",
                 "-n",
                 "<NUM_CPU>",
                 "<OPTS>",
@@ -317,11 +314,11 @@ def __init__(self) -> None:
                 str(
                     (
                         Path(__file__)
-                        / "../../../resources/forward_models/res/script/flow.py"
+                        / "../../../resources/forward_models/run_reservoirsimulator.py"
                     ).resolve()
                 ),
+                "flow",
                 "<ECLBASE>",
-                "-v",
                 "<VERSION>",
                 "-n",
                 "<NUM_CPU>",
@@ -629,31 +626,23 @@ def installable_forward_model_steps() -> List[Type[ForwardModelStepPlugin]]:
 
 
 def _available_eclrun_versions(simulator: Literal["eclipse", "e300"]) -> List[str]:
-    if shutil.which("eclrun") is None:
-        return []
     pm = ErtPluginManager()
-    ecl_config_path = (
-        pm.get_ecl100_config_path()
-        if simulator == "eclipse"
-        else pm.get_ecl300_config_path()
+    eclrun_env = os.environ.copy()
+    _simulator_to_fm_stepname = {"eclipse": "ECLIPSE100", "e300": "ECLIPSE300"}
+    eclrun_path = (
+        pm.get_forward_model_configuration()
+        .get(_simulator_to_fm_stepname[simulator], {})
+        .get("ECLRUN_PATH", "")
+    )
+    eclrun_env["PATH"] = os.pathsep.join(
+        [eclrun_path, *os.getenv("PATH", "").split(os.pathsep)]
     )
-
-    if not ecl_config_path:
-        return []
-    eclrun_env = {"PATH": os.getenv("PATH", "")}
-
-    with open(ecl_config_path, encoding="utf-8") as f:
-        try:
-            config = yaml.safe_load(f)
-        except yaml.YAMLError as e:
-            raise ValueError(f"Failed parse: {ecl_config_path} as yaml") from e
-    ecl_install_path = config.get("eclrun_env", {}).get("PATH", "")
-    eclrun_env["PATH"] = eclrun_env["PATH"] + os.pathsep + ecl_install_path
-
     try:
         return (
             subprocess.check_output(
-                ["eclrun", "--report-versions", simulator],
+                # It is not sufficient to just give the correct path to eclrun, its PATH must include its location
+                # for versions to be displayed. Ensure this is covered in tests.
+                ["eclrun", simulator, "--report-versions"],
                 env=eclrun_env,
             )
             .decode("utf-8")

src/ert/plugins/hook_specifications/__init__.py

@@ -4,6 +4,7 @@
     flow_config_path,
 )
 from .forward_model_steps import (
+    forward_model_configuration,
     installable_forward_model_steps,
 )
 from .help_resources import help_links
@@ -21,6 +22,7 @@
     "ecl100_config_path",
     "ecl300_config_path",
     "flow_config_path",
+    "forward_model_configuration",
     "help_links",
     "installable_forward_model_steps",
     "installable_jobs",

src/ert/plugins/hook_specifications/forward_model_steps.py

@@ -18,3 +18,11 @@ def installable_forward_model_steps() -> (
     :return: List of forward model step plugins in the form of subclasses of the
         ForwardModelStepPlugin class
     """
+
+
+@no_type_check
+@hook_specification
+def forward_model_configuration() -> PluginResponse[List[Type[ForwardModelStepPlugin]]]:
+    """
+    :return: List of configurations to be merged to be provided to forward model steps.
+    """

src/ert/plugins/plugin_manager.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import collections
 import logging
 import os
 import shutil
@@ -139,6 +140,50 @@ def get_flow_config_path(self) -> Optional[str]:
             hook=self.hook.flow_config_path, config_name="flow"
         )
 
+    def get_forward_model_configuration(self) -> Dict[str, Dict[str, Any]]:
+        response: List[PluginResponse[Dict[str, str]]] = (
+            self.hook.forward_model_configuration()
+        )
+        if response == []:
+            return {}
+
+        fm_configs: Dict[str, Dict[str, Any]] = collections.defaultdict(dict)
+        for res in response:
+            if not isinstance(res.data, dict):
+                raise TypeError(
+                    f"{res.plugin_metadata.plugin_name} did not provide "
+                    "dict[str, dict]"
+                )
+
+            for fmstep_name, fmstep_config in res.data.items():
+                if not isinstance(fmstep_name, str):
+                    raise TypeError(
+                        f"{res.plugin_metadata.plugin_name} did not "
+                        "provide dict[str, dict[str, Any]]"
+                    )
+                if not isinstance(fmstep_config, dict):
+                    raise TypeError(
+                        f"{res.plugin_metadata.plugin_name} did not "
+                        "provide dict[str, dict[str, Any]]"
+                    )
+                for key, value in fmstep_config.items():
+                    if not isinstance(key, str):
+                        raise TypeError(
+                            f"{res.plugin_metadata.plugin_name} did not "
+                            "provide dict[str, dict[str, Any]]"
+                        )
+                    if key.lower() in [
+                        existing.lower() for existing in fm_configs[fmstep_name]
+                    ]:
+                        raise RuntimeError(
+                            "Duplicate configuration or fm_step "
+                            f"{fmstep_name} for key {key} when parsing plugin "
+                            f"{res.plugin_metadata.plugin_name}, it is already "
+                            "registered by another plugin."
+                        )
+                    fm_configs[fmstep_name][key] = value
+        return fm_configs
+
     def _site_config_lines(self) -> List[str]:
         try:
             plugin_responses = self.hook.site_config_lines()