Fix .bam file handling for visualizations

refinery/analysis_manager/migrations/0009_analysisstatus_auxiliary_file_task_group_id.py

@@ -0,0 +1,19 @@
+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('analysis_manager', '0008_analysisstatus_galaxy_workflow_task_group_id'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='analysisstatus',
+            name='auxiliary_file_task_group_id',
+            field=models.UUIDField(null=True, editable=False),
+        ),
+    ]

refinery/analysis_manager/models.py

@@ -25,6 +25,7 @@ class AnalysisStatus(models.Model):
     galaxy_import_task_group_id = models.UUIDField(null=True, editable=False)
     galaxy_export_task_group_id = models.UUIDField(null=True, editable=False)
     galaxy_workflow_task_group_id = models.UUIDField(null=True, editable=False)
+    auxiliary_file_task_group_id = models.UUIDField(null=True, editable=False)
     #: state of Galaxy file imports
     galaxy_import_state = CharField(max_length=10, blank=True,
                                     choices=GALAXY_HISTORY_STATES)

refinery/analysis_manager/tasks.py

@@ -146,9 +146,48 @@ def _attach_workflow_outputs(analysis_uuid):
     """
     analysis = _get_analysis(analysis_uuid)
     analysis_status = _get_analysis_status(analysis_uuid)
-
     if analysis.workflow.type == Workflow.ANALYSIS_TYPE:
-        analysis.attach_derived_nodes_to_dataset()
+        if not analysis_status.auxiliary_file_task_group_id:
+            auxiliary_file_tasks_signatures = \
+                analysis.attach_derived_nodes_to_dataset()
+            logger.info(
+                "Starting auxiliary file generation and import for analysis "
+                "'%s'", analysis)
+            auxiliary_file_tasks = TaskSet(
+                tasks=auxiliary_file_tasks_signatures
+            ).apply_async()
+            auxiliary_file_tasks.save()
+            analysis_status.auxiliary_file_task_group_id = (
+                auxiliary_file_tasks.taskset_id
+            )
+            analysis_status.save()
+            run_analysis.retry(countdown=RETRY_INTERVAL)
+        # check if analysis results have finished downloading from Galaxy
+        auxiliary_file_tasks = get_taskset_result(
+            analysis_status.auxiliary_file_task_group_id
+        )
+        if not auxiliary_file_tasks.ready():
+            logger.debug("Auxiliary file import and generation "
+                         "running for analysis '%s'", analysis)
+            run_analysis.retry(countdown=RETRY_INTERVAL)
+        elif not auxiliary_file_tasks.successful():
+            error_msg = ("Analysis '{}' failed while generating "
+                         "auxiliary file".format(analysis))
+            logger.error(error_msg)
+            analysis.set_status(Analysis.FAILURE_STATUS, error_msg)
+            analysis.send_email()
+
+            get_taskset_result(
+                analysis_status.refinery_import_task_group_id
+            ).delete()
+            get_taskset_result(
+                analysis_status.galaxy_import_task_group_id
+            ).delete()
+            get_taskset_result(
+                analysis_status.galaxy_export_task_group_id
+            ).delete()
+            auxiliary_file_tasks.delete()
+            analysis.galaxy_cleanup()
     elif analysis.workflow.type == Workflow.DOWNLOAD_TYPE:
         analysis.attach_outputs_downloads()
     else:

refinery/core/models.py

@@ -1358,7 +1358,8 @@ def attach_derived_nodes_to_dataset(self):
             )
         )
         self._create_derived_data_file_nodes(graph_with_input_nodes_linked)
-        self._create_annotated_nodes()
+        auxiliary_file_import_tasks = self._create_annotated_nodes()
+        return auxiliary_file_import_tasks
 
     def attach_outputs_downloads(self):
         if self.results.all().count() == 0:
@@ -1409,6 +1410,7 @@ def _prepare_annotated_nodes(self, node_uuids):
         Call order is ensured through:
         core.tests.test__prepare_annotated_nodes_calls_methods_in_proper_order
         """
+        auxiliary_file_tasks = []
         for result in self.results.all():
             try:
                 item = FileStoreItem.objects.get(uuid=result.file_store_uuid)
@@ -1424,8 +1426,11 @@ def _prepare_annotated_nodes(self, node_uuids):
                              item.uuid, exc)
             else:
                 if node.is_derived():
-                    node.run_generate_auxiliary_node_task()
+                    subtask = node.run_generate_auxiliary_node_task()
+                    if subtask is not None:
+                        auxiliary_file_tasks += [subtask]
         index_annotated_nodes_selection(node_uuids)
