Merge pull request #3116 from antgonza/add-default-processing

Add default processing

Author: wasade

qiita_db/metadata_template/prep_template.py

@@ -713,3 +713,196 @@ def modification_timestamp(self):
     @staticmethod
     def max_samples():
         return qdb.util.max_preparation_samples()
+
+    def add_default_workflow(self, user):
+        """The modification timestamp of the prep information
+
+        Parameters
+        ----------
+        user : qiita_db.user.User
+            The user that requested to add the default workflows
+
+        Returns
+        -------
+        ProcessingWorkflow
+            The workflow created
+
+        Raises
+        ------
+        ValueError
+            a. If this preparation doesn't have valid workflows
+            b. This preparation has been fully processed (no new steps needed)
+            c. If there is no valid initial artifact to start the workflow
+        """
+        # helper functions to avoid duplication of code
+
+        def _get_node_info(workflow, node):
+            # retrieves the merging scheme of a node
+            parent = list(workflow.graph.predecessors(node))
+            if parent:
+                parent = parent.pop()
+                pdp = parent.default_parameter
+                pcmd = pdp.command
+                pparams = pdp.values
+            else:
+                pcmd = None
+                pparams = {}
+
+            dp = node.default_parameter
+            cparams = dp.values
+            ccmd = dp.command
+
+            parent_cmd_name = None
+            parent_merging_scheme = None
+            if pcmd is not None:
+                parent_cmd_name = pcmd.name
+                parent_merging_scheme = pcmd.merging_scheme
+
+            return qdb.util.human_merging_scheme(
+                ccmd.name, ccmd.merging_scheme, parent_cmd_name,
+                parent_merging_scheme, cparams, [], pparams)
+
+        def _get_predecessors(workflow, node):
+            # recursive method to get predecessors of a given node
+            pred = []
+            for pnode in workflow.graph.predecessors(node):
+                pred = _get_predecessors(workflow, pnode)
+                cxns = {x[0]: x[2]
+                        for x in workflow.graph.get_edge_data(
+                            pnode, node)['connections'].connections}
+                data = [pnode, node, cxns]
+                if pred is None:
+                    pred = [data]
+                else:
+                    pred.append(data)
+                return pred
+
+        # Note: we are going to use the final BIOMs to figure out which
+        #       processing is missing from the back/end to the front, as this
+        #       will prevent generating unnecessary steps (AKA already provided
+        #       by another command), like "Split Library of Demuxed",
+        #       when "Split per Sample" is alrady generated
+        #
+        # The steps to generate the default workflow are as follow:
+        # 1. retrieve all valid merging schemes from valid jobs in the
+        #    current preparation
+        # 2. retrive all the valid workflows for the preparation data type and
+        #    find the final BIOM missing from the valid available merging
+        #    schemes
+        # 3. loop over the missing merging schemes and create the commands
+        #    missing to get to those processed samples and add them to a new
+        #    workflow
+
+        # 1.
+        prep_jobs = [j for c in self.artifact.descendants.nodes()
+                     for j in c.jobs(show_hidden=True)
+                     if j.command.software.type == 'artifact transformation']
+        merging_schemes = {
+            qdb.archive.Archive.get_merging_scheme_from_job(j): {
+                x: y.id for x, y in j.outputs.items()}
+            for j in prep_jobs if j.status == 'success' and not j.hidden}
+
+        # 2.
+        pt_dt = self.data_type()
+        workflows = [wk for wk in qdb.software.DefaultWorkflow.iter()
+                     if pt_dt in wk.data_type]
+        if not workflows:
+            # raises option a.
+            raise ValueError(f'This preparation data type: "{pt_dt}" does not '
+                             'have valid workflows')
+        missing_artifacts = dict()
+        for wk in workflows:
+            missing_artifacts[wk] = dict()
+            for node, degree in wk.graph.out_degree():
+                if degree != 0:
+                    continue
+                mscheme = _get_node_info(wk, node)
+                if mscheme not in merging_schemes:
+                    missing_artifacts[wk][mscheme] = node
+            if not missing_artifacts[wk]:
+                del missing_artifacts[wk]
+        if not missing_artifacts:
+            # raises option b.
+            raise ValueError('This preparation is complete')
+
+        # 3.
+        workflow = None
+        for wk, wk_data in missing_artifacts.items():
+            previous_jobs = dict()
+            for ma, node in wk_data.items():
+                predecessors = _get_predecessors(wk, node)
+                predecessors.reverse()
+                cmds_to_create = []
+                init_artifacts = None
+                for i, (pnode, cnode, cxns) in enumerate(predecessors):
+                    cdp = cnode.default_parameter
+                    cdp_cmd = cdp.command
+                    params = cdp.values.copy()
+
+                    icxns = {y: x for x, y in cxns.items()}
+                    reqp = {x: icxns[y[1][0]]
+                            for x, y in cdp_cmd.required_parameters.items()}
+                    cmds_to_create.append([cdp_cmd, params, reqp])
+
+                    info = _get_node_info(wk, pnode)
+                    if info in merging_schemes:
+                        if set(merging_schemes[info]) >= set(cxns):
+                            init_artifacts = merging_schemes[info]
+                            break
+                if init_artifacts is None:
+                    pdp = pnode.default_parameter
+                    pdp_cmd = pdp.command
+                    params = pdp.values.copy()
+                    reqp = {x: y[1][0]
+                            for x, y in pdp_cmd.required_parameters.items()}
+                    cmds_to_create.append([pdp_cmd, params, reqp])
+
+                    init_artifacts = {
+                        self.artifact.artifact_type: self.artifact.id}
+
+                cmds_to_create.reverse()
+                current_job = None
+                for i, (cmd, params, rp) in enumerate(cmds_to_create):
+                    previous_job = current_job
+                    if previous_job is None:
+                        req_params = dict()
+                        for iname, dname in rp.items():
+                            if dname not in init_artifacts:
+                                msg = (f'Missing Artifact type: "{dname}" in '
+                                       'this preparation; are you missing a '
+                                       'step to start?')
+                                # raises option c.
+                                raise ValueError(msg)
+                            req_params[iname] = init_artifacts[dname]
+                    else:
+                        req_params = dict()
+                        connections = dict()
+                        for iname, dname in rp.items():
+                            req_params[iname] = f'{previous_job.id}{dname}'
+                            connections[dname] = iname
+                    params.update(req_params)
+                    job_params = qdb.software.Parameters.load(
+                        cmd, values_dict=params)
+
+                    if job_params in previous_jobs.values():
+                        for x, y in previous_jobs.items():
+                            if job_params == y:
+                                current_job = x
+                        continue
+
+                    if workflow is None:
+                        PW = qdb.processing_job.ProcessingWorkflow
+                        workflow = PW.from_scratch(user, job_params)
+                        current_job = [j for j in workflow.graph.nodes()][0]
+                    else:
+                        if previous_job is None:
+                            current_job = workflow.add(
+                                job_params, req_params=req_params)
+                        else:
+                            current_job = workflow.add(
+                                job_params, req_params=req_params,
+                                connections={previous_job: connections})
+
+                    previous_jobs[current_job] = job_params
+
+        return workflow

