easyvvuq.params_specification

  1"""Data structures to ensure consistency during serialization for databases.
  2
  3"""
  4import logging
  5import cerberus
  6import json
  7import numpy
  8from .utils.discrete_validation import is_integer_valued, convert_to_integer
  9
 10__copyright__ = """
 11
 12    Copyright 2018 Robin A. Richardson, David W. Wright
 13
 14    This file is part of EasyVVUQ
 15
 16    EasyVVUQ is free software: you can redistribute it and/or modify
 17    it under the terms of the Lesser GNU General Public License as published by
 18    the Free Software Foundation, either version 3 of the License, or
 19    (at your option) any later version.
 20
 21    EasyVVUQ is distributed in the hope that it will be useful,
 22    but WITHOUT ANY WARRANTY; without even the implied warranty of
 23    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 24    Lesser GNU General Public License for more details.
 25
 26    You should have received a copy of the Lesser GNU General Public License
 27    along with this program.  If not, see <https://www.gnu.org/licenses/>.
 28
 29"""
 30__license__ = "LGPL"
 31
 32logger = logging.getLogger(__name__)
 33
 34
 35class EasyVVUQValidator(cerberus.Validator):
 36    def __init__(self, *args, **kwargs):
 37        super(EasyVVUQValidator, self).__init__(*args, **kwargs)
 38
 39        # Add numpy.int64 as an acceptable 'integer' type
 40        integer_type = cerberus.TypeDefinition('integer', (int, numpy.int64), ())
 41        cerberus.Validator.types_mapping['integer'] = integer_type
 42
 43        # Add 'fixture' type (for now, it's expected just to be a string)
 44        fixture_type = cerberus.TypeDefinition('fixture', (str), ())
 45        cerberus.Validator.types_mapping['fixture'] = fixture_type
 46
 47    def _validate_type_integer(self, value):
 48        """
 49        Enhanced integer validation that handles discrete distributions.
 50        
 51        This method allows float values that represent integers (e.g., 2.0)
 52        to pass validation for integer parameters. This is necessary because
 53        chaospy returns float arrays for discrete distributions when mixed
 54        with continuous distributions.
 55        """
 56        # First check if it's already an integer type
 57        if isinstance(value, (int, numpy.int64)):
 58            return True
 59        
 60        # Check if it's a float that represents an integer
 61        if is_integer_valued(value):
 62            return True
 63        
 64        # Fall back to standard validation
 65        return super()._validate_type_integer(value)
 66
 67
 68class ParamsSpecification:
 69
 70    def __init__(self, params, appname=None):
 71
 72        if not isinstance(params, dict):
 73            msg = "params must be of type 'dict'"
 74            logger.error(msg)
 75            raise Exception(msg)
 76
 77        if not params:
 78            msg = ("params must not be empty. At least one parameter "
 79                   "should be specified.")
 80            logger.error(msg)
 81            raise Exception(msg)
 82
 83        # Check each param has a dict as a value, and that dict has a "default" defined
 84        for param_key, param_def in params.items():
 85            if not isinstance(param_def, dict):
 86                msg = f"Entry for param '{param_key}' must be a dictionary"
 87                logger.error(msg)
 88                raise Exception(msg)
 89            if "default" not in param_def:
 90                msg = (
 91                    f"Entry for param '{param_key}' must be a dictionary"
 92                    f"defining a 'default' value for this parameter."
 93                )
 94                logger.error(msg)
 95                raise Exception(msg)
 96
 97        self.params_dict = params
 98        self.appname = appname
 99
100        # Create a validator for the schema defined by params_dict
101        self.cerberus_validator = EasyVVUQValidator(self.params_dict)
102
103    def process_run(self, new_run, verify=True):
104        # If necessary parameter names are missing, fill them in from the
105        # default values in params_info
106        for param in self.params_dict.keys():
107            if param not in new_run.keys():
108                default_val = self.params_dict[param]["default"]
109                new_run[param] = default_val
110
111        # Convert float values to integers for integer parameters when they represent integers
112        # This handles the case where chaospy returns float arrays for discrete distributions
113        # Do this BEFORE validation to avoid type errors
114        for param_name, value in new_run.items():
115            if param_name in self.params_dict:
116                param_def = self.params_dict[param_name]
117                if param_def.get('type') == 'integer' and is_integer_valued(value):
118                    new_run[param_name] = convert_to_integer(value)
119
120        # Optionally verify that all params are known for this app, that the types are
121        # correct, params are within specified ranges etc. Uses cerberus for this.
122        if verify:
123            if not self.cerberus_validator.validate(new_run):
124                errors = self.cerberus_validator.errors
125                msg = (
126                    f"Error when verifying the following new run:\n"
127                    f"{new_run}\n"
128                    f"Identified errors were:\n"
129                    f"{errors}\n")
130
131                errors_list = [error[0] for error in self.cerberus_validator.errors.values()]
132                if 'unknown field' in errors_list:
133                    msg += (
134                        f"The allowed parameter names for this app are:\n"
135                        f"{list(self.params_dict.keys())}")
136
137                logger.error(msg)
138                raise RuntimeError(msg)
139
140        return new_run
141
142    def serialize(self):
143        return json.dumps(self.params_dict)
144
145    @staticmethod
146    def deserialize(serialized_params):
147        return ParamsSpecification(json.loads(serialized_params))