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/observation_converters/dispatcher.py

@@ -9,18 +9,21 @@
 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"
 
 
 ConverterFunction = Callable[[str], None]
 
 _SUPPORTED_CONVERSIONS: dict[SupportedFormat, ConverterFunction] = {
     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,70 @@
+from pathlib import Path
+from typing import Literal, TypedDict
+
+from ruamel.yaml import YAML
+
+from ert.cli.main import ErtCliError
+from ert.config import ErtConfig, Observation
+
+
+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 found.\nExiting ...")
+
+        self.summary_observations = summary_observations
+
+    def _summary_to_yaml_dict(self) -> YamlDict:
+        summary_observations = [
+            o for o in self.summary_observations if o.type == "summary_observation"
+        ]
+        summary_keys: set[str] = {o.key for o in summary_observations}
+        summary_list: list[SummaryDict] = []
+        for key in summary_keys:
+            observations_with_key = [o for o in summary_observations if o.key == key]
+            obs_dicts: list[YamlObservation] = [
+                {"date": o.date, "value": o.value, "error": o.error}
+                for o in observations_with_key
+            ]
+            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()
+        if Path(self.target_file).is_file():
+            raise ErtCliError(
+                f"A file with name '{self.target_file}' already exists. "
+                f"Will not overwrite it and exit instead."
+            )
+        with Path(self.target_file).open("w", encoding="utf-8") as f:
+            yaml.dump(yaml_dict, f)
+        print(f"Successfully wrote summary observations to '{self.target_file}'.")
+
+
+def convert_summary_to_yaml(config: str) -> None:
+    ert_config = ErtConfig.from_file(config)
+    yaml_exporter = YamlConverter(
+        observations=ert_config.observation_declarations,
+    )
+    yaml_exporter.export_yaml()

tests/ert/unit_tests/cli/test_summary_to_bulk.py

@@ -88,8 +88,8 @@ def test_that_happy_path_on_snake_oil_produces_csv_and_stdout(capsys):
         assert old_obs == new_obs
 
 
-@pytest.fixture(name="patched_csv_writer")
-def patched_csv_writing(monkeypatch):
+@pytest.fixture(name="patched_writer")
+def patched_writing(monkeypatch):
     """Avoid writing to file.
     Fixture mock can be used to assert what has been written to file.
     """
@@ -112,7 +112,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,
@@ -135,7 +135,7 @@ def _make_breakthrough_obs(
     )
 
 
-@pytest.mark.usefixtures("patched_csv_writer")
+@pytest.mark.usefixtures("patched_writer")
 def test_that_convert_summary_observations_extracts_localization_information(capsys):
     shape_registry = ShapeRegistry()
 
@@ -166,9 +166,9 @@ def test_that_convert_summary_observations_extracts_localization_information(cap
     assert expected_print_with_localization in capsys.readouterr().out
 
 
-@pytest.mark.usefixtures("patched_csv_writer")
+@pytest.mark.usefixtures("patched_writer")
 def test_that_convert_summary_observations_produces_natsorted_csv_rows(
-    monkeypatch, patched_csv_writer
+    monkeypatch, patched_writer
 ):
     observations = [
         _make_summary_obs("OP30"),
@@ -182,15 +182,15 @@ def test_that_convert_summary_observations_produces_natsorted_csv_rows(
     ).write_csv()
 
     ordered_wells = ["OP2", "OP4", "OP10", "OP30"]
-    csv_content = patched_csv_writer.getvalue()
+    csv_content = patched_writer.getvalue()
     obs_rows = csv_content.strip().split("\n")[1:]
     csv_well_ordering = [row.split(",")[0].strip() for row in obs_rows]
     assert csv_well_ordering == ordered_wells
 
 
-@pytest.mark.usefixtures("patched_csv_writer")
+@pytest.mark.usefixtures("patched_writer")
 def test_that_convert_summary_observations_chronologically_sorts_within_well(
-    monkeypatch, patched_csv_writer
+    monkeypatch, patched_writer
 ):
     observations = [
         _make_summary_obs(well="OP1", date="2010-01-01"),
@@ -206,7 +206,7 @@ def test_that_convert_summary_observations_chronologically_sorts_within_well(
     ).write_csv()
 
     ordered_wells = ["OP1"] * 3 + ["OP2"] * 3
-    csv_content = patched_csv_writer.getvalue()
+    csv_content = patched_writer.getvalue()
     obs_rows = csv_content.strip().split("\n")[1:]
     csv_well_ordering = [row.split(",")[1].strip() for row in obs_rows]
     assert csv_well_ordering == ordered_wells
@@ -217,7 +217,7 @@ def test_that_convert_summary_observations_chronologically_sorts_within_well(
     assert ordered_dates == csv_date_ordering
 
 
-@pytest.mark.usefixtures("patched_csv_writer")
+@pytest.mark.usefixtures("patched_writer")
 def test_that_localization_can_be_gathered_from_breakthrough(capsys):
     shape_registry = ShapeRegistry()
     shape_id = shape_registry.register(
@@ -242,7 +242,7 @@ def test_that_localization_can_be_gathered_from_breakthrough(capsys):
     ) in capsys.readouterr().out
 
 
-@pytest.mark.usefixtures("patched_csv_writer")
+@pytest.mark.usefixtures("patched_writer")
 def test_that_multiple_breakthrough_observations_for_the_same_well_raises_cli_error(
     monkeypatch,
 ):
@@ -257,7 +257,7 @@ def test_that_multiple_breakthrough_observations_for_the_same_well_raises_cli_er
         BulkConfigConverter([brt1, brt2])
 
 
-@pytest.mark.usefixtures("patched_csv_writer")
+@pytest.mark.usefixtures("patched_writer")
 def test_that_the_correct_number_of_observations_are_mentioned_in_helper_text(capsys):
     observations = [
         _make_summary_obs(well="OP1"),
@@ -269,7 +269,7 @@ def test_that_the_correct_number_of_observations_are_mentioned_in_helper_text(ca
     assert "4 observations can be replaced" in capsys.readouterr().out
 
 
-@pytest.mark.usefixtures("patched_csv_writer")
+@pytest.mark.usefixtures("patched_writer")
 def test_that_bpr_observation_populates_ijk_columns_while_others_are_left_empty(
     patched_csv_writer,
 ):

tests/ert/unit_tests/cli/test_summary_to_yaml.py

@@ -0,0 +1,129 @@
+import io
+from contextlib import contextmanager
+from pathlib import Path
+from textwrap import dedent
+from unittest.mock import MagicMock
+
+import pytest
+
+from ert.cli.main import ErtCliError
+from ert.config._observations import GeneralObservation
+from ert.observation_converters import convert_observations
+from ert.observation_converters.summary_to_yaml import YamlConverter
+from tests.ert.unit_tests.cli.test_summary_to_bulk import _make_summary_obs
+
+
+@pytest.fixture(name="patched_writer")
+def patched_writing(monkeypatch):
+    """Avoid writing to file.
+    Fixture mock can be used to assert what has been written to file.
+    """
+    write_buffer = io.StringIO()
+
+    @contextmanager
+    def mock_open(*args, **kwargs):
+        yield write_buffer
+
+    monkeypatch.setattr(Path, "open", mock_open)
+    return write_buffer
+
+
+@pytest.mark.usefixtures("snake_oil_case")
+def test_that_happy_path_on_snake_oil_produces_yaml_and_stdout(capsys):
+    args = MagicMock(format="yaml", config="snake_oil.ert")
+    convert_observations(args)
+
+    expected_yaml_content = dedent("""\
+    smry:
+    - key: WOPR:OP1
+      observations:
+      - date: '2010-03-31T00:00:00'
+        value: 0.1
+        error: 0.05
+      - date: '2010-12-26T00:00:00'
+        value: 0.7
+        error: 0.07
+      - date: '2011-12-21T00:00:00'
+        value: 0.5
+        error: 0.05
+      - date: '2012-12-15T00:00:00'
+        value: 0.3
+        error: 0.075
+      - date: '2013-12-10T00:00:00'
+        value: 0.2
+        error: 0.035
+      - date: '2015-03-15T00:00:00'
+        value: 0.015
+        error: 0.01
+    """)
+    yaml_content = Path("summary_observations.yaml").read_text(encoding="utf-8")
+    assert yaml_content == expected_yaml_content
+
+    expected_stdout = (
+        f"Successfully wrote summary observations to '{YamlConverter.target_file}'."
+    )
+    stdout = capsys.readouterr().out
+    assert expected_stdout in stdout
+
+
+def test_that_empty_observations_raises_ert_cli_error():
+    with pytest.raises(ErtCliError, match="No summary observations found"):
+        YamlConverter([])
+
+
+def test_that_no_summary_observations_raises_ert_cli_error():
+    gen_obs = GeneralObservation(
+        name="foo", data="foo", value=1.0, error=1.0, restart=5, index=5
+    )
+    with pytest.raises(ErtCliError, match="No summary observations found"):
+        YamlConverter([gen_obs])
+
+
+def test_that_observations_with_same_summary_key_are_gathered_in_yaml_dict(use_tmpdir):
+    k1, k2 = "foo", "bar"
+    observations = 2 * [
+        _make_summary_obs(key=k1, well=None),
+        _make_summary_obs(key=k2, well=None),
+    ]
+
+    converter = YamlConverter(observations=observations)
+    summary_dicts = converter._summary_to_yaml_dict()
+
+    keys = [entry["key"] for entry in summary_dicts["smry"]]
+    assert len(keys) == 2
+    assert set(keys) == {k1, k2}
+
+    observations = [s_d["observations"] for s_d in summary_dicts["smry"]]
+    assert all(len(o) == 2 for o in observations)
+
+
+def test_that_observations_with_different_summary_keys_are_seperated_in_yaml_dict():
+    k1, k2 = "foo", "bar"
+    observations = [
+        _make_summary_obs(key=k1, well=None),
+        _make_summary_obs(key=k2, well=None),
+    ]
+
+    converter = YamlConverter(observations=observations)
+    result = converter._summary_to_yaml_dict()
+
+    keys = [entry["key"] for entry in result["smry"]]
+    assert len(keys) == 2
+    assert set(keys) == {k1, k2}
+
+
+def test_that_dumping_to_yaml_is_skipped_when_file_already_exists(use_tmpdir):
+    observations = [_make_summary_obs()]
+
+    Path("summary_observations.yaml").write_text("existing", encoding="utf-8")
+    assert Path("summary_observations.yaml").is_file()
+
+    converter = YamlConverter(observations=observations)
+    with pytest.raises(
+        ErtCliError,
+        match=(
+            r"A file with name 'summary_observations.yaml' already exists. "
+            "Will not overwrite it and exit instead."
+        ),
+    ):
+        converter.export_yaml()