Source code for energym.envs.utils.kpi

[docs]class KPI(object): """A class to track the KPIs while running simulations. The KPIs are customizable by specifying them in the kpi_options dict. An example for the dict looks as follows: kpi_options = { "kpi1":{"name":"Z01_T" ,"type":"avg_dev", "target":[19,23]}, "kpi2":{"name":"Z01_T" ,"type":"tot_viol", "target":[19,23]}, "kpi3":{"name":"Fa_Pw_All","type":"avg"}, "kpi4":{"name":"Fa_Pw_All", "type":"sum"} } Possible KPI types are: * "avg": The average over the given steps. * "sum": The sum over the given steps. * "avg_dev": The average deviation from a target (float or list). * "tot_viol": The number of interval constraint violations for a target interval (list). Attributes ---------- kpi_dict: dict Dict of kpi related outputs with list of observations num_obs: int Number of saved observations Methods ------- add_observations(obs) Extracts relevant observations and appends them to the corresponding list. get_kpi(start_ind, end_ind) Computes the KPIs for a given time window. get_cumulative_kpi(phrase, kpi_type, out_type) Computes cumulative KPIs over multiple variables. reset() Frees up the storages in the object. """ def __init__(self, kpi_options): """ Parameters ---------- kpi_options : dict Dict to specify the tracked KPIs. """ self.kpi_options = kpi_options self._initialize_kpi_dict() self.num_obs = 0 def _initialize_kpi_dict(self): self.kpi_dict = {} for key in self.kpi_options: self.kpi_dict[key] = []
[docs] def add_observation(self, obs): """Extracts relevant observations and appends them to the corresponding list. Parameters ---------- obs : dict Observation object to be tracked. """ added_obs = 0 for key in self.kpi_options: if type(obs[self.kpi_options[key]["name"]]) == list: self.kpi_dict[key].extend( obs[self.kpi_options[key]["name"]] ) added_obs = len(obs[self.kpi_options[key]["name"]]) else: self.kpi_dict[key].append( obs[self.kpi_options[key]["name"]] ) added_obs = 1 self.num_obs = self.num_obs + added_obs
[docs] def get_kpi(self, start_ind=0, end_ind=-1): """Computes the KPIs for a given time window. Parameters ---------- start_ind : int, optional Index determining the start of the time window, by default 0 end_ind : int, optional Index determining the end of the time window, by default -1 Returns ------- kpi_summary : dict Dict containing the KPIs. Raises ------ IndexError If no observations are stored. """ if self.num_obs == 0: raise IndexError("Can't compute KPIs without observations!") else: if end_ind > self.num_obs: end_ind = self.num_obs elif end_ind < 0: end_ind = self.num_obs + 1 + end_ind if start_ind < 0: start_ind = self.num_obs + start_ind assert start_ind < end_ind kpi_summary = self.kpi_options.copy() for key in self.kpi_options: kpi_val = self._compute_kpi( self.kpi_dict[key][start_ind:end_ind], self.kpi_options[key], end_ind - start_ind, ) kpi_summary[key]["kpi"] = kpi_val return kpi_summary
def _compute_kpi(self, obs, opts, length): if opts["type"] == "avg": obs_abs = map(abs, obs) kpi = sum(obs_abs) / length elif opts["type"] == "sum": kpi = sum(obs) elif opts["type"] == "avg_dev": kpi = 0 if type(opts["target"]) == list: for val in obs: if val < opts["target"][0]: kpi += opts["target"][0] - val elif val > opts["target"][1]: kpi += val - opts["target"][1] else: for val in obs: kpi += abs(val - opts["target"]) kpi = kpi / length elif opts["type"] == "tot_viol": kpi = 0 for val in obs: if val < opts["target"][0] or val > opts["target"][1]: kpi += 1 return kpi
[docs] def get_cumulative_kpi(self, names, kpi_type, out_type): """Computes cumulative KPIs over multiple variables. Parameters ---------- names : list or str List of variable names or common string to filter the variables. kpi_type : str One of the 4 KPI types to filter the variables. out_type : str Cumulative KPI type ("avg" or "sum"). Returns ------- float or int Cumulative KPI value. Raises ------ IndexError If no variables fitting the criteria are detected. """ cumulative_kpi = 0 num = 0 kpi_summary = self.get_kpi() if isinstance(names, list): for key in kpi_summary: for name in names: if ( name == kpi_summary[key]["name"] and kpi_summary[key]["type"] == kpi_type ): cumulative_kpi += kpi_summary[key]["kpi"] num += 1 else: for key in kpi_summary: if ( names in kpi_summary[key]["name"] and kpi_summary[key]["type"] == kpi_type ): cumulative_kpi += kpi_summary[key]["kpi"] num += 1 if num == 0: raise IndexError( "No KPIs corresponding to {} found!".format(names) ) if out_type == "avg": cumulative_kpi = cumulative_kpi / num return cumulative_kpi
[docs] def reset(self): """Frees up the storages in the object.""" self.num_obs = 0 for key in self.kpi_dict: self.kpi_dict[key] = []