Make updating work with failed realizations

src/ert/config/gen_kw_config.py

@@ -26,6 +26,8 @@
 
     from ert.storage import EnsembleReader
 
+_logger = logging.getLogger(__name__)
+
 
 class PriorDict(TypedDict):
     key: str
@@ -181,7 +183,7 @@ def sample_or_load(
         if self.forward_init_file:
             return self.read_from_runpath(Path(), real_nr)
 
-        logging.info(f"Sampling parameter {self.name} for realization {real_nr}")
+        _logger.info(f"Sampling parameter {self.name} for realization {real_nr}")
         keys = [e.name for e in self.transfer_functions]
         parameter_value = self._sample_value(
             self.name,

src/ert/run_models/ensemble_smoother.py

@@ -117,8 +117,11 @@ def run_experiment(
                 prior_ensemble=prior,
             ),
             runpaths=self.run_paths,
-            initial_mask=prior_context.sim_fs.get_realization_mask_with_parameters()
-            + prior_context.sim_fs.get_realization_mask_with_responses(),
+            initial_mask=(
+                prior_context.sim_fs.get_realization_mask_with_parameters()
+                * prior_context.sim_fs.get_realization_mask_with_responses()
+                * prior_context.sim_fs.get_realization_mask_without_failure()
+            ),
             iteration=1,
         )
         try:

src/ert/run_models/iterated_ensemble_smoother.py

@@ -180,8 +180,11 @@ def run_experiment(
             posterior_context = RunContext(
                 sim_fs=posterior,
                 runpaths=self.run_paths,
-                initial_mask=prior_context.sim_fs.get_realization_mask_with_parameters()
-                + prior_context.sim_fs.get_realization_mask_with_responses(),
+                initial_mask=(
+                    prior_context.sim_fs.get_realization_mask_with_parameters()
+                    * prior_context.sim_fs.get_realization_mask_with_responses()
+                    * prior_context.sim_fs.get_realization_mask_without_failure()
+                ),
                 iteration=current_iter,
             )
             update_success = False

src/ert/run_models/multiple_data_assimilation.py

@@ -161,8 +161,11 @@ def run_experiment(
                     prior_ensemble=prior_context.sim_fs,
                 ),
                 runpaths=self.run_paths,
-                initial_mask=prior_context.sim_fs.get_realization_mask_with_parameters()
-                + prior_context.sim_fs.get_realization_mask_with_responses(),
+                initial_mask=(
+                    prior_context.sim_fs.get_realization_mask_with_parameters()
+                    * prior_context.sim_fs.get_realization_mask_with_responses()
+                    * prior_context.sim_fs.get_realization_mask_without_failure()
+                ),
                 iteration=iteration + 1,
             )
             smoother_snapshot = self.update(

src/ert/storage/local_ensemble.py

@@ -109,6 +109,18 @@ def get_realization_mask_without_parent_failure(self) -> npt.NDArray[np.bool_]:
             ]
         )
 
+    def get_realization_mask_without_failure(self) -> npt.NDArray[np.bool_]:
+        return np.array(
+            [
+                e
+                not in [
+                    RealizationStorageState.PARENT_FAILURE,
+                    RealizationStorageState.LOAD_FAILURE,
+                ]
+                for e in self.get_ensemble_state()
+            ]
+        )
+
     def get_realization_mask_with_parameters(self) -> npt.NDArray[np.bool_]:
         return np.array(
             [

tests/integration_tests/analysis/test_es_update.py

@@ -1,5 +1,8 @@
+import os
+import stat
 from argparse import ArgumentParser
 from pathlib import Path
+from textwrap import dedent
 
 import numpy as np
 import pytest
@@ -23,6 +26,7 @@
 from ert.config import AnalysisConfig, ErtConfig, GenDataConfig, GenKwConfig
 from ert.config.analysis_module import ESSettings
 from ert.storage import open_storage
+from ert.storage.realization_storage_state import RealizationStorageState
 
 
 @pytest.fixture
@@ -437,3 +441,63 @@ def test_update_subset_parameters(storage, uniform_parameter, obs):
     assert not prior.load_parameters("PARAMETER", 0)["values"].equals(
         posterior_ens.load_parameters("PARAMETER", 0)["values"]
     )
+
+
+def test_that_update_works_with_failed_realizations(copy_case):
+    copy_case("poly_example")
+
+    with open("poly_eval.py", "w", encoding="utf-8") as f:
+        f.write(
+            dedent(
+                """\
+                #!/usr/bin/env python
+                import numpy as np
+                import sys
+                import json
+
+                def _load_coeffs(filename):
+                    with open(filename, encoding="utf-8") as f:
+                        return json.load(f)["COEFFS"]
+
+                def _evaluate(coeffs, x):
+                    return coeffs["a"] * x**2 + coeffs["b"] * x + coeffs["c"]
+
+                if __name__ == "__main__":
+                    if np.random.random(1) > 0.5:
+                        sys.exit(1)
+                    coeffs = _load_coeffs("parameters.json")
+                    output = [_evaluate(coeffs, x) for x in range(10)]
+                    with open("poly.out", "w", encoding="utf-8") as f:
+                        f.write("\\n".join(map(str, output)))
+                """
+            )
+        )
+    os.chmod(
+        "poly_eval.py",
+        os.stat("poly_eval.py").st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH,
+    )
+
+    parser = ArgumentParser(prog="test_main")
+    parsed = ert_parser(
+        parser,
+        [
+            ENSEMBLE_SMOOTHER_MODE,
+            "poly.ert",
+            "--target-case",
+            "posterior",
+        ],
+    )
+    run_cli(parsed)
+
+    ert_config = ErtConfig.from_file("poly.ert")
+
+    with open_storage(ert_config.ens_path) as storage:
+        prior = storage.get_ensemble_by_name("default")
+        posterior = storage.get_ensemble_by_name("posterior")
+
+        assert all(
+            posterior.get_ensemble_state()[idx]
+            == RealizationStorageState.PARENT_FAILURE
+            for idx, v in enumerate(prior.get_ensemble_state())
+            if v == RealizationStorageState.LOAD_FAILURE
+        )