csworkflow.py

Go to the documentation of this file.

#! /usr/bin/env python
# ==========================================================================
# Executes analysis workflow
#
# Copyright (C) 2016-2022 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 <http://www.gnu.org/licenses/>.
#
# ==========================================================================
import sys
import gammalib
import ctools
import cscripts


# ================ #
# csworkflow class #
# ================ #
class csworkflow(ctools.cscript):
    """
    Executes an analysis workflow.

    The ``csworkflow`` script executes an analysis workflow defined in an
    XML file.
    """

    # Constructor
    def __init__(self, *argv):
        """
        Constructor.

        Parameters
        ----------
        argv : list of str
            List of IRAF command line parameter strings of the form
            ``parameter=3``.

        Raises
        ------
        TypeError
            An invalid number of command line arguments was provided.
        """
        # Initialise application by calling the base class constructor
        self._init_cscript(self.__class__.__name__, ctools.__version__, argv)

        # Set members
        self._workflow = gammalib.GXml()
        self._actors   = []

        # Return
        return


    # Private methods
    #
    # Query all user parameters
    def _get_parameters(self):

        # Load workflow XML file
        xmlfile        = self['inflow'].filename()
        self._workflow = gammalib.GXml(xmlfile.url())

        #  Write input parameters into logger
        self._log_parameters(gammalib.TERSE)

        # Return
        return

    # Parse the workflow XML file
    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_par   = actor_inputs.element('parameter', k)
                    input_name  = input_par.attribute('name')
                    input_value = input_par.attribute('value')
                    input_actor = input_par.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_par   = actor_output.element('parameter', k)
                    output_name  = output_par.attribute('name')
                    output_value = output_par.attribute('value')
                    output_actor = output_par.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
            self._log_value(gammalib.NORMAL, 'Actor "%s"' % name, tool)

            # Compute list of predecessors
            if num_inputs == 0:
                predecessors = 'none'
            else:
                predecessors = ''
                for k in range(num_inputs):
                    if k > 0:
                        predecessors += ', '
                    predecessors += '"'+input_actors[k]+'"'
 
            # Log predecessors
            self._log_value(gammalib.NORMAL,
                            gammalib.number('  Predecessor', num_inputs),
                            predecessors)

        # Return
        return

    # Execute a workflow
    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
                    self._log_value(gammalib.NORMAL, 'Execute actor',
                                    '"%s"' % actor['name'])

                    # Execute actor
                    self._execute_actor(actor)

                    # Log execution finish
                    self._log_value(gammalib.NORMAL, 'Finished actor execution',
                                    '"%s"' % actor['name'])

                    # 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

    # Execute an actor
    def _execute_actor(self, actor):

        # Log input parameters
        pars = actor['input_parameters']
        for par in pars:
            self._log_value(gammalib.NORMAL, '  Input parameter', '%s=%s' %
                            (par['name'], self._get_parameter_value(par)))

        # Log output parameters
        pars = actor['output_parameters']
        for par in pars:
            self._log_value(gammalib.NORMAL, '  Output parameter', '%s=%s' %
                            (par['name'], self._get_parameter_value(par)))

        # 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
        tool = eval(tool_eval)

        # Set actor input parameters
        pars = actor['input_parameters']
        for par in pars:
            app_par = tool[par['name']]
            app_par.value(self._get_parameter_value(par))
            tool[par['name']] = app_par

        # Set actor output parameters
        pars = actor['output_parameters']
        for par in pars:
            app_par = tool[par['name']]
            app_par.value(self._get_parameter_value(par))
            tool[par['name']] = app_par

        # Execute actor
        tool.logFileOpen()
        tool.execute()
        
        # Return
        return

    # Get the parameter value from the respective actor
    def _get_parameter_value(self, par):

        # 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

    # Get the actor with the specified name
    def _get_actor(self, 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


    # Public methods
    def process(self):
        """
        Process the script
        """
        # Get parameters
        self._get_parameters()

        # 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


# ======================== #
# Main routine entry point #
# ======================== #
if __name__ == '__main__':

    # Create instance of application
    app = csworkflow(sys.argv)

    # Execute application
    app.execute()