Add yaml obs converter

The purpose of this converter class is to support conversion of
observation configurations to a format supported by webviz.

For now, only summary observations are of interest. RFTs are
manually loaded through other workflows and other observations
are not of interest as of now.

Localization is not supported in webviz, so those attributes are
left out.

Author: SAKavli

src/ert/__main__.py

@@ -496,7 +496,7 @@ def get_ert_parser(parser: ArgumentParser | None = None) -> ArgumentParser:
         "observation format to summary, but this can be "
         "configured using the --format flag to specify "
         "which format to convert to."
-        "Valid formats are: bulk, summary"
+        "Valid formats are: bulk, summary, yaml"
     )
     convert_obs_parser = subparsers.add_parser(
         "convert_observations",

src/ert/observation_converters/dispatcher.py

@@ -11,16 +11,19 @@
 from .summary_to_bulk import (
     convert_summary_to_bulk,
 )
+from .summary_to_yaml import convert_summary_to_yaml
 
 
 class SupportedFormat(StrEnum):
     SUMMARY = "summary"
     BULK = "bulk"
+    YAML = "yaml"
 
 
 _SUPPORTED_CONVERSIONS: dict[SupportedFormat, Callable[..., None]] = {
     SupportedFormat.BULK: convert_summary_to_bulk,
     SupportedFormat.SUMMARY: convert_history_to_summary,
+    SupportedFormat.YAML: convert_summary_to_yaml,
 }
 
 

src/ert/observation_converters/summary_to_yaml.py

@@ -0,0 +1,89 @@
+import warnings
+from pathlib import Path
+from typing import Literal, TypedDict
+
+from natsort import natsorted
+from ruamel.yaml import YAML
+
+from ert.cli.main import ErtCliError
+from ert.config import ErtConfig, Observation
+from ert.plugins import ErtRuntimePlugins
+
+
+class YamlObservation(TypedDict):
+    date: str
+    value: float
+    error: float
+
+
+class SummaryDict(TypedDict):
+    key: str
+    observations: list[YamlObservation]
+
+
+YamlDict = dict[Literal["smry"], list[SummaryDict]]
+
+
+class YamlConverter:
+    TARGET_FILE = "summary_observations.yaml"
+
+    def __init__(self, observations: list[Observation]) -> None:
+        summary_observations = [
+            o for o in observations if o.type == "summary_observation"
+        ]
+        if not summary_observations:
+            raise ErtCliError("No summary observations in configuration.\nExiting ...")
+
+        self.summary_observations = summary_observations
+
+    def _summary_to_yaml_dict(self) -> YamlDict:
+        summary_keys: set[str] = {o.key for o in self.summary_observations}
+        summary_list: list[SummaryDict] = []
+        for key in natsorted(summary_keys):
+            observations_with_key = [
+                o for o in self.summary_observations if o.key == key
+            ]
+            chronological_observations = sorted(
+                observations_with_key, key=lambda o: o.date
+            )
+            # Round dates without HH/MM/SS to just date
+            for o in chronological_observations:
+                if o.date.endswith("T00:00:00"):
+                    o.date = o.date.split("T")[0]
+            obs_dicts: list[YamlObservation] = [
+                {
+                    "date": o.date,
+                    "value": o.value,
+                    "error": o.error,
+                }
+                for o in chronological_observations
+            ]
+            summary_dict: SummaryDict = {"key": key, "observations": obs_dicts}
+            summary_list.append(summary_dict)
+        return {"smry": summary_list}
+
+    def export_yaml(self) -> None:
+        yaml = YAML()
+        yaml_dict = self._summary_to_yaml_dict()
+        try:
+            with Path(self.TARGET_FILE).open("x", encoding="utf-8") as f:
+                yaml.dump(yaml_dict, f)
+        except FileExistsError as error:
+            raise ErtCliError(
+                f"A file with name '{self.TARGET_FILE}' already exists. "
+                "Will not overwrite it and exit instead."
+            ) from error
+        print(f"Successfully wrote summary observations to '{self.TARGET_FILE}'.")
+
+
+def convert_summary_to_yaml(config: str, site_plugins: ErtRuntimePlugins) -> None:
+    with warnings.catch_warnings():
+        warnings.filterwarnings(action="ignore")
+        ert_config = ErtConfig.with_plugins(site_plugins).from_file(config)
+
+    observations = ert_config.observation_declarations
+
+    yaml_exporter = YamlConverter(
+        observations=observations,
+    )
+    yaml_exporter.export_yaml()

tests/ert/unit_tests/cli/test_converters.py

@@ -0,0 +1,45 @@
+from pathlib import Path
+from unittest.mock import MagicMock
+
+from ert.observation_converters import convert_observations
+from ert.plugins import get_site_plugins
+
+
+def test_that_convert_observations_does_not_fail_when_config_has_hooked_workflows(
+    use_tmpdir,
+):
+    """This reproduces the case where ErtConfig.from_file() is called without
+    plugins while hooked workflows reference plugin-provided jobs.
+    """
+    site_plugins = get_site_plugins()
+
+    arbitrary_existing_job = next(iter(site_plugins.installed_workflow_jobs))
+
+    workflow_file = Path("my_hook_workflow")
+    workflow_file.write_text(f"{arbitrary_existing_job}\n", encoding="utf-8")
+
+    obs_config = "foo.txt"
+    summary_obs = (
+        "SUMMARY_OBSERVATION { KEY = FOPR; VALUE = 10; ERROR = 5; DATE = 2000-01-01; };"
+    )
+    Path(obs_config).write_text(
+        summary_obs,
+        encoding="utf-8",
+    )
+
+    ert_config = "config.ert"
+    minimal_workflow_config = f"""\
+    NUM_REALIZATIONS 10
+    ECLBASE foo
+    OBS_CONFIG {obs_config}
+    LOAD_WORKFLOW {workflow_file} MY_HOOK
+    HOOK_WORKFLOW MY_HOOK PRE_SIMULATION
+    """
+    Path(ert_config).write_text(
+        minimal_workflow_config,
+        encoding="utf-8",
+    )
+
+    for format_ in ["summary", "bulk", "yaml"]:
+        args = MagicMock(format=format_, config=ert_config)
+        convert_observations(args, site_plugins)

tests/ert/unit_tests/cli/test_summary_to_bulk.py

@@ -17,7 +17,7 @@
     BulkConfigConverter,
     _breakthrough_to_string,
 )
