corems.mass_spectra.input.brukerSolarix

  1__author__ = "Yuri E. Corilo"
  2__date__ = "Oct 29, 2019"
  3
  4from threading import Thread
  5from pathlib import Path
  6from s3path import S3Path
  7
  8# import h5py
  9
 10# from corems.encapsulation.constant import Labels
 11from corems.mass_spectra.factory.lc_class import LCMSBase
 12
 13# from corems.encapsulation.factory.parameters import default_parameters
 14from corems.transient.input.brukerSolarix import ReadBrukerSolarix
 15
 16from corems.mass_spectra.input.brukerSolarix_utils import get_scan_attributes
 17
 18
 19class ReadBruker_SolarixTransientMassSpectra(Thread):
 20    """
 21    Class for reading Bruker Solarix Transient Mass Spectra.
 22
 23    Parameters
 24    ----------
 25    d_directory_location : str, pathlib.Path, or s3path.S3Path
 26        Path object from pathlib containing the file location.
 27    analyzer : str, optional
 28        Type of analyzer used in the mass spectrometer. Defaults to "ICR".
 29    instrument_label : str, optional
 30        Label for the instrument. Defaults to "15T".
 31    auto_process : bool, optional
 32        Flag indicating whether to automatically process the mass spectra. Defaults to True.
 33    keep_profile : bool, optional
 34        Flag indicating whether to keep the profile data in the mass spectra. Defaults to False.
 35    """
 36
 37    def __init__(
 38        self,
 39        d_directory_location: str | Path | S3Path,
 40        analyzer="ICR",
 41        instrument_label="15T",
 42        auto_process=True,
 43        keep_profile=False,
 44    ):
 45        Thread.__init__(self)
 46
 47        if isinstance(d_directory_location, str):
 48            # if obj is a string it defaults to create a Path obj, pass the S3Path if needed
 49            d_directory_location = Path(d_directory_location)
 50
 51        if not d_directory_location.exists():
 52            raise FileNotFoundError("File does not exist: " + str(d_directory_location))
 53
 54        self.scan_attr = d_directory_location / "scan.xml"
 55        self.imaging_info_attr = d_directory_location / "ImagingInfo.xml"
 56
 57        if not self.scan_attr.exists():
 58            raise FileExistsError(
 59                "%s does not seem to be a valid Solarix Mass Spectra Experiment,\
 60                                maybe an Imaging experiment?\
 61                                please ReadBruker_SolarixTransientImage class for Imaging dataset "
 62                % d_directory_location
 63            )
 64
 65        self.lcms = LCMSBase(d_directory_location, analyzer, instrument_label)
 66
 67        self.auto_process = auto_process
 68        self.keep_profile = keep_profile
 69
 70    def get_scan_attr(self) -> dict:
 71        """
 72        Get the scan attributes from the scan.xml or ImagingInfo.xml file.
 73        If the scan.xml file exists, it will be used; otherwise, it will look for ImagingInfo.xml.
 74        If neither file exists, a FileNotFoundError will be raised.
 75
 76
 77        TODO: - This function is replicated in the corems.transient.input.brukerSolarix module,
 78                consider refactoring to avoid duplication.
 79
 80        Returns
 81        -------
 82        dict
 83            Dictionary containing the scan number as key and a tuple of retention time and TIC as value.
 84        """
 85        return get_scan_attributes(self.scan_attr, self.imaging_info_attr)
 86
 87    def import_mass_spectra(self) -> None:
 88        """
 89        Import the mass spectra from the scan.xml file.
 90        """
 91        dict_scan_rt_tic = self.get_scan_attr()
 92
 93        list_rt, list_tic = (
 94            list(),
 95            list(),
 96        )
 97
 98        list_scans = sorted(list(dict_scan_rt_tic.keys()))
 99
100        for scan_number in list_scans:
101            mass_spec = self.get_mass_spectrum(scan_number)
102
103            self.lcms.add_mass_spectrum(mass_spec)
104
105            list_rt.append(dict_scan_rt_tic.get(scan_number)[0])
106
107            list_tic.append(dict_scan_rt_tic.get(scan_number)[1])
108
109        self.lcms.retention_time = list_rt
110        self.lcms.tic = list_tic
111        self.lcms.scans_number = list_scans
112
113    def get_mass_spectrum(self, scan_number: int):
114        """
115        Get the mass spectrum for a given scan number.
116
117        Parameters
118        ----------
119        scan_number : int
120            Scan number.
121
122        """
123        bruker_reader = ReadBrukerSolarix(self.lcms.file_location)
124
125        bruker_transient = bruker_reader.get_transient(scan_number)
126
127        mass_spec = bruker_transient.get_mass_spectrum(
128            plot_result=False,
129            auto_process=self.auto_process,
130            keep_profile=self.keep_profile,
131        )
132
133        return mass_spec
134
135    def run(self):
136        """
137        Run the import_mass_spectra method.
138        """
139        self.import_mass_spectra()
140
141    def get_lcms_obj(self):
142        """
143        Get the LCMSBase object.
144
145        Raises
146        ------
147        Exception
148            If the LCMSBase object is empty.
149        """
150        if self.lcms:
151            return self.lcms
152        else:
153            raise Exception("Returning an empty LCMSBase class.")