+        return auxiliary_file_tasks
 
     def _get_output_connection_to_analysis_result_mapping(self):
         """Create and return a dict mapping each "output" type
@@ -1627,7 +1632,8 @@ def _create_annotated_nodes(self):
             self.get_input_node_study().uuid,
             self.get_input_node_assay().uuid
         )
-        self._prepare_annotated_nodes(node_uuids)
+        auxiliary_file_tasks = self._prepare_annotated_nodes(node_uuids)
+        return auxiliary_file_tasks
 
     def get_refinery_import_task_signatures(self):
         """Create and return a list of file import task signatures for the

refinery/data_set_manager/models.py

@@ -14,6 +14,7 @@
 from django.dispatch import receiver
 
 from celery.result import AsyncResult
+from celery import chain
 from django_extensions.db.fields import UUIDField
 import requests
 from requests.exceptions import HTTPError
@@ -22,7 +23,7 @@
 from core.utils import delete_analysis_index, skip_if_test_run
 import data_set_manager
 from file_store.models import FileStoreItem
-
+from file_store.tasks import FileImportTask
 """
 TODO: Refactor import data_set_manager. Importing
 data_set_manager.tasks.generate_auxiliary_file()
@@ -497,6 +498,9 @@ class Node(models.Model):
     TYPES = ASSAYS | FILES | {
         SOURCE, SAMPLE, EXTRACT, LABELED_EXTRACT, SCAN, NORMALIZATION,
         DATA_TRANSFORMATION}
+    # Currently we only need to create an auxiliary file for bam, but WIG
+    # needs an index file as well
+    AUXILIARY_FILES_NEEDED_FOR_VISUALIZATION = ['bam']
 
     uuid = UUIDField(unique=True, auto=True)
     study = models.ForeignKey(Study, db_index=True)
@@ -598,11 +602,18 @@ def _create_and_associate_auxiliary_node(self, filestore_item):
                 self.add_child(node_object)
                 return node_object
 
-    def get_children(self):
+    def get_children(self, auxiliary_filter=None):
         """
         Return a list of child Node's uuids for a given Node
         """
-        return [child.uuid for child in self.children.all()]
+        if auxiliary_filter is None:
+            return [child.uuid for child in self.children.all()]
+        else:
+            return [
+                child.uuid for child in self.children.filter(
+                    is_auxiliary_node=auxiliary_filter
+                )
+            ]
 
     def get_parents(self):
         """
@@ -646,19 +657,28 @@ def run_generate_auxiliary_node_task(self):
                 self.file_item.filetype.used_for_visualization and
                 self.file_item.datafile and
                 settings.REFINERY_AUXILIARY_FILE_GENERATION ==
-                'on_file_import'):
+                'on_file_import' and
+                self.file_item.get_extension().lower() in
+                self.AUXILIARY_FILES_NEEDED_FOR_VISUALIZATION):
             # Create an empty FileStoreItem (we do the datafile association
             # within the generate_auxiliary_file task
             auxiliary_file_store_item = FileStoreItem.objects.create()
 
             auxiliary_node = self._create_and_associate_auxiliary_node(
                 auxiliary_file_store_item
             )