qiita_db/metadata_template/test/test_prep_template.py

@@ -1340,11 +1340,36 @@ def test_artifact_setter_error(self):
 
     def test_artifact_setter(self):
         pt = qdb.metadata_template.prep_template.PrepTemplate.create(
-            self.metadata, self.test_study, self.data_type_id)
+            self.metadata, self.test_study, '16S')
         self.assertEqual(pt.artifact, None)
         artifact = qdb.artifact.Artifact.create(
             self.filepaths, "FASTQ", prep_template=pt)
         self.assertEqual(pt.artifact, artifact)
+
+        # here we can test that we can properly create a workflow
+        wk = pt.add_default_workflow(qdb.user.User('[email protected]'))
+        self.assertEqual(len(wk.graph.nodes), 2)
+        self.assertEqual(len(wk.graph.edges), 1)
+        self.assertEqual(
+            [x.command.name for x in wk.graph.nodes],
+            ['Split libraries FASTQ', 'Pick closed-reference OTUs'])
+
+        # now let's try to generate again and it should fail cause the jobs
+        # are alrady created
+        with self.assertRaisesRegex(ValueError, "Cannot create job because "
+                                    "the parameters are the same as jobs"):
+            pt.add_default_workflow(qdb.user.User('[email protected]'))
+
+        # now let's test that an error is raised when there is no valid initial
+        # input data; this moves the data type from FASTQ to taxa_summary
+        qdb.sql_connection.perform_as_transaction(
+            'UPDATE qiita.artifact SET artifact_type_id = 10 WHERE '
+            f'artifact_id = {pt.artifact.id}')
+        with self.assertRaisesRegex(ValueError, 'Missing Artifact type: '
+                                    '"FASTQ" in this preparation; are you '
+                                    'missing a step to start?'):
+            pt.add_default_workflow(qdb.user.User('[email protected]'))
+
         # cleaning
         qdb.artifact.Artifact.delete(artifact.id)
         qdb.metadata_template.prep_template.PrepTemplate.delete(pt.id)

