add feature table uploader functionality basics

Author: adamcantor22

mmeds/database/documents.py

@@ -35,7 +35,6 @@ class MMEDSDoc(men.Document):
     testing = men.BooleanField(required=True)
     owner = men.StringField(max_length=100, required=True)
     email = men.StringField(max_length=100)
-    path = men.StringField(max_length=300)
     access_code = men.StringField(max_length=50)
     doc_type = men.EnumField(DocType, required=True)
     is_alive = men.BooleanField()
@@ -166,14 +165,14 @@ class DataDoc(MMEDSDoc):
     data_type = men.StringField()
     studies = men.ListField(men.ReferenceField(MMEDSDoc))
     files = men.MapField(field=men.FileField())
-    latest_version = men.BooleanField()
 
 
 class FeatureTableDoc(MMEDSDoc):
     """
     MongoDB Document for storing feature tables
     """
     table_name = men.StringField()
+    table_type = men.StringField()
     studies = men.ListField(men.ReferenceField(MMEDSDoc))
     from_analysis = men.ReferenceField(MMEDSDoc)
     table = men.FileField()

mmeds/database/feature_table_uploader.py

@@ -9,7 +9,7 @@
 from datetime import datetime
 from pathlib import Path
 from multiprocessing import Process
-from mmeds.error import NoResultError, InvalidUploadError
+from mmeds.error import NoResultError
 from mmeds.logging import Logger
 from mmeds.util import (send_email, create_local_copy)
 
@@ -18,15 +18,15 @@ class FeatureTableUploader(Process):
     """
     This class handles the processing and uploading of observation matrix tables
     """
-    def __init__(self, access_code, owner, data_name, data_type, data_files, public, testing):
+    def __init__(self, access_code, owner, studies, table_name, table_type, table_file, public, testing):
         warnings.simplefilter('ignore')
         super().__init__()
-        Logger.debug('DataUploader created with params')
+        Logger.debug('FeatureTableUploader created with params')
         Logger.debug({
             'owner': owner,
-            'data_name': data_name,
-            'data_type': data_type,
-            'data_files': data_files,
+            'table_name': table_name,
+            'table_type': table_type,
+            'table_file': table_file,
             'public': public,
             'testing': testing
         })
@@ -35,24 +35,8 @@ def __init__(self, access_code, owner, data_name, data_type, data_files, public,
         self.access_code = access_code
         self.owner = owner
         self.testing = testing
-        self.data_name = data_name
+        self.table_name = table_name
         self.public = public
-        self.datafiles = data_files
-
-        if data_type == "MultiplexedPairedEndSingleBarcodes":
-            self.reads_type = "paired"
-            self.barcodes_type = "single"
-            self.demultiplexed = False
-        elif data_type == "MultiplexedPairedEndDualBarcodes":
-            self.reads_type = "paired"
-            self.barcodes_type = "dual"
-            self.demultiplexed = False
-        elif data_type == "DemultiplexedPairedEnd":
-            self.reads_type = None
-            self.barcodes_type = None
-            self.demultiplexed = True
-        else:
-            raise InvalidUploadError(f"Got unexpected data type '{data_type}'")
 
         # If testing connect to test server
         if testing:
@@ -82,34 +66,24 @@ def __init__(self, access_code, owner, data_name, data_type, data_files, public,
                                      host=sec.MONGO_HOST)
 
         # Create the document
-        self.data = docs.DataDoc(created=datetime.utcnow(),
-                                 last_accessed=datetime.utcnow(),
-                                 testing=self.testing,
-                                 doc_type=docs.DocType.DATA,
-                                 access_code=self.access_code,
-                                 data_name=self.data_name,
-                                 data_type=self.data_type,
-                                 files=[],
-                                 owner=self.owner,
-                                 public=self.public)
-
-        self.data.save()
-
-        count = 0
-        new_dir = fig.SEQUENCING_DIR / ('{}_{}_{}'.format(self.owner, self.data_name, count))
-
-        while new_dir.is_dir():
-            count += 1
-            new_dir = fig.SEQUENCING_DIR / ('{}_{}_{}'.format(self.owner, self.data_name, count))
-        new_dir.mkdir()
-
-        self.path = Path(new_dir) / 'database_dir'
+        self.doc = docs.FeatureTableDoc(created=datetime.utcnow(),
+                                        last_accessed=datetime.utcnow(),
+                                        testing=self.testing,
+                                        doc_type=docs.DocType.FEATURE_TABLE,
+                                        access_code=self.access_code,
+                                        table_name=self.table_name,
+                                        table_type=self.table_type,
+                                        studies=self.studies,
+                                        owner=self.owner,
+                                        public=self.public,
+                                        latest_version=True)
+        self.doc.save()
 
     def get_info(self):
         """ Method to return a dictionary of relevant info for the process log """
         info = {
             'created': self.created,
-            'type': 'upload-run',
+            'type': 'upload-feature-table',
             'owner': self.owner,
             'pid': self.pid,
             'name': self.name,
@@ -121,9 +95,9 @@ def run(self):
         """
         Thread that handles the upload of sequencing run files.
         """
-        self.data.update(is_alive=True)
-        self.data.save()
-        Logger.debug('Handling upload for sequencing run {} for user {}'.format(self.sequencing_run_name, self.owner))
+        self.doc.update(is_alive=True)
+        self.doc.save()
+        Logger.debug('Handling upload for feature table {} for user {}'.format(self.table_name, self.owner))
 
         # If the owner is None set user_id to 0
         if self.owner is None:
@@ -142,52 +116,26 @@ def run(self):
             self.user_id = int(result[0])
             self.email = result[1]
 
-        # If the metadata is to be made public overwrite the user_id
-        if self.public:
-            self.user_id = 1
-        self.check_file = fig.DATABASE_DIR / 'last_check.dat'
-
         if not self.path.is_dir():
             self.path.mkdir()
-        self.data.update(path=str(self.path.parent))
-        self.data.save()
-
-        # Create a copy of the Data files
-        datafile_copies = {key: create_local_copy(Path(filepath).read_bytes(),
-                                                  f"{Path(filepath).name}", self.path.parent, False)
-                           for key, filepath in self.datafiles.items()
-                           if filepath is not None}
-
-        for key, filepath in self.datafiles.items():
-            filepath = Path(filepath)
-            with open(filepath, 'rb') as f:
-                grid_file = men.GridFSProxy()
-                grid_file.put(f, content_type='text/plain', filename=filepath.name)
-                self.data.files.append(grid_file)
-        self.data.save()
+        self.doc.update(path=str(self.path.parent))
+        self.doc.save()
 
-        # Create sequencing run directory file
-        with open(self.path.parent / fig.SEQUENCING_DIRECTORY_FILE, "wt") as f:
-            for key, filepath in self.datafiles.items():
-                adjusted = key
-                if key == 'for_reads':
-                    adjusted = 'forward'
-                elif key == 'rev_reads':
-                    adjusted = 'reverse'
-                f.write(f"{adjusted}: {Path(filepath).name}\n")
-
-        self.mongo_import(**datafile_copies)
+        self.mongo_import(self.table_file)
 
         # Send the confirmation email
-        send_email(self.email, self.owner, message='upload-run', run=self.sequencing_run_name,
-                   code=self.access_code, testing=self.testing)
+        send_email(self.email, self.owner, message='upload-feature_table', study=self.studies[0],
+                   table_name=self.table_name, code=self.access_code, testing=self.testing)
         # Update the doc to reflect the successful upload
-        self.data.update(is_alive=False, exit_code=0)
-        self.data.save()
+        self.doc.update(is_alive=False, exit_code=0)
+        self.doc.save()
         return 0
 
-    def mongo_import(self, **kwargs):
+    def mongo_import(self, table_file):
         """ Imports additional columns into the NoSQL database. """
-        self.data.files.update(kwargs)
-        self.data.update(email=self.email, path=str(self.path.parent))
-        self.data.save()
+        self.doc.table = men.GridFSProxy()
+        with open(table_file, "rb") as f:
+            self.doc.table.put(f, content_type="text/plain")
+
+        self.doc.update(email=self.email, path=str(self.path.parent))
+        self.doc.save()

mmeds/spawn.py

@@ -18,6 +18,7 @@
 from mmeds.database.data_uploader import DataUploader
 from mmeds.database.metadata_adder import MetaDataAdder
 from mmeds.database.study_creator import StudyCreator
+from mmeds.database.feature_table_uploader import FeatureTableUploader
 from mmeds.error import AnalysisError, MissingUploadError
 
 from mmeds.tools.analysis import Analysis
@@ -331,19 +332,26 @@ def handle_upload(self, process):
 
             # Add new sequencing run
             elif 'run' in process[0]:
-                (ptype, sequencing_run_name, username, reads_type, barcodes_type,
-                 datafiles, public) = process
+                (ptype, sequencing_run_name, username, reads_type, barcodes_type, datafiles, public) = process
 
                 p = DataUploader(new_access_code, username, sequencing_run_name, reads_type,
                                  datafiles, public, self.testing)
+
+            # Add new feature table
+            elif 'feature-table' in process[0]:
+                (ptype, username, study, table_name, table_type, table_file, public) = process
+
+                p = FeatureTableUploader(new_access_code, username, [study], table_name, table_type, table_file,
+                                         public, self.testing)
+
             # Add new study
             else:
                 Logger.debug(f"length: {len(process)}")
                 (ptype, study_doc, subject_metadata, specimen_metadata, subject_type,
                     username, meta_study, temporary, public) = process
                 # Start a process to handle loading the data
-                p = MetaDataUploader(new_access_code, subject_metadata, subject_type, specimen_metadata, username, 'qiime',
-                                     study_doc, meta_study, temporary, public, self.testing)
+                p = MetaDataUploader(new_access_code, subject_metadata, subject_type, specimen_metadata, username,
+                                     'qiime', study_doc, meta_study, temporary, public, self.testing)
         self.db_lock.acquire()
         p.start()
         self.add_process(ptype, p.access_code)

mmeds/util.py

@@ -804,6 +804,13 @@ def send_email(toaddr, user, message='upload', testing=False, **kwargs):
                'the following access code:\n{code}\n\nBest,\nMmeds Team\n\n' +\
                'If you have any issues please email: {cemail} with a description of your problem.\n'
         subject = 'New Sequencing Run Uploaded'
+    elif message == 'upload-feature-table':
+        body = 'Hello {email}, \nthe user {user} uploaded a feature table "{table_name}" for the {study} study ' +
+                'to the mmeds database server.\nThis table can now be used for downstream analyses.' +\
+               'In order to gain access to this data without the password to\n{user} you must provide ' +\
+               'the following access code:\n{code}\n\nBest,\nMmeds Team\n\n' +\
+               'If you have any issues please email: {cemail} with a description of your problem.\n'
+        subject = 'New Feature Table Uploaded'
     elif message == 'ids_generated':
         body = 'Hello {email},\nthe user {user} uploaded {id_type}s for the study {study}. \n' +\
                'The aliquots are added and the IDs have been generated.\n\nBest,\nMmeds Team\n\n' +\
@@ -1487,6 +1494,15 @@ def upload_sequencing_run_local(queue, run_name, user, datafiles, reads_type, ba
     return 0
 
 
+def upload_feature_table_local(queue, user, study, table_name, table_type, table_file):
+    """
+    Directly upload a local feature table using the watcher, bypassing the server
+    """
+    queue.put(('upload-feature-table', user, study, table_name, table_type, table_file, False))
+    Logger.debug("Feature table sent to queue directly")
+    return 0
+
+
 def process_sequencing_runs_local(runs_path, queue):
     """ Uploads sequencing runs included in a dump file """
     for run in runs_path.glob("*"):

scripts/upload_feature_table.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python3
+
+import click
+import mmeds.util as util
+from mmeds.spawn import Watcher
+from pathlib import Path
+
+CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help'])
+
+
+@click.command(context_settings=CONTEXT_SETTINGS)
+@click.option('-n', '--table-name', required=True, help='Name of table')
+@click.option('-t', '--table-type', required=True, help='Type of table', default="taxonomic")
+@click.option('-f', '--table-file', required=True, help='Filepath containing table')
+@click.option('-s', '--study', required=True, help='Study to be associated with feature table')
+@click.option('-u', '--user', required=True, help='User for upload')
+def upload_sequencing_run(table_name, table_type, table_file, study, user):
+    """
+    Uploads a sequencing run directly from the command line, bypassing the server
+    """
+    q = get_queue()
+
+    result = util.upload_feature_table_local(q, user, study, table_name, table_type, table_file)
+    assert result == 0
+
+
+def get_queue():
+    """ Connects to a watcher queue to be sent to the analysis function in util.py. Those functions
+    cannot connect to Watcher queue on their own due to their needing a recursive import
+    """
+    watcher = Watcher()
+    watcher.connect()
+    queue = watcher.get_queue()
+    return queue
+
+
+if __name__ == '__main__':
+    upload_sequencing_run()