Public Member Functions
	__init__ (self, path)

	run (self, working_dir, verbose=False)

Data Fields
	json_contents

	adj

	workflow_sequence

Static Public Attributes
str	WORKFLOWS_K = 'workflows'

str	DEPENDENCIES_K = 'run-after'

str	CFG_REFERENCE_K = 'cfg-reference'

str	WORKFLOW_STEPS_K = 'steps'

str	ONE_CMD_TOOL_K = 'one-cmd'

str	COMMANDS_K = 'commands'

Protected Member Functions
	_check_cycle (self)

	_verify_workflow (self, json_contents)

Detailed Description

Definition at line 26 of file WorkflowRunner.py.

Constructor & Destructor Documentation

◆ init()

WorkflowRunner.WorkflowRunner.__init__	(	self,
		path
	)

Definition at line 34 of file WorkflowRunner.py.

    def __init__(self, path):
        try:
            with open(path) as f:
                self.json_contents = json.load(f)
        except FileNotFoundError:
            raise FileNotFoundError("Not found given workflow file")
        except json.decoder.JSONDecodeError:
            raise ImportError("Invalid workflow file")
 
        self._verify_workflow(self.json_contents)
 
        workflows = self.json_contents[self.WORKFLOWS_K]
        self.adj = dict.fromkeys(workflows, [])
        # decide the order according to the dependencies of each workflow.
        helper = TopologicalSortHelper(workflows)
        for workflow_k in workflows:
            workflow = self.json_contents[workflow_k]
            if self.DEPENDENCIES_K in workflow:
                for previous_workflow in workflow[self.DEPENDENCIES_K]:
                    helper.add_edge(previous_workflow, workflow_k)
                    self.adj[previous_workflow].append(workflow_k)
        self.workflow_sequence = helper.sort()
 
        self._check_cycle()
 

Member Function Documentation

◆ _check_cycle()

WorkflowRunner.WorkflowRunner._check_cycle ( self )

protected

Definition at line 59 of file WorkflowRunner.py.

    def _check_cycle(self):
        pos = dict()
        index = 0
        workflow_num = len(self.workflow_sequence)
        # number the order
        for seq_idx in range(workflow_num):
            pos[self.workflow_sequence[seq_idx]] = index
            index += 1
 
        for seq_idx in range(workflow_num):
            first_wf = self.workflow_sequence[seq_idx]
            for adj_wf in self.adj[first_wf]:
                first_pos = 0 if first_wf not in pos else pos[first_wf]
                second_pos = 0 if adj_wf not in pos else pos[adj_wf]
                if (first_pos > second_pos):
                    raise RuntimeError("Workflows should not have a cycle")
 

References WorkflowRunner.WorkflowRunner.adj, and WorkflowRunner.WorkflowRunner.workflow_sequence.

◆ _verify_workflow()

WorkflowRunner.WorkflowRunner._verify_workflow	(	self,
		json_contents
	)

protected

Definition at line 76 of file WorkflowRunner.py.

    def _verify_workflow(self, json_contents):
        # workflow file should have WORKFLOWS_K
        if not self.WORKFLOWS_K in json_contents:
            raise ValueError("Not found \"" + self.WORKFLOWS_K +
                             "\" key in workflow file")
 
        workflows = json_contents[self.WORKFLOWS_K]
        # workflow file should have keys listed in WORKFLOWS_K
        for workflow_k in workflows:
            if not workflow_k in json_contents:
                raise ValueError("Not found " + workflow_k + " key listed in \"" +
                                 self.WORKFLOWS_K + "\"")
 
        # each workflow should have either WORKFLOW_STEPS_K or CFG_REFERENCE_K
        for workflow_k in workflows:
            if not self.WORKFLOW_STEPS_K in json_contents[
                    workflow_k] and not self.CFG_REFERENCE_K in json_contents[workflow_k]:
                raise ValueError("Each workflow should have either \"" +
                                 self.WORKFLOW_STEPS_K + "\" or \"" +
                                 self.CFG_REFERENCE_K + "\"")
        for workflow_k in workflows:
            if self.WORKFLOW_STEPS_K in json_contents[
                    workflow_k] and self.CFG_REFERENCE_K in json_contents[workflow_k]:
                raise ValueError("\"" + self.WORKFLOW_STEPS_K + "\" and \"" +
                                 self.CFG_REFERENCE_K + "\" are exclusive key")
 
        # each step should have ONE_CMD_TOOL_K and COMMANDS_K
        for workflow_k in workflows:
            workflow = json_contents[workflow_k]
            if self.WORKFLOW_STEPS_K in workflow:
                step_keys = workflow[self.WORKFLOW_STEPS_K]
                for step_k in step_keys:
                    step = workflow[step_k]
                    if not self.ONE_CMD_TOOL_K in step or not self.COMMANDS_K in step:
                        raise ValueError("Each step should have \"" +
                                         self.ONE_CMD_TOOL_K + "\"" + " and \"" +
                                         self.COMMANDS_K + "\"")
 

