ng6workflow.py

#
# Copyright (C) 2012 INRA
# 
# 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 inspect
import os
import pickle
import datetime

from jflow.workflow import Workflow
from jflow.parameter import Parameter, date

from ng6.project import Project
from ng6.run import Run


class NG6Workflow (Workflow):
    """
    """             

    def __init__(self, args={}, id=None, function= "process", parameters_section="parameters", add_run_params=True):
        Workflow.__init__(self, args, id, function, parameters_section)
        if add_run_params: 
            self.parameters.extend(self._add_run_parameters())
            self.args = self._extend_and_format_args(self.parameters, args)

    def _add_run_parameters(self):
        orequired = []
        # add run parameters
        orequired.append(Parameter("project_id", "The project id the run belongs to", "--project-id", required=True, type=int))
        orequired.append(Parameter("run_name", "Give a name to your run", "--name", required=True))
        orequired.append(Parameter("run_description", "Give a description to your run", "--description", required=True))
        orequired.append(Parameter("run_date", "When were the data produced","--date",  required=True, type=date))
        orequired.append(Parameter("data_nature", "Are Sequences cDNA, genomique, RNA, ...", "--data-nature", required=True))
        orequired.append(Parameter("sequencer", "Which sequencer produced the data", "--sequencer", required=True))
        orequired.append(Parameter("species", "Which species has been sequenced", "--species", required=True))
        orequired.append(Parameter("type", "What type of data is it (1 lane, 1 region)", "--type", required=True))        
        return orequired

    def pre_process(self):

        self.project = None
        self.runobj = None

        # if starting from an existing project
        if self.args.has_key("project_id"):
            self.project = Project.get_from_id(self.args["project_id"])
            self.metadata.append("project_id="+str(self.args["project_id"]))
        # if starting from a brand new project
        elif self.args.has_key("project_name") and self.args.has_key("project_description") and self.args.has_key("admin_login"):
            self.project = Project(self.args["project_name"], self.args["project_description"], self.args["admin_login"])
            self.metadata.append("project_id="+str(self.project.id))
        # if starting from an existing run
        if self.args.has_key("run_id"):
            self.runobj = Run.get_from_id(self.args["run_id"])
            self.metadata.append("run_id="+str(self.args["run_id"]))
        # if starting from a brand new run
        elif self.args.has_key("run_date") and self.args.has_key("run_name") and self.args.has_key("species") and self.args.has_key("data_nature") \
            and self.args.has_key("type") and self.args.has_key("run_description") and self.args.has_key("sequencer") and self.args.has_key("project_id"):
            # build the project
            self.project = Project.get_from_id(self.args["project_id"])
            # and the run
            self.runobj = Run(self.args["run_name"], self.args["run_date"], self.args["species"], self.args["data_nature"], 
                              self.args["type"], self.args["run_description"], self.args["sequencer"])
            # then add the run to the project
            self.project.add_run(self.runobj)
            # if it's a run with some index, let's write down description
            if self.args.has_key("sample") and self.args["sample"]:
                try:
                    samples = {}
                    for sample in self.args["sample"]:
                        samples[sample["sample_name"]] = sample["sample_description"]
                    self.runobj.add_mids_description(samples)
                except: pass
            self.metadata.append("run_id="+str(self.runobj.id))
    
    def add_component(self, component_name, args=[], kwargs={}, component_prefix="default", parent=None, addto="run"):
        # first build and check if this component is OK
        if self.comp_pckg.has_key(component_name):
            my_pckge = __import__(self.comp_pckg[component_name], globals(), locals(), [component_name], -1)
            # build the object and define required field
            cmpt_object = getattr(my_pckge, component_name)()
            cmpt_object.output_directory = self.get_component_output_directory(component_name, component_prefix)
            cmpt_object.prefix = component_prefix
            if kwargs: cmpt_object.define_parameters(**kwargs)
            else: cmpt_object.define_parameters(*args)
            # if the built object is an analysis
            for derived_class in cmpt_object.__class__.__bases__:
                if derived_class.__name__ == "Analysis":
                    # add parent and project/run information
                    cmpt_object.parent = parent
                    if addto == "project": cmpt_object.project = self.project
                    elif addto == "run": cmpt_object.run = self.runobj  
            # add the component
            self.components.append(cmpt_object)
            # if this one require a dynamic pipeline, execute the first part
            if cmpt_object.is_dynamic():
                self._execute_weaver()
                # update outputs
                for output in cmpt_object.get_dynamic_outputs():
                    output.update()
            return cmpt_object
        else:
            raise ImportError(component_name + " component cannot be loaded, available components are: {0}".format(", ".join(self.comp_pckg.keys())))
        
    def post_process(self):
        # once everything done, sync directories
        if self.runobj:
            self.runobj.sync()
        elif self.project:
            self.project.sync()