Add validation of parsed general input

src/semeio/fmudesign/fmudesignrunner.py

@@ -27,6 +27,7 @@
 from pathlib import Path
 
 from packaging.version import Version
+from pydantic import ValidationError
 
 import semeio
 from semeio.fmudesign import DesignMatrix, excel_to_dict
@@ -312,6 +313,7 @@ def main() -> None:
     """semeio.fmudesign is a command line utility for generating design matrices
 
     Wrapper for the the semeio.fmudesign module"""
+
     warnings.filterwarnings("ignore", category=DeprecationWarning)
     warnings.filterwarnings("ignore", category=FutureWarning)
 
@@ -329,18 +331,27 @@ def main() -> None:
         parser.print_help()
         sys.exit(0)
 
+    err_guide_msg = (
+        "\n \n"
+        "fmudesign failed. Read the error message above and fix the input file.\n"
+        " - Documentation:           https://equinor.github.io/fmu-tools/fmudesign.html\n"
+        " - Course docs:             https://fmu-docs.equinor.com/docs/fmu-coursedocs/fmu-howto/sensitivities/index.html \n"  # ruff: ignore[line-too-long]
+        " - Issues/feature requests: https://github.com/equinor/semeio/issues\n"
+        "If you believe this error is a bug or are unable to fix it, create an issue or contact the scout team \n"  # ruff: ignore[line-too-long]
+    )
     try:
         args.func(args)
+    except ValidationError as e:
+        for err in e.errors(include_url=False):
+            print(
+                f"Validation error for '{err['loc'][0]}': "
+                f"{err['msg']}, was '{err['input']}'"
+            )
+        print(err_guide_msg)
+        sys.exit(1)
     except Exception:  # ruff: ignore[blind-except]
         traceback.print_exc()
-        print(
-            "\n \n",
-            "fmudesign failed. Read the error message above and fix the input file.\n",
-            " - Documentation:           https://equinor.github.io/fmu-tools/fmudesign.html\n",
-            " - Course docs:             https://fmu-docs.equinor.com/docs/fmu-coursedocs/fmu-howto/sensitivities/index.html \n",  # ruff: ignore[line-too-long]
-            " - Issues/feature requests: https://github.com/equinor/semeio/issues\n",
-            "If you believe this error is a bug or are unable to fix it, create an issue or contact the scout team \n",  # ruff: ignore[line-too-long]
-        )
+        print(err_guide_msg)
         sys.exit(1)  # Exit with a non-zero status code (required for smoke tests!)
 
     print(

src/semeio/fmudesign/general_input.py

@@ -0,0 +1,88 @@
+from collections.abc import Collection
+from pathlib import Path
+from typing import Any, Literal, Self
+
+import pandas as pd
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    FilePath,
+    NonNegativeInt,
+    PositiveInt,
+    field_serializer,
+)
+
+from semeio.fmudesign.config_validation import SeedStrategy
+from semeio.fmudesign.utils import resolve_path
+
+
+def parse_value(value: object) -> object:
+    if isinstance(value, str):
+        return value.strip()
+    # pd.isna(Collection) -> NDArray, which is ambiguous
+    if not isinstance(value, Collection) and pd.isna(value):  # type: ignore[call-overload]
+        return None
+    return value
+
+
+class GeneralInput(BaseModel):
+    designtype: Literal["onebyone"]
+    repeats: PositiveInt
+    distribution_seed: NonNegativeInt | None
+    rms_seeds: FilePath | Literal["default"] | None
+    correlation_iterations: NonNegativeInt = 0
+    seed_strategy: SeedStrategy = Field(
+        default=SeedStrategy.JOINT, validate_default=True
+    )
+    background: FilePath | str | None = None
+
+    model_config = ConfigDict(extra="forbid", use_enum_values=True)
+
+    @classmethod
+    def from_dict(cls, inputdict: dict[str, Any], input_filename: str = "") -> Self:
+        general_input: dict[str, Any] = {
+            str(key).strip(): parse_value(value) for key, value in inputdict.items()
+        }
+
+        # Boolean values are interpreted as valid ints, and are not caught by
+        # pydantic's validation. It can be caught using strict=True, but that
+        # removes the flexibility of allowing numeric strings for numeric fields.
+        # No values should be boolean, so we check for all.
+        for key, value in general_input.items():
+            if isinstance(value, bool):
+                raise ValueError(
+                    f"key '{key}' cannot have boolean value, got '{value}'"
+                )
+
+        for key in ["seed_strategy", "correlation_iterations"]:
+            val = general_input.get(key)
+            is_none = general_input.get(key) is None
+            is_none_str = isinstance(val, str) and val.lower() == "none"
+            if is_none or is_none_str:
+                print(
+                    f"'{key}' not set in general input sheet. "
+                    f"Setting to default "
+                    f"{GeneralInput.model_fields[key].default}."
+                )
+                general_input.pop(key, None)
+
+        for key in ["rms_seeds", "background"]:
+            val = general_input.get(key)
+            if isinstance(val, str):
+                resolved = resolve_path(input_filename, val)
+                assert isinstance(resolved, str)
+                if Path(resolved).exists():
+                    general_input[key] = Path(resolved)
+                elif resolved.lower() == "none":
+                    general_input[key] = None
+                else:
+                    general_input[key] = resolved
+
+        return cls(**general_input)
+
+    @field_serializer("rms_seeds", "background")
+    def serialize_paths(self, field: Path | None) -> str | None:  # ruff: ignore[no-self-use]
+        if field is None:
+            return None
+        return str(field)

src/semeio/fmudesign/read_background.py

@@ -0,0 +1,107 @@
+from typing import Any
+
+import pandas as pd
+
+from semeio.fmudesign.read_correlations import parse_sensitivity_correlations
+from semeio.fmudesign.utils import _has_value, _is_int, find_sheet
+
+
+def read_background(inp_filename: str, bck_sheet: str) -> dict[str, Any]:
+    """Reads excel sheet with background parameters and distributions
+
+    Args:
+        inp_filename (str): name of Excel workbook
+        bck_sheet (str): name of sheet with background parameters
+
+    Returns:
+        dict with parameter names and distributions
+    """
+    backdict: dict[str, Any] = {}
+    paramdict: dict[str, Any] = {}
+    with pd.ExcelFile(inp_filename, engine="openpyxl") as workbook:
+        sheet_names = [str(name) for name in workbook.sheet_names]
+    try:
+        bck_sheet = find_sheet(bck_sheet, names=sheet_names)
+    except ValueError as err:
+        raise ValueError(
+            f"Sheet {bck_sheet!r} with background parameters, specified in the "
+            f"general input sheet, was not found in {inp_filename!r}.\n"
+            f"Sheets in workbook: {sheet_names}\n"
+            "Use 'None' as background in the general input sheet if no "
+            "background parameters are wanted."
+        ) from err
+    bck_input = (
+        pd.read_excel(inp_filename, bck_sheet, engine="openpyxl")
+        .dropna(axis=0, how="all")
+        .loc[:, lambda df: ~df.columns.astype(str).str.contains("^Unnamed")]
+    )
+
+    backdict["correlations"] = None
+    if "corr_sheet" in bck_input:
+        backdict["correlations"] = parse_sensitivity_correlations(
+            bck_input, inp_filename, group_description=f"background sheet {bck_sheet!r}"
+        )
+
+    if "dist_param1" not in bck_input.columns.to_numpy():
+        bck_input["dist_param1"] = float("NaN")
+    if "dist_param2" not in bck_input.columns.to_numpy():
+        bck_input["dist_param2"] = float("NaN")
+    if "dist_param3" not in bck_input.columns.to_numpy():
+        bck_input["dist_param3"] = float("NaN")
+    if "dist_param4" not in bck_input.columns.to_numpy():
+        bck_input["dist_param4"] = float("NaN")
+
+    for row in bck_input.itertuples():
+        if not _has_value(row.param_name):
+            raise ValueError(
+                "Background parameters specified "
+                "where one line has empty parameter "
+                "name "
+            )
+        if not _has_value(row.dist_param1):
+            raise ValueError(
+                f"Parameter {row.param_name} has been input "
+                "in background sheet but with empty "
+                "first distribution parameter "
+            )
+        if not _has_value(row.dist_param2) and _has_value(row.dist_param3):
+            raise ValueError(
+                f"Parameter {row.param_name} has been input in "
+                "background sheet with "
+                'value for "dist_param3" while '
+                '"dist_param2" is empty. This is not '
+                "allowed"
+            )
+        if not _has_value(row.dist_param3) and _has_value(row.dist_param4):
+            raise ValueError(
+                f"Parameter {row.param_name} has been input in "
+                "background sheet with "
+                'value for "dist_param4" while '
+                '"dist_param3" is empty. This is not '
+                "allowed"
+            )
+        distparams = [
+            item
+            for item in [
+                row.dist_param1,
+                row.dist_param2,
+                row.dist_param3,
+                row.dist_param4,
+            ]
+            if _has_value(item)
+        ]
+        if "corr_sheet" in bck_input:
+            corrsheet = None if not _has_value(row.corr_sheet) else row.corr_sheet
+        else:
+            corrsheet = None
+        paramdict[str(row.param_name)] = [str(row.dist_name), distparams, corrsheet]
+    backdict["parameters"] = paramdict
+
+    if "decimals" in bck_input:
+        decimals: dict[str, Any] = {}
+        for row in bck_input.itertuples():
+            if _has_value(row.decimals) and _is_int(row.decimals):  # type: ignore[arg-type]
+                decimals[row.param_name] = int(row.decimals)  # type: ignore[arg-type, index]
+        backdict["decimals"] = decimals
+
+    return backdict

