Have ert show hooked workflows in GUI

This commit changes the GUI to:
* Show iterations and updates in tabs to accommodate showing pre/post
  run/update workflows
* Show pre-experiment and post-experiment workflows on main tab-bar
* Pressing terminate experiment during hooked workflows cancels the
  workflow, and marks it as cancelled by user. Following workflows in
  the same hook will be marked as cancelled, as to not mislead the user.

Author: jonathan-eq

docs/ert/getting_started/configuration/poly_new/minimal/ci_cmp_simulations.png

@@ -1,8 +1,11 @@
 from datetime import UTC, datetime
+from enum import StrEnum
 from typing import Annotated, Any, Final, Literal
 
 from pydantic import BaseModel, ConfigDict, Field, TypeAdapter
 
+from _ert.hook_runtime import HookRuntime
+
 
 class Id:
     FORWARD_MODEL_STEP_START_TYPE = Literal["forward_model_step.start"]
@@ -239,3 +242,56 @@ def dispatcher_event_from_json(raw_msg: str | bytes) -> DispatcherEvent:
 
 def dispatcher_event_to_json(event: DispatcherEvent) -> str:
     return event.model_dump_json()
+
+
+class WorkflowStatus(StrEnum):
+    PENDING = "Pending"
+    RUNNING = "Running"
+    FINISHED = "Finished"
+    FAILED = "Failed"
+    CANCELLED = "Cancelled"
+
+
+class _WorkflowEvent(BaseModel):
+    hook: HookRuntime | None = None
+    iteration: int | None = None
+
+
+class WorkflowBatchStartedEvent(_WorkflowEvent):
+    event_type: Literal["WorkflowBatchStartedEvent"] = "WorkflowBatchStartedEvent"
+    workflow_names: list[str]
+
+
+class WorkflowBatchFinishedEvent(_WorkflowEvent):
+    event_type: Literal["WorkflowBatchFinishedEvent"] = "WorkflowBatchFinishedEvent"
+    status: WorkflowStatus
+    workflow_names: list[str]
+
+
+class WorkflowStartedEvent(_WorkflowEvent):
+    event_type: Literal["WorkflowStartedEvent"] = "WorkflowStartedEvent"
+    workflow_name: str
+
+
+class WorkflowFinishedEvent(_WorkflowEvent):
+    event_type: Literal["WorkflowFinishedEvent"] = "WorkflowFinishedEvent"
+    workflow_name: str
+    status: WorkflowStatus
+    stdout: str | None = None
+    stderr: str | None = None
+
+
+class WorkflowCancelledEvent(_WorkflowEvent):
+    event_type: Literal["WorkflowCancelledEvent"] = "WorkflowCancelledEvent"
+    workflow_name: str
+    stdout: str | None = None
+    stderr: str | None = None
+
+
+WorkflowEvent = (
+    WorkflowBatchStartedEvent
+    | WorkflowBatchFinishedEvent
+    | WorkflowStartedEvent
+    | WorkflowFinishedEvent
+    | WorkflowCancelledEvent
+)

src/_ert/events.py

@@ -0,0 +1,22 @@
+from enum import StrEnum
+
+
+class HookRuntime(StrEnum):
+    PRE_SIMULATION = "PRE_SIMULATION"
+    POST_SIMULATION = "POST_SIMULATION"
+    PRE_UPDATE = "PRE_UPDATE"
+    POST_UPDATE = "POST_UPDATE"
+    PRE_FIRST_UPDATE = "PRE_FIRST_UPDATE"
+    PRE_EXPERIMENT = "PRE_EXPERIMENT"
+    POST_EXPERIMENT = "POST_EXPERIMENT"
+
+    def workflow_tab_title(self) -> str:
+        return {
+            HookRuntime.PRE_EXPERIMENT: "Pre-experiment workflows",
+            HookRuntime.POST_EXPERIMENT: "Post-experiment workflows",
+            HookRuntime.PRE_SIMULATION: "Pre-simulation workflows",
+            HookRuntime.POST_SIMULATION: "Post-simulation workflows",
+            HookRuntime.PRE_FIRST_UPDATE: "Pre-first-update workflows",
+            HookRuntime.PRE_UPDATE: "Pre-update workflows",
+            HookRuntime.POST_UPDATE: "Post-update workflows",
+        }[self]

