in lib/ramble/ramble/application.py [0:0]
def create_experiment_chain(self, workspace):
"""Create the necessary chained experiments for this instance
This method determines which experiments need to be chained, grabs the
base instance from the experiment set, creates a copy of it (with a
unique name), injects the copy back into the experiment set,
and builds an internal mapping from unique name to the chaining definition.
"""
if not self.chained_experiments or self.is_template:
return
# Build initial stack. Uses a reversal of the current instance's
# chained experiments
parent_namespace = self.expander.experiment_namespace
classes_in_stack = {self}
chain_idx = 0
chain_stack = []
for exp in reversed(self.chained_experiments):
for exp_name in self.experiment_set.search_primary_experiments(exp["name"]):
child_inst = self.experiment_set.get_experiment(exp_name)
if child_inst in classes_in_stack:
raise ChainCycleDetectedError(
"Cycle detected in experiment chain:\n"
+ f" Primary experiment {parent_namespace}\n"
+ f" Chained expeirment name: {exp_name}\n"
+ f" Chain definition: {str(exp)}"
)
chain_stack.append((exp_name, exp.copy()))
parent_run_dir = self.expander.expand_var(
self.expander.expansion_str(self.keywords.experiment_run_dir)
)
# Continue until the stack is empty
while len(chain_stack) > 0:
cur_exp_name = chain_stack[-1][0]
cur_exp_def = chain_stack[-1][1]
# Perform basic validation on the chained experiment definition
if "name" not in cur_exp_def:
raise InvalidChainError(
"Invalid experiment chain defined:\n"
+ f" Primary experiment {parent_namespace}\n"
+ f" Chain definition: {str(exp)}\n"
+ ' "name" keyword must be defined'
)
if "order" in cur_exp_def:
possible_orders = ["after_chain", "after_root", "before_chain", "before_root"]
if cur_exp_def["order"] not in possible_orders:
raise InvalidChainError(
"Invalid experiment chain defined:\n"
+ f" Primary experiment {parent_namespace}\n"
+ f" Chain definition: {str(exp)}\n"
+ ' Optional keyword "order" must '
+ f"be one of {str(possible_orders)}\n"
)
if "command" not in cur_exp_def.keys():
raise InvalidChainError(
"Invalid experiment chain defined:\n"
+ f" Primary experiment {parent_namespace}\n"
+ f" Chain definition: {str(exp)}\n"
+ ' "command" keyword must be defined'
)
if "variables" in cur_exp_def:
if not isinstance(cur_exp_def["variables"], dict):
raise InvalidChainError(
"Invalid experiment chain defined:\n"
+ f" Primary experiment {parent_namespace}\n"
+ f" Chain definition: {str(exp)}\n"
+ ' Optional keyword "variables" '
+ "must be a dictionary"
)
base_inst = self.experiment_set.get_experiment(cur_exp_name)
if base_inst in classes_in_stack:
chain_stack.pop()
classes_in_stack.remove(base_inst)
order = "after_root"
if "order" in cur_exp_def:
order = cur_exp_def["order"]
chained_name = f"chain.{chain_idx}.{cur_exp_name}"
new_name = f"{parent_namespace}.{chained_name}"
new_run_dir = os.path.join(
parent_run_dir, namespace.chained_experiments, chained_name
)
if order == "before_chain":
self.chain_prepend.insert(0, new_name)
elif order == "before_root":
self.chain_prepend.append(new_name)
elif order == "after_root":
self.chain_append.insert(0, new_name)
elif order == "after_chain":
self.chain_append.append(new_name)
self.chain_commands[new_name] = cur_exp_def[namespace.command]
# Skip editing the new instance if the base_inst doesn't work
# This happens if the originating command is `workspace info`
# The printing experiment set doesn't have access to all
# of the experiment, so the base_inst command above
# doesn't get an application instance.
if base_inst:
new_inst = base_inst.copy()
if namespace.variables in cur_exp_def:
for var, val in cur_exp_def[namespace.variables].items():
new_inst.variables[var] = val
new_inst.expander._experiment_namespace = new_name
new_inst.variables[self.keywords.experiment_run_dir] = new_run_dir
new_inst.variables[self.keywords.experiment_name] = new_name
new_inst.variables[self.keywords.experiment_index] = (
self.expander.expand_var_name(self.keywords.experiment_index)
)
new_inst.repeats = self.repeats
new_inst.read_status()
# Extract inherited variables
if namespace.inherit_variables in cur_exp_def:
for inherit_var in cur_exp_def[namespace.inherit_variables]:
new_inst.variables[inherit_var] = self.variables[inherit_var]
# Expand the chained experiment vars, so we can build the execution command
new_inst.add_expand_vars(workspace)
chain_cmd = new_inst.expander.expand_var(cur_exp_def[namespace.command])
self.chain_commands[new_name] = chain_cmd
cur_exp_def[namespace.command] = chain_cmd
self.experiment_set.add_chained_experiment(new_name, new_inst)
chain_idx += 1
else:
# Avoid cycles, from children
if base_inst in classes_in_stack:
chain_stack.pop()
else:
if base_inst.chained_experiments:
for exp in reversed(base_inst.chained_experiments):
for exp_name in self.experiment_set.search_primary_experiments(
exp["name"]
):
child_inst = self.experiment_set.get_experiment(exp_name)
if child_inst in classes_in_stack:
raise ChainCycleDetectedError(
"Cycle detected in "
+ "experiment chain:\n"
+ " Primary experiment "
+ f"{parent_namespace}\n"
+ " Chained expeirment name: "
+ f"{cur_exp_name}\n"
+ " Chain definition: "
+ f"{str(cur_exp_def)}"
)
chain_stack.append((exp_name, exp))
classes_in_stack.add(base_inst)
# Create the final chain order
for exp in self.chain_prepend:
self.chain_order.append(exp)
self.chain_order.append(self.expander.experiment_namespace)
for exp in self.chain_append:
self.chain_order.append(exp)
# Inject the chain order into the children experiments
for exp in self.chain_prepend:
exp_inst = self.experiment_set.get_experiment(exp)
if exp_inst:
exp_inst.chain_order = self.chain_order.copy()
for exp in self.chain_append:
exp_inst = self.experiment_set.get_experiment(exp)
if exp_inst:
exp_inst.chain_order = self.chain_order.copy()