using threads for raw files

Author: hesspnnl

metaMS/lcms_functions.py

@@ -30,10 +30,11 @@ def instantiate_lcms_obj(file_in, spectra="ms1"):
     myLCMSobj : corems LCMS object
         LCMS object with unprocessed ms1 spectra included as an attribute
     """
+    click.echo("within instantiate_lcms_obj")
     # Instantiate parser based on binary file type
     if ".raw" in str(file_in):
         parser = ImportMassSpectraThermoMSFileReader(file_in)
-
+        click.echo("returning parser from corems")
     if ".mzML" in str(file_in):
         parser = MZMLSpectraParser(file_in)
 

metaMS/lcms_lipidomics_workflow.py

@@ -2,6 +2,7 @@
 import toml
 from pathlib import Path
 from multiprocessing import Pool
+from concurrent.futures import ThreadPoolExecutor
 import click
 import warnings
 import pandas as pd
@@ -10,6 +11,7 @@
 from corems.mass_spectra.input.corems_hdf5 import ReadCoreMSHDFMassSpectra
 from corems.mass_spectra.output.export import LipidomicsExport
 
+
 from metaMS.lipid_metadata_prepper import get_lipid_library, _to_flashentropy
 from metaMS.lcms_functions import (
     instantiate_lcms_obj,
@@ -139,13 +141,17 @@ def run_lipid_sp_ms1(file_in, out_path, params_toml, scan_translator):
     mz_dict : dict
         Dict with keys "positive" and "negative" and values of lists of precursor mzs
     """  
-
-    myLCMSobj = instantiate_lcms_obj(file_in)           
+    click.echo("in run_lipid_sp_ms1....")
+    myLCMSobj = instantiate_lcms_obj(file_in)   
+    click.echo("calling set_params_on_lcms_obj")        
     set_params_on_lcms_obj(myLCMSobj, params_toml)
+    click.echo("calling check_scan_translator")
     check_scan_translator(myLCMSobj, scan_translator)
+    click.echo("calling add_mass_features")
     add_mass_features(myLCMSobj, scan_translator)
+    click.echo("calling remove_unprocessed_data")
     myLCMSobj.remove_unprocessed_data()
-    #Finally, perform molecular formula search on all ms1 spectra associated with mass features
+    # Finally, perform molecular formula search on all ms1 spectra associated with mass features
     molecular_formula_search(myLCMSobj)
     export_results(myLCMSobj, out_path=out_path, final=False)
     precursor_mz_list = list(
@@ -422,10 +428,10 @@ def run_lcms_lipidomics_workflow(
     cores = lipid_workflow_params.cores
     params_toml = lipid_workflow_params.corems_toml_path
     scan_translator = lipid_workflow_params.scan_translator_path
-
+    use_threads = False
     # Limit cores to 1 if the file type is .raw
-    if any(".raw" in file for file in files_list):
-        cores = 1
+    if any(".raw" in file.name for file in files_list):
+        use_threads = True
 
     click.echo("Starting lipidomics workflow for " + str(len(files_list)) + " file(s), using " +  str(cores) + " core(s)")
     # Run signal processing, get associated ms1, add associated ms2, do ms1 molecular search, and export intermediate results
@@ -440,17 +446,35 @@ def run_lcms_lipidomics_workflow(
             )
             mz_dicts.append(mz_dict)
     elif cores > 1:
-        with Pool(cores) as pool:
-            args = [
-                (
-                    str(file_in),
-                    str(file_out),
-                    params_toml,
-                    scan_translator,
-                )
-                for file_in, file_out in zip(files_list, out_paths_list)
-            ]
-            mz_dicts = pool.starmap(run_lipid_sp_ms1, args)
+        click.echo("Entering multiple cores if condition....")
+        if use_threads:
+            click.echo("Using threads...")
+            with ThreadPoolExecutor(max_workers=cores) as executor:
+                click.echo("In with executor...")
+                args = [
+                    (
+                        str(file_in),
+                        str(file_out),
+                        params_toml,
+                        scan_translator,
+                    )
+                    for file_in, file_out in zip(files_list, out_paths_list)
+                ]
+                mz_dicts = list(executor.map(lambda arg: run_lipid_sp_ms1(*arg), args))
+                click.echo("Out with executor...")
+        else:
+            with Pool(cores) as pool:
+                click.echo("In with processing pool...")
+                args = [
+                    (
+                        str(file_in),
+                        str(file_out),
+                        params_toml,
+                        scan_translator,
+                    )
+                    for file_in, file_out in zip(files_list, out_paths_list)
+                ]
+                mz_dicts = pool.starmap(run_lipid_sp_ms1, args)
 
     # Prepare metadata for searching
     click.echo("Preparing metadata for ms2 spectral search")
@@ -469,4 +493,4 @@ def run_lcms_lipidomics_workflow(
             args = [(file_out, metadata, scan_translator) for file_out in out_paths_list]
             pool.starmap(run_lipid_ms2, args)
 
-    click.echo("Lipidomics workflow complete")
+    click.echo("Lipidomics workflow complete")

metaMS/tiny_tester.py

@@ -16,14 +16,13 @@
 import click
 
 def func(file_in):
-    click.echo("creating parser")
-    parser = ImportMassSpectraThermoMSFileReader(file_in)
-    click.echo("returning parser")
-    return parser
+   click.echo(f"Processing file: {file_in}")
+   parser = ImportMassSpectraThermoMSFileReader(file_in)
+   myLCMSobj = parser.get_lcms_obj(spectra="ms1")    
+   return myLCMSobj
 
 cores = 2
 files = ["Blanch_Nat_Lip_C_4_AB_M_08_NEG_25Jan18_Brandi-WCSH5801.raw", "Blanch_Nat_Lip_C_4_AB_M_08_NEG_25Jan18_Brandi-WCSH5801-copy-2.raw"]
 with Pool(cores) as pool:
     click.echo("In with pool...")
-    mz_dicts = pool.map(func, files)
-    
+    mz_dicts = pool.map(func, files)