src/_ert/hook_runtime.py

@@ -31,6 +31,8 @@ def run(self, *args: Any) -> None:
         sys.stdout.write(self._stdoutdata)
 
         if self.__job.returncode != 0:
+            if self.isCancelled():
+                return None
             raise RuntimeError(self._stderrdata)
 
     def cancel(self) -> Any:

src/ert/config/external_ert_script.py

@@ -1,3 +1,5 @@
+from _ert.hook_runtime import HookRuntime
+
 from ._parse_zonemap import parse_zonemap
 from .config_dict import ConfigDict
 from .config_errors import ConfigValidationError, ConfigWarning
@@ -8,7 +10,6 @@
 from .forward_model_keywords import ForwardModelStepKeys
 from .forward_model_schema import init_forward_model_schema
 from .history_source import HistorySource
-from .hook_runtime import HookRuntime
 from .lark_parser import parse, parse_contents, read_file
 from .observations_parser import (
     ObservationConfigError,

src/ert/config/parsing/__init__.py

@@ -1,3 +1,5 @@
+from _ert.hook_runtime import HookRuntime
+
 from .config_dict import ConfigDict
 from .config_keywords import ConfigKeys
 from .config_schema_deprecations import deprecated_keywords_list
@@ -15,7 +17,6 @@
     string_keyword,
 )
 from .history_source import HistorySource
-from .hook_runtime import HookRuntime
 from .observations_parser import parse_observations
 from .queue_system import QueueSystem
 from .schema_dict import SchemaItemDict

src/ert/config/parsing/config_schema.py

@@ -8,7 +8,7 @@
 
 from typing_extensions import TypedDict
 
-from ert.config.parsing.hook_runtime import HookRuntime
+from _ert.hook_runtime import HookRuntime
 
 if TYPE_CHECKING:
     from PyQt6.QtWidgets import QWidget

src/ert/config/parsing/hook_runtime.py

@@ -31,7 +31,15 @@
 )
 from typing_extensions import override
 