References WorkflowRunner.WorkflowRunner.CFG_REFERENCE_K, WorkflowRunner.WorkflowRunner.COMMANDS_K, WorkflowRunner.WorkflowRunner.ONE_CMD_TOOL_K, WorkflowRunner.WorkflowRunner.WORKFLOW_STEPS_K, and WorkflowRunner.WorkflowRunner.WORKFLOWS_K.

◆ run()

WorkflowRunner.WorkflowRunner.run	(	self,
		working_dir,
		verbose = `False`
	)

Definition at line 114 of file WorkflowRunner.py.

    def run(self, working_dir, verbose=False):
        # run workflows in sequence
        for workflow_k in self.workflow_sequence:
            workflow = self.json_contents[workflow_k]
            if self.WORKFLOW_STEPS_K in workflow:
                steps = workflow[self.WORKFLOW_STEPS_K]
                for step_k in steps:
                    step = workflow[step_k]
                    commands = step[self.COMMANDS_K]
                    driver_name = step[self.ONE_CMD_TOOL_K]
                    option_builder = OptionBuilder(driver_name)
                    options = option_builder.build(commands)
                    # get the absolute path of the caller
                    driver_path = os.path.join(working_dir, driver_name)
                    cmd = [driver_path] + options
                    oneutils.run(cmd)
            elif self.CFG_REFERENCE_K in workflow:
                cfg_path = workflow[self.CFG_REFERENCE_K]['path']
                runner = CfgRunner(cfg_path)
                runner.run(working_dir, verbose)

References WorkflowRunner.WorkflowRunner.CFG_REFERENCE_K, WorkflowRunner.WorkflowRunner.COMMANDS_K, WorkflowRunner.WorkflowRunner.json_contents, WorkflowRunner.WorkflowRunner.ONE_CMD_TOOL_K, WorkflowRunner.WorkflowRunner.workflow_sequence, and WorkflowRunner.WorkflowRunner.WORKFLOW_STEPS_K.

Field Documentation

◆ adj

WorkflowRunner.WorkflowRunner.adj

Definition at line 46 of file WorkflowRunner.py.

Referenced by WorkflowRunner.WorkflowRunner._check_cycle().

◆ CFG_REFERENCE_K

str WorkflowRunner.WorkflowRunner.CFG_REFERENCE_K = 'cfg-reference'

static

Definition at line 29 of file WorkflowRunner.py.

Referenced by WorkflowRunner.WorkflowRunner._verify_workflow(), and WorkflowRunner.WorkflowRunner.run().

◆ COMMANDS_K

str WorkflowRunner.WorkflowRunner.COMMANDS_K = 'commands'

static

Definition at line 32 of file WorkflowRunner.py.

Referenced by WorkflowRunner.WorkflowRunner._verify_workflow(), and WorkflowRunner.WorkflowRunner.run().

◆ DEPENDENCIES_K

str WorkflowRunner.WorkflowRunner.DEPENDENCIES_K = 'run-after'

static

Definition at line 28 of file WorkflowRunner.py.

◆ json_contents

WorkflowRunner.WorkflowRunner.json_contents

Definition at line 37 of file WorkflowRunner.py.

Referenced by WorkflowRunner.WorkflowRunner.run().

◆ ONE_CMD_TOOL_K

str WorkflowRunner.WorkflowRunner.ONE_CMD_TOOL_K = 'one-cmd'

static

Definition at line 31 of file WorkflowRunner.py.

Referenced by WorkflowRunner.WorkflowRunner._verify_workflow(), and WorkflowRunner.WorkflowRunner.run().

◆ workflow_sequence

WorkflowRunner.WorkflowRunner.workflow_sequence

Definition at line 55 of file WorkflowRunner.py.

Referenced by WorkflowRunner.WorkflowRunner._check_cycle(), and WorkflowRunner.WorkflowRunner.run().

◆ WORKFLOW_STEPS_K

str WorkflowRunner.WorkflowRunner.WORKFLOW_STEPS_K = 'steps'

static

Definition at line 30 of file WorkflowRunner.py.

Referenced by WorkflowRunner.WorkflowRunner._verify_workflow(), and WorkflowRunner.WorkflowRunner.run().

◆ WORKFLOWS_K

str WorkflowRunner.WorkflowRunner.WORKFLOWS_K = 'workflows'

static

Definition at line 27 of file WorkflowRunner.py.

Referenced by WorkflowRunner.WorkflowRunner._verify_workflow().

The documentation for this class was generated from the following file:

compiler/one-cmds/onelib/WorkflowRunner.py

Public Member Functions

Data Fields

Static Public Attributes

Protected Member Functions

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ _check_cycle()

◆ _verify_workflow()

◆ run()

Field Documentation

◆ adj

◆ CFG_REFERENCE_K

◆ COMMANDS_K

◆ DEPENDENCIES_K

◆ json_contents

◆ ONE_CMD_TOOL_K

◆ workflow_sequence

◆ WORKFLOW_STEPS_K

◆ WORKFLOWS_K

◆ init()