in lib/ramble/ramble/workspace/workspace.py [0:0]
def dump_results(self, output_formats=None, print_results=False, summary_only=False):
"""
Write out result file in desired format
This attempts to avoid the loss of previous results data by appending
the datetime to the filename, but is willing to clobber the file
results.latest.<extension>
"""
if output_formats is None:
output_formats = ["text"]
if not self.results:
self.results = {}
results = _filter_results(self.results, summary_only=summary_only)
results_written = []
symlinks_updated = []
dt = self.date_string()
inner_delim = "."
filename_base = "results" + inner_delim + dt
latest_base = "results" + inner_delim + "latest"
software_key = ramble.experiment_result._OUTPUT_MAPPING["software"]
if "text" in output_formats:
file_extension = ".txt"
out_file = os.path.join(self.root, filename_base + file_extension)
latest_file = os.path.join(self.root, latest_base + file_extension)
results_written.append(out_file)
with open(out_file, "w+") as f:
f.write(f"From Workspace: {self.name} (hash: {results['workspace_hash']})\n")
if "experiments" in results:
for exp in results["experiments"]:
f.write("Experiment %s figures of merit:\n" % exp["name"])
f.write(" Status = %s\n" % exp["RAMBLE_STATUS"])
if "TAGS" in exp:
f.write(f' Tags = {exp["TAGS"]}\n')
if exp["N_REPEATS"] > 0: # this is a base exp with summary of repeats
for context in exp["CONTEXTS"]:
f.write(f' {context["display_name"]} figures of merit:\n')
fom_summary = {}
for fom in context["foms"]:
name = fom["name"]
if name not in fom_summary.keys():
fom_summary[name] = []
stat_name = fom["origin_type"]
value = fom["value"]
units = fom["units"]
output = f"{stat_name} = {value} {units}\n"
fom_summary[name].append(output)
for fom_name, fom_val_list in fom_summary.items():
f.write(f" {fom_name}:\n")
for fom_val in fom_val_list:
f.write(f" {fom_val.strip()}\n")
if software_key in exp and exp[software_key]:
self.write_software_info(f, exp)
else:
for context in exp["CONTEXTS"]:
f.write(f' {context["display_name"]} figures of merit:\n')
for fom in context["foms"]:
name = fom["name"]
if fom["origin_type"] == "modifier":
delim = "::"
mod = fom["origin"]
name = f"{fom['origin_type']}{delim}{mod}{delim}{name}"
output = "{} = {} {}".format(name, fom["value"], fom["units"])
f.write(" %s\n" % (output.strip()))
if software_key in exp and exp[software_key]:
self.write_software_info(f, exp)
else:
logger.msg("No results to write")
symlinks_updated.append(latest_file)
self.symlink_result(out_file, latest_file)
# Convert SoftwareInfo classes to dicts
for exp in results["experiments"]:
for key, pkg_list in exp[software_key].items():
exp[software_key][key] = [pkg.to_dict() for pkg in pkg_list]
if "json" in output_formats:
file_extension = ".json"
out_file = os.path.join(self.root, filename_base + file_extension)
latest_file = os.path.join(self.root, latest_base + file_extension)
results_written.append(out_file)
with open(out_file, "w+") as f:
sjson.dump(results, f)
symlinks_updated.append(latest_file)
self.symlink_result(out_file, latest_file)
if "yaml" in output_formats:
file_extension = ".yaml"
out_file = os.path.join(self.root, filename_base + file_extension)
latest_file = os.path.join(self.root, latest_base + file_extension)
results_written.append(out_file)
from ruamel.yaml import RoundTripDumper
class RambleSafeDumper(RoundTripDumper):
def ignore_aliases(self, _data):
"""Make the dumper NEVER print YAML aliases."""
return True
def call_value(dumper, data):
return dumper.represent_data(data.value)
RambleSafeDumper.add_representer(ramble.experiment_result.ExperimentStatus, call_value)
with open(out_file, "w+") as f:
syaml.dump(results, stream=f, Dumper=RambleSafeDumper)
symlinks_updated.append(latest_file)
self.symlink_result(out_file, latest_file)
if not results_written:
logger.die("Results were not written.")
logger.all_msg("Results are written to:")
for out_file in results_written:
logger.all_msg(f" {out_file}")
logger.all_msg("Symlinks updated:")
for symlink_path in symlinks_updated:
logger.all_msg(f" {symlink_path}")
if print_results:
with open(results_written[0]) as f:
# Use tty directly to avoid cluttering the analyze log
tty.msg(f"Results from the analysis pipeline:\n{f.read()}")
return filename_base