qiita_pet/handlers/api_proxy/studies.py

@@ -236,6 +236,9 @@ def study_prep_get_req(study_id, user_id):
                 info['start_artifact'] = None
                 info['start_artifact_id'] = None
                 info['youngest_artifact'] = None
+                info['num_artifact_children'] = 0
+                info['youngest_artifact_name'] = None
+                info['youngest_artifact_type'] = None
                 info['ebi_experiment'] = 0
 
             dtype_infos.append(info)

qiita_pet/handlers/api_proxy/tests/test_studies.py

@@ -238,6 +238,9 @@ def test_study_prep_get_req_failed_EBI(self):
              'start_artifact_id': None,
              'creation_timestamp': pt.creation_timestamp,
              'modification_timestamp': pt.modification_timestamp,
+             'num_artifact_children': 0,
+             'youngest_artifact_name': None,
+             'youngest_artifact_type': None,
              'total_samples': 3}]
 
         exp = {
@@ -577,6 +580,9 @@ def test_study_prep_get_req(self):
                              'start_artifact_id': None,
                              'start_artifact': None,
                              'youngest_artifact': None,
+                             'num_artifact_children': 0,
+                             'youngest_artifact_name': None,
+                             'youngest_artifact_type': None,
                              'ebi_experiment': 0}]
         exp = {'status': 'success',
                'message': '',

qiita_pet/handlers/study_handlers/__init__.py

@@ -16,7 +16,7 @@
                    DataTypesMenuAJAX, StudyFilesAJAX, StudyGetTags, StudyTags,
                    Study)
 from .prep_template import (
-    PrepTemplateAJAX, PrepFilesHandler,
+    PrepTemplateAJAX, PrepFilesHandler, AddDefaultWorkflowHandler,
     NewPrepTemplateAjax, PrepTemplateSummaryAJAX)
 from .processing import (ListCommandsHandler, ListOptionsHandler,
                          WorkflowHandler, WorkflowRunHandler, JobAJAX)
@@ -31,7 +31,7 @@
            'VAMPSHandler', 'ListStudiesAJAX', 'ArtifactGraphAJAX',
            'ArtifactAdminAJAX', 'StudyIndexHandler', 'StudyBaseInfoAJAX',
            'SampleTemplateHandler', 'SampleTemplateOverviewHandler',
