Update everest monitoring code

- Replace/remove the realization numbers for finished/failed
  realizations with just their count, because these numbers are only
  used internally by everest and do not correspond with the everest
  model realization numbers.
- Add the numbers for failed realizations and for failed perturbations
  to the monitoring output for each finalized batch.

Author: verveerpj

src/ert/run_models/event.py

@@ -51,6 +51,7 @@ class EverestBatchResultEvent(BaseModel, extra="forbid"):
     everest_event: Literal["OPTIMIZATION_RESULT",]
     result_type: Literal["FunctionResult", "GradientResult"]
     results: dict[str, Any] | None = None
+    failures: dict[int, list[int]] | None = None
 
 
 class RunModelTimeEvent(RunModelEvent):

src/ert/run_models/everest_run_model.py

@@ -626,23 +626,25 @@ def _handle_optimizer_results(self, results: tuple[Results, ...]) -> None:
             target_ensemble.save_batch_dataframes(dataframes=batch_dict)
             target_ensemble.update_improvement_flag(is_improvement=False)
 
-        for r in results:
+        for result in results:
             batches = (
                 self._experiment.ensembles_with_function_results
-                if isinstance(r, FunctionResults)
+                if isinstance(result, FunctionResults)
                 else self._experiment.ensembles_with_gradient_results
             )
-            ens = next((ens for ens in batches if ens.iteration == r.batch_id), None)
+            ens = next(
+                (ens for ens in batches if ens.iteration == result.batch_id), None
+            )
             if ens is None:
                 continue
 
             results_dict: dict[str, Any] | None = None
-            if isinstance(r, FunctionResults):
+            if isinstance(result, FunctionResults):
                 results_dict = {}
                 if ens.realization_controls is not None:
                     results_dict |= {
                         "controls": ens.realization_controls.drop(
-                            "realization", "simulation_id"
+                            "batch_id", "realization", "simulation_id"
                         ).to_dicts()[0],
                     }
 
@@ -719,14 +721,25 @@ def _handle_optimizer_results(self, results: tuple[Results, ...]) -> None:
                         "perturbation_constraints": perturbation_gradient_dicts
                     }
 
+            failures = defaultdict(list)
+            if ens.everest_realization_info is not None:
+                for r in ens.everest_realization_info:
+                    if ens.has_failure(r):
+                        m = ens.everest_realization_info[r]["model_realization"]
+                        p = ens.everest_realization_info[r]["perturbation"]
+                        failures[m].append(p)
+
             self.send_event(
                 EverestBatchResultEvent(
-                    batch=r.batch_id,
+                    batch=result.batch_id,
                     everest_event="OPTIMIZATION_RESULT",
-                    result_type="FunctionResult"
-                    if isinstance(r, FunctionResults)
-                    else "GradientResult",
+                    result_type=(
+                        "FunctionResult"
+                        if isinstance(result, FunctionResults)
+                        else "GradientResult"
+                    ),
                     results=results_dict,
+                    failures={k: v for k, v in failures.items() if v} or None,
                 )
             )
 

src/everest/bin/utils.py

@@ -8,7 +8,6 @@
 from collections.abc import Generator, Sequence
 from contextlib import contextmanager
 from dataclasses import dataclass, field
-from itertools import groupby
 from pathlib import Path
 from textwrap import dedent
 from typing import Any, ClassVar
@@ -40,6 +39,7 @@
     wait_for_server_to_stop,
 )
 from everest.strings import EVEREST, OPT_PROGRESS_ID, SIM_PROGRESS_ID
+from everest.util import format_list
 
 JOB_SUCCESS = "Finished"
 JOB_RUNNING = "Running"
@@ -139,25 +139,6 @@ def _get_max_width(sequence: list[Any]) -> int:
     return max(len(item) for item in sequence)
 
 
-def _format_list(values: Sequence[int]) -> str:
-    """Formats a sequence of integers into a comma separated string of ranges.
-
-    For instance: {1, 3, 4, 5, 7, 8, 10} -> "1, 3-5, 7-8, 10"
-    """
-    grouped = (
-        tuple(y for _, y in x)
-        for _, x in groupby(enumerate(sorted(values)), lambda x: x[0] - x[1])
-    )
-    return ", ".join(
-        (
-            "-".join([str(sub_group[0]), str(sub_group[-1])])
-            if len(sub_group) > 1
-            else str(sub_group[0])
-        )
-        for sub_group in grouped
-    )
-
-
 @dataclass
 class JobProgress:
     name: str
@@ -177,7 +158,7 @@ class JobProgress:
         JOB_FAILURE: ansi.RED,
     }
 
