Source code for l2l.utils.trajectory

import time
from l2l.utils.groups import ParameterGroup, ResultGroup, ParameterDict
from l2l.utils.individual import Individual
import logging

logging = logging.getLogger("Trajectory")


[docs]class Trajectory: """ The trajectory is a class which holds the history of the parameter space exploration, defines the current parameters to be explored and holds the results from each execution. Based on the pypet trajectory concept: https://github.com/SmokinCaterpillar/pypet """ def __init__(self, **keyword_args): """ Initializes the trajectory. Some parameters are kept to match the interface with the pypet trajectory. TODO: remove all irrelevant attributes and simplify the class """ if 'name' in keyword_args: self._name = keyword_args['name'] self._timestamp = time.time() self._parameters = ParameterDict(self) # Contains all parameters self._results = {} # Contains all results self.individual = Individual() self.results = ResultGroup() self.results.f_add_result_group('all_results', "Contains all the results") self.current_results = {} self._parameters.parameter_group = {} self._parameters.parameter = {} self.individuals = {} self.v_idx = 0
[docs] def f_add_parameter_group(self, name, comment=""): """ Adds a new parameter group :param name: name of the new parameter group :param comment: ignored for the moment. Kept to match pypet interface. """ self._parameters[name] = ParameterGroup() logging.info("Added new parameter group: " + name)
[docs] def f_add_parameter_to_group(self, group_name, key, val): """ Adds a parameter to an already existing group. :param group_name: Name of the group where the parameter should be added :param key: Name of the parameter to be added :param val: Value of the parameter Throws an exception if the group does not exist """ if group_name in self._parameters.keys(): self._parameters[group_name].f_add_parameter(key, val) else: # LOG("Key not found when adding to result group") raise Exception("Group name not found when adding value to result group")
[docs] def f_add_result(self,key, val, comment=""): """ Adds a result to the trajectory :param key: it identifies either a generation params result group or another result :param val: The value to be added to the results TODO: verify where is the generation_params call performed """ if key == 'generation_params': self.results[key] = ResultGroup() else: self._results[key] = val
[docs] def f_add_parameter(self, key, val, comment=""): """ Adds a parameter to the trajectory :param key: Name of the parameter :param val: Value of the parameter :param comment """ self._parameters[key] = val
[docs] def f_add_derived_parameter(self, key, val, comment=""): """ Adds a derived parameter to the trajectory. Match the previous pypet interface. :param key: Name of the parameter :param val: Value of the parameter :param comment: """ self.f_add_parameter(key,val,comment)
[docs] def f_expand(self, build_dict, fail_safe=True): """ The expand function takes care of adding a new generation and individuals to the trajectory This is a critical function to allow the addition of a new generation, called by the optimizer from the postprocessing function :param build_dict: The dictionary containing the new generation id and its individuals :param fail_safe: Currently ignored """ params = {} gen = [] ind_idx = [] for key in build_dict.keys(): if key == 'generation': gen = build_dict['generation'] elif key == 'ind_idx': ind_idx = build_dict['ind_idx'] else: params[key] = build_dict[key] generation = gen[0] self.individuals[generation] = [] for i in ind_idx: ind = Individual(generation,i,[]) for j in params: ind.f_add_parameter(j, params[j][i]) self.individuals[generation].append(ind) logging.info("Expanded trajectory for generation: " + str(generation))
def __str__(self): return str(self._parameters) def __getattr__(self, attr): """ Handle attribute access like a sdict :param attr: The attribute to be accessed :return: the value of this attributes """ if '.' in attr: # This is triggered exclusively in the case where __getattr__ is called from __getitem__ attrs = attr.split('.') ret = self._parameters.get(attrs[0]) for at in attrs[1:]: ret = ret[at] elif attr == 'par' or attr == 'parameters': ret = self._parameters else: ret = self._parameters.get(attr,default_value=None) return ret def __getitem__(self, key): return self.__getattr__(key) def __getstate__(self): # print(self.__dict__) return self.__dict__ def __setstate__(self, d): self.__dict__.update(d)