RE2022-272: Add a bulk version of genbank_to_genome in GFU

GenomeFileUtil.spec

@@ -65,6 +65,38 @@ module GenomeFileUtil {
     funcdef genbank_to_genome(GenbankToGenomeParams params)
                 returns (GenomeSaveResult result) authentication required;
 
+    typedef structure {
+        File file;
+        string genome_name;
+
+        string source;
+        string taxon_wsname;
+        string taxon_id;
+
+        string release;
+        string generate_ids_if_needed;
+        int    genetic_code;
+        string scientific_name;
+        usermeta metadata;
+        boolean generate_missing_genes;
+        string use_existing_assembly;
+    } GenbankToGenomeInput;
+
+    typedef structure {
+        int workspace_id;
+        list<GenbankToGenomeInput> inputs;
+    } GenbanksToGenomesParams;
+
+    /* Results for the genbanks_to_genomes function.
+        results - the results of the save operation in the same order as the input.
+    */
+    typedef structure {
+        list<GenomeSaveResult> results;
+    } GenomeSaveResults;
+
+    funcdef genbanks_to_genomes(GenbanksToGenomesParams params)
+                returns (GenomeSaveResults results) authentication required;
+
     /*
         is_gtf - optional flag switching export to GTF format (default is 0,
             which means GFF)

lib/GenomeFileUtil/GenomeFileUtilImpl.py

@@ -8,13 +8,13 @@
 from pprint import pprint
 
 from GenomeFileUtil.core.FastaGFFToGenome import FastaGFFToGenome
-from GenomeFileUtil.core.GenbankToGenome import GenbankToGenome
-from GenomeFileUtil.core.GenomeFeaturesToFasta import GenomeFeaturesToFasta
-from GenomeFileUtil.core.GenomeInterface import (
-    GenomeInterface,
+from GenomeFileUtil.core.GenbankToGenome import (
+    GenbankToGenome,
     MAX_THREADS_DEFAULT,
     THREADS_PER_CPU_DEFAULT,
 )
+from GenomeFileUtil.core.GenomeFeaturesToFasta import GenomeFeaturesToFasta
+from GenomeFileUtil.core.GenomeInterface import GenomeInterface
 from GenomeFileUtil.core.GenomeToGFF import GenomeToGFF
 from GenomeFileUtil.core.GenomeToGenbank import GenomeToGenbank
 from installed_clients.AssemblyUtilClient import AssemblyUtil
@@ -92,7 +92,6 @@
         #END_CONSTRUCTOR
         pass
 
-
     def genbank_to_genome(self, ctx, params):
         """
         :param params: instance of type "GenbankToGenomeParams" (genome_name
@@ -132,7 +131,7 @@
         pprint(params)
 
         importer = GenbankToGenome(self.cfg)
-        result = importer.refactored_import(ctx, params)
+        result = importer.refactored_import(params)
 
         print('import complete -- result = ')
         pprint(result)
@@ -145,6 +144,60 @@
         # return the results
         return [result]
 
+    def genbanks_to_genomes(self, ctx, params):
+        """
+        :param params: instance of type "GenbanksToGenomesParams" -> structure:
+           parameter "workspace_id" of Long, parameter "inputs" of list of
+           type "GenbankToGenomeInput" (genome_name - becomes the name of the
+           object source - Source of the file typically something like RefSeq
+           or Ensembl taxon_ws_name - where the reference taxons are :
+           ReferenceTaxons taxon_id - if defined, will try to link the Genome
+           to the specified taxonomy id in lieu of performing the lookup
+           during upload release - Release or version number of the data per
+           example Ensembl has numbered releases of all their data: Release 31
+           generate_ids_if_needed - If field used for feature id is not there,
+           generate ids (default behavior is raising an exception) genetic_code
+           - Genetic code of organism. Overwrites determined GC from taxon
+           object scientific_name - will be used to set the scientific name of
+           the genome and link to a taxon generate_missing_genes - If the file
+           has CDS or mRNA with no corresponding gene, generate a spoofed gene.
+           use_existing_assembly - Supply an existing assembly reference) ->
+           structure: parameter "file" of type "File" -> structure: parameter
+           "path" of String, parameter "shock_id" of String, parameter
+           "ftp_url" of String, parameter "genome_name" of String, parameter
+           "source" of String, parameter "taxon_wsname" of String, parameter
+           "taxon_id" of String, parameter "release" of String, parameter
+           "generate_ids_if_needed" of String, parameter "genetic_code" of
+           Long, parameter "scientific_name" of String, parameter "metadata"
+           of type "usermeta" -> mapping from String to String, parameter
+           "generate_missing_genes" of type "boolean" (A boolean - 0 for false,
+           1 for true. @range (0, 1)), parameter "use_existing_assembly" of
+           String
+        :returns: instance of type "GenomeSaveResults" -> structure: parameter
+           "results" of list of type "GenomeSaveResult" -> structure: parameter
+           "genome_ref" of String
+        """
+        # ctx is the context object
+        # return variables are: result
+        #BEGIN genbanks_to_genomes
+        print('genbanks_to_genomes -- paramaters = ')
+        pprint(params)
+
+        results = {
+            'results': GenbankToGenome(self.cfg).refactored_import_mass(params)
+        }
+
+        print('import complete -- results = ')
+        pprint(results)
+        #END genbanks_to_genomes
+
+        # At some point might do deeper type checking...
+        if not isinstance(results, dict):
+            raise ValueError('Method genbank_to_genome return value ' +
+                             'result is not type dict as required.')
+        # return the results
+        return [results]
+
     def genome_to_gff(self, ctx, params):
         """
         :param params: instance of type "GenomeToGFFParams" (is_gtf -

lib/GenomeFileUtil/GenomeFileUtilServer.py

@@ -342,6 +342,10 @@ def __init__(self):
                              name='GenomeFileUtil.genbank_to_genome',
                              types=[dict])
         self.method_authentication['GenomeFileUtil.genbank_to_genome'] = 'required'  # noqa
+        self.rpc_service.add(impl_GenomeFileUtil.genbanks_to_genomes,
+                             name='GenomeFileUtil.genbanks_to_genomes',
+                             types=[dict])
+        self.method_authentication['GenomeFileUtil.genbanks_to_genomes'] = 'required'  # noqa
         self.rpc_service.add(impl_GenomeFileUtil.genome_to_gff,
                              name='GenomeFileUtil.genome_to_gff',
                              types=[dict])

lib/GenomeFileUtil/core/GenbankToGenome.py

@@ -28,6 +28,17 @@
 MAX_MISC_FEATURE_SIZE = 10000
 MAX_PARENT_LOOKUPS = 5
 
+# catalog params
+MAX_THREADS_DEFAULT = 10
+THREADS_PER_CPU_DEFAULT = 1
+
+_WSID = 'workspace_id'
+_INPUTS = 'inputs'
+
+
+def _upa(object_info):
+    return f'{object_info[6]}/{object_info[0]}/{object_info[4]}'
+
 
 class GenbankToGenome:
     def __init__(self, config):
@@ -36,11 +47,15 @@
         self.dfu = DataFileUtil(config.callbackURL)
         self.aUtil = AssemblyUtil(config.callbackURL)
         self.ws = Workspace(config.workspaceURL)
+        self.re_api_url = config.re_api_url
+        yml_text = open('/kb/module/kbase.yml').read()
+        self.version = re.search("module-version:\n\W+(.+)\n", yml_text).group(1)
+        self.reset_attributes()
+
+    def reset_attributes(self):
         self._messages = []
         self.time_string = str(datetime.datetime.fromtimestamp(
             time.time()).strftime('%Y_%m_%d_%H_%M_%S'))
-        yml_text = open('/kb/module/kbase.yml').read()
-        self.version = re.search("module-version:\n\W+(.+)\n", yml_text).group(1)
         self.generate_parents = False
         self.generate_ids = False
         self.genes = OrderedDict()
@@ -62,7 +77,6 @@
         self.excluded_features = ('source', 'exon', 'fasta_record')
         self.ont_mappings = load_ontology_mappings('/kb/module/data')
         self.code_table = 11
-        self.re_api_url = config.re_api_url
         # dict with feature 'id's that have been used more than once.
         self.used_twice_identifiers = {}
         self.default_params = {
@@ -84,53 +98,121 @@
     def messages(self):
         return "\n".join(self._messages)
 
-    def refactored_import(self, ctx, params):
-        # 1) validate parameters and extract defaults
-        self.validate_params(params)
-
-        # 2) construct the input directory staging area
-        input_directory = self.stage_input(params)
-
-        # 3) update default params
-        self.default_params.update(params)
-        params = self.default_params
-        self.generate_parents = params.get('generate_missing_genes')
-        self.generate_ids = params.get('generate_ids_if_needed')
-        if params.get('genetic_code'):
-            self.code_table = params['genetic_code']
-
-        # 4) Do the upload
-        files = self._find_input_files(input_directory)
-        consolidated_file = self._join_files_skip_empty_lines(files)
-        genome = self.parse_genbank(consolidated_file, params)
-        if params.get('genetic_code'):
-            genome["genetic_code"] = params['genetic_code']
-
-        result = self.gi.save_one_genome({
-            'workspace': params['workspace_name'],
-            'name': params['genome_name'],
-            'data': genome,
-            "meta": params['metadata'],
-        })
-        ref = f"{result['info'][6]}/{result['info'][0]}/{result['info'][4]}"
-        logging.info(f"Genome saved to {ref}")
-
-        # 5) clear the temp directory
-        shutil.rmtree(input_directory)
-
-        # 6) return the result
-        info = result['info']
-        details = {
-            'genome_ref': ref,
-            'genome_info': info
-        }
-
+    def refactored_import(self, params):
+        print('validating parameters')
+        mass_params = self._set_up_single_params(params)
+        return self._refactored_import_mass(mass_params)[0]
+
+    def refactored_import_mass(self, params):
+        print('validating parameters')
+        self._validate_mass_params(params)
+        return self._refactored_import_mass(params)
+
+    def _set_up_single_params(self, params):
+        inputs = dict(params)
+        self.validate_params(inputs)
+        ws_id = self._get_int(inputs.pop(_WSID, None), _WSID)
+        ws_name = inputs.pop('workspace_name', None)
+        if (bool(ws_id) == bool(ws_name)):  # xnor
+            raise ValueError(f"Exactly one of a '{_WSID}' or a 'workspace' parameter must be provided")
+        if not ws_id:
+            print(f"Translating workspace name {ws_name} to a workspace ID. Prefer submitting "
+                  + "a workspace ID over a mutable workspace name that may cause race conditions")
+            ws_id = self.dfu.ws_name_to_id(ws_name)
+        mass_params = {_WSID: ws_id, _INPUTS: [inputs]}
+        return mass_params
+
+    def _validate_mass_params(self, params):
+        ws_id = self._get_int(params.get(_WSID), _WSID)
+        if not ws_id:
+            raise ValueError(f"{_WSID} is required")
+        inputs = params.get(_INPUTS)
+        if not inputs or type(inputs) != list:
+            raise ValueError(f"{_INPUTS} field is required and must be a non-empty list")
+        for i, inp in enumerate(inputs, start=1):
+            if type(inp) != dict:
+                raise ValueError(f"Entry #{i} in {_INPUTS} field is not a mapping as required")
+            self.validate_params(inp)
+
+    def _get_int(self, putative_int, name, minimum=1):
+        if putative_int is not None:
+            if type(putative_int) != int:
+                raise ValueError(f"{name} must be an integer, got: {putative_int}")
+            if putative_int < minimum:
+                raise ValueError(f"{name} must be an integer >= {minimum}")
+        return putative_int
+
+    def _refactored_import_mass(self, params):
+
+        workspace_id = params[_WSID]
+        inputs = params[_INPUTS]
+
+        genome_names = []
+        genome_data = []
+        genome_meta = []
+
+        for input_params in inputs:
+            # 1) construct the input directory staging area
+            input_directory = self.stage_input(input_params)
+
+            # 2) update default params
+            input_params = {**self.default_params, **input_params}
+            self.generate_parents = input_params.get('generate_missing_genes')
+            self.generate_ids = input_params.get('generate_ids_if_needed')
+            if input_params.get('genetic_code'):
+                self.code_table = input_params['genetic_code']
+
+            # 3) Do the upload
+            files = self._find_input_files(input_directory)
+            consolidated_file = self._join_files_skip_empty_lines(files)
+            genome = self.parse_genbank(
+                workspace_id, consolidated_file, input_params
+            )
+            if input_params.get('genetic_code'):
+                genome["genetic_code"] = input_params['genetic_code']
+
+            # 4) clear the temp directory and reset attributes
+            shutil.rmtree(input_directory)
+            self.reset_attributes()
+
+            genome_data.append(genome)
+            genome_names.append(input_params['genome_name'])
+            genome_meta.append(input_params['metadata'])
+
+        results = self._save_genomes(
+            workspace_id, genome_names, genome_data, genome_meta
+        )
+
+        # 5) return the result
+        details = [
+            {'genome_ref': _upa(result["info"]), 'genome_info': result["info"]}
+            for result in results
+        ]
+        for detail in details:
+            logging.info(f"Genome saved to {detail['genome_ref']}")
         return details
 
+    def _save_genomes(
+        self,
+        workspace_id,
+        genome_names,
+        genome_data,
+        genome_meta
+    ):
+        results = [
+            self.gi.save_one_genome(
+                {
+                    'workspace': workspace_id,
+                    'name': name,
+                    'data': data,
+                    "meta": meta,
+                }
+            ) for name, data, meta in zip(genome_names, genome_data, genome_meta)
+        ]
+        return results
+
     @staticmethod
     def validate_params(params):
-        if 'workspace_name' not in params:
-            raise ValueError('required "workspace_name" field was not defined')
         if 'genome_name' not in params:
             raise ValueError('required "genome_name" field was not defined')
         if 'file' not in params:
@@ -206,15 +288,15 @@
 
         return input_directory
 
-    def parse_genbank(self, file_path, params):
+    def parse_genbank(self, workspace_id, file_path, params):
         logging.info("Saving original file to shock")
         shock_res = self.dfu.file_to_shock({
             'file_path': file_path,
             'make_handle': 1,
             'pack': 'gzip',
         })
         # Write and save assembly file
-        assembly_ref = self._save_assembly(file_path, params)
+        assembly_ref = self._save_assembly(workspace_id, file_path, params)
         assembly_data = self.dfu.get_objects(
             {'object_refs': [assembly_ref],
              'ignore_errors': 0})['data'][0]['data']
@@ -319,7 +401,7 @@
         logging.info(f"Feature Counts: {genome['feature_counts']}")
         return genome
 
-    def _save_assembly(self, genbank_file, params):
+    def _save_assembly(self, workspace_id, genbank_file, params):
         """Convert genbank file to fasta and sve as assembly"""
         contigs = Bio.SeqIO.parse(genbank_file, "genbank")
         assembly_id = f"{params['genome_name']}_assembly"
@@ -367,7 +449,7 @@
         Bio.SeqIO.write(out_contigs, fasta_file, "fasta")
         assembly_ref = self.aUtil.save_assembly_from_fasta(
             {'file': {'path': fasta_file},
-             'workspace_name': params['workspace_name'],
+             'workspace_id': workspace_id,
              'assembly_name': assembly_id,
              'type': params.get('genome_type', 'isolate'),
              'contig_info': extra_info})