corems.mass_spectra.input.massList

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