From 59741c971b05e5af0a34bfceddf9bb896ff9cfe3 Mon Sep 17 00:00:00 2001 From: castaneda Date: Fri, 21 Mar 2025 15:06:00 +0100 Subject: [PATCH] Update scripts/NNDy_Interface.py complete commenting added handling of uncertainities --- scripts/NNDy_Interface.py | 207 ++++++++++++++++++++++---------------- 1 file changed, 118 insertions(+), 89 deletions(-) diff --git a/scripts/NNDy_Interface.py b/scripts/NNDy_Interface.py index ed20b3d..e2fb977 100644 --- a/scripts/NNDy_Interface.py +++ b/scripts/NNDy_Interface.py @@ -1,90 +1,119 @@ -#imports for runmanager - labscript -from runmanager_remote import run_experiment -import time -import numpy as np - -#Imports for M-LOOP -import mloop.interfaces as mli - - - -#importlib allows to import the costfunction defined in cost_model.py -import importlib -_module_cache = {} #avoid multiple calls of cost function for same routine - - -#Declare your custom class that inherits from the Interface class -class NNDy_Interface(mli.Interface): - - def __init__(self, routine_name, cost_model, hyperpars): - #You must include the super command to call the parent class, Interface, constructor - super(NNDy_Interface,self).__init__() - - #Attributes of the interface can be added here - self.exp_global_par = hyperpars['globalPar'] - self.input_names = hyperpars['inputPar_names'] - self.routine = routine_name - self.cost_model = cost_model - - def cost(self, parameters): - module_name = self.cost_model - if module_name not in _module_cache: - try: - module = importlib.import_module(module_name) - cost_func = getattr(module, 'cost') - #analysis_func = getattr(self.module, 'analysis') - _module_cache[module_name] = { - "cost_func": cost_func , - # "analysis_func": analysis_func - } - except (ModuleNotFoundError, AttributeError) as e: - raise ImportError(f'Failed to load cost function from "{module_name}.py": {e}') - - cost_model = _module_cache[module_name]["cost_func"] - - return cost_model(parameters) - - - - #the method that runs the experiment given a set of parameters and returns a cost - def get_next_cost_dict(self,In_params_raw): - - #The parameters come in a dictionary and are provided in a numpy array - In_params = In_params_raw['params'] - #print(In_params) - #optimization parameters to be send back to labscript are converted back into dictionaries - if len(In_params) != len(self.input_names): - raise Exception('number of optimized parameters and names do not match') - In_params_dict = {} - for par,name in zip(In_params, self.input_names) : - In_params_dict.update({name: par}) - #merge with fixed global variables - global_group = In_params_dict | self.exp_global_par - - - #Here you can include the code to run your experiment given a particular set of parameters - - #run_experiment runs the routine specified by routine name with global variables equal to the new set of parameters given by the optimizer - #this means that the experiment parameters - In_params - are chosen among the global variablesof this labscript routine and are passed as a dictionary - #print('running the experiment') - results = { - 'cost': np.inf, - 'bad': False - } - try: - hdf_output_file = run_experiment(self.routine, global_var = global_group) - results = self.cost(hdf_output_file) - except Exception as e: - print(f"Exception type '{e}', considered as bad run!") - results['bad'] = True - - #self.analysis(hdf_output_file) - #print('cost is computed') - - uncer = 0 - - time.sleep(0.001) - - #The cost, uncertainty and bad boolean must all be returned as a dictionary - cost_dict = {'cost':results['cost'], 'uncer':uncer, 'bad':results['bad']} +import time +import numpy as np + +#imports for runmanager - labscript +from runmanager_remote import run_experiment + +#Imports for M-LOOP +import mloop.interfaces as mli + + +# THIS FILE SHOULDN'T BE MODIFIED +# if you're trying to improve the design of the interface define it in a new script +# possible improvements of this version +# - have the interface run the experiment many times to gather statistics +# - run two or more optimizers in parallel with different settings (like trust regions, update of parameters) for efficiency + +#only exception +# if during debugging you want to know where the errors are generated in the experiment look for this mark $%$%$ + + + +#importlib allows to import the costfunction defined in cost_model.py +import importlib +_module_cache = {} #avoid multiple calls of cost function for same routine + + +#Declare your custom class that inherits from the Interface class +class NNDy_Interface(mli.Interface): + + def __init__(self, routine_name, cost_model, hyperpars): + #You must include the super command to call the parent class, Interface, constructor + super(NNDy_Interface,self).__init__() + + #Attributes of the interface can be added here + + #here the interface is passed global variables eventually set from the main and the names of the input parameters that the controller controller will optimize + self.exp_global_par = hyperpars['globalPar'] + self.input_names = hyperpars['inputPar_names'] + + self.routine = routine_name + self.cost_model = cost_model + + #the cost function is retrieved from /{cost_model}.py + #in such file there should be a function called cost that takes as input the hdf5_file of the run and returns a dictionary with arguments 'bad', 'cost', 'uncer' + def cost(self, hdf5_file): + module_name = self.cost_model + + #looks in the cache if the file was accessed already, otherwise imports the cost function + if module_name not in _module_cache: + try: + module = importlib.import_module(module_name) + cost_func = getattr(module, 'cost') + _module_cache[module_name] = { + "cost_func": cost_func , + } + except (ModuleNotFoundError, AttributeError) as e: + raise ImportError(f'Failed to load cost function from "{module_name}.py": {e}') + + cost_model = _module_cache[module_name]["cost_func"] + + return cost_model(hdf5_file) + + + + #the method that runs the experiment given a set of parameters and returns a cost + def get_next_cost_dict(self,In_params_raw): + + #The parameters come in a dictionary and are provided in a numpy array + In_params = In_params_raw['params'] + #print(In_params) + + #optimization parameters to be send back to labscript are converted back into dictionaries + if len(In_params) != len(self.input_names): + raise Exception('number of optimized parameters and names do not match') + In_params_dict = {} + for par,name in zip(In_params, self.input_names) : + In_params_dict.update({name: par}) + + #merge with fixed global variables + global_group = In_params_dict | self.exp_global_par + + + #Here you can include the code to run your experiment given a particular set of parameters + + # default values + results = { + 'bad': False, + 'cost': float('inf'), #may cause some problem, in case try with an absurdly high but finite value, or with 0 if cost is always negative + 'uncer': 0 + } + + #run_experiment runs the routine specified by self.routine with global variables equal to the new set of parameters given by the optimizer + #print('running the experiment') + + # $%$%$ disable this expection catch if the experiment is always giving 'bad' results + try: + hdf_output_file = run_experiment(self.routine, global_var = global_group) + #results should be a dictionary that contains at least a 'cost' item + results = self.cost(hdf_output_file) + + + except Exception as e: + print(f"Exception type '{e}', considered as bad run!") + results['bad'] = True + + #print('cost is computed') + + #sets uncer to 0 if it's not returned + try: + uncer = results['uncer'] + except Exception as e: + uncer = 0 + #print(f"Exception type '{e}', no uncertainity set") + + time.sleep(0.001) + + #The cost, uncertainty and bad boolean must all be returned as a dictionary + cost_dict = {'cost':results['cost'], 'uncer':uncer, 'bad':results['bad']} return cost_dict \ No newline at end of file