Add ScalarParameters config

 - transformations become dataclases -> TODO: pydantic
 - ScalarParameter represents a single parameter instance with a
   datasource (DESIGN_MATRIX, SAMPLED)
 - all ScalarParameter instances are bound by
   ScalarParameters(ParameterConfig) and are stored for all realizations
in a single pl.Dataframe (parquet) file.
 - API to load / save scalar parameters is in local_ensemble load|save_param_scalar
 - DESIGN_MATRIX now creates only ScalarParameters instances and
   modifies the ScalarParameterc Config.
 - Reimplement load_all_gen_kwn_data

Author: xjules

src/ert/analysis/_es_update.py

@@ -5,12 +5,7 @@
 import time
 from collections.abc import Callable, Iterable, Sequence
 from fnmatch import fnmatch
-from typing import (
-    TYPE_CHECKING,
-    Generic,
-    Self,
-    TypeVar,
-)
+from typing import TYPE_CHECKING, Generic, Self, TypeVar
 
 import iterative_ensemble_smoother as ies
 import numpy as np
@@ -19,7 +14,13 @@
 import scipy
 from iterative_ensemble_smoother.experimental import AdaptiveESMDA
 
-from ert.config import ESSettings, GenKwConfig, ObservationGroups, UpdateSettings
+from ert.config import (
+    ESSettings,
+    GenKwConfig,
+    ObservationGroups,
+    ScalarParameters,
+    UpdateSettings,
+)
 
 from . import misfit_preprocessor
 from .event import (
@@ -31,10 +32,7 @@
     AnalysisTimeEvent,
     DataSection,
 )
-from .snapshots import (
-    ObservationAndResponseSnapshot,
-    SmootherSnapshot,
-)
+from .snapshots import ObservationAndResponseSnapshot, SmootherSnapshot
 
 if TYPE_CHECKING:
     import numpy.typing as npt