-            result = data_set_manager.tasks.generate_auxiliary_file.delay(
-                auxiliary_node, self.file_item
+            generate = data_set_manager.tasks.generate_auxiliary_file.subtask(
+                (auxiliary_node, self.file_item,)
             )
-            auxiliary_file_store_item.import_task_id = result.task_id
-            auxiliary_file_store_item.save()
+            file_import = FileImportTask().subtask(
+                (auxiliary_node.file_item.uuid, None,),
+                immutable=True
+            )
+            generate_and_import = chain(generate, file_import)
+            return generate_and_import
+        else:
+            logger.debug("No auxiliary Node needs be generated")
+            return None
 
     def get_auxiliary_file_generation_task_state(self):
         """Return the generate_auxiliary_file task state for a given auxiliary

refinery/data_set_manager/tasks.py

@@ -1,18 +1,21 @@
 from datetime import date
 import logging
+import os
 import time
 
+from django.conf import settings
 from django.contrib.auth.models import User
 from django.db import transaction
 
 import botocore
 import celery
 from celery.task import task
 import pysam
+import tempfile
 
 from core.models import DataSet, ExtendedGroup, FileStoreItem
 from file_store.models import FileExtension, generate_file_source_translator
-from file_store.tasks import FileImportTask
+from file_store.tasks import FileImportTask, download_s3_object
 
 from .isa_tab_parser import IsaTabParser
 from .models import Investigation, Node, initialize_attribute_order
@@ -273,7 +276,7 @@ def parse_isatab(username, public, path, identity_id=None,
         return data_set_uuid
 
 
-@task()
+@task(soft_time_limit=3600)
 def generate_auxiliary_file(auxiliary_node, parent_node_file_store_item):
     """Task that will generate an auxiliary file for visualization purposes
     with specific file generation tasks going on for different FileTypes
@@ -287,8 +290,13 @@ def generate_auxiliary_file(auxiliary_node, parent_node_file_store_item):
     :type parent_node_file_store_item: FileStoreItem
     """
     generate_auxiliary_file.update_state(state=celery.states.STARTED)
+    datafile = parent_node_file_store_item.datafile
+
     try:
-        datafile_path = parent_node_file_store_item.datafile.path
+        if not settings.REFINERY_S3_USER_DATA:
+            datafile_path = datafile.path
+        else:
+            datafile_path = datafile.name
     except (NotImplementedError, ValueError):
         datafile_path = None
     try:
@@ -305,6 +313,7 @@ def generate_auxiliary_file(auxiliary_node, parent_node_file_store_item):
 
         logger.debug("Auxiliary file for %s generated in %s "
                      "seconds." % (datafile_path, time.time() - start_time))
+
     except Exception as e:
         logger.error(
             "Something went wrong while trying to generate the auxiliary file "
@@ -332,13 +341,25 @@ def generate_bam_index(auxiliary_file_store_item_uuid, datafile_path):
     # fail if we can't get what we want.
     bam_index_file_extension = FileExtension.objects.get(name="bai").name
     auxiliary_file_store_item = FileStoreItem.objects.get(
-        uuid=auxiliary_file_store_item_uuid)
+        uuid=auxiliary_file_store_item_uuid
+    )
 
     # Leverage pysam library to generate bam index file
     # FIXME: This should be refactored once we don't have a need for
     # Standalone IGV because this is creating a bam_index file in the same
     # directory as it's bam file
-    pysam.index(bytes(datafile_path))
+    if settings.REFINERY_S3_USER_DATA:
+        key = datafile_path
+        bucket = settings.MEDIA_BUCKET
+        temp_file = os.path.join(tempfile.gettempdir(), key)
+        os.makedirs(os.path.abspath(os.path.join(temp_file, os.pardir)))
+        with open(temp_file, 'wb') as destination:
+            download_s3_object(bucket, key, destination)
+        pysam.index(bytes(temp_file))
+        datafile_path = temp_file
+        os.remove(temp_file)
+    else:
+        pysam.index(bytes(datafile_path))
 
     # Map source field of FileStoreItem to path of newly created bam index file
     auxiliary_file_store_item.source = "{}.{}".format(

refinery/data_set_manager/test_models.py

@@ -102,6 +102,14 @@ def test_get_children(self):
 
         # Check inverse relationship:
         self.assertEqual(self.node.uuid, self.another_node.get_parents()[0])
+        auxiliary_node = Node.objects.create(
+            assay=self.assay,
+            study=self.study,
+            is_auxiliary_node=True
+        )
+        self.node.add_child(auxiliary_node)
+        self.assertEqual(self.node.get_children(auxiliary_filter=True),
+                         [auxiliary_node.uuid])
 
     def test_get_parents(self):
         self.assertEqual(self.another_node.get_parents(), [])

refinery/tool_manager/models.py

@@ -487,6 +487,7 @@ class VisualizationTool(Tool):
     """
     API_PREFIX = "api_prefix"
     FILE_URL = "file_url"
+    AUXILIARY_FILE_LIST = "auxiliary_file_list"
     INPUT_NODE_INFORMATION = "node_info"
     NODE_SOLR_INFO = "node_solr_info"
     ALL_NODE_INFORMATION = "all_node_info"
@@ -553,7 +554,13 @@ def _get_detailed_nodes_dict(self, node_uuid_list,
                 self.FILE_URL: get_file_url_from_node_uuid(
                     node["uuid"],
                     require_valid_url=require_valid_urls
-                )
+                ),
+                self.AUXILIARY_FILE_LIST: [
+                    get_file_url_from_node_uuid(
+                        child_uuid, require_valid_url=require_valid_urls
+                    ) for child_uuid in Node.objects.get(uuid=node["uuid"]).
+                    get_children(auxiliary_filter=True)
+                ]
             }
             for node in solr_response_json["nodes"]
         }

refinery/tool_manager/tests.py

@@ -719,6 +719,10 @@ def _create_detailed_nodes_dict(self, nodes):
         return {
             node.uuid: {
                 'file_url': self.node.file_item.get_datafile_url(),
+                'auxiliary_file_list': [
+                    child_uuid for child_uuid in
+                    self.node.get_children(auxiliary_filter=True)
+                ],
                 VisualizationTool.NODE_SOLR_INFO: {
                     'uuid': node.uuid,
                     'name': node.name,