-           'SampleTemplateColumnsHandler',
+           'SampleTemplateColumnsHandler', 'AddDefaultWorkflowHandler',
            'PrepTemplateAJAX', 'NewArtifactHandler', 'PrepFilesHandler',
            'ListCommandsHandler', 'ListOptionsHandler', 'SampleAJAX',
            'StudyDeleteAjax', 'NewPrepTemplateAjax',

qiita_pet/handlers/study_handlers/prep_template.py

@@ -15,6 +15,7 @@
 from qiita_pet.handlers.base_handlers import BaseHandler
 from qiita_db.util import (get_files_from_uploads_folders, get_mountpoint,
                            supported_filepath_types)
+from qiita_db.metadata_template.prep_template import PrepTemplate
 from qiita_pet.handlers.api_proxy import (
     prep_template_ajax_get_req, new_prep_template_get_req,
     prep_template_summary_get_req)
@@ -32,6 +33,22 @@ def get(self):
                     study_id=study_id)
 
 
+class AddDefaultWorkflowHandler(BaseHandler):
+    @authenticated
+    def post(self):
+        prep_id = self.get_argument('prep_id')
+        msg_error = None
+        data = None
+        try:
+            workflow = PrepTemplate(prep_id).add_default_workflow(
+                self.current_user)
+            data = workflow.id
+        except Exception as error:
+            msg_error = str(error)
+
+        self.write({'data': data, 'msg_error': msg_error})
+
+
 class PrepTemplateSummaryAJAX(BaseHandler):
     @authenticated
     def get(self):

qiita_pet/static/js/networkVue.js