@@ -111,16 +109,26 @@ def _all_parameters(
 
 
 def _save_param_ensemble_array_to_disk(
-    ensemble: Ensemble,
+    source_ensemble: Ensemble,
+    target_ensemble: Ensemble,
     param_ensemble_array: npt.NDArray[np.float64],
     param_group: str,
     iens_active_index: npt.NDArray[np.int_],
 ) -> None:
-    config_node = ensemble.experiment.parameter_configuration[param_group]
-    for i, realization in enumerate(iens_active_index):
-        config_node.save_parameters(
-            ensemble, param_group, realization, param_ensemble_array[:, i]
+    config_node = target_ensemble.experiment.parameter_configuration[param_group]
+    if isinstance(config_node, ScalarParameters):
+        config_node.save_updated_parameters_and_copy_remaining(
+            source_ensemble,
+            target_ensemble,
+            param_group,
+            iens_active_index,
+            param_ensemble_array,
         )
+    else:
+        for i, realization in enumerate(iens_active_index):
+            config_node.save_parameters(
+                target_ensemble, param_group, realization, param_ensemble_array[:, i]
+            )
 
 
 def _load_param_ensemble_array(
@@ -129,7 +137,11 @@ def _load_param_ensemble_array(
     iens_active_index: npt.NDArray[np.int_],
 ) -> npt.NDArray[np.float64]:
     config_node = ensemble.experiment.parameter_configuration[param_group]
-    return config_node.load_parameters(ensemble, param_group, iens_active_index)
+    if isinstance(config_node, ScalarParameters):
+        return config_node.load_parameters_to_update(ensemble, iens_active_index)
+    dataset = config_node.load_parameters(ensemble, param_group, iens_active_index)
+    assert isinstance(dataset, np.ndarray), "dataset is not an numpy array"
+    return dataset
 
 
 def _expand_wildcards(
@@ -608,9 +620,12 @@ def correlation_callback(
         logger.info(log_msg)
         progress_callback(AnalysisStatusEvent(msg=log_msg))
         start = time.time()
-
         _save_param_ensemble_array_to_disk(
-            target_ensemble, param_ensemble_array, param_group, iens_active_index
+            source_ensemble,
+            target_ensemble,
+            param_ensemble_array,
+            param_group,
+            iens_active_index,
         )
         logger.info(
             f"Storing data for {param_group} completed in {(time.time() - start) / 60} minutes"

src/ert/config/__init__.py

@@ -31,6 +31,7 @@
 from .parsing.observations_parser import ObservationType
 from .queue_config import QueueConfig
 from .response_config import InvalidResponseFile, ResponseConfig
+from .scalar_parameter import DataSource, ScalarParameter, ScalarParameters
 from .summary_config import SummaryConfig
 from .summary_observation import SummaryObservation
 from .surface_config import SurfaceConfig
@@ -44,6 +45,7 @@
     "ConfigValidationError",
     "ConfigValidationError",
     "ConfigWarning",
+    "DataSource",
     "DesignMatrix",
     "ESSettings",
     "EnkfObs",
@@ -70,6 +72,8 @@
     "QueueConfig",
     "QueueSystem",
     "ResponseConfig",
+    "ScalarParameter",
+    "ScalarParameters",
     "SummaryConfig",
     "SummaryObservation",
     "SurfaceConfig",

src/ert/config/design_matrix.py

@@ -8,13 +8,18 @@
 import pandas as pd
 from pandas.api.types import is_integer_dtype
 
-from ert.config.gen_kw_config import GenKwConfig, TransformFunctionDefinition
-
 from ._option_dict import option_dict
 from .parsing import ConfigValidationError, ErrorInfo
+from .scalar_parameter import (
+    SCALAR_PARAMETERS_NAME,
+    DataSource,
+    ScalarParameter,
+    ScalarParameters,
+    TransRawSettings,
+)
 
 if TYPE_CHECKING:
-    from ert.config import ParameterConfig
+    pass
 
 DESIGN_MATRIX_GROUP = "DESIGN_MATRIX"
 
@@ -32,7 +37,7 @@ def __post_init__(self) -> None:
             (
                 self.active_realizations,
                 self.design_matrix_df,
-                self.parameter_configuration,
+                self.scalars,
             ) = self.read_design_matrix()
         except (ValueError, AttributeError) as exc:
             raise ConfigValidationError.with_context(
@@ -102,64 +107,57 @@ def merge_with_other(self, dm_other: DesignMatrix) -> None:
         except ValueError as exc:
             errors.append(ErrorInfo(f"Error when merging design matrices {exc}!"))
 
-        for tfd in dm_other.parameter_configuration.transform_function_definitions:
-            self.parameter_configuration.transform_function_definitions.append(tfd)
+        for param in dm_other.scalars:
+            self.scalars.append(param)
 
         if errors:
             raise ConfigValidationError.from_collected(errors)
 
     def merge_with_existing_parameters(
-        self, existing_parameters: list[ParameterConfig]
-    ) -> tuple[list[ParameterConfig], GenKwConfig]:
+        self, existing_scalars: ScalarParameters
+    ) -> ScalarParameters:
         """
         This method merges the design matrix parameters with the existing parameters and
-        returns the new list of existing parameters, wherein we drop GEN_KW group having a full overlap with the design matrix group.
-        GEN_KW group that was dropped will acquire a new name from the design matrix group.
-        Additionally, the ParameterConfig which is the design matrix group is returned separately.
-
+        returns the new list of existing parameters.
         Args:
-            existing_parameters (List[ParameterConfig]): List of existing parameters
+            existing_scalars (ScalarParameters): existing scalar parameters
 
-        Raises:
-            ConfigValidationError: If there is a partial overlap between the design matrix group and any existing GEN_KW group
 
         Returns:
-            tuple[List[ParameterConfig], ParameterConfig]: List of existing parameters and the dedicated design matrix group
+            ScalarParameters: new set of ScalarParameters
         """
 
-        new_param_config: list[ParameterConfig] = []
+        all_params: list[ScalarParameter] = []
 
-        design_parameter_group = self.parameter_configuration
-        design_keys = [e.name for e in design_parameter_group.transform_functions]
-
-        design_group_added = False
-        for parameter_group in existing_parameters:
-            if not isinstance(parameter_group, GenKwConfig):
-                new_param_config += [parameter_group]
+        overlap_set = set()
+        for parameter_sampled in existing_scalars.scalars:
+            if parameter_sampled.input_source == DataSource.DESIGN_MATRIX:
                 continue
-            existing_keys = [e.name for e in parameter_group.transform_functions]
-            if set(existing_keys) == set(design_keys):
-                if design_group_added:
-                    raise ConfigValidationError(
-                        "Multiple overlapping groups with design matrix found in existing parameters!\n"
-                        f"{design_parameter_group.name} and {parameter_group.name}"
-                    )
-
-                design_parameter_group.name = parameter_group.name
-                design_group_added = True
-            elif set(design_keys) & set(existing_keys):
-                raise ConfigValidationError(
-                    "Overlapping parameter names found in design matrix!\n"
-                    f"{DESIGN_MATRIX_GROUP}:{design_keys}\n{parameter_group.name}:{existing_keys}"
-                    "\nThey need to match exactly or not at all."
-                )
-            else:
-                new_param_config += [parameter_group]
-        return new_param_config, design_parameter_group
+            overlap = False
+            for parameter_design in self.scalars:
+                if parameter_sampled.param_name == parameter_design.param_name:
+                    parameter_design.group_name = parameter_sampled.group_name
+                    all_params.append(parameter_design)
+                    overlap = True
+                    overlap_set.add(parameter_sampled.param_name)
+                    break
+            if not overlap:
+                all_params.append(parameter_sampled)
+
+        for parameter_design in self.scalars:
+            if parameter_design.param_name not in overlap_set:
+                all_params.append(parameter_design)
+
+        return ScalarParameters(
+            name=SCALAR_PARAMETERS_NAME,
+            forward_init=False,
+            update=True,
+            scalars=all_params,
+        )
 
     def read_design_matrix(
         self,
-    ) -> tuple[list[bool], pd.DataFrame, GenKwConfig]:
+    ) -> tuple[list[bool], pd.DataFrame, list[ScalarParameter]]:
         # Read the parameter names (first row) as strings to prevent pandas from modifying them.
         # This ensures that duplicate or empty column names are preserved exactly as they appear in the Excel sheet.
         # By doing this, we can properly validate variable names, including detecting duplicates or missing names.
@@ -205,29 +203,25 @@ def read_design_matrix(
 
         design_matrix_df = pd.concat([design_matrix_df, default_df], axis=1)
 
-        transform_function_definitions: list[TransformFunctionDefinition] = []
+        scalars: list[ScalarParameter] = []
         for parameter in design_matrix_df.columns:
-            transform_function_definitions.append(
-                TransformFunctionDefinition(
-                    name=parameter,
-                    param_name="RAW",
-                    values=[],
+            scalars.append(
+                ScalarParameter(
+                    param_name=parameter,
+                    group_name=DESIGN_MATRIX_GROUP,
+                    input_source=DataSource.DESIGN_MATRIX,
+                    distribution=TransRawSettings(),
+                    template_file=None,
+                    output_file=None,
+                    update=False,
                 )
             )
-        parameter_configuration = GenKwConfig(
-            name=DESIGN_MATRIX_GROUP,
-            forward_init=False,
-            template_file=None,
-            output_file=None,
-            transform_function_definitions=transform_function_definitions,
-            update=False,
-        )
 
         reals = design_matrix_df.index.tolist()
         return (
             [x in reals for x in range(max(reals) + 1)],
             design_matrix_df,
-            parameter_configuration,
+            scalars,
         )
 
     @staticmethod

src/ert/config/ensemble_config.py

@@ -16,6 +16,7 @@
 from .parsing import ConfigDict, ConfigKeys, ConfigValidationError
 from .refcase import Refcase
 from .response_config import ResponseConfig
+from .scalar_parameter import ScalarParameters
 from .summary_config import SummaryConfig
 from .surface_config import SurfaceConfig
 
@@ -47,7 +48,8 @@ class EnsembleConfig:
         default_factory=dict
     )
     parameter_configs: dict[
-        str, GenKwConfig | FieldConfig | SurfaceConfig | ExtParamConfig
+        str,
+        GenKwConfig | FieldConfig | SurfaceConfig | ExtParamConfig | ScalarParameters,
     ] = field(default_factory=dict)
     refcase: Refcase | None = None
 
@@ -134,7 +136,7 @@ def make_field(field_list: list[str]) -> FieldConfig:
             return FieldConfig.from_config_list(grid_file_path, dims, field_list)
 
         parameter_configs = (
-            [GenKwConfig.from_config_list(g) for g in gen_kw_list]
+            [ScalarParameters.from_config_list(gen_kw_list)]
             + [SurfaceConfig.from_config_list(s) for s in surface_list]
             + [make_field(f) for f in field_list]
         )
@@ -180,12 +182,12 @@ def hasNodeGenData(self, key: str) -> bool:
         config = self.response_configs["gen_data"]
         return key in config.keys
 
+    # TODO: This might not be needed but it retrieves the group names for genkw config
     def get_keylist_gen_kw(self) -> list[str]:
-        return [
-            val.name
-            for val in self.parameter_configuration
-            if isinstance(val, GenKwConfig)
-        ]
+        for val in self.parameter_configuration:
+            if isinstance(val, ScalarParameters):
+                return list(val.groups.keys())
+        return []
 
     @property
     def parameters(self) -> list[str]: