diff --git a/src/semeio/fmudesign/_excel_to_dict.py b/src/semeio/fmudesign/_excel_to_dict.py index dc800d046..c5f7bf4da 100644 --- a/src/semeio/fmudesign/_excel_to_dict.py +++ b/src/semeio/fmudesign/_excel_to_dict.py @@ -3,21 +3,43 @@ by the DesignMatrix class to generate design matrices. """ -import collections -import contextlib -import math from collections import Counter from collections.abc import Hashable, Sequence from pathlib import Path -from typing import Any, cast +from typing import Any, Literal, cast -import numpy as np import openpyxl import pandas as pd import yaml -from semeio.fmudesign.design_distributions import read_correlations -from semeio.fmudesign.utils import seeds_from_extern +from semeio.fmudesign.general_input import GeneralInput +from semeio.fmudesign.read_background import read_background +from semeio.fmudesign.read_correlations import parse_sensitivity_correlations +from semeio.fmudesign.utils import ( + _has_value, + find_sheet, + resolve_path, + seeds_from_extern, +) + + +def _read_general_input( + input_filename: str, general_input_sheet: str +) -> dict[str, Any]: + general_input = ( + pd.read_excel( + input_filename, + general_input_sheet, + header=None, + engine="openpyxl", + ) + .dropna(axis=0, how="all") + .dropna(axis=1, how="all") + .set_index(0) + .loc[:, 1] + .to_dict() + ) + return {str(k): v for k, v in general_input.items()} def excel_to_dict( @@ -49,29 +71,13 @@ def excel_to_dict( design_input_sheet = find_sheet(design_input_sheet, names=xlsx.sheetnames) default_values_sheet = find_sheet(default_values_sheet, names=xlsx.sheetnames) - generalinput = ( - pd.read_excel( - input_filename, - general_input_sheet, - header=None, - index_col=0, - engine="openpyxl", - ) - .dropna(axis=0, how="all") - .dropna(axis=1, how="all") - .loc[:, 1] - .to_dict() - ) + general_input_dict = _read_general_input(input_filename, general_input_sheet) - if (design_type := generalinput.get("designtype")) != "onebyone": - raise ValueError( - "Generation of DesignMatrix only implemented " - f"for type 'onebyone', not {design_type}" - ) + general_input = GeneralInput.from_dict(general_input_dict, input_filename) return _excel_to_dict_onebyone( input_filename=input_filename, - general_input_sheet=general_input_sheet, + general_input=general_input, design_input_sheet=design_input_sheet, default_values_sheet=default_values_sheet, ) @@ -88,30 +94,6 @@ def inputdict_to_yaml(inputdict: dict[str, Any], filename: str) -> None: yaml.dump(inputdict, stream) -def find_sheet(name: str, names: list[str]) -> str: - """Search for Excel sheets with a soft matching. Raises ValueError if zero - or more than one match is found. - - Examples: - >>> find_sheet('general_input', ['generalinput', 'designinput', 'defaultinput']) - 'generalinput' - >>> find_sheet('variable_input', ['generalinput', 'designinput', 'defaultinput']) - Traceback (most recent call last): - ... - ValueError: No match for variable_input: ['generalinput', 'designinput', 'defaultinput'] - """ # ruff: ignore[line-too-long] - - def sanitize(inputstring: str) -> str: - return inputstring.lower().strip().replace("_", "") - - found = [name_i for name_i in names if sanitize(name) == sanitize(name_i)] - if len(found) > 1: - raise ValueError(f"More than one match for {name}: {found}") - if len(found) == 0: - raise ValueError(f"No match for {name}: {names}") - return found[0] - - def _check_designinput(dsgn_input: pd.DataFrame) -> None: """Checks for valid input in designinput sheet""" # Filter out rows where sensname has no value @@ -150,39 +132,10 @@ def _check_for_mixed_sensitivities(sens_name: str, sens_group: pd.DataFrame) -> ) -def resolve_path(input_filename: str, reference: str | None) -> str | None: - """The path `input_filename` is an Excel sheet, and `reference` is a cell - value that *might* be a reference to another file. Resolve the path to - `reference` and return. If no such file exists, return `reference`. - """ - # The reference is None, so just return it back - if reference is None: - return reference - - # It's a string but not a reference to another file - if not str(reference).endswith(("xlsx", "csv")): - return reference - - # If the reference is e.g. 'C:/Users/USER/files/doe1.xlsx' - reference_path = Path(reference) - if reference_path.is_absolute() and reference_path.exists(): - return str(reference_path.resolve()) - - # If the reference is e.g. 'doe1.xlsx' - full_path = Path(input_filename).parent / reference_path - if full_path.exists(): - return str(full_path.resolve()) - - if reference_path.exists(): - return str(reference_path.resolve()) - - raise ValueError(f"Failed to resolve path for file: {reference}") - - def _excel_to_dict_onebyone( input_filename: str, *, - general_input_sheet: str, + general_input: GeneralInput, design_input_sheet: str, default_values_sheet: str, ) -> dict[str, Any]: @@ -190,103 +143,39 @@ def _excel_to_dict_onebyone( Args: input_filename (str): Name of excel workbook - general_input_sheet (str): name of general input sheet + general_input (GeneralInput): Validated general input design_input_sheet (str): name of design input sheet default_values_sheet (str): name of default value sheet Returns: dict on format for DesignMatrix.generate """ - output: dict[str, Any] = { - "input_file": input_filename - } # This is the config that we read and return - # Read the general input sheet to a dictionary - generalinput = ( - pd.read_excel( - input_filename, - general_input_sheet, - header=None, - engine="openpyxl", - ) - .dropna(axis=0, how="all") - .dropna(axis=1, how="all") - .set_index(0) - .loc[:, 1] - .to_dict() - ) + if isinstance(seeds := general_input.rms_seeds, Path): + rms_seeds: Literal["default"] | list[int] | None = seeds_from_extern(seeds) + else: + rms_seeds = seeds - def parse_value(value: object) -> object: - if pd.isna(value): # type: ignore[call-overload] - return None - if isinstance(value, str): - return value.strip() - return value - - # Convert NaN values to None and strip other values - generalinput = { - str(key).strip(): parse_value(value) for (key, value) in generalinput.items() - } - - # Check that there are no wrong keys or typos, e.g. 'repets' - ALLOWED_KEYS = { - "designtype", - "repeats", - "correlation_iterations", - "distribution_seed", - "seed_strategy", - "rms_seeds", - "background", - } - extra_keys = set(generalinput.keys()) - set(ALLOWED_KEYS) - if extra_keys: - msg = ( - "In the general input sheet, the following parameter(s) are not" - f"recognized and cannot be parsed:\n{extra_keys!r}\n" - f"Allowed keys:{ALLOWED_KEYS!r}" - ) - raise LookupError(msg) - - # Copy keys over if they exist - keys = [ - "designtype", - "repeats", - "correlation_iterations", - "distribution_seed", - "seed_strategy", - ] - for key in keys: - if key not in generalinput: - continue - output[key] = generalinput[key] - - # Copy the 'rms_seeds' key over. It is called 'seeds' further down in - # the code for historical reasons. - key = "seeds" - with contextlib.suppress(KeyError): - output[key] = generalinput["rms_seeds"] - - # If 'seeds' / 'rms_seed' is a file, then read it - if key in output: - maybe_path = resolve_path(input_filename, output[key]) - if isinstance(maybe_path, str) and Path(maybe_path).exists(): - output[key] = seeds_from_extern(maybe_path) - - # The 'background' key is either blank/'None', a reference to another file - # or the name of a sheet in this workbook. - key = "background" - value = generalinput.get(key) - background = "" if value is None else str(value).strip() - if background.lower() in {"", "none"}: - output[key] = None - elif background.endswith(("csv", "xlsx")): - output[key] = {"extern": resolve_path(input_filename, background)} + if isinstance(bgr := general_input.background, Path): + background = {"extern": str(bgr)} + elif isinstance(bgr, str): + background = read_background(input_filename, bgr) else: - output[key] = _read_background(input_filename, background) + background = None - output["defaultvalues"] = _read_defaultvalues(input_filename, default_values_sheet) + output: dict[str, Any] = { + "input_file": input_filename, + "designtype": general_input.designtype, + "repeats": general_input.repeats, + "distribution_seed": general_input.distribution_seed, + "background": background, + "seeds": rms_seeds, + "correlation_iterations": general_input.correlation_iterations, + "seed_strategy": general_input.seed_strategy, + "defaultvalues": _read_defaultvalues(input_filename, default_values_sheet), + "sensitivities": {}, + } # This is the config that we read and return - output["sensitivities"] = {} designinput = ( pd.read_excel(input_filename, design_input_sheet, engine="openpyxl") .dropna(axis=0, how="all") @@ -343,7 +232,9 @@ def parse_value(value: object) -> object: sensdict["parameters"] = _read_dist_sensitivity(group) sensdict["correlations"] = None if "corr_sheet" in group: - sensdict["correlations"] = _read_correlations(group, input_filename) + sensdict["correlations"] = parse_sensitivity_correlations( + group, input_filename + ) elif sens_type == "extern": sensdict["extern_file"] = resolve_path( @@ -457,107 +348,6 @@ def _read_dependencies( return depend_dict -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"] = _read_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 - - def _read_scenario_sensitivity(sensgroup: pd.DataFrame) -> dict[str, Any]: """Reads parameters and values for scenario sensitivities @@ -704,77 +494,6 @@ def _read_dist_sensitivity(sensgroup: pd.DataFrame) -> dict[str, Any]: return paramdict -def _read_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 = collections.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 - - -def _has_value(value: Any) -> bool: # ruff: ignore[any-type] - """Returns False only if the argument is np.nan""" - try: - return not np.isnan(value) - except TypeError: - return True - - -def _is_int(teststring: str) -> bool: - """Test if string is a finite integer""" - try: - if not np.isnan(int(teststring)): - return math.isclose((float(teststring) % 1), 0, abs_tol=1e-14) - return False # It was a "number", but it was NaN. - except ValueError: - return False - - def _raise_if_duplicates(container: Sequence[Hashable]) -> None: """Raises a descriptive error if there are duplicates in the container.""" duplicates = {k: v for (k, v) in Counter(container).items() if v > 1} diff --git a/src/semeio/fmudesign/fmudesignrunner.py b/src/semeio/fmudesign/fmudesignrunner.py index 83345e2fe..dce105e7a 100644 --- a/src/semeio/fmudesign/fmudesignrunner.py +++ b/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( diff --git a/src/semeio/fmudesign/general_input.py b/src/semeio/fmudesign/general_input.py new file mode 100644 index 000000000..c528b8195 --- /dev/null +++ b/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) diff --git a/src/semeio/fmudesign/read_background.py b/src/semeio/fmudesign/read_background.py new file mode 100644 index 000000000..7594eae41 --- /dev/null +++ b/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 diff --git a/src/semeio/fmudesign/read_correlations.py b/src/semeio/fmudesign/read_correlations.py new file mode 100644 index 000000000..e6b148f2b --- /dev/null +++ b/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 diff --git a/src/semeio/fmudesign/utils.py b/src/semeio/fmudesign/utils.py index f35ab2063..57420474f 100644 --- a/src/semeio/fmudesign/utils.py +++ b/src/semeio/fmudesign/utils.py @@ -2,8 +2,11 @@ Module for utility functions that do not belong elsewhere. """ +import math +from pathlib import Path from typing import Any +import numpy as np import pandas as pd @@ -32,7 +35,7 @@ def parameters_from_extern(filename: str) -> pd.DataFrame: ) -def seeds_from_extern(filename: str) -> list[int]: +def seeds_from_extern(filename: Path | str) -> list[int]: """Read parameter values or background values from specified file. Format either Excel ('xlsx') or csv. @@ -214,3 +217,74 @@ def map_dependencies( print(f" {from_} => {to_}") return df + + +def find_sheet(name: str, names: list[str]) -> str: + """Search for Excel sheets with a soft matching. Raises ValueError if zero + or more than one match is found. + + Examples: + >>> find_sheet('general_input', ['generalinput', 'designinput', 'defaultinput']) + 'generalinput' + >>> find_sheet('variable_input', ['generalinput', 'designinput', 'defaultinput']) + Traceback (most recent call last): + ... + ValueError: No match for variable_input: ['generalinput', 'designinput', 'defaultinput'] + """ # ruff: ignore[line-too-long] + + def sanitize(inputstring: str) -> str: + return inputstring.lower().strip().replace("_", "") + + found = [name_i for name_i in names if sanitize(name) == sanitize(name_i)] + if len(found) > 1: + raise ValueError(f"More than one match for {name}: {found}") + if len(found) == 0: + raise ValueError(f"No match for {name}: {names}") + return found[0] + + +def _has_value(value: Any) -> bool: # ruff: ignore[any-type] + """Returns False only if the argument is np.nan""" + try: + return not np.isnan(value) + except TypeError: + return True + + +def _is_int(teststring: str) -> bool: + """Test if string is a finite integer""" + try: + if not np.isnan(int(teststring)): + return math.isclose((float(teststring) % 1), 0, abs_tol=1e-14) + return False # It was a "number", but it was NaN. + except ValueError: + return False + + +def resolve_path(input_filename: str, reference: str | None) -> str | None: + """The path `input_filename` is an Excel sheet, and `reference` is a cell + value that *might* be a reference to another file. Resolve the path to + `reference` and return. If no such file exists, return `reference`. + """ + # The reference is None, so just return it back + if reference is None: + return reference + + # It's a string but not a reference to another file + if not str(reference).endswith(("xlsx", "csv")): + return reference + + # If the reference is e.g. 'C:/Users/USER/files/doe1.xlsx' + reference_path = Path(reference) + if reference_path.is_absolute() and reference_path.exists(): + return str(reference_path.resolve()) + + # If the reference is e.g. 'doe1.xlsx' + full_path = Path(input_filename).parent / reference_path + if full_path.exists(): + return str(full_path.resolve()) + + if reference_path.exists(): + return str(reference_path.resolve()) + + raise ValueError(f"Failed to resolve path for file: {reference}") diff --git a/tests/fmudesign/data/config/design_input_onebyone.xlsx b/tests/fmudesign/data/config/design_input_onebyone.xlsx index 7331fc219..6bda60b2e 100644 Binary files a/tests/fmudesign/data/config/design_input_onebyone.xlsx and b/tests/fmudesign/data/config/design_input_onebyone.xlsx differ diff --git a/tests/fmudesign/test_general_input.py b/tests/fmudesign/test_general_input.py new file mode 100644 index 000000000..489933dd2 --- /dev/null +++ b/tests/fmudesign/test_general_input.py @@ -0,0 +1,421 @@ +from pathlib import Path +from types import NoneType +from typing import Any + +import hypothesis.strategies as st +import numpy as np +import pandas as pd +import pytest +from hypothesis import assume, given +from pydantic import TypeAdapter, ValidationError + +from semeio.fmudesign._excel_to_dict import GeneralInput +from semeio.fmudesign.config_validation import SeedStrategy +from semeio.fmudesign.general_input import parse_value + +PYDANTIC_PATH_ERROR = "Path does not point to a file|Input is not a valid path" + + +def base_general_input_dict(): + return { + "designtype": "onebyone", + "repeats": 10, + "distribution_seed": None, + "rms_seeds": None, + "correlation_iterations": 1, + "seed_strategy": SeedStrategy.JOINT, + "background": None, + } + + +ANY_TYPE = st.one_of( + st.integers(), + st.floats(allow_nan=False), + st.text(), + st.booleans(), + st.none(), + st.lists(st.integers(), min_size=1), + st.dictionaries(st.text(), st.integers() | st.text(), min_size=1), +) + + +_int_adapter = TypeAdapter(int) +_float_adapter = TypeAdapter(float) + + +def is_pydantic_numeric(x): + for adapter in (_int_adapter, _float_adapter): + try: + adapter.validate_python(x) + return True + except ValidationError: + pass + return False + + +NON_NUMERIC = ANY_TYPE.filter(lambda x: not is_pydantic_numeric(x)) + + +BOOLEAN_ERROR = "cannot have boolean value" + + +@pytest.mark.parametrize( + "nan_value", + [ + float("nan"), + np.nan, + np.float64("nan"), + pd.NaT, + None, + ], +) +def test_that_parse_value_converts_nan_formats_to_none(nan_value): + assert parse_value(nan_value) is None + + +@pytest.mark.parametrize( + "required_key", + (key for key, info in GeneralInput.model_fields.items() if info.is_required()), +) +def test_that_missing_required_keys_raises_validation_error(required_key): + general_input_dict = base_general_input_dict() + general_input_dict.pop(required_key) + with pytest.raises(ValidationError): + GeneralInput.from_dict(general_input_dict) + + +@pytest.mark.parametrize( + "optional_key", + (key for key, info in GeneralInput.model_fields.items() if not info.is_required()), +) +def test_that_missing_optional_keys_does_not_raise_validation_error(optional_key): + general_input_dict = base_general_input_dict() + general_input_dict.pop(optional_key) + GeneralInput.from_dict(general_input_dict) + + +def test_that_extra_key_raises_value_error(): + extra = {"extra_key": "foo"} + with pytest.raises(ValidationError, match="Extra inputs are not permitted"): + GeneralInput( + designtype="onebyone", + repeats=1, + distribution_seed=42, + rms_seeds="default", + **extra, + ) + + +def test_that_whitespace_around_keys_is_stripped(): + general_input_dict = base_general_input_dict() + general_input_dict[" repeats "] = general_input_dict.pop("repeats") + result = GeneralInput.from_dict(general_input_dict) + assert result.repeats == 10 + + +def test_that_designtype_onebyone_is_accepted(): + result = GeneralInput.from_dict(base_general_input_dict()) + assert result.designtype == "onebyone" + + +@given(ANY_TYPE) +def test_that_other_design_types_than_onebyone_raises_validation_error(text): + assume(text != "onebyone") + general_input_dict = base_general_input_dict() | {"designtype": text} + with pytest.raises(ValueError, match=f"Input should be 'onebyone'|{BOOLEAN_ERROR}"): + GeneralInput.from_dict(general_input_dict) + + +@given(st.integers(min_value=1, max_value=10000)) +def test_that_positive_int_repeats_is_accepted(positive_int): + general_input_dict = base_general_input_dict() | {"repeats": positive_int} + result = GeneralInput.from_dict(general_input_dict) + assert result.repeats == positive_int + + +def test_that_zero_repeats_raises_validation_error(): + general_input_dict = base_general_input_dict() | {"repeats": 0} + with pytest.raises(ValidationError, match="repeats"): + GeneralInput.from_dict(general_input_dict) + + +def test_that_negative_repeats_raises_validation_error(): + general_input_dict = base_general_input_dict() | {"repeats": -1} + with pytest.raises(ValidationError): + GeneralInput.from_dict(general_input_dict) + + +@given(NON_NUMERIC) +def test_that_non_integer_repeats_raises_validation_error(non_int): + general_input_dict = base_general_input_dict() | {"repeats": non_int} + with pytest.raises( + ValidationError, + match=r"Input should be greater than 0|Input should be a valid integer", + ): + GeneralInput.from_dict(general_input_dict) + + +@pytest.mark.parametrize("bool_", [True, False]) +def test_that_true_false_are_rejected_for_repeats(bool_): + input_dict = base_general_input_dict() | {"repeats": bool_} + with pytest.raises( + ValueError, + match=BOOLEAN_ERROR, + ): + GeneralInput.from_dict(input_dict) + + +@given(st.integers(min_value=0)) +def test_that_non_negative_distribution_seed_is_accepted(value): + general_input_dict = base_general_input_dict() | {"distribution_seed": value} + result = GeneralInput.from_dict(general_input_dict) + assert result.distribution_seed == value + + +def test_that_none_distribution_seed_is_accepted(): + general_input_dict = base_general_input_dict() | {"distribution_seed": None} + result = GeneralInput.from_dict(general_input_dict) + assert result.distribution_seed is None + + +@given(st.integers(max_value=-1)) +def test_that_negative_distribution_seed_raises_validation_error(value): + general_input_dict = base_general_input_dict() | {"distribution_seed": value} + with pytest.raises(ValidationError): + GeneralInput.from_dict(general_input_dict) + + +@given(NON_NUMERIC.filter(lambda x: not isinstance(x, NoneType))) +def test_that_invalid_distribution_seed_types_raises_validation_error( + invalid_distribution_seed, +): + general_input_dict = base_general_input_dict() | { + "distribution_seed": invalid_distribution_seed + } + with pytest.raises(ValidationError, match="Input should be a valid integer"): + GeneralInput.from_dict(general_input_dict) + + +@pytest.mark.parametrize("bool_", [True, False]) +def test_that_true_false_are_rejected_for_distribution_seed(bool_): + input_dict = base_general_input_dict() | {"distribution_seed": bool_} + with pytest.raises( + ValueError, + match=BOOLEAN_ERROR, + ): + GeneralInput.from_dict(input_dict) + + +def test_that_rms_seeds_none_is_accepted(): + general_input_dict = base_general_input_dict() | {"rms_seeds": None} + result = GeneralInput.from_dict(general_input_dict) + assert result.rms_seeds is None + + +def test_that_rms_seeds_default_is_accepted(): + general_input_dict = base_general_input_dict() | {"rms_seeds": "default"} + result = GeneralInput.from_dict(general_input_dict) + assert result.rms_seeds == "default" + + +def test_that_rms_seeds_existing_file_is_accepted(use_tmpdir): + seeds_file_name = "seeds.csv" + Path(seeds_file_name).touch() + general_input_dict = base_general_input_dict() | { + "rms_seeds": seeds_file_name, + } + result = GeneralInput.from_dict(general_input_dict) + assert result.rms_seeds == Path(seeds_file_name).resolve() + + +def test_that_rms_seeds_non_existing_file_raises(use_tmpdir): + seeds_file_name = "seeds.csv" + general_input_dict = base_general_input_dict() | { + "rms_seeds": seeds_file_name, + } + with pytest.raises(ValueError, match=r"Failed to resolve path for file: seeds.csv"): + GeneralInput.from_dict(general_input_dict) + + +def test_that_rms_seeds_from_extern_csv_file_in_subdirectory_is_accepted(use_tmpdir): + subdir = Path("subdir") + subdir.mkdir() + seeds_file_name = "seeds.csv" + (subdir / seeds_file_name).touch() + general_input_dict = base_general_input_dict() | { + "rms_seeds": str(seeds_file_name), + } + result = GeneralInput.from_dict( + general_input_dict, input_filename=str(subdir / "input.xlsx") + ) + assert result.rms_seeds == (subdir / seeds_file_name).resolve() + + +@given(ANY_TYPE.filter(lambda x: not isinstance(x, list | NoneType))) +def test_that_invalid_rms_seeds_types_raises_validation_error(invalid_rms_seeds): + assume(invalid_rms_seeds != "default") + general_input_dict = base_general_input_dict() | {"rms_seeds": invalid_rms_seeds} + with pytest.raises( + ValueError, + match=f"Path does not point to a file" + f"|Input is not a valid path" + f"|{BOOLEAN_ERROR}", + ): + GeneralInput.from_dict(general_input_dict) + + +def test_that_correlation_iterations_defaults_to_zero(): + general_input_dict = base_general_input_dict() + general_input_dict.pop("correlation_iterations") + result = GeneralInput.from_dict(general_input_dict) + assert result.correlation_iterations == 0 + + +@given(st.integers(min_value=0)) +def test_that_non_negative_correlation_iterations_is_accepted(value): + general_input_dict = base_general_input_dict() | { + "correlation_iterations": value, + } + result = GeneralInput.from_dict(general_input_dict) + assert result.correlation_iterations == value + + +@given(NON_NUMERIC.filter(lambda x: not isinstance(x, NoneType))) +def test_that_invalid_correlation_iterations_raises_validation_error( + invalid_correlation_iterations, +): + general_input_dict = base_general_input_dict() | { + "correlation_iterations": invalid_correlation_iterations, + } + with pytest.raises(ValidationError): + GeneralInput.from_dict(general_input_dict) + + +def test_that_negative_correlation_iterations_raises_validation_error(): + general_input_dict = base_general_input_dict() | { + "correlation_iterations": -1, + } + with pytest.raises(ValidationError): + GeneralInput.from_dict(general_input_dict) + + +@pytest.mark.parametrize("bool_", [True, False]) +def test_that_true_false_are_rejected_for_correlation_iterations(bool_): + input_dict = base_general_input_dict() | {"correlation_iterations": bool_} + with pytest.raises( + ValueError, + match=BOOLEAN_ERROR, + ): + GeneralInput.from_dict(input_dict) + + +def test_that_seed_strategy_defaults_to_joint(): + general_input_dict = base_general_input_dict() + general_input_dict.pop("seed_strategy") + result = GeneralInput.from_dict(general_input_dict) + assert result.seed_strategy == SeedStrategy.JOINT + + +def test_that_seed_strategy_defaults_to_string_type(): + general_input_dict = base_general_input_dict() + general_input_dict.pop("seed_strategy") + result = GeneralInput.from_dict(general_input_dict) + assert type(result.seed_strategy) is str + + +@pytest.mark.parametrize("strategy", list(SeedStrategy)) +def test_that_valid_seed_strategies_are_accepted(strategy): + general_input_dict = base_general_input_dict() | {"seed_strategy": strategy.value} + result = GeneralInput.from_dict(general_input_dict) + assert result.seed_strategy == strategy + + +def _not_seed_strategy_or_none_str(x: Any): + return not isinstance(x, str) or x.lower() not in {*SeedStrategy, "none"} + + +@given(ANY_TYPE.filter(_not_seed_strategy_or_none_str).filter(lambda x: x is not None)) +def test_that_invalid_seed_strategy_raises_validation_error(invalid_seed_strategy): + general_input_dict = base_general_input_dict() | { + "seed_strategy": invalid_seed_strategy + } + seed_strategy_error = "Input should be 'joint' or 'independent'" + with pytest.raises(ValueError, match=f"{BOOLEAN_ERROR}|{seed_strategy_error}"): + GeneralInput.from_dict(general_input_dict) + + +def test_that_seed_strategy_is_serialized_as_str(): + gi = GeneralInput.from_dict(base_general_input_dict()) + assert type(gi.seed_strategy) is str + + +def test_that_background_none_is_accepted(): + general_input_dict = base_general_input_dict() | {"background": None} + result = GeneralInput.from_dict(general_input_dict) + assert result.background is None + + +@pytest.mark.parametrize("value", ["None", "none", "NONE"]) +def test_that_background_none_like_strings_are_treated_as_none(value): + general_input_dict = base_general_input_dict() | {"background": value} + result = GeneralInput.from_dict(general_input_dict) + assert result.background is None + + +def test_that_existing_background_csv_path_is_accepted(use_tmpdir): + bg_file = "background.csv" + Path(bg_file).touch() + general_input_dict = base_general_input_dict() | { + "background": bg_file, + } + result = GeneralInput.from_dict(general_input_dict) + assert result.background == Path(bg_file).resolve() + + +def test_that_non_existing_background_csv_path_is_accepted_as_str(): + """Background is either file (Path) or excel sheet name (str). + The validation should allow strings as it may be a valid background sheet. + Whether this is the case is responsibility outside of the scope of GeneralInput.""" + bg_file = "background" + general_input_dict = base_general_input_dict() | { + "background": bg_file, + } + result = GeneralInput.from_dict(general_input_dict) + assert isinstance(result.background, str) + assert result.background == bg_file + + +def test_that_background_file_with_rel_path_to_input_file_is_resolved_to_path_from_cwd( + use_tmpdir, +): + """Tests that background file with relative path to input_filename is resolved + to path.""" + subdir = Path("subdir") + subdir.mkdir() + input_file = subdir / "input.csv" + bg_file = "background.csv" + (subdir / bg_file).touch() + result = GeneralInput.from_dict( + base_general_input_dict() + | { + "background": bg_file, + }, + input_filename=str(input_file), + ) + assert result.background == Path(subdir / bg_file).resolve() + + +def test_that_background_path_is_serialized_as_string(): + bg_file = "background.csv" + Path(bg_file).touch() + general_input_dict = base_general_input_dict() | { + "background": bg_file, + } + gi = GeneralInput.from_dict(general_input_dict) + assert isinstance(gi.background, Path) + assert isinstance(gi.model_dump()["background"], str) + + +def test_that_serialize_paths_doesnt_fail_given_none(): + gi = GeneralInput.from_dict(base_general_input_dict()) + assert gi.model_dump()["background"] is None