Unify ahm and pred ert subprocess call

src/flownet/ahm/_assisted_history_matching.py

@@ -6,14 +6,13 @@
 import pathlib
 import re
 import shutil
-import subprocess
 from typing import List, Dict, Optional, Tuple
 
 import jinja2
 import numpy as np
 import pandas as pd
 
-from ..ert import create_ert_setup
+from ..ert import create_ert_setup, run_ert_subprocess
 from ..realization import Schedule
 from ..network_model import NetworkModel
 from ..parameters import Parameter
@@ -120,37 +119,11 @@ def run_ert(self, weights: List[float]):
                 )
             )
 
-        # Ignore deprecation warnings (ERT as of August 2019 has a lot of them
-        # due to transition to Python3)
-
-        # Should revert here to use the much simpler subprocess.run when
-        # https://github.com/equinor/libres/issues/984 is closed. See
-        # https://github.com/equinor/flownet/pull/119 on changes to revert.
-        with subprocess.Popen(
-            "export PYTHONWARNINGS=ignore::DeprecationWarning;"
+        run_ert_subprocess(
             f"ert es_mda --weights {','.join(map(str, weights))!r} ahm_config.ert",
             cwd=self.output_folder,
-            shell=True,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            universal_newlines=True,
-        ) as process:
-            for line in process.stdout:  # type: ignore
-                print(line, end="")
-                if (
-                    "active realisations left, which is less than "
-                    "the minimum specified - stopping assimilation." in line
-                    or "All realizations failed!" in line
-                ):
-                    process.terminate()
-                    error_files = glob.glob(
-                        str(
-                            self.output_folder
-                            / self._ert_config["runpath"].replace("%d", "*")
-                            / "ERROR"
-                        )
-                    )
-                    raise RuntimeError(pathlib.Path(error_files[0]).read_text())
+            runpath=self._ert_config["runpath"],
+        )
 
     def report(self):
         """

src/flownet/ert/__init__.py

@@ -1,2 +1,3 @@
 from ._create_synthetic_refcase import create_synthetic_refcase
 from ._create_ert_setup import create_ert_setup
+from ._run_subprocess import run_ert_subprocess

src/flownet/ert/_run_subprocess.py

@@ -0,0 +1,37 @@
+import glob
+import pathlib
+import subprocess
+
+
+def run_ert_subprocess(command: str, cwd: pathlib.Path, runpath: str) -> None:
+    """
+    Helper function to run a ERT setup.
+    Args:
+        command: Command to run.
+        cwd: The folder to run the command from.
+        runpath: Runpath variable given to ERT.
+    Returns:
+        Nothing
+    """
+
+    # Should revert here to use the much simpler subprocess.run when
+    # https://github.com/equinor/libres/issues/984 is closed. See
+    # https://github.com/equinor/flownet/pull/119 on changes to revert.
+    with subprocess.Popen(
+        command,
+        cwd=cwd,
+        shell=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        universal_newlines=True,
+    ) as process:
+        for line in process.stdout:  # type: ignore
+            print(line, end="")
+            if (
+                "active realisations left, which is less than "
+                "the minimum specified - stopping assimilation." in line
+                or "All realizations failed!" in line
+            ):
+                process.terminate()
+                error_files = glob.glob(str(cwd / runpath.replace("%d", "*") / "ERROR"))
+                raise RuntimeError(pathlib.Path(error_files[0]).read_text())

src/flownet/prediction/_run_pred.py

@@ -1,37 +1,16 @@
 import pickle
-import subprocess
 import shutil
-import pathlib
 
 import jinja2
 
-from ..ert import create_ert_setup
+from ..ert import create_ert_setup, run_ert_subprocess
 
 _TEMPLATE_ENVIRONMENT = jinja2.Environment(
     loader=jinja2.PackageLoader("flownet", "templates"),
     undefined=jinja2.StrictUndefined,
 )
 
 
-def _run_ert(output_folder: pathlib.Path) -> None:
-    """
-    Helper function to run the ERT prediction setup.
-
-    Args:
-        output_folder: Path to the output folder.
-
-    Returns:
-        Nothing
-
-    """
-    subprocess.run(
-        "ert ensemble_experiment pred_config.ert",
-        cwd=output_folder,
-        shell=True,
-        check=True,
-    )
-
-
 def run_flownet_prediction(config, args):
     """
     Create prediction ERT setup, and runs it.
@@ -68,4 +47,8 @@ def run_flownet_prediction(config, args):
             )
         )
 
-    _run_ert(args.output_folder)
+    run_ert_subprocess(
+        "ert ensemble_experiment pred_config.ert",
+        cwd=args.output_folder,
+        runpath=config.ert.runpath,
+    )