-from _ert.events import EnsembleEvaluationWarning
+from _ert.events import (
+    EnsembleEvaluationWarning,
+    WorkflowBatchFinishedEvent,
+    WorkflowBatchStartedEvent,
+    WorkflowCancelledEvent,
+    WorkflowFinishedEvent,
+    WorkflowStartedEvent,
+)
+from _ert.hook_runtime import HookRuntime
 from ert.config import ErrorInfo, WarningInfo
 from ert.ensemble_evaluator import (
     EndEvent,
@@ -80,10 +88,12 @@
 from .queue_emitter import QueueEmitter
 from .view import (
     DiskSpaceWidget,
+    IterationWidget,
     ProgressWidget,
     RealizationWidget,
     RunpathProgressWidget,
     UpdateWidget,
+    WorkflowWidget,
 )
 from .view.disk_space_widget import MountType
 
@@ -389,10 +399,15 @@ def is_experiment_done(self) -> bool:
 
     def _current_tab_changed(self, index: int) -> None:
         widget = self._tab_widget.widget(index)
-        if isinstance(widget, RealizationWidget):
-            widget.refresh_current_selection()
+        if isinstance(widget, IterationWidget):
+            current_widget = widget.current_widget()
+            if isinstance(current_widget, RealizationWidget):
+                current_widget.refresh_current_selection()
+
+            self.fm_step_frame.setHidden(isinstance(current_widget, UpdateWidget))
+            return
 
-        self.fm_step_frame.setHidden(isinstance(widget, UpdateWidget))
+        self.fm_step_frame.setHidden(False)
 
     @Slot(QModelIndex, int, int)
     def on_snapshot_new_iteration(
@@ -402,26 +417,21 @@ def on_snapshot_new_iteration(
             index = self._snapshot_model.index(start, 0, parent)
             iteration = int(cast(IterNode, index.internalPointer()).id_)
             self._latest_iteration = iteration
-            iter_row = start
             self._iteration_progress_label.setText(
                 f"Progress for iteration {iteration}"
                 if not self.is_everest
                 else f"Progress for batch {iteration}"
             )
 
-            widget = RealizationWidget(iter_row)
+            iteration_widget = self._get_or_create_iteration_tab(
+                iteration, is_update=False
+            )
+            widget = iteration_widget.select_or_create_realization_tab()
             widget.setSnapshotModel(self._snapshot_model)
             widget.itemClicked.connect(self._select_real)
             widget.setProperty("identifier", f"tab-iter-{iteration}")
             self._select_real(widget._real_list_model.index(0, 0))
-            tab_index = self._tab_widget.addTab(
-                widget,
-                f"Realizations for iteration {iteration}"
-                if not self.is_everest
-                else f"Batch {iteration}...",
-            )
-            if self._tab_widget.currentIndex() == self._tab_widget.count() - 2:
-                self._tab_widget.setCurrentIndex(tab_index)
+            self._tab_widget.setCurrentWidget(iteration_widget)
 
             if self.is_everest:
                 self._batch_result_types.append(set())
@@ -579,6 +589,24 @@ def _on_event(self, event: object) -> None:
                 self.post_experiment_warnings.append(msg)
             case EnsembleEvaluationWarning(warning_message=msg):
                 self._show_warning(msg)
+            case WorkflowBatchStartedEvent(
+                hook=hook, iteration=iteration, workflow_names=workflow_names
+            ):
+                self._select_or_create_workflow_tab(hook, iteration, workflow_names)
+            case WorkflowStartedEvent(hook=hook, iteration=iteration):
+                self._select_or_create_workflow_tab(hook, iteration).handle_event(event)
+            case WorkflowFinishedEvent(
+                hook=hook,
+                iteration=iteration,
+            ):
+                self._select_or_create_workflow_tab(hook, iteration).handle_event(event)
+            case WorkflowCancelledEvent(
+                hook=hook,
+                iteration=iteration,
+            ):
+                self._select_or_create_workflow_tab(hook, iteration).handle_event(event)
+            case WorkflowBatchFinishedEvent(hook=hook, iteration=iteration):
+                self._select_or_create_workflow_tab(hook, iteration).handle_event(event)
 
             case FullSnapshotEvent(
                 status_count=status_count, realization_count=realization_count
@@ -604,22 +632,23 @@ def _on_event(self, event: object) -> None:
                 )
                 self.progress_update_event.emit(status_count, realization_count)
             case RunModelUpdateBeginEvent(iteration=iteration):
-                widget = UpdateWidget(iteration)
-                tab_index = self._tab_widget.addTab(widget, f"Update {iteration}")
-                if self._tab_widget.currentIndex() == self._tab_widget.count() - 2:
-                    self._tab_widget.setCurrentIndex(tab_index)
+                iteration_widget = self._get_or_create_iteration_tab(
+                    iteration, is_update=True
+                )
+                widget = iteration_widget.select_or_create_update_tab()
+                self._tab_widget.setCurrentWidget(iteration_widget)
                 widget.begin(event)
             case RunModelUpdateEndEvent():
                 self._progress_widget.stop_waiting_progress_bar()
-                self._get_update_widget(event.iteration).end(event)
+                self._get_or_create_update_tab(event.iteration).end(event)
                 event.write_as_csv(self.output_path)
             case RunModelStatusEvent() | RunModelTimeEvent():
-                self._get_update_widget(event.iteration).update_status(event)
+                self._get_or_create_update_tab(event.iteration).update_status(event)
             case RunModelDataEvent():
-                self._get_update_widget(event.iteration).add_table(event)
+                self._get_or_create_update_tab(event.iteration).add_table(event)
                 event.write_as_csv(self.output_path)
             case RunModelErrorEvent():
-                self._get_update_widget(event.iteration).error(event)
+                self._get_or_create_update_tab(event.iteration).error(event)
                 event.write_as_csv(self.output_path)
             case EverestBatchResultEvent():
                 batch_types = self._batch_result_types[event.batch]
@@ -652,12 +681,85 @@ def _on_event(self, event: object) -> None:
                 ):
                     runpath_widget.advance()
 
-    def _get_update_widget(self, iteration: int) -> UpdateWidget:
-        for i in range(self._tab_widget.count()):
-            widget = self._tab_widget.widget(i)
-            if isinstance(widget, UpdateWidget) and widget.iteration == iteration:
+    def _get_or_create_update_tab(self, iteration: int) -> UpdateWidget:
+        return self._get_or_create_iteration_tab(
+            iteration, is_update=True
+        ).select_or_create_update_tab()
+
+    def _get_or_create_iteration_tab(
+        self, iteration: int, is_update: bool
+    ) -> IterationWidget:
+        for index in range(self._tab_widget.count()):
+            widget = self._tab_widget.widget(index)
+            if (
+                isinstance(widget, IterationWidget)
+                and widget.iteration == iteration
+                and widget.is_update_page == is_update
+            ):
                 return widget
-        raise ValueError("Could not find UpdateWidget")
+
+        widget = IterationWidget(iteration, self)
+        widget.is_update_page = is_update
+        widget.currentTabChanged.connect(
+            lambda iteration_widget=widget: self._on_iteration_tab_changed(
+                iteration_widget
+            )
+        )
+
+        self._tab_widget.addTab(
+            widget,
+            f"update-{iteration}" if is_update else f"iteration-{iteration}",
+        )
+        return widget
+
+    def _on_iteration_tab_changed(self, iteration_widget: IterationWidget) -> None:
+        if self._tab_widget.currentWidget() is iteration_widget:
+            self._current_tab_changed(self._tab_widget.currentIndex())
+
+    @staticmethod
+    def _workflow_belongs_to_update(hook: HookRuntime) -> bool:
+        return hook in {
+            HookRuntime.PRE_FIRST_UPDATE,
+            HookRuntime.PRE_UPDATE,
+            HookRuntime.POST_UPDATE,
+        }
+
+    def _select_or_create_workflow_tab(
+        self,
+        hook: HookRuntime | None,
+        iteration: int | None,
+        workflow_names: list[str] | None = None,
+    ) -> WorkflowWidget:
+        assert hook is not None
+        if iteration is not None:
+            iteration_widget = (
+                self._get_or_create_iteration_tab(iteration, is_update=True)
+                if self._workflow_belongs_to_update(hook)
+                else self._get_or_create_iteration_tab(iteration, is_update=False)
+            )
+            widget = iteration_widget.select_or_create_workflow_tab(
+                hook, workflow_names
+            )
+            self._tab_widget.setCurrentWidget(iteration_widget)
+            return widget
+
+        for index in range(self._tab_widget.count()):
+            existing_widget = self._tab_widget.widget(index)
+            if (
+                isinstance(existing_widget, WorkflowWidget)
+                and existing_widget.hook == hook
+            ):
+                return existing_widget
+
+        if workflow_names is None:
+            raise RuntimeError(
+                "Workflow tab must be created from WorkflowBatchStartedEvent"
+            )
+
+        widget = WorkflowWidget(hook, workflow_names, parent=self)
+        tab_index = self._tab_widget.addTab(widget, hook.workflow_tab_title())
+        self._tab_widget.setCurrentIndex(tab_index)
+        return widget
 
     def update_total_progress(
         self, progress_value: float, iteration_label: str, iteration: int | None = None

src/ert/config/workflow_fixtures.py

@@ -1,13 +1,17 @@
 from .disk_space_widget import DiskSpaceWidget
+from .iteration import IterationWidget
 from .progress_widget import ProgressWidget
 from .realization import RealizationWidget
 from .runpath_progress_widget import RunpathProgressWidget
 from .update import UpdateWidget
+from .workflow import WorkflowWidget
 
 __all__ = [
     "DiskSpaceWidget",
+    "IterationWidget",
     "ProgressWidget",
     "RealizationWidget",
     "RunpathProgressWidget",
     "UpdateWidget",
+    "WorkflowWidget",
 ]