src.fairreckitlib.experiment.experiment_config_parser
This module contains a parser for the experiment configuration.
Classes:
ExperimentConfigParser: parse an experiment configuration from a dictionary or yml.
This program has been developed by students from the bachelor Computer Science at Utrecht University within the Software Project course. © Copyright Utrecht University (Department of Information and Computing Sciences)
1"""This module contains a parser for the experiment configuration. 2 3Classes: 4 5 ExperimentConfigParser: parse an experiment configuration from a dictionary or yml. 6 7This program has been developed by students from the bachelor Computer Science at 8Utrecht University within the Software Project course. 9© Copyright Utrecht University (Department of Information and Computing Sciences) 10""" 11 12from typing import Any, Dict, Optional, Union 13 14from ..core.config.config_factories import GroupFactory 15from ..core.config.config_option_param import create_bool_param 16from ..core.config.config_value_param import ConfigNumberParam 17from ..core.core_constants import KEY_TYPE, TYPE_PREDICTION, TYPE_RECOMMENDATION, VALID_TYPES 18from ..core.core_constants import KEY_NAME, KEY_TOP_K, DEFAULT_TOP_K, MIN_TOP_K, MAX_TOP_K 19from ..core.core_constants import KEY_RATED_ITEMS_FILTER, DEFAULT_RATED_ITEMS_FILTER 20from ..core.events.event_dispatcher import EventDispatcher 21from ..core.io.io_utility import load_yml 22from ..core.parsing.parse_assert import assert_is_type, assert_is_key_in_dict,assert_is_one_of_list 23from ..core.parsing.parse_event import ON_PARSE, print_parse_event 24from ..core.parsing.parse_config_params import parse_config_param 25from ..data.data_factory import KEY_DATA 26from ..data.filter.filter_constants import KEY_DATA_SUBSET 27from ..data.pipeline.data_config_parsing import parse_data_config 28from ..data.set.dataset_registry import DataRegistry 29from ..evaluation.evaluation_factory import KEY_EVALUATION 30from ..evaluation.pipeline.evaluation_config_parsing import parse_evaluation_config 31from ..model.model_factory import KEY_MODELS 32from ..model.pipeline.model_config_parsing import parse_models_config 33from .experiment_config import PredictorExperimentConfig, RecommenderExperimentConfig 34 35 36class ExperimentConfigParser: 37 """Experiment Configuration Parser. 38 39 Public methods: 40 41 parse_experiment_config 42 parse_experiment_config_from_yml 43 """ 44 45 def __init__(self, verbose: bool): 46 """Construct the ExperimentConfigParser. 47 48 Args: 49 verbose: whether the parser should give verbose output. 50 """ 51 handle_parse_event = lambda parser, args: \ 52 print_parse_event(args) if parser.verbose else None 53 54 self.verbose = verbose 55 self.event_dispatcher = EventDispatcher() 56 self.event_dispatcher.add_listener(ON_PARSE, self, (handle_parse_event, None)) 57 58 def parse_experiment_config( 59 self, 60 experiment_config: Any, 61 data_registry: DataRegistry, 62 experiment_factory: GroupFactory 63 ) -> Optional[Union[PredictorExperimentConfig, RecommenderExperimentConfig]]: 64 """Parse an experiment configuration. 65 66 Args: 67 experiment_config: the experiment's total configuration. 68 data_registry: the data registry containing the available datasets. 69 experiment_factory: the factory containing all three pipeline factories. 70 71 Returns: 72 the parsed configuration or None on failure. 73 """ 74 # assert experiment_config is a dict 75 if not assert_is_type( 76 experiment_config, 77 dict, 78 self.event_dispatcher, 79 'PARSE ERROR: invalid experiment config type' 80 ): return None 81 82 # attempt to parse experiment name (required) 83 experiment_name = self.parse_experiment_name(experiment_config) 84 if experiment_name is None: 85 return None 86 87 # attempt to parse experiment type (required) 88 experiment_type = self.parse_experiment_type(experiment_config) 89 if experiment_type is None: 90 return None 91 92 # attempt to parse experiment datasets (required) 93 experiment_datasets = parse_data_config( 94 experiment_config, 95 data_registry, 96 experiment_factory.get_factory(KEY_DATA), 97 self.event_dispatcher 98 ) 99 if experiment_datasets is None: 100 return None 101 102 # attempt to parse experiment models (required) 103 experiment_models = parse_models_config( 104 experiment_config, 105 experiment_factory.get_factory(KEY_MODELS).get_factory(experiment_type), 106 self.event_dispatcher 107 ) 108 if experiment_models is None: 109 return None 110 111 # attempt to parse experiment evaluation (optional) 112 experiment_evaluation = parse_evaluation_config( 113 data_registry, 114 experiment_factory.get_factory(KEY_DATA).get_factory(KEY_DATA_SUBSET), 115 experiment_config, 116 experiment_factory.get_factory(KEY_EVALUATION).get_factory(experiment_type), 117 self.event_dispatcher 118 ) 119 120 parsed_config = None 121 122 if experiment_type == TYPE_PREDICTION: 123 parsed_config = PredictorExperimentConfig( 124 experiment_datasets, 125 experiment_models, 126 experiment_evaluation, 127 experiment_name 128 ) 129 elif experiment_type == TYPE_RECOMMENDATION: 130 parsed_config = RecommenderExperimentConfig( 131 experiment_datasets, 132 experiment_models, 133 experiment_evaluation, 134 experiment_name, 135 self.parse_experiment_top_k( 136 experiment_config 137 ), 138 self.parse_experiment_rated_items_filter( 139 experiment_config 140 ) 141 ) 142 143 return parsed_config 144 145 def parse_experiment_config_from_yml( 146 self, 147 file_path: str, 148 data_registry: DataRegistry, 149 experiment_factory: GroupFactory 150 ) -> Optional[Union[PredictorExperimentConfig, RecommenderExperimentConfig]]: 151 """Parse an experiment configuration from a yml file. 152 153 Args: 154 file_path: path to the yml file without extension. 155 data_registry: the data registry containing the available datasets. 156 experiment_factory: the factory containing all three pipeline factories. 157 158 Returns: 159 the parsed configuration or None on failure. 160 """ 161 experiment_config = load_yml(file_path + '.yml') 162 return self.parse_experiment_config(experiment_config, data_registry, experiment_factory) 163 164 def parse_experiment_name(self, experiment_config: Dict[str, Any]) -> Optional[str]: 165 """Parse the name of the experiment. 166 167 Args: 168 experiment_config: the experiment's total configuration. 169 170 Returns: 171 the name of the experiment or None on failure. 172 """ 173 if not assert_is_key_in_dict( 174 KEY_NAME, 175 experiment_config, 176 self.event_dispatcher, 177 'PARSE ERROR: missing experiment key \'' + KEY_NAME + '\' (required)' 178 ): return None 179 180 if not assert_is_type( 181 experiment_config[KEY_NAME], 182 str, 183 self.event_dispatcher, 184 'PARSE ERROR: invalid value for experiment key \'' + KEY_NAME + '\'' 185 ): return None 186 187 return experiment_config[KEY_NAME] 188 189 def parse_experiment_rated_items_filter( 190 self, recommender_experiment_config: Dict[str, Any]) -> bool: 191 """Parse the rated items filter of the recommender experiment. 192 193 Args: 194 recommender_experiment_config: the experiment's total configuration. 195 196 Returns: 197 the rated items filter of the experiment or True on failure. 198 """ 199 _, experiment_rated_items_filter = parse_config_param( 200 recommender_experiment_config, 201 'recommender experiment', 202 create_bool_param(KEY_RATED_ITEMS_FILTER, DEFAULT_RATED_ITEMS_FILTER), 203 self.event_dispatcher 204 ) 205 206 return experiment_rated_items_filter 207 208 def parse_experiment_top_k(self, recommender_experiment_config: Dict[str, Any]) -> int: 209 """Parse the top K of the recommender experiment. 210 211 Args: 212 recommender_experiment_config: the experiment's total configuration. 213 214 Returns: 215 the topK of the experiment or default_top_k on failure. 216 """ 217 _, experiment_top_k = parse_config_param( 218 recommender_experiment_config, 219 'recommender experiment', 220 ConfigNumberParam( 221 KEY_TOP_K, 222 int, 223 DEFAULT_TOP_K, 224 (MIN_TOP_K, MAX_TOP_K) 225 ), 226 self.event_dispatcher 227 ) 228 229 return experiment_top_k 230 231 def parse_experiment_type(self, experiment_config: Dict[str, Any]) -> Optional[str]: 232 """Parse the type of the experiment. 233 234 Args: 235 experiment_config: the experiment's total configuration. 236 237 Returns: 238 the type of the experiment or None on failure. 239 """ 240 # assert KEY_TYPE is present 241 if not assert_is_key_in_dict( 242 KEY_TYPE, 243 experiment_config, 244 self.event_dispatcher, 245 'PARSE ERROR: missing experiment key \'' + KEY_TYPE + '\' (required)', 246 one_of_list=VALID_TYPES 247 ): return None 248 249 experiment_type = experiment_config[KEY_TYPE] 250 251 # assert experiment_type is valid 252 if not assert_is_one_of_list( 253 experiment_type, 254 VALID_TYPES, 255 self.event_dispatcher, 256 'PARSE ERROR: invalid value for experiment ' + KEY_TYPE + 257 ' \'' + str(experiment_type) + '\'' 258 ): return None 259 260 return experiment_type
37class ExperimentConfigParser: 38 """Experiment Configuration Parser. 39 40 Public methods: 41 42 parse_experiment_config 43 parse_experiment_config_from_yml 44 """ 45 46 def __init__(self, verbose: bool): 47 """Construct the ExperimentConfigParser. 48 49 Args: 50 verbose: whether the parser should give verbose output. 51 """ 52 handle_parse_event = lambda parser, args: \ 53 print_parse_event(args) if parser.verbose else None 54 55 self.verbose = verbose 56 self.event_dispatcher = EventDispatcher() 57 self.event_dispatcher.add_listener(ON_PARSE, self, (handle_parse_event, None)) 58 59 def parse_experiment_config( 60 self, 61 experiment_config: Any, 62 data_registry: DataRegistry, 63 experiment_factory: GroupFactory 64 ) -> Optional[Union[PredictorExperimentConfig, RecommenderExperimentConfig]]: 65 """Parse an experiment configuration. 66 67 Args: 68 experiment_config: the experiment's total configuration. 69 data_registry: the data registry containing the available datasets. 70 experiment_factory: the factory containing all three pipeline factories. 71 72 Returns: 73 the parsed configuration or None on failure. 74 """ 75 # assert experiment_config is a dict 76 if not assert_is_type( 77 experiment_config, 78 dict, 79 self.event_dispatcher, 80 'PARSE ERROR: invalid experiment config type' 81 ): return None 82 83 # attempt to parse experiment name (required) 84 experiment_name = self.parse_experiment_name(experiment_config) 85 if experiment_name is None: 86 return None 87 88 # attempt to parse experiment type (required) 89 experiment_type = self.parse_experiment_type(experiment_config) 90 if experiment_type is None: 91 return None 92 93 # attempt to parse experiment datasets (required) 94 experiment_datasets = parse_data_config( 95 experiment_config, 96 data_registry, 97 experiment_factory.get_factory(KEY_DATA), 98 self.event_dispatcher 99 ) 100 if experiment_datasets is None: 101 return None 102 103 # attempt to parse experiment models (required) 104 experiment_models = parse_models_config( 105 experiment_config, 106 experiment_factory.get_factory(KEY_MODELS).get_factory(experiment_type), 107 self.event_dispatcher 108 ) 109 if experiment_models is None: 110 return None 111 112 # attempt to parse experiment evaluation (optional) 113 experiment_evaluation = parse_evaluation_config( 114 data_registry, 115 experiment_factory.get_factory(KEY_DATA).get_factory(KEY_DATA_SUBSET), 116 experiment_config, 117 experiment_factory.get_factory(KEY_EVALUATION).get_factory(experiment_type), 118 self.event_dispatcher 119 ) 120 121 parsed_config = None 122 123 if experiment_type == TYPE_PREDICTION: 124 parsed_config = PredictorExperimentConfig( 125 experiment_datasets, 126 experiment_models, 127 experiment_evaluation, 128 experiment_name 129 ) 130 elif experiment_type == TYPE_RECOMMENDATION: 131 parsed_config = RecommenderExperimentConfig( 132 experiment_datasets, 133 experiment_models, 134 experiment_evaluation, 135 experiment_name, 136 self.parse_experiment_top_k( 137 experiment_config 138 ), 139 self.parse_experiment_rated_items_filter( 140 experiment_config 141 ) 142 ) 143 144 return parsed_config 145 146 def parse_experiment_config_from_yml( 147 self, 148 file_path: str, 149 data_registry: DataRegistry, 150 experiment_factory: GroupFactory 151 ) -> Optional[Union[PredictorExperimentConfig, RecommenderExperimentConfig]]: 152 """Parse an experiment configuration from a yml file. 153 154 Args: 155 file_path: path to the yml file without extension. 156 data_registry: the data registry containing the available datasets. 157 experiment_factory: the factory containing all three pipeline factories. 158 159 Returns: 160 the parsed configuration or None on failure. 161 """ 162 experiment_config = load_yml(file_path + '.yml') 163 return self.parse_experiment_config(experiment_config, data_registry, experiment_factory) 164 165 def parse_experiment_name(self, experiment_config: Dict[str, Any]) -> Optional[str]: 166 """Parse the name of the experiment. 167 168 Args: 169 experiment_config: the experiment's total configuration. 170 171 Returns: 172 the name of the experiment or None on failure. 173 """ 174 if not assert_is_key_in_dict( 175 KEY_NAME, 176 experiment_config, 177 self.event_dispatcher, 178 'PARSE ERROR: missing experiment key \'' + KEY_NAME + '\' (required)' 179 ): return None 180 181 if not assert_is_type( 182 experiment_config[KEY_NAME], 183 str, 184 self.event_dispatcher, 185 'PARSE ERROR: invalid value for experiment key \'' + KEY_NAME + '\'' 186 ): return None 187 188 return experiment_config[KEY_NAME] 189 190 def parse_experiment_rated_items_filter( 191 self, recommender_experiment_config: Dict[str, Any]) -> bool: 192 """Parse the rated items filter of the recommender experiment. 193 194 Args: 195 recommender_experiment_config: the experiment's total configuration. 196 197 Returns: 198 the rated items filter of the experiment or True on failure. 199 """ 200 _, experiment_rated_items_filter = parse_config_param( 201 recommender_experiment_config, 202 'recommender experiment', 203 create_bool_param(KEY_RATED_ITEMS_FILTER, DEFAULT_RATED_ITEMS_FILTER), 204 self.event_dispatcher 205 ) 206 207 return experiment_rated_items_filter 208 209 def parse_experiment_top_k(self, recommender_experiment_config: Dict[str, Any]) -> int: 210 """Parse the top K of the recommender experiment. 211 212 Args: 213 recommender_experiment_config: the experiment's total configuration. 214 215 Returns: 216 the topK of the experiment or default_top_k on failure. 217 """ 218 _, experiment_top_k = parse_config_param( 219 recommender_experiment_config, 220 'recommender experiment', 221 ConfigNumberParam( 222 KEY_TOP_K, 223 int, 224 DEFAULT_TOP_K, 225 (MIN_TOP_K, MAX_TOP_K) 226 ), 227 self.event_dispatcher 228 ) 229 230 return experiment_top_k 231 232 def parse_experiment_type(self, experiment_config: Dict[str, Any]) -> Optional[str]: 233 """Parse the type of the experiment. 234 235 Args: 236 experiment_config: the experiment's total configuration. 237 238 Returns: 239 the type of the experiment or None on failure. 240 """ 241 # assert KEY_TYPE is present 242 if not assert_is_key_in_dict( 243 KEY_TYPE, 244 experiment_config, 245 self.event_dispatcher, 246 'PARSE ERROR: missing experiment key \'' + KEY_TYPE + '\' (required)', 247 one_of_list=VALID_TYPES 248 ): return None 249 250 experiment_type = experiment_config[KEY_TYPE] 251 252 # assert experiment_type is valid 253 if not assert_is_one_of_list( 254 experiment_type, 255 VALID_TYPES, 256 self.event_dispatcher, 257 'PARSE ERROR: invalid value for experiment ' + KEY_TYPE + 258 ' \'' + str(experiment_type) + '\'' 259 ): return None 260 261 return experiment_type
Experiment Configuration Parser.
Public methods:
parse_experiment_config parse_experiment_config_from_yml
46 def __init__(self, verbose: bool): 47 """Construct the ExperimentConfigParser. 48 49 Args: 50 verbose: whether the parser should give verbose output. 51 """ 52 handle_parse_event = lambda parser, args: \ 53 print_parse_event(args) if parser.verbose else None 54 55 self.verbose = verbose 56 self.event_dispatcher = EventDispatcher() 57 self.event_dispatcher.add_listener(ON_PARSE, self, (handle_parse_event, None))
Construct the ExperimentConfigParser.
Args: verbose: whether the parser should give verbose output.
59 def parse_experiment_config( 60 self, 61 experiment_config: Any, 62 data_registry: DataRegistry, 63 experiment_factory: GroupFactory 64 ) -> Optional[Union[PredictorExperimentConfig, RecommenderExperimentConfig]]: 65 """Parse an experiment configuration. 66 67 Args: 68 experiment_config: the experiment's total configuration. 69 data_registry: the data registry containing the available datasets. 70 experiment_factory: the factory containing all three pipeline factories. 71 72 Returns: 73 the parsed configuration or None on failure. 74 """ 75 # assert experiment_config is a dict 76 if not assert_is_type( 77 experiment_config, 78 dict, 79 self.event_dispatcher, 80 'PARSE ERROR: invalid experiment config type' 81 ): return None 82 83 # attempt to parse experiment name (required) 84 experiment_name = self.parse_experiment_name(experiment_config) 85 if experiment_name is None: 86 return None 87 88 # attempt to parse experiment type (required) 89 experiment_type = self.parse_experiment_type(experiment_config) 90 if experiment_type is None: 91 return None 92 93 # attempt to parse experiment datasets (required) 94 experiment_datasets = parse_data_config( 95 experiment_config, 96 data_registry, 97 experiment_factory.get_factory(KEY_DATA), 98 self.event_dispatcher 99 ) 100 if experiment_datasets is None: 101 return None 102 103 # attempt to parse experiment models (required) 104 experiment_models = parse_models_config( 105 experiment_config, 106 experiment_factory.get_factory(KEY_MODELS).get_factory(experiment_type), 107 self.event_dispatcher 108 ) 109 if experiment_models is None: 110 return None 111 112 # attempt to parse experiment evaluation (optional) 113 experiment_evaluation = parse_evaluation_config( 114 data_registry, 115 experiment_factory.get_factory(KEY_DATA).get_factory(KEY_DATA_SUBSET), 116 experiment_config, 117 experiment_factory.get_factory(KEY_EVALUATION).get_factory(experiment_type), 118 self.event_dispatcher 119 ) 120 121 parsed_config = None 122 123 if experiment_type == TYPE_PREDICTION: 124 parsed_config = PredictorExperimentConfig( 125 experiment_datasets, 126 experiment_models, 127 experiment_evaluation, 128 experiment_name 129 ) 130 elif experiment_type == TYPE_RECOMMENDATION: 131 parsed_config = RecommenderExperimentConfig( 132 experiment_datasets, 133 experiment_models, 134 experiment_evaluation, 135 experiment_name, 136 self.parse_experiment_top_k( 137 experiment_config 138 ), 139 self.parse_experiment_rated_items_filter( 140 experiment_config 141 ) 142 ) 143 144 return parsed_config
Parse an experiment configuration.
Args: experiment_config: the experiment's total configuration. data_registry: the data registry containing the available datasets. experiment_factory: the factory containing all three pipeline factories.
Returns: the parsed configuration or None on failure.
146 def parse_experiment_config_from_yml( 147 self, 148 file_path: str, 149 data_registry: DataRegistry, 150 experiment_factory: GroupFactory 151 ) -> Optional[Union[PredictorExperimentConfig, RecommenderExperimentConfig]]: 152 """Parse an experiment configuration from a yml file. 153 154 Args: 155 file_path: path to the yml file without extension. 156 data_registry: the data registry containing the available datasets. 157 experiment_factory: the factory containing all three pipeline factories. 158 159 Returns: 160 the parsed configuration or None on failure. 161 """ 162 experiment_config = load_yml(file_path + '.yml') 163 return self.parse_experiment_config(experiment_config, data_registry, experiment_factory)
Parse an experiment configuration from a yml file.
Args: file_path: path to the yml file without extension. data_registry: the data registry containing the available datasets. experiment_factory: the factory containing all three pipeline factories.
Returns: the parsed configuration or None on failure.
165 def parse_experiment_name(self, experiment_config: Dict[str, Any]) -> Optional[str]: 166 """Parse the name of the experiment. 167 168 Args: 169 experiment_config: the experiment's total configuration. 170 171 Returns: 172 the name of the experiment or None on failure. 173 """ 174 if not assert_is_key_in_dict( 175 KEY_NAME, 176 experiment_config, 177 self.event_dispatcher, 178 'PARSE ERROR: missing experiment key \'' + KEY_NAME + '\' (required)' 179 ): return None 180 181 if not assert_is_type( 182 experiment_config[KEY_NAME], 183 str, 184 self.event_dispatcher, 185 'PARSE ERROR: invalid value for experiment key \'' + KEY_NAME + '\'' 186 ): return None 187 188 return experiment_config[KEY_NAME]
Parse the name of the experiment.
Args: experiment_config: the experiment's total configuration.
Returns: the name of the experiment or None on failure.
190 def parse_experiment_rated_items_filter( 191 self, recommender_experiment_config: Dict[str, Any]) -> bool: 192 """Parse the rated items filter of the recommender experiment. 193 194 Args: 195 recommender_experiment_config: the experiment's total configuration. 196 197 Returns: 198 the rated items filter of the experiment or True on failure. 199 """ 200 _, experiment_rated_items_filter = parse_config_param( 201 recommender_experiment_config, 202 'recommender experiment', 203 create_bool_param(KEY_RATED_ITEMS_FILTER, DEFAULT_RATED_ITEMS_FILTER), 204 self.event_dispatcher 205 ) 206 207 return experiment_rated_items_filter
Parse the rated items filter of the recommender experiment.
Args: recommender_experiment_config: the experiment's total configuration.
Returns: the rated items filter of the experiment or True on failure.
209 def parse_experiment_top_k(self, recommender_experiment_config: Dict[str, Any]) -> int: 210 """Parse the top K of the recommender experiment. 211 212 Args: 213 recommender_experiment_config: the experiment's total configuration. 214 215 Returns: 216 the topK of the experiment or default_top_k on failure. 217 """ 218 _, experiment_top_k = parse_config_param( 219 recommender_experiment_config, 220 'recommender experiment', 221 ConfigNumberParam( 222 KEY_TOP_K, 223 int, 224 DEFAULT_TOP_K, 225 (MIN_TOP_K, MAX_TOP_K) 226 ), 227 self.event_dispatcher 228 ) 229 230 return experiment_top_k
Parse the top K of the recommender experiment.
Args: recommender_experiment_config: the experiment's total configuration.
Returns: the topK of the experiment or default_top_k on failure.
232 def parse_experiment_type(self, experiment_config: Dict[str, Any]) -> Optional[str]: 233 """Parse the type of the experiment. 234 235 Args: 236 experiment_config: the experiment's total configuration. 237 238 Returns: 239 the type of the experiment or None on failure. 240 """ 241 # assert KEY_TYPE is present 242 if not assert_is_key_in_dict( 243 KEY_TYPE, 244 experiment_config, 245 self.event_dispatcher, 246 'PARSE ERROR: missing experiment key \'' + KEY_TYPE + '\' (required)', 247 one_of_list=VALID_TYPES 248 ): return None 249 250 experiment_type = experiment_config[KEY_TYPE] 251 252 # assert experiment_type is valid 253 if not assert_is_one_of_list( 254 experiment_type, 255 VALID_TYPES, 256 self.event_dispatcher, 257 'PARSE ERROR: invalid value for experiment ' + KEY_TYPE + 258 ' \'' + str(experiment_type) + '\'' 259 ): return None 260 261 return experiment_type
Parse the type of the experiment.
Args: experiment_config: the experiment's total configuration.
Returns: the type of the experiment or None on failure.