corems.mass_spectra.input.massList

  1__author__ = "Yuri E. Corilo"
  2__date__ = "Jun 12, 2019"
  3
  4import sys
  5
  6from pathlib import Path
  7from threading import Thread
  8import warnings
  9from corems.mass_spectrum.input.massList import ReadCoremsMasslist
 10from corems.mass_spectra.factory.lc_class import LCMSBase
 11
 12
 13class ReadCoremsMassSpectraText(ReadCoremsMasslist, Thread):
 14    """
 15    Class for reading CoreMS mass spectra from a text file.
 16
 17    Parameters
 18    ----------
 19    file_location : str, pathlib.Path, or s3path.S3Path
 20        Path object from pathlib containing the file location
 21    analyzer : str, optional
 22        Name of the analyzer, by default 'Unknown'
 23    instrument_label : str, optional
 24        Label of the instrument, by default 'Unknown'
 25
 26    Attributes
 27    ----------
 28    lcms : LCMSBase
 29        LCMSBase object for storing the mass spectra data.
 30
 31    Methods
 32    -------
 33    * get_scans_filename(). Get the filenames of all the scan files associated with the CoreMS file.
 34    * set_filepath_datatype_and_delimiter(file_path_obj). Set the file path, data type, and delimiter based on the file path object.
 35    * import_mass_spectra(). Import the mass spectra from the scan files and add them to the LCMSBase object.
 36    * run(). Run the import_mass_spectra method to create the LCMSBase object.
 37    * get_lcms_obj(). Get the LCMSBase object.
 38    """
 39
 40    def __init__(self, file_location, analyzer="Unknown", instrument_label="Unknown"):
 41        if isinstance(file_location, str):
 42            # if obj is a string it defaults to create a Path obj, pass the S3Path if needed
 43            file_location = Path(file_location)
 44
 45        if not file_location.exists():
 46            raise FileNotFoundError("%s not found" % file_location)
 47
 48        if not file_location.suffix == ".corems":
 49            raise TypeError("%s is not a valid CoreMS file" % file_location)
 50
 51        Thread.__init__(self)
 52
 53        ReadCoremsMasslist.__init__(self, file_location)
 54
 55        self.lcms = LCMSBase(
 56            self.file_location, analyzer=analyzer, instrument_label=instrument_label
 57        )
 58
 59    def get_scans_filename(self) -> list:
 60        all_other = self.file_location.glob("*_scan*[!.json]")
 61
 62        scans_filepath = [
 63            (file_path_obj.stem.split("scan")[1], file_path_obj)
 64            for file_path_obj in all_other
 65        ]
 66
 67        scans_filepath.sort(key=lambda m: int(m[0]))
 68
 69        return scans_filepath
 70
 71    def set_filepath_datatype_and_delimiter(self, file_path_obj) -> None:
 72        self.file_location = file_path_obj
 73
 74        if file_path_obj.suffix == ".pkl":
 75            self.data_type == "dataframe"
 76
 77        else:
 78            if file_path_obj.suffix == ".csv":
 79                self.data_type == "txt"
 80                self.delimiter = ","
 81
 82            elif file_path_obj.suffix == ".xlsx":
 83                self.data_type == "excel"
 84                self.delimiter = ","
 85
 86            elif file_path_obj.suffix == ".txt":
 87                self.data_type == "txt"
 88                self.delimiter = "\t"
 89                warnings.warn("using tab as delimiter")
 90            else:
 91                raise NotImplementedError(
 92                    "%s data not yet supported " % file_path_obj.suffix
 93                )
 94
 95    def import_mass_spectra(self) -> None:
 96        list_rt, list_tic, list_scan = list(), list(), list()
 97
 98        for scan_number, file_path_obj in self.get_scans_filename():
 99            self.set_filepath_datatype_and_delimiter(file_path_obj)
100
101            mass_spec = self.get_mass_spectrum(int(scan_number))
102
103            list_scan.append(int(scan_number))
104
105            list_rt.append(mass_spec.retention_time)
106
107            list_tic.append(mass_spec.tic)
108
109            self.lcms.add_mass_spectrum(mass_spec)
110
111        self.lcms.retention_time = list_rt
112        self.lcms.tic_list = list_tic  # TODO: check if this is correct
113        self.lcms.scans_number = list_scan
114
115    def run(self) -> None:
116        """Creates the LCMS object and imports mass spectra."""
117
118        self.import_mass_spectra()
119
120    def get_lcms_obj(self) -> LCMSBase:
121        """
122        Returns the LCMSBase object associated with the massList.
123
124        If the LCMSBase object is already initialized, it is returned.
125        Otherwise, an exception is raised.
126
127        Raises:
128            Exception: If the LCMSBase object is not initialized.
129        """
130        if self.lcms:
131            return self.lcms
132        else:
133            raise Exception("returning an empty lcms class")