src/semeio/fmudesign/read_correlations.py

@@ -0,0 +1,60 @@
+from collections import defaultdict
+from typing import Any
+
+import pandas as pd
+
+from semeio.fmudesign.design_distributions import read_correlations
+from semeio.fmudesign.utils import _has_value
+
+
+def parse_sensitivity_correlations(
+    sensgroup: pd.DataFrame, inputfile: str, group_description: str | None = None
+) -> dict[str, Any] | None:
+    """Parse correlation information from a sensitivity group.
+
+    Args:
+        sensgroup: rows describing the parameters, either a sensitivity group
+            from the designinput sheet or the background sheet.
+        inputfile: name of the Excel workbook holding the correlation sheets.
+        group_description: how to refer to `sensgroup` in error messages.
+            Defaults to the sensname of the group.
+    """
+
+    # No correlation sheet column exists
+    if "corr_sheet" not in sensgroup.columns:
+        return None
+
+    # The column exists, but it is all blank
+    if sensgroup["corr_sheet"].dropna().empty:
+        return None
+
+    if group_description is None:
+        group_description = f"sensitivity group {sensgroup['sensname'].iloc[0]!r}"
+
+    correlations: dict[str, Any] = {"inputfile": inputfile}
+
+    # Create a mapping 'corr_to_params' like:
+    # {'corr1': ['var_A', 'var_B', ...], ...}
+    corr_to_params = defaultdict(list)
+    for _, row in sensgroup.iterrows():
+        if not _has_value(row["corr_sheet"]):
+            continue
+        corr_to_params[row["corr_sheet"]].append(row["param_name"])
+
+    # Open the correlation sheet and peek at it
+    # We want to verify that if variables ['A', 'B'] point to the corr sheet,
+    # then exactly those variables are also defined in the sheet
+    for corr_sheet, parameters in corr_to_params.items():
+        df_corr = read_correlations(excel_filename=inputfile, corr_sheet=corr_sheet)
+        if set(df_corr.columns) != set(parameters):
+            msg = f"Mismatch between parameters in {group_description} "
+            msg += f"pointing to\ncorrelation sheet {corr_sheet!r} and "
+            msg += "parameters specified in that correlation sheet.\n"
+            msg += f"Parameters in {group_description}: {sorted(set(parameters))}\n"
+            msg += f"Parameters in correlation sheet: {sorted(set(df_corr.columns))}\n"
+            msg += "These parameters must be specified one-to-one."
+            raise ValueError(msg)
+
+    correlations["sheetnames"] = list(set(corr_to_params.keys()))
+
+    return correlations