in lib/ramble/ramble/application.py [0:0]
def calculate_statistics(self, workspace):
"""Calculate statistics for results of repeated experiments
When repeated experiments are used, this method aggregates the results of
each experiment's repeats and calculates statistics for each numeric FOM.
If a FOM is non-numeric, no calculations are performed.
Statistics are injected into the results file under the base experiment
namespace.
"""
def is_numeric_fom(fom):
"""Returns true if a fom value is numeric, and of an applicable type"""
value = fom["value"]
try:
value = float(value)
if (
fom["fom_type"]["name"] is FomType.CATEGORY.name
or fom["fom_type"]["name"] is FomType.INFO.name
):
return False
return True
except (ValueError, TypeError):
return False
if not self.repeats.is_repeat_base:
return
repeat_experiments = {}
repeat_foms = {}
first_repeat_exp = ""
# repeat_experiments dict = {repeat_experiment_namespace: {dict}}
# repeat_foms dict = {context: {(fom_name, units, origin, origin_type): [list of values]}}
# origin_type is generated as 'summary::stat_name'
base_exp_name = self.expander.experiment_name
base_exp_namespace = self.expander.experiment_namespace
# Create a list of all repeats by inserting repeat index
for n in range(1, self.repeats.n_repeats + 1):
if (
base_exp_name in self.experiment_set.chained_experiments.keys()
and base_exp_name not in self.experiment_set.experiments.keys()
):
insert_idx = base_exp_name.find(".chain")
repeat_exp_namespace = (
base_exp_name[:insert_idx] + f".{n}" + base_exp_name[insert_idx:]
)
else:
base_exp_namespace = self.expander.experiment_namespace
repeat_exp_namespace = f"{base_exp_namespace}.{n}"
repeat_experiments[repeat_exp_namespace] = {}
repeat_experiments[repeat_exp_namespace]["base_exp"] = base_exp_namespace
if n == 1:
first_repeat_exp = repeat_exp_namespace
# If repeat_success_strict is true, one failed experiment will fail the whole set
# If repeat_success_strict is false, any passing experiment will pass the whole set
repeat_success = False
exp_status_list = []
for exp in repeat_experiments.keys():
if exp in self.experiment_set.experiments.keys():
exp_inst = self.experiment_set.experiments[exp]
elif exp in self.experiment_set.chained_experiments.keys():
exp_inst = self.experiment_set.chained_experiments[exp]
else:
continue
exp_status_list.append(exp_inst.get_status())
if workspace.repeat_success_strict:
if ExperimentStatus.FAILED in exp_status_list:
repeat_success = False
else:
repeat_success = True
else:
if ExperimentStatus.SUCCESS in exp_status_list:
repeat_success = True
else:
repeat_success = False
if repeat_success:
self.set_status(status=ExperimentStatus.SUCCESS)
else:
self.set_status(status=ExperimentStatus.FAILED)
self._init_result()
logger.debug(
f"Calculating statistics for {self.repeats.n_repeats} repeats of " f"{base_exp_name}"
)
results = []
# Iterate through repeat experiment instances, extract foms, and aggregate them
for exp in repeat_experiments.keys():
if exp in self.experiment_set.experiments.keys():
exp_inst = self.experiment_set.experiments[exp]
elif exp in self.experiment_set.chained_experiments.keys():
exp_inst = self.experiment_set.chained_experiments[exp]
else:
continue
# When strict success is off for repeats (loose success), skip failed exps
if exp_inst.result.status == ExperimentStatus.FAILED:
continue
if exp_inst.result.contexts:
for context in exp_inst.result.contexts:
context_name = context["name"]
if context_name not in repeat_foms.keys():
repeat_foms[context_name] = {}
for fom in context["foms"]:
fom_key = (
fom["name"],
fom["units"],
fom["origin"],
fom["origin_type"],
)
# Stats will not be calculated for non-numeric foms so they're skipped
if fom_key not in repeat_foms[context_name].keys():
repeat_foms[context_name][fom_key] = {
"fom_type": fom["fom_type"],
"fom_values": [],
}
if is_numeric_fom(fom):
repeat_foms[context_name][fom_key]["fom_is_numeric"] = True
else:
repeat_foms[context_name][fom_key]["fom_is_numeric"] = False
fom_contents = (False, fom["value"], fom["fom_type"])
if repeat_foms[context_name][fom_key]["fom_is_numeric"]:
repeat_foms[context_name][fom_key]["fom_values"].append(
float(fom["value"])
)
else:
repeat_foms[context_name][fom_key]["fom_values"].append(fom["value"])
# Iterate through the aggregated foms, calculate stats, and insert into results
for context, fom_dict in repeat_foms.items():
if not fom_dict:
continue
context_map = {
"name": context,
"foms": [],
"display_name": _get_context_display_name(context),
}
summary_foms = []
if context == _NULL_CONTEXT:
# Use the app name as the origin of the FOM
summary_origin = self.name
n_total_dict = {
"value": self.repeats.n_repeats,
"units": "repeats",
"origin": summary_origin,
"origin_type": "summary::n_total_repeats",
"name": "Experiment Summary",
"fom_type": FomType.MEASURE.to_dict(),
}
summary_foms.append(n_total_dict)
n_success = exp_status_list.count(ExperimentStatus.SUCCESS)
n_success_dict = {
"value": n_success,
"units": "repeats",
"origin": summary_origin,
"origin_type": "summary::n_successful_repeats",
"name": "Experiment Summary",
"fom_type": FomType.MEASURE.to_dict(),
}
summary_foms.append(n_success_dict)
for fom_key, fom_contents in fom_dict.items():
fom_name, fom_units, fom_origin, fom_origin_type = fom_key
fom_type = fom_contents["fom_type"]
fom_values = fom_contents["fom_values"]
if fom_contents["fom_is_numeric"]:
calcs = []
for statistic in ramble.util.stats.all_stats:
calcs.append(statistic.report(fom_values, fom_units))
for calc in calcs:
fom_calc_dict = {
"value": calc[0],
"units": calc[1],
"origin": fom_origin,
"origin_type": calc[2],
"name": fom_name,
"fom_type": fom_type,
}
context_map["foms"].append(fom_calc_dict)
else:
# Only elevate non-numeric FOMs when they have the same value for all repeats
if len(set(fom_values)) == 1:
fom_str_dict = {
"value": fom_values[0],
"units": fom_units,
"origin": fom_origin,
"origin_type": fom_origin_type,
"name": fom_name,
"fom_type": fom_type,
}
context_map["foms"].append(fom_str_dict)
else:
continue
# Display the FOMs in alphabetic order, even if the corresponding log entries
# may be in different ordering.
context_map["foms"].sort(key=operator.itemgetter("name"))
if context == _NULL_CONTEXT:
context_map["foms"] = summary_foms + context_map["foms"]
results.append(context_map)
if results:
self.result.contexts = results
workspace.insert_result(self.result.to_dict(), first_repeat_exp)