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