Simplify EverestControl to represent a single scalar value

There is now one EverestControl instance per control variable. This aligns its structure with scalar parameter types in ERT. Runpath parameter file generation logic has been adapted to handle this new structure while ensuring the final output format remains backward compatible.

Author: yngve-sk

src/ert/config/everest_control.py

@@ -1,10 +1,8 @@
 from __future__ import annotations
 
 import importlib
-import json
 import logging
 from collections.abc import Iterator, Mapping, MutableMapping
-from dataclasses import field
 from pathlib import Path
 from textwrap import dedent
 from typing import TYPE_CHECKING, Any, Literal, Self
@@ -16,8 +14,6 @@
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 from ropt.workflow import find_sampler_plugin
 
-from ert.substitutions import substitute_runpath_name
-
 from .parameter_config import ParameterCardinality, ParameterConfig
 
 if TYPE_CHECKING:
@@ -144,37 +140,38 @@ def validate_backend_and_method(self) -> Self:
 
 
 class EverestControl(ParameterConfig):
-    """Create an EverestControl for @key with the given @input_keys
+    """Create an EverestControl for a single control variable.
 
-    @input_keys can be either a list of keys as strings or a dict with
-    keys as strings and a list of suffixes for each key.
-    If a list of strings is given, the order is preserved.
+    Each EverestControl represents one scalar value. Multiple controls can
+    share the same group name to indicate they belong to the same logical group.
     """
 
     type: Literal["everest_parameters"] = "everest_parameters"
-    input_keys: list[str] = field(default_factory=list)
+    dimensionality: Literal[1] = 1
+    input_key: str
     forward_init: bool = False
     output_file: str = ""
     forward_init_file: str = ""
     update: bool = False
-    types: list[Literal["well_control", "generic_control"]]
-    initial_guesses: list[float]
-    control_types: list[Literal["real", "integer"]]
-    enabled: list[bool]
-    min: list[float]
-    max: list[float]
-    perturbation_types: list[Literal["absolute", "relative"]]
-    perturbation_magnitudes: list[float]
-    scaled_ranges: list[tuple[float, float]]
-    samplers: list[SamplerConfig | None]
+    control_type_: Literal["well_control", "generic_control"]
+    initial_guess: float
+    control_type: Literal["real", "integer"]
+    enabled: bool
+    min: float
+    max: float
+    perturbation_type: Literal["absolute", "relative"]
+    perturbation_magnitude: float
+    scaled_range: tuple[float, float]
+    sampler: SamplerConfig | None
+    group: str
 
     # Set up for deprecation, but has to live here until support for the
     # "dotdash" notation is removed for everest controls via everest config.
-    input_keys_dotdash: list[str] = field(default_factory=list)
+    input_key_dotdash: str = ""
 
     @property
     def parameter_keys(self) -> list[str]:
-        return self.input_keys
+        return [self.input_key]
 
     @property
     def cardinality(self) -> ParameterCardinality:
@@ -194,31 +191,12 @@ def load_parameter_graph(self) -> nx.Graph[int]:
         raise NotImplementedError
 
     def __len__(self) -> int:
-        return len(self.input_keys)
+        return 1
 
     def write_to_runpath(
         self, run_path: Path, real_nr: int, ensemble: Ensemble
-    ) -> None:
-        file_path: Path = run_path / substitute_runpath_name(
-            self.output_file, real_nr, ensemble.iteration
-        )
-        Path.mkdir(file_path.parent, exist_ok=True, parents=True)
-
-        data: dict[str, Any] = {}
-        df = ensemble.load_parameters(self.name, real_nr)
-        assert isinstance(df, pl.DataFrame)
-        df = df.drop("realization")
-        df = df.rename({col: col.replace(f"{self.name}.", "", 1) for col in df.columns})
-        for c in df.columns:
-            if "." in c:
-                top_key, sub_key = c.split(".", 1)
-                if top_key not in data:
-                    data[top_key] = {}
-                data[top_key][sub_key] = df[c].item()
-            else:
-                data[c] = df[c].item()
-
-        file_path.write_text(json.dumps(data), encoding="utf-8")
+    ) -> dict[str, dict[str, float | str]] | None:
+        raise NotImplementedError
 
     def create_storage_datasets(
         self,
@@ -228,7 +206,7 @@ def create_storage_datasets(
         df = pl.DataFrame(
             {
                 "realization": iens_active_index,
-                **{k: from_data[:, i] for i, k in enumerate(self.parameter_keys)},
+                self.input_key: pl.Series(from_data.flatten()),
             },
             strict=False,
         )

src/ert/run_models/_create_run_path.py

@@ -13,6 +13,7 @@
 from typing import TYPE_CHECKING, Any
 
 import orjson
+import polars as pl
 
 from _ert.utils import file_safe_timestamp
 from ert.config import (
@@ -140,6 +141,9 @@ def _generate_parameter_files(
     exports: dict[str, dict[str, float | str]] = {}
     log_exports: dict[str, dict[str, float | str]] = {}
 
+    # Group EverestControls by output_file for aggregated JSON writing
+    everest_controls_by_file: dict[str, list[EverestControl]] = defaultdict(list)
+
     for param in parameter_configs:
         # For the first iteration we do not write the parameter
         # to run path, as we expect to read if after the forward
@@ -148,8 +152,12 @@ def _generate_parameter_files(
             continue
         start_time = time.perf_counter()
         export_values: dict[str, dict[str, float | str]] | None = None
-        log_export_values: dict[str, dict[str, float | str]] | None = {}
-        if param.name in scalar_data:
+        log_export_values: dict[str, dict[str, float | str]] = {}
+
+        if isinstance(param, EverestControl):
+            everest_controls_by_file[param.output_file].append(param)
+            continue
+        elif param.name in scalar_data:
             scalar_value = scalar_data[param.name]
             export_values = {param.group_name: {param.name: scalar_value}}
             if isinstance(param, GenKwConfig) and isinstance(
@@ -175,6 +183,39 @@ def _generate_parameter_files(
                 log_exports.setdefault(group, {}).update(vals)
         export_timings[param.type] += time.perf_counter() - start_time
         continue
+
+    # Write aggregated EverestControl JSON files
+    start_time = time.perf_counter()
+    for output_file, controls in everest_controls_by_file.items():
+        file_path: Path = Path(run_path) / substitute_runpath_name(
+            output_file, iens, iteration
+        )
+        file_path.parent.mkdir(exist_ok=True, parents=True)
+
+        data: dict[str, Any] = {}
+        for control in controls:
+            df = fs.load_parameters(control.name, iens)
+            assert isinstance(df, pl.DataFrame)
+            value = df[control.input_key].item()
+
+            key_without_group = control.input_key.replace(f"{control.group}.", "", 1)
+
+            if "." in key_without_group:
+                parts = key_without_group.split(".")
+                current = data
+                for part in parts[:-1]:
+                    if part not in current:
+                        current[part] = {}
+                    current = current[part]
+                current[parts[-1]] = value
+            else:
+                data[key_without_group] = value
+
+        file_path.write_text(json.dumps(data, indent=2), encoding="utf-8")
+
+    if everest_controls_by_file:
+        export_timings["everest_parameters"] += time.perf_counter() - start_time
+
     start_time = time.perf_counter()
     _value_export_txt(run_path, export_base_name, exports | log_exports)
     export_timings["value_export_txt"] = time.perf_counter() - start_time

src/ert/run_models/everest_run_model.py

@@ -296,7 +296,9 @@ def create(
         )
 
         parameter_configs = [
-            control.to_ert_parameter_config() for control in everest_config.controls
+            ert_control
+            for control in everest_config.controls
+            for ert_control in control.to_ert_parameter_config()
         ]
 
         response_configs: list[ResponseConfig] = []

src/ert/storage/local_storage.py

@@ -31,7 +31,7 @@
 
 logger = logging.getLogger(__name__)
 
-_LOCAL_STORAGE_VERSION = 25
+_LOCAL_STORAGE_VERSION = 26
 
 
 class _Migrations(BaseModel):
@@ -515,6 +515,7 @@ def _migrate(self, version: int) -> None:
             to23,
             to24,
             to25,
+            to26,
         )
 
         try:
@@ -569,6 +570,7 @@ def _migrate(self, version: int) -> None:
                     22: to23,
                     23: to24,
                     24: to25,
+                    25: to26,
                 }
                 for from_version in range(version, _LOCAL_STORAGE_VERSION):
                     migrations[from_version].migrate(self.path)

src/ert/storage/migration/to26.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+info = "Unroll EverestControl to one config per parameter"
+
+
+def migrate_everest_control_format(path: Path) -> None:
+    experiments_dir = path / "experiments"
+    if not experiments_dir.exists():
+        return
+
+    for exp_dir in experiments_dir.iterdir():
+        if not exp_dir.is_dir():
+            continue
+
+        index_file = exp_dir / "index.json"
+        index_data = json.loads(index_file.read_text(encoding="utf-8"))
+
+        experiment_data = index_data.get("experiment")
+        params_config = experiment_data.get("parameter_configuration")
+        new_params_config: list[dict[str, Any]] = []
+        modified = False
+
+        for param in params_config:
+            if param.get("type") == "everest_parameters" and "input_keys" in param:
+                modified = True
+
+                common_fields = {
+                    "forward_init": False,
+                    "output_file": param["output_file"],
+                    "forward_init_file": "",
+                    "update": False,
+                    "type": "everest_parameters",
+                    "dimensionality": 1,
+                }
+
+                group_name = param["name"]
+                input_keys = param["input_keys"]
+
+                for i, input_key in enumerate(input_keys):
+                    new_params_config.append(
+                        {
+                            **common_fields,
+                            "input_key": input_key,
+                            "group": group_name,
+                            "name": input_key,
+                            "control_type_": param["types"][i],
+                            "initial_guess": param["initial_guesses"][i],
+                            "control_type": param["control_types"][i],
+                            "enabled": param["enabled"][i],
+                            "min": param["min"][i],
+                            "max": param["max"][i],
+                            "perturbation_type": param["perturbation_types"][i],
+                            "perturbation_magnitude": param["perturbation_magnitudes"][
+                                i
+                            ],
+                            "scaled_range": param["scaled_ranges"][i],
+                            "sampler": param["samplers"][i],
+                            "input_key_dotdash": param["input_keys_dotdash"][i],
+                        }
+                    )
+            else:
+                new_params_config.append(param)
+
+        if modified:
+            experiment_data["parameter_configuration"] = new_params_config
+            index_file.write_text(json.dumps(index_data, indent=2), encoding="utf-8")
+
+
+def migrate(path: Path) -> None:
+    migrate_everest_control_format(path)