castaneda/NNDy
1
0
Fork 0
Code Issues Pull Requests Actions Packages repo.project_board Releases Wiki Activity

Update scripts/NNDy_Interface.py

Browse Source 
complete commenting
added handling of uncertainities
This commit is contained in:
castaneda 2025-03-21 15:06:00 +01:00
parent 48d73f6811
commit 59741c971b
1 changed files with 118 additions and 89 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

207
scripts/NNDy_Interface.py
View File

@ -1,90 +1,119 @@
#imports for runmanager - labscript import time
from runmanager_remote import run_experiment import numpy as np
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 #Imports for M-LOOP
import mloop.interfaces as mli
#importlib allows to import the costfunction defined in cost_model.py # THIS FILE SHOULDN'T BE MODIFIED
import importlib # if you're trying to improve the design of the interface define it in a new script
_module_cache = {} #avoid multiple calls of cost function for same routine # 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
#Declare your custom class that inherits from the Interface class
class NNDy_Interface(mli.Interface): #only exception
# if during debugging you want to know where the errors are generated in the experiment look for this mark $%$%$
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__()
#importlib allows to import the costfunction defined in cost_model.py
#Attributes of the interface can be added here import importlib
self.exp_global_par = hyperpars['globalPar'] _module_cache = {} #avoid multiple calls of cost function for same routine
self.input_names = hyperpars['inputPar_names']
self.routine = routine_name
self.cost_model = cost_model #Declare your custom class that inherits from the Interface class
class NNDy_Interface(mli.Interface):
def cost(self, parameters):
module_name = self.cost_model def __init__(self, routine_name, cost_model, hyperpars):
if module_name not in _module_cache: #You must include the super command to call the parent class, Interface, constructor
try: super(NNDy_Interface,self).__init__()
module = importlib.import_module(module_name)
cost_func = getattr(module, 'cost') #Attributes of the interface can be added here
#analysis_func = getattr(self.module, 'analysis')
_module_cache[module_name] = { #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
"cost_func": cost_func , self.exp_global_par = hyperpars['globalPar']
# "analysis_func": analysis_func self.input_names = hyperpars['inputPar_names']
}
except (ModuleNotFoundError, AttributeError) as e: self.routine = routine_name
raise ImportError(f'Failed to load cost function from "{module_name}.py": {e}') self.cost_model = cost_model
cost_model = _module_cache[module_name]["cost_func"] #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'
return cost_model(parameters) 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
#the method that runs the experiment given a set of parameters and returns a cost if module_name not in _module_cache:
def get_next_cost_dict(self,In_params_raw): try:
module = importlib.import_module(module_name)
#The parameters come in a dictionary and are provided in a numpy array cost_func = getattr(module, 'cost')
In_params = In_params_raw['params'] _module_cache[module_name] = {
#print(In_params) "cost_func": cost_func ,
#optimization parameters to be send back to labscript are converted back into dictionaries }
if len(In_params) != len(self.input_names): except (ModuleNotFoundError, AttributeError) as e:
raise Exception('number of optimized parameters and names do not match') raise ImportError(f'Failed to load cost function from "{module_name}.py": {e}')
In_params_dict = {}
for par,name in zip(In_params, self.input_names) : cost_model = _module_cache[module_name]["cost_func"]
In_params_dict.update({name: par})
#merge with fixed global variables return cost_model(hdf5_file)
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 #the method that runs the experiment given a set of parameters and returns a cost
def get_next_cost_dict(self,In_params_raw):
#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 #The parameters come in a dictionary and are provided in a numpy array
#print('running the experiment') In_params = In_params_raw['params']
results = { #print(In_params)
'cost': np.inf,
'bad': False #optimization parameters to be send back to labscript are converted back into dictionaries
} if len(In_params) != len(self.input_names):
try: raise Exception('number of optimized parameters and names do not match')
hdf_output_file = run_experiment(self.routine, global_var = global_group) In_params_dict = {}
results = self.cost(hdf_output_file) for par,name in zip(In_params, self.input_names) :
except Exception as e: In_params_dict.update({name: par})
print(f"Exception type '{e}', considered as bad run!")
results['bad'] = True #merge with fixed global variables
global_group = In_params_dict | self.exp_global_par
#self.analysis(hdf_output_file)
#print('cost is computed')
#Here you can include the code to run your experiment given a particular set of parameters
uncer = 0
# default values
time.sleep(0.001) results = {
'bad': False,
#The cost, uncertainty and bad boolean must all be returned as a dictionary '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
cost_dict = {'cost':results['cost'], 'uncer':uncer, 'bad':results['bad']} '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 return cost_dict