-    def _status_string(self, max_widths: dict[str, int]) -> str:
+    def progress_str(self, max_widths: dict[str, int]) -> str:
         string = []
         for state in [JOB_RUNNING, JOB_SUCCESS, JOB_FAILURE]:
             number_of_simulations = len(self.status[state])
@@ -186,27 +167,16 @@ def _status_string(self, max_widths: dict[str, int]) -> str:
             string.append(f"{color}{number_of_simulations:>{width}}{ansi.RESET}")
         return "/".join(string)
 
-    def progress_str(self, max_widths: dict[str, int]) -> str:
-        msg = ""
-        for state in [JOB_SUCCESS, JOB_FAILURE]:
-            simulations_list = _format_list(self.status[state])
-            width = _get_max_width([simulations_list])
-            if width > 0:
-                color = self.STATUS_COLOR[state]
-                msg += f" | {color}{state}: {simulations_list:<{width}}{ansi.RESET}"
-
-        return self._status_string(max_widths) + msg
-
 
 class _DetachedMonitor:
-    WIDTH = 78
+    WIDTH = 60
     INDENT = 2
     FLOAT_FMT = ".5g"
 
     def __init__(self) -> None:
         self._clear_lines: int = 0
-        self._batches_done = set[int]()
         self._last_reported_batch: int = -1
+        self._last_reported_opt_progress: int = -1
         self._snapshots: dict[int, EnsembleSnapshot] = {}
 
     def update(self, status: dict[str, Any]) -> None:
@@ -215,8 +185,9 @@ def update(self, status: dict[str, Any]) -> None:
                 opt_status = status[OPT_PROGRESS_ID]
                 if opt_status:
                     msg = self._get_opt_progress_single_batch(opt_status)
-                    ansi.ansi_print(msg + "\n")
-                    self._clear_lines = 0
+                    if msg:
+                        ansi.ansi_print(msg + "\n")
+                        self._clear_lines = 0
             if SIM_PROGRESS_ID in status:
                 match status[SIM_PROGRESS_ID]:
                     case EndEvent(msg=msg):
@@ -254,19 +225,6 @@ def update(self, status: dict[str, Any]) -> None:
         except Exception:
             logging.getLogger(EVEREST).debug(traceback.format_exc())
 
-    def get_opt_progress(self, context_status: dict[str, Any]) -> tuple[str, int]:
-        cli_monitor_data = context_status["cli_monitor_data"]
-        messages = []
-        first_batch = -1
-        for idx, batch in enumerate(cli_monitor_data["batches"]):
-            if batch not in self._batches_done:
-                if first_batch < 0:
-                    first_batch = batch
-                self._batches_done.add(batch)
-                msg = self._get_opt_progress_batch(cli_monitor_data, batch, idx)
-                messages.append(msg)
-        return self._join_two_newlines(messages), first_batch
-
     def _get_opt_progress_batch(
         self, cli_monitor_data: dict[str, Any], batch: int, idx: int
     ) -> str:
