Replace interface towards eclrun and flow using fmstep_config

.github/workflows/build_and_test.yml

@@ -69,6 +69,18 @@ jobs:
       python-version: ${{ matrix.python-version }}
     secrets: inherit
 
+  test-ert-with-flow:
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ ubuntu-latest ]
+        python-version: [ '3.11', '3.12' ]
+    uses: ./.github/workflows/test_ert_with_flow.yml
+    with:
+      os: ${{ matrix.os }}
+      python-version: ${{ matrix.python-version }}
+    secrets: inherit
+
   test-mac-ert:
     if: github.ref_type != 'tag' # when not tag
     strategy:

.github/workflows/test_ert_with_flow.yml

@@ -0,0 +1,64 @@
+on:
+  workflow_call:
+    inputs:
+      os:
+        type: string
+      python-version:
+        type: string
+
+env:
+  UV_SYSTEM_PYTHON: 1
+
+jobs:
+  test-ert-with-flow:
+    name: Run ert tests
+    timeout-minutes: 20
+    runs-on: ${{ inputs.os }}
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - uses: actions/setup-python@v5
+      id: setup_python
+      with:
+        python-version: ${{ inputs.python-version }}
+
+    - name: Install uv
+      uses: astral-sh/setup-uv@v4
+
+    - name: Install ert and everest
+      run: |
+        uv pip install ".[everest,dev]"
+        uv pip install git+https://github.com/equinor/everest-models
+
+    - name: Install flow
+      run: |
+        set -e
+        sudo apt install software-properties-common
+        sudo apt-add-repository ppa:opm/ppa
+        sudo apt update
+        sudo apt install mpi-default-bin
+        sudo apt install libopm-simulators-bin python3-opm-common
+
+        which flow
+        flow --version
+
+    - name: Run integration tests towards OPM flow without flowrun
+      run: |
+        set -e
+        pytest tests/ert/unit_tests/resources/test_run_flow_simulator.py
+
+    - name: Run Ert on an example configuration with flow
+      run: |
+        pushd test-data/ert/flow_example
+        perl -p -i -e 's/NUM_REALIZATIONS\s*12/NUM_REALIZATIONS 2/g' flow.ert
+        ert ensemble_experiment flow.ert
+        popd
+
+    - name: Run Everest on an example configuration with flow
+      run: |
+        set +e
+        pushd test-data/everest/egg/everest/model
+        everest lint config_flow.yml
+        everest run config_flow.yml
+        popd

src/ert/config/ert_config.py

@@ -191,7 +191,15 @@ def handle_default(fm_step: ForwardModelStep, arg: str) -> str:
                 for arg in fm_step.arglist
             ],
             "environment": substituter.filter_env_dict(
-                dict(env_pr_fm_step.get(fm_step.name, {}), **fm_step.environment)
+                dict(
+                    **{
+                        key: value
+                        for key, value in env_pr_fm_step.get(fm_step.name, {}).items()
+                        # Plugin settings can not override anything:
+                        if key not in env_vars and key not in fm_step.environment
+                    },
+                    **fm_step.environment,
+                )
             ),
             "exec_env": substituter.filter_env_dict(fm_step.exec_env),
             "max_running_minutes": fm_step.max_running_minutes,
@@ -734,10 +742,15 @@ def _create_list_of_forward_model_steps_to_run(
         cls,
         installed_steps: dict[str, ForwardModelStep],
         substitutions: Substitutions,
-        config_dict,
+        config_dict: dict,
     ) -> list[ForwardModelStep]:
         errors = []
         fm_steps = []
+
+        env_vars = {}
+        for key, val in config_dict.get("SETENV", []):
+            env_vars[key] = val
+
         for fm_step_description in config_dict.get(ConfigKeys.FORWARD_MODEL, []):
             if len(fm_step_description) > 1:
                 unsubstituted_step_name, args = fm_step_description
@@ -791,9 +804,14 @@ def _create_list_of_forward_model_steps_to_run(
                         context=substitutions,
                         forward_model_steps=[fm_step],
                         skip_pre_experiment_validation=True,
+                        env_vars=env_vars,
                     )