-from ert.plugins import ErtRuntimePlugins, get_site_plugins
+from ert.plugins import ErtRuntimePlugins
 
 
 @pytest.mark.usefixtures("snake_oil_case")
@@ -113,7 +113,7 @@ def _make_summary_obs(
     key += f":{well}" if well else ""
     return SummaryObservation(
         name="foo",
-        key=f"{key}:{well}",
+        key=key,
         value=0.5,
         error=0.02,
         date=date,
@@ -397,42 +397,3 @@ def test_that_no_summary_observations_raises_ert_cli_error():
     args = MagicMock(format="bulk", config="snake_oil.ert")
     with pytest.raises(ErtCliError, match="No summary observations found"):
         convert_observations(args, ErtRuntimePlugins())
-
-
-def test_that_convert_observations_does_not_fail_when_config_has_hooked_workflows(
-    use_tmpdir,
-):
-    """This reproduces the case where ErtConfig.from_file() is called without
-    plugins while hooked workflows reference plugin-provided jobs.
-    """
-    site_plugins = get_site_plugins()
-
-    arbitrary_existing_job = next(iter(site_plugins.installed_workflow_jobs))
-
-    workflow_file = Path("my_hook_workflow")
-    workflow_file.write_text(f"{arbitrary_existing_job}\n", encoding="utf-8")
-
-    obs_config = "foo"
-    summary_obs = (
-        "SUMMARY_OBSERVATION { KEY = FOPR; VALUE = 10; ERROR = 5; DATE = 2000-01-01; };"
-    )
-    Path(obs_config).write_text(
-        summary_obs,
-        encoding="utf-8",
-    )
-
-    ert_config = "config.ert"
-    minimal_workflow_config = f"""\
-    NUM_REALIZATIONS 10
-    ECLBASE foo
-    OBS_CONFIG {obs_config}
-    LOAD_WORKFLOW {workflow_file} MY_HOOK
-    HOOK_WORKFLOW MY_HOOK PRE_SIMULATION
-    """
-    Path(ert_config).write_text(
-        minimal_workflow_config,
-        encoding="utf-8",
-    )
-
-    args = MagicMock(format="bulk", config=ert_config)
-    convert_observations(args, site_plugins)