90 lines
3.9 KiB
Python
90 lines
3.9 KiB
Python
#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']}
|
|
return cost_dict |