-                    job_json = substituted_json["jobList"][0]
-                    fm_step.validate_pre_experiment(job_json)
+                    fm_json_for_validation = dict(substituted_json["jobList"][0])
+                    fm_json_for_validation["environment"] = {
+                        **substituted_json["global_environment"],
+                        **fm_json_for_validation["environment"],
+                    }
+                    fm_step.validate_pre_experiment(fm_json_for_validation)
                 except ForwardModelStepValidationError as err:
                     errors.append(
                         ConfigValidationError.with_context(

src/ert/plugins/hook_implementations/forward_model_steps.py

@@ -1,20 +1,16 @@
-import os
 import shutil
 import subprocess
 from pathlib import Path
 from textwrap import dedent
 from typing import Literal
 
-import yaml
-
 from ert import (
     ForwardModelStepDocumentation,
     ForwardModelStepJSON,
     ForwardModelStepPlugin,
     ForwardModelStepValidationError,
     plugin,
 )
-from ert.plugins import ErtPluginManager
 
 
 class CarefulCopyFile(ForwardModelStepPlugin):
@@ -207,11 +203,12 @@ def __init__(self) -> None:
                 str(
                     (
                         Path(__file__)
-                        / "../../../resources/forward_models/res/script/ecl100.py"
+                        / "../../../resources/forward_models/run_reservoirsimulator.py"
                     ).resolve()
                 ),
+                "eclipse",
                 "<ECLBASE>",
-                "-v",
+                "--version",
                 "<VERSION>",
                 "-n",
                 "<NUM_CPU>",
@@ -220,18 +217,20 @@ def __init__(self) -> None:
             default_mapping={"<NUM_CPU>": 1, "<OPTS>": ""},
         )
 
-    def validate_pre_experiment(self, _: ForwardModelStepJSON) -> None:
+    def validate_pre_experiment(self, fm_json: ForwardModelStepJSON) -> None:
         if "<VERSION>" not in self.private_args:
             raise ForwardModelStepValidationError(
                 "Forward model step ECLIPSE100 must be given a VERSION argument"
             )
         version = self.private_args["<VERSION>"]
-        available_versions = _available_eclrun_versions(simulator="eclipse")
+        available_versions = _available_eclrun_versions(
+            simulator="eclipse", env_vars=fm_json["environment"]
+        )
 
         if available_versions and version not in available_versions:
             raise ForwardModelStepValidationError(
-                f"Unavailable ECLIPSE100 version {version} current supported "
-                f"versions {available_versions}"
+                f"Unavailable ECLIPSE100 version {version}. "
+                f"Available versions: {available_versions}"
             )
 
     @staticmethod
@@ -265,11 +264,12 @@ def __init__(self) -> None:
                 str(
                     (
                         Path(__file__)
-                        / "../../../resources/forward_models/res/script/ecl300.py"
+                        / "../../../resources/forward_models/run_reservoirsimulator.py"
                     ).resolve()
                 ),
+                "e300",
                 "<ECLBASE>",
-                "-v",
+                "--version",
                 "<VERSION>",
                 "-n",
                 "<NUM_CPU>",
@@ -278,17 +278,22 @@ def __init__(self) -> None:
             default_mapping={"<NUM_CPU>": 1, "<OPTS>": "", "<VERSION>": "version"},
         )
 
-    def validate_pre_experiment(self, _: ForwardModelStepJSON) -> None:
+    def validate_pre_experiment(
+        self,
+        fm_step_json: ForwardModelStepJSON,
+    ) -> None:
         if "<VERSION>" not in self.private_args:
             raise ForwardModelStepValidationError(
                 "Forward model step ECLIPSE300 must be given a VERSION argument"
             )
         version = self.private_args["<VERSION>"]
-        available_versions = _available_eclrun_versions(simulator="e300")
+        available_versions = _available_eclrun_versions(
+            simulator="e300", env_vars=fm_step_json["environment"]
+        )
         if available_versions and version not in available_versions:
             raise ForwardModelStepValidationError(
-                f"Unavailable ECLIPSE300 version {version} current supported "
-                f"versions {available_versions}"
+                f"Unavailable ECLIPSE300 version {version}. "
+                f"Available versions: {available_versions}"
             )
 
     @staticmethod
@@ -317,11 +322,12 @@ 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",
                 "<VERSION>",
                 "-n",
                 "<NUM_CPU>",
@@ -628,33 +634,22 @@ def installable_forward_model_steps() -> list[type[ForwardModelStepPlugin]]:
     return [*_UpperCaseFMSteps, *_LowerCaseFMSteps]
 
 
-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()
-    )
-
-    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
-
+def _available_eclrun_versions(
+    simulator: Literal["eclipse", "e300"], env_vars: dict[str, str]
+) -> list[str]:
+    eclrun_path = env_vars.get("ECLRUN_PATH", "")
     try:
+        eclrun_abspath = shutil.which(Path(eclrun_path) / "eclrun")
+        if eclrun_abspath is None:
+            return []
         return (
             subprocess.check_output(
-                ["eclrun", "--report-versions", simulator],
-                env=eclrun_env,
+                [
+                    eclrun_abspath,
+                    simulator,
+                    "--report-versions",
+                ],
+                env=env_vars,
             )
             .decode("utf-8")
             .strip()