@@ -110,7 +110,7 @@ Vue.component('processing-graph', {
                 '</div>' +
               '</div>' +
             '</div>',
-  props: ['portal', 'graph-endpoint', 'jobs-endpoint', 'no-init-jobs-callback', 'is-analysis-pipeline'],
+  props: ['portal', 'graph-endpoint', 'jobs-endpoint', 'no-init-jobs-callback', 'is-analysis-pipeline', 'element-id'],
   methods: {
     /**
      *
@@ -997,6 +997,11 @@ Vue.component('processing-graph', {
           $("#processing-network-instructions-div").show();
           $("#show-hide-network-btn").show();
           $("#processing-job-div").hide();
+          if (vm.workflowId === null && vm.isAnalysisPipeline === false) {
+            $("#add-default-workflow").show();
+          } else {
+            $("#add-default-workflow").hide();
+          }
         }
       })
         .fail(function(object, status, error_msg) {
@@ -1138,6 +1143,10 @@ Vue.component('processing-graph', {
       '<tr>' +
         '<td><small>Job status (circles):</small></td>' +
         '<td>' + circle_statuses.join('') + '</td>' +
+        '<td rowspan="2" width="20px">&nbsp;</td>' +
+        '<td rowspan="2">' +
+            '<a class="btn btn-success form-control" id="add-default-workflow"><span class="glyphicon glyphicon-flash"></span> Add Default Workflow</a>' +
+        '</td>' +
       '</tr>' +
       '<tr>' +
         '<td><small>Artifact status (triangles):</small>' +
@@ -1146,6 +1155,20 @@ Vue.component('processing-graph', {
     '</table>';
     $('#circle-explanation').html(full_text);
 
+    $('#add-default-workflow').on('click', function () {
+      $('#add-default-workflow').attr('disabled', true);
+      document.getElementById('add-default-workflow').innerHTML = 'Submitting!';
+      $.post(vm.portal + '/study/process/workflow/default/', {prep_id: vm.elementId}, function(data) {
+        if (data['msg_error'] !== null){
+          $('#add-default-workflow').attr('disabled', false);
+          bootstrapAlert('Error generating workflow: ' + data['msg_error'].replace("\n", "<br/>"));
+        } else {
+          vm.updateGraph();
+        }
+      });
+      document.getElementById('add-default-workflow').innerHTML = ' Add Default Workflow';
+    });
+
     // This call to udpate graph will take care of updating the jobs
     // if the graph is not available
     vm.updateGraph();

qiita_pet/templates/analysis_description.html

@@ -91,7 +91,7 @@ <h2>
   <hr/>
 </div>
 <div id='analysis-graph-vue' style="margin-left: 15px">
-  <processing-graph v-bind:is-analysis-pipeline='true' ref="procGraph" portal="{% raw qiita_config.portal_dir %}" graph-endpoint="/analysis/description/{{analysis_id}}/graph/" jobs-endpoint="/analysis/description/{{analysis_id}}/jobs/"></processing-graph>
+  <processing-graph v-bind:is-analysis-pipeline='true' ref="procGraph" portal="{% raw qiita_config.portal_dir %}" graph-endpoint="/analysis/description/{{analysis_id}}/graph/" jobs-endpoint="/analysis/description/{{analysis_id}}/jobs/" element-id="{{analysis_id}}"></processing-graph>
 </div>
 <div class="row" id='processing-content-div'></div>
 

qiita_pet/templates/study_ajax/data_type_menu.html

@@ -62,7 +62,7 @@ <h4 class="panel-title">
                   <span id="prep-header-{{prep['id']}}">
                     {{prep['name']}} - ID: {{prep['id']}} - {{prep['status']}}
                   </span><br/>
-                    Raw files {% if prep['start_artifact'] == 'FASTQ' %}<i class="fa fa-check green"></i>{% else %}<i class="fa fa-times red"></i>{% end %}, processed {% if prep['num_artifact_children'] > 0 %}<i class="fa fa-check green"></i> {% if prep['num_artifact_children'] > 1 %} <i class="fa fa-check green"></i>{% end%}{% else %}<i class="fa fa-times red"></i>{% end %}, BIOM {% if prep['youngest_artifact_type'] == 'BIOM' %}<i class="fa fa-check green"></i>{% else %}<i class="fa fa-times red"></i>{% end %}
+                    Raw files {% if prep['start_artifact'] in ('per_sample_FASTQ', 'FASTA', 'FASTQ') %}<i class="fa fa-check green"></i>{% else %}<i class="fa fa-times red"></i>{% end %}, processed {% if prep['num_artifact_children'] > 0 %}<i class="fa fa-check green"></i> {% if prep['num_artifact_children'] > 1 %} <i class="fa fa-check green"></i>{% end%}{% else %}<i class="fa fa-times red"></i>{% end %}, BIOM {% if prep['youngest_artifact_type'] == 'BIOM' %}<i class="fa fa-check green"></i>{% else %}<i class="fa fa-times red"></i>{% end %}
                   <br />
                   Created: {{prep['creation_timestamp'].strftime('%B %-d, %Y')}}, last updated: {{prep['modification_timestamp'].strftime('%B %-d, %Y')}}
                 </a>

qiita_pet/templates/study_ajax/prep_summary.html

@@ -557,7 +557,7 @@ <h5>
         </div>
       </div>
       <div id="processing-graph-vue" class="tab-pane fade in active">
-        <processing-graph ref="procGraph" v-bind:is-analysis-pipeline='false' v-bind:no-init-jobs-callback="load_new_artifact" portal="{% raw qiita_config.portal_dir %}" graph-endpoint="/prep_template/{{prep_id}}/graph/" jobs-endpoint="/prep_template/{{prep_id}}/jobs/"></processing-graph>
+        <processing-graph ref="procGraph" v-bind:is-analysis-pipeline='false' v-bind:no-init-jobs-callback="load_new_artifact" portal="{% raw qiita_config.portal_dir %}" graph-endpoint="/prep_template/{{prep_id}}/graph/" jobs-endpoint="/prep_template/{{prep_id}}/jobs/" element-id="{{prep_id}}"></processing-graph>
       </div>
     </div>
   </div>