"""NRV-:class:`.Problem` handling."""importnumpyasnpimportfaulthandlerimporttracebackfrom..backend._NRV_ClassimportNRV_classfrom..backend._MCoreimportMCH,synchronize_processesfrom..backend._log_interfaceimportrise_error,pass_debug_info,set_log_levelfrom.optim_utils._OptimResultsimportoptim_resultsfrom._CostFunctionsimportcost_functionfrom._OptimizersimportOptimizerimportsys# enable faulthandler to ease 'segmentation faults' debugfaulthandler.enable()
[docs]defcost_function_swarm_from_particle(cost_function_part,**kwargs):""" Generate a cost function for a swarm from a cost function for a particle Parameters ---------- cost_function_part : func cost function return a cost (float) from a particle (1-dimensional array) verbose : tupple if True, print a progress bar for updated whenby default True Returns ------- part : np.ndarray vector of values of the part in the len_part dimension """defcost_function_swarm(swarm):L=len(swarm)costs=np.zeros((L))foriinrange(L):particle=swarm[i][:]costs[i]=cost_function_part(particle,**kwargs)# print("part=", particle, "c=", costs[i] )returncostsreturncost_function_swarm
[docs]classProblem(NRV_class):""" Problem Class A class to describe problems that should be optimized with the NRV Framework. The problem should be described with a simulation and a cost, using the object cost_function, and various optimization algorithms can be used to find optimal solution. This class is abstract and is not supposed to be used directly by the end user. NRV can handle two types of problems: - problems where a geometric parameter can be optimized: please refer to ... - problems where the waveform can be optimized: please refer to ... """
[docs]def__init__(self,cost_function:cost_function=None,optimizer:Optimizer=None,save_problem_results=False,problem_fname="optim.json",):super().__init__()self._CostFunction=cost_functionself._Optimizer=optimizer# For cases where optimisation is done on a swarm(groupe) of particleself.swarm_optimizer=Falseself._SwarmCostFunction=Noneself.save_problem_results=save_problem_resultsself.problem_fname=problem_fname
# Handling the cost_function attribute@propertydefcostfunction(self):""" Cost function of a Problem, the cost function should be a CosFunction object, it should return a scalar. NRV function should be prefered """returnself._CostFunction@costfunction.setterdefcostfunction(self,cf:cost_function):# need to add a verification that the cost function is a scallar and so onself._CostFunction=cf@costfunction.deleterdefcostfunction(self):self._CostFunction=None
# Handling the Optimizer attribute@propertydefoptimizer(self):""" Optimizer of the problem, the Optimizer should be an Optimizer object. It has reference to optimization methods and constraints """returnself._Optimizer@optimizer.setterdefoptimizer(self,optim:Optimizer):self._Optimizer=optimself.swarm_optimizer=self._Optimizer.swarm_optimizer@optimizer.deleterdefoptmizer(self):# self._Optimizer = Nonepass# Call method is where the magic happensdef__call__(self,**kwargs)->optim_results:""" Perform the optimization: minimze the `cost_function` using `optmizer` Parameters ---------- kwargs containing parameters of the optimizer to change Returns ------- optim_results results of the optimization Raises ------ KeyboardInterrupt _description_ """ifMCH.do_master_only_work():try:kwargs=self.__update_saving_parameters(**kwargs)ifnotself.swarm_optimizer:results=self._Optimizer(self._CostFunction,**kwargs)else:self._SwarmCostFunction=cost_function_swarm_from_particle(self._CostFunction)results=self._Optimizer(self._SwarmCostFunction,**kwargs)MCH.master_broadcasts_to_all({"status":"Completed"})exceptKeyboardInterrupt:raiseKeyboardInterruptexcept:MCH.master_broadcasts_to_all({"status":"Error"})rise_error(traceback.format_exc())elifself.__check_MCore_CostFunction():self.__wait_for_simulation()else:passset_log_level("INFO")ifMCH.do_master_only_work():ifself.save_problem_results:results.save(save=True,fname=self.problem_fname)returnresultselse:returnoptim_results({"dummy_res":1})# Mcore handlingdef__check_MCore_CostFunction(self):""" ch """returngetattr(self._CostFunction,"_MCore_CostFunction",False)def__wait_for_simulation(self):slave_status={"status":"Wait"}try:set_log_level("WARNING")whileslave_status["status"]=="Wait":slave_status=MCH.master_broadcasts_to_all(slave_status)pass_debug_info(MCH.rank,slave_status)sys.stdout.flush()ifslave_status["status"]=="Simulate":self._CostFunction(slave_status["X"])slave_status["status"]="Wait"exceptKeyboardInterrupt:raiseKeyboardInterruptpass_debug_info(MCH.rank,slave_status)# additional methodsdef__update_saving_parameters(self,**kwargs):""" internal use only: update the results saving parameters and remove the corresponding keys from kwargs """if"save_problem_results"inkwargs:self.save_problem_results=kwargs.pop("save_problem_results")if"problem_fname"inkwargs:self.problem_fname=kwargs.pop("problem_fname")returnkwargs