@@ -296,29 +254,57 @@ def _get_opt_progress_batch(
 
     def _get_opt_progress_single_batch(self, cli_monitor_data: dict[str, Any]) -> str:
         batch: int = cli_monitor_data.get("batch", 0)
-        header = self._make_header(f"Optimization progress (Batch #{batch})")
-        width = _get_max_width(cli_monitor_data["controls"].keys())
-        controls = self._join_one_newline_indent(
-            [
-                f"{name:>{width}}: {value:{self.FLOAT_FMT}}"
-                for name, value in cli_monitor_data["controls"].items()
-            ]
-        )
-        expected_objectives = cli_monitor_data["expected_objectives"]
-        width = _get_max_width(expected_objectives.keys())
-        objectives = self._join_one_newline_indent(
-            [
-                f"{name:>{width}}: {value:{self.FLOAT_FMT}}"
-                for name, value in expected_objectives.items()
-            ]
-        )
-        objective_value = cli_monitor_data["objective_value"]
-        total_objective = (
-            f"Total normalized objective: {objective_value:{self.FLOAT_FMT}}"
-        )
-        return self._join_two_newlines_indent(
-            (header, controls, objectives, total_objective)
-        )
+        if batch == self._last_reported_opt_progress:
+            return ""
+
+        lines = [self._make_header(f"Optimization progress (Batch #{batch})")]
+
+        if controls := cli_monitor_data.get("controls"):
+            width = _get_max_width(controls.keys())
+            lines.append(
+                self._join_one_newline_indent(
+                    [
+                        f"{name:>{width}}: {value:{self.FLOAT_FMT}}"
+                        for name, value in controls.items()
+                    ]
+                )
+            )
+        if expected_objectives := cli_monitor_data.get("expected_objectives"):
+            width = _get_max_width(expected_objectives.keys())
+            lines.append(
+                self._join_one_newline_indent(
+                    [
+                        f"{name:>{width}}: {value:{self.FLOAT_FMT}}"
+                        for name, value in expected_objectives.items()
+                    ]
+                )
+            )
+        if objective_value := cli_monitor_data.get("objective_value"):
+            lines.append(
+                f"Total normalized objective: {objective_value:{self.FLOAT_FMT}}"
+            )
+
+        if failures := cli_monitor_data.get("failures", {}):
+            failed_lines = []
+            if failed_functions := [r for r, p in failures.items() if -1 in p]:
+                s = "s" if len(failed_functions) > 1 else ""
+                failed_lines.append(
+                    f"{ansi.RED}Failed function evaluation{s} for realization{s}: "
+                    f"{format_list(failed_functions)}{ansi.RESET}"
+                )
+            for k, v in failures.items():
+                if p := [item for item in v if item >= 0]:
+                    s = "s" if len(p) > 1 else ""
+                    failed_lines.append(
+                        f"{ansi.RED}Failed perturbation{s} for realization {k}: "
+                        f"{format_list(p)}{ansi.RESET}"
+                    )
+            if failed_lines:
+                lines.append(self._join_one_newline_indent(failed_lines))
+
+        self._last_reported_opt_progress = batch
+
+        return self._join_two_newlines_indent(lines)
 
     @staticmethod
     def _get_progress_summary(status: dict[str, int]) -> str:
@@ -362,8 +348,7 @@ def _get_job_states(cls, snapshot: EnsembleSnapshot) -> str:
                 if job.errors:
                     print_lines.extend(
                         [
-                            f"{ansi.RED}{job.name:>{width}}: Failed: {err}, "
-                            f"realizations: {_format_list(job.errors[err])}{ansi.RESET}"
+                            f"{ansi.RED}{job.name:>{width}}: {err}{ansi.RESET}"
                             for err in job.errors
                         ]
                     )

src/everest/detached/client.py

@@ -216,11 +216,18 @@ def get_opt_status_from_batch_result_event(
 
     assert event.batch is not None
 
+    if event.result_type == "FunctionResult":
+        return {
+            "batch": event.batch,
+            "controls": event.results["controls"],
+            "objective_value": event.results["total_objective_value"],
+            "expected_objectives": event.results["objectives"],
+            "failures": event.failures,
+        }
+
     return {
         "batch": event.batch,
-        "controls": event.results["controls"],
-        "objective_value": event.results["total_objective_value"],
-        "expected_objectives": event.results["objectives"],
+        "failures": event.failures,
     }
 
 
@@ -253,14 +260,13 @@ def start_monitor(
                     message = websocket.recv(timeout=1.0)
                     event = status_event_from_json(message)
                     if isinstance(event, EverestBatchResultEvent):
-                        if event.result_type == "FunctionResult":
-                            callback(
-                                {
-                                    OPT_PROGRESS_ID: get_opt_status_from_batch_result_event(  # noqa: E501
-                                        event
-                                    )
-                                }
-                            )
+                        callback(
+                            {
+                                OPT_PROGRESS_ID: get_opt_status_from_batch_result_event(
+                                    event
+                                )
+                            }
+                        )
                     else:
                         callback({SIM_PROGRESS_ID: event})
                 except TimeoutError:

src/everest/util/__init__.py

@@ -1,5 +1,12 @@
 from ropt.version import version as ropt_version
 
+from ._utils import format_list
+
+__all__ = [
+    "format_list",
+]
+
+
 try:
     from ert.shared.version import version as ert_version
 except ImportError:

src/everest/util/_utils.py

@@ -0,0 +1,21 @@
+from collections.abc import Sequence
+from itertools import groupby
+
+
+def format_list(values: Sequence[int]) -> str:
+    """Formats a sequence of integers into a comma separated string of ranges.
+
+    For instance: {1, 3, 4, 5, 7, 8, 10} -> "1, 3-5, 7-8, 10"
+    """
+    grouped = (
+        tuple(y for _, y in x)
+        for _, x in groupby(enumerate(sorted(values)), lambda x: x[0] - x[1])
+    )
+    return ", ".join(
+        (
+            "-".join([str(sub_group[0]), str(sub_group[-1])])
+            if len(sub_group) > 1
+            else str(sub_group[0])
+        )
+        for sub_group in grouped
+    )