From d294f6196a3e55ad3e0f33be978cce540756a86e Mon Sep 17 00:00:00 2001 From: Jürgen Knödlseder Date: Wed, 24 Feb 2016 01:50:20 +0100 Subject: [PATCH] Add csworkflow script as prove of principle for a XML file-based analysis workflow. --- cscripts/csworkflow.py | 420 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ test/data/workflow.xml | 129 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 549 insertions(+), 0 deletions(-) create mode 100755 cscripts/csworkflow.py create mode 100644 test/data/workflow.xml diff --git a/cscripts/csworkflow.py b/cscripts/csworkflow.py new file mode 100755 index 0000000..9a5c205 --- /dev/null +++ b/cscripts/csworkflow.py @@ -0,0 +1,420 @@ +#! /usr/bin/env python +# ========================================================================== +# This script executes a workflow defined in an XML file. +# +# Copyright (C) 2016 Juergen Knoedlseder +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# +# ========================================================================== +import gammalib +import ctools +import cscripts +import sys + + +# ================ # +# csworkflow class # +# ================ # +class csworkflow(ctools.cscript): + """ + """ + def __init__(self, *argv): + """ + Constructor. + """ + # Set name and version + self.name = "csworkflow" + self.version = "1.0.0" + + # Set members + self.workflow = gammalib.GXml() + self.actors = [] + + # Make sure that parfile exists + self.parfile() + + # Initialise application + if len(argv) == 0: + ctools.cscript.__init__(self, self.name, self.version) + elif len(argv) ==1: + ctools.cscript.__init__(self, self.name, self.version, *argv) + else: + raise TypeError("Invalid number of arguments given.") + + # Set logger properties + self.log_header() + self.log.date(True) + + # Return + return + + def __del__(self): + """ + Destructor. + """ + # Return + return + + def parfile(self): + """ + Check if parfile exists. If parfile does not exist then create a + default parfile. This kluge avoids shipping the cscript with a parfile. + """ + # Set parfile name + parfile = self.name+".par" + + # Load parfile or create one + try: + pars = gammalib.GApplicationPars(parfile) + except: + # Signal if parfile was not found + print("Parfile "+parfile+" not found. Create default parfile.") + + # Create default parfile + pars = gammalib.GApplicationPars() + pars.append(gammalib.GApplicationPar("inflow","f","h","test/data/workflow.xml","","","Input workflow XML file")) + pars.append_standard() + pars.append(gammalib.GApplicationPar("logfile","f","h","csworkflow.log","","","Log filename")) + pars.save(parfile) + + # Return + return + + def get_parameters(self): + """ + Get parameters from parfile. + """ + # Load workflow XML file + xmlfile = self["inflow"].filename() + self.workflow = gammalib.GXml(xmlfile.url()) + + # Return + return + + def execute(self): + """ + Execute the script. + """ + # Run the script + self.run() + + # Return + return + + def run(self): + """ + Run the script. + """ + # Switch screen logging on in debug mode + if self.logDebug(): + self.log.cout(True) + + # Get parameters + self.get_parameters() + + # Write input parameters into logger + if self.logTerse(): + self.log_parameters() + self.log("\n") + + # Parse XML file + if self.logTerse(): + self.log("\n") + self.log.header1("Parse workflow XML file") + self.parse_workflow() + + # Execute workflow + if self.logTerse(): + self.log("\n") + self.log.header1("Execute workflow") + self.execute_workflow() + + # Return + return + + def parse_workflow(self): + """ + """ + # Get workflow + workflow = self.workflow.element("workflow") + + # Get number of actors + num_actors = workflow.elements("actor") + + # Initialise actors + self.actors = [] + + # Loop over all actors + for i in range(num_actors): + + # Get actor + actor = workflow.element("actor", i) + + # Initialise parameters and input actors + input_parameters = [] + output_parameters = [] + input_actors = [] + output_actors = [] + + # Get actor attributes + name = actor.attribute("name") + tool = actor.attribute("tool") + + # Get actor input parameters + if actor.elements("input") > 0: + actor_inputs = actor.element("input") + num_inputs = actor_inputs.elements("parameter") + for k in range(num_inputs): + input = actor_inputs.element("parameter", k) + input_name = input.attribute("name") + input_value = input.attribute("value") + input_actor = input.attribute("actor") + parameter = {'name': input_name, \ + 'value': input_value, \ + 'actor': input_actor} + input_parameters.append(parameter) + if input_actor != "": + if input_actor not in input_actors: + input_actors.append(input_actor) + + # Get actor output parameters + if actor.elements("output") > 0: + actor_output = actor.element("output") + num_outputs = actor_output.elements("parameter") + for k in range(num_outputs): + output = actor_output.element("parameter", k) + output_name = output.attribute("name") + output_value = output.attribute("value") + output_actor = output.attribute("actor") + parameter = {'name': output_name, \ + 'value': output_value, \ + 'actor': output_actor} + output_parameters.append(parameter) + if output_actor != "": + if output_actor not in output_actors: + output_actors.append(output_actor) + + # Determine number of dependencies + num_inputs = len(input_actors) + + # Set actor status + if num_inputs > 0: + status = 'waiting for input' + else: + status = 'ready' + + # Create actor entry + entry = {'name': name, + 'tool': tool, + 'input_parameters': input_parameters, + 'input_actors': input_actors, + 'output_parameters': output_parameters, + 'output_actors': output_actors, + 'status': status} + + # Append entry + self.actors.append(entry) + + # Log information about actors + if self.logNormal(): + self.log.parformat("Actor \""+name+"\"") + self.log(tool) + self.log("\n") + if num_inputs == 0: + self.log.parformat(" Predecessor") + self.log("none\n") + else: + if num_inputs == 1: + self.log.parformat(" Predecessor") + else: + self.log.parformat(" Predecessors") + for k in range(num_inputs): + if k > 0: + self.log(", ") + self.log("\""+input_actors[k]+"\"") + self.log("\n") + + # Return + return + + def execute_workflow(self): + """ + """ + # Continue while there are actors + while len(self.actors) > 0: + + # Find actors which are ready + num_executed = 0 + for actor in self.actors: + + # If actor is ready then execute it + if actor['status'] == 'ready': + + # Log execution start + if self.logNormal(): + self.log.parformat("Execute actor") + self.log("\""+actor['name']+"\"\n") + + # Execute actor + self.execute_actor(actor) + + # Log execution finish + if self.logNormal(): + self.log.parformat("Finished actor execution") + self.log("\""+actor['name']+"\"\n") + + # Set actor status to finished + actor['status'] = 'finished' + + # Increment number of executed actors + num_executed += 1 + + # Break if no actors have been executed + if num_executed < 1: + break + + # Update actor status + for actor in self.actors: + if actor['status'] != 'finished': + input_actors = actor['input_actors'] + ready = True + for input_actor in input_actors: + a = self.get_actor(input_actor) + if a['status'] != 'finished': + ready = False + break + if ready: + actor['status'] = 'ready' + + # Return + return + + def execute_actor(self, actor): + """ + Excute an actor. + """ + # Log input parameters + if self.logNormal(): + pars = actor['input_parameters'] + for par in pars: + self.log.parformat(" Input parameter") + self.log(par['name']) + self.log("=") + self.log(self.get_parameter_value(par)) + self.log("\n") + + # Log output parameters + if self.logNormal(): + pars = actor['output_parameters'] + for par in pars: + self.log.parformat(" Output parameter") + self.log(par['name']) + self.log("=") + self.log(self.get_parameter_value(par)) + self.log("\n") + + # Set actor tool + if 'tool' in actor: + tool = actor['tool'] + if tool[:2] == 'ct': + tool_eval = 'ctools.'+tool+'()' + elif tool[:2] == 'cs': + tool_eval = 'cscripts.'+tool+'()' + else: + tool_eval = "" + if tool_eval == "": + return + + # Create actor tool object + object = eval(tool_eval) + + # Set actor input parameters + pars = actor['input_parameters'] + for par in pars: + app_par = object[par['name']] + app_par.value(self.get_parameter_value(par)) + object[par['name']] = app_par + + # Set actor output parameters + pars = actor['output_parameters'] + for par in pars: + app_par = object[par['name']] + app_par.value(self.get_parameter_value(par)) + object[par['name']] = app_par + + # Execute actor + object.logFileOpen() + object.execute() + + # Return + return + + def get_parameter_value(self, par): + """ + Get the parameter value from the respective actor. + """ + # If the actor is empty then simply return the value + if (par['actor'] == ""): + value = par['value'] + + # ... otherwise get the value from the actor + else: + found = False + actor = self.get_actor(par['actor']) + for output in actor['output_parameters']: + if output['name'] == par['value']: + value = output['value'] + found = True + break + if not found: + msg = "Parameter \""+par['value']+"\" is not an output "\ + "parameter of actor \""+par['actor']+"\"." + raise RuntimeError(msg) + + # Return value + return value + + def get_actor(self, name): + """ + Get the actor with the specified name. + """ + # Return actor with specified name + for actor in self.actors: + if actor['name'] == name: + return actor + + # Throw an exception + msg = "Actor \""+name+"\" not found." + raise RuntimeError(msg) + + # Return + return + + +# ======================== # +# Main routine entry point # +# ======================== # +if __name__ == '__main__': + """ + Executes workflow. + """ + # Create instance of application + app = csworkflow(sys.argv) + + # Open logfile + app.logFileOpen() + + # Execute application + app.execute() diff --git a/test/data/workflow.xml b/test/data/workflow.xml new file mode 100644 index 0000000..1f494a8 --- /dev/null +++ b/test/data/workflow.xml @@ -0,0 +1,129 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + -- libgit2 0.21.4