corems.mass_spectra.input.boosterHDF5

  1__author__ = "Yuri E. Corilo"
  2__date__ = "Oct 29, 2019"
  3
  4from threading import Thread
  5from pathlib import Path
  6from io import BytesIO
  7
  8import h5py
  9from s3path import S3Path
 10
 11from corems.encapsulation.constant import Labels
 12from corems.mass_spectrum.factory.MassSpectrumClasses import MassSpecProfile
 13from corems.mass_spectra.factory.lc_class import LCMSBase
 14from corems.encapsulation.factory.parameters import default_parameters
 15
 16
 17class ReadHDF_BoosterMassSpectra(Thread):
 18    """
 19    Class for reading HDF5 files containing booster mass spectra.
 20
 21    Parameters
 22    ----------
 23    file_location : Path or S3Path
 24        The full path to the HDF5 file.
 25    analyzer : str, optional
 26        The type of analyzer used for the mass spectra. Defaults to "ICR".
 27    instrument_label : str, optional
 28        The label of the instrument. Defaults to "21T".
 29    auto_process : bool, optional
 30        Whether to automatically process the mass spectra. Defaults to True.
 31    """
 32
 33    def __init__(
 34        self,
 35        file_location: Path | S3Path,
 36        analyzer="ICR",
 37        instrument_label="21T",
 38        auto_process=True,
 39    ):
 40        """
 41        Initialize the ReadHDF_BoosterMassSpectra class.
 42
 43        Parameters
 44        ----------
 45        file_location : Path or S3Path
 46            The full path to the HDF5 file.
 47        analyzer : str, optional
 48            The type of analyzer used for the mass spectra. Defaults to "ICR".
 49        instrument_label : str, optional
 50            The label of the instrument. Defaults to "21T".
 51        auto_process : bool, optional
 52            Whether to automatically process the mass spectra. Defaults to True.
 53        """
 54        Thread.__init__(self)
 55
 56        if isinstance(file_location, str):
 57            # if obj is a string it defaults to create a Path obj, pass the S3Path if needed
 58            self.file_location = Path(file_location)
 59
 60        self.lcms = LCMSBase(
 61            file_location, analyzer=analyzer, instrument_label=instrument_label
 62        )
 63
 64        if isinstance(file_location, S3Path):
 65            data = BytesIO(file_location.open("rb").read())
 66        else:
 67            data = file_location
 68
 69        self.hdf_obj = h5py.File(data, "r")
 70
 71        self.list_scans = sorted([int(i) for i in list(self.hdf_obj.keys())])
 72
 73        self.initial_scan_number = self.list_scans[0]
 74
 75        self.final_scan_number = self.list_scans[-1]
 76
 77        self.file_location = file_location
 78
 79        self.auto_process = True
 80
 81        self.analyzer = analyzer
 82
 83        self.instrument_label = instrument_label
 84
 85    def get_polarity(self, file_location: Path | S3Path, scan: int):
 86        """
 87        Get the polarity of a scan.
 88
 89        Parameters
 90        ----------
 91        file_location : Path or S3Path
 92            The full path to the HDF5 file.
 93        scan : int
 94            The scan number.
 95
 96        """
 97        if isinstance(file_location, S3Path):
 98            data = BytesIO(file_location.open("rb").read())
 99        else:
100            data = file_location
101
102        self.h5pydata = h5py.File(data, "r")
103
104        self.scans = list(self.h5pydata.keys())
105
106        polarity = self.get_attr_data(scan, "r_h_polarity")
107
108        if polarity == "negative scan":
109            return -1
110        else:
111            return +1
112
113    def get_attr_data(self, scan, attr_srt):
114        """
115        Get the attribute data of a scan.
116
117        Parameters
118        ----------
119        scan : int
120            The scan number.
121        attr_srt : str
122            The attribute name.
123
124        """
125        return self.hdf_obj[str(scan)].attrs[attr_srt]
126
127    def import_mass_spectra(self, d_params: dict):
128        """
129        Import the mass spectra from the HDF5 file.
130
131        Parameters
132        ----------
133        d_params : dict
134            The parameters for importing the mass spectra.
135        """
136        list_rt, list_tic = list(), list()
137
138        for scan_number in self.list_scans:
139            d_params["rt"] = list_rt.append(
140                self.get_attr_data(scan_number, "r_h_start_time")
141            )
142
143            d_params["scan_number"] = scan_number
144
145            d_params["label"] = Labels.booster_profile
146
147            d_params["polarity"] = self.get_polarity(self.file_location, scan_number)
148
149            d_params["Aterm"] = self.get_attr_data(scan_number, "r_cparams")[0]
150
151            d_params["Bterm"] = self.get_attr_data(scan_number, "r_cparams")[1]
152
153            d_params["analyzer"] = self.analyzer
154
155            d_params["instrument_label"] = self.instrument_label
156
157            list_rt.append(d_params["rt"])
158
159            list_tic.append(self.get_attr_data(scan_number, "r_h_tic"))
160
161            mass_spec = self.get_mass_spectrum(scan_number, d_params)
162
163            self.lcms.add_mass_spectrum(mass_spec)
164
165        self.lcms.retention_time = list_rt
166        self.lcms.tic = list_tic
167        self.lcms.scans_number = self.list_scans
168
169    def get_mass_spectrum(self, scan: int, d_params: dict):
170        """
171        Get the mass spectrum for a scan.
172
173        Parameters
174        ----------
175        scan : int
176            The scan number.
177        d_params : dict
178            The parameters for creating the mass spectrum.
179
180        """
181        booster_data = self.hdf_obj[str(scan)]
182
183        if booster_data.shape[0] != 2:
184            raise NotImplementedError(
185                "opening transient, needs read raw file here, get bandwidth, create transient class and then the mass spectrum"
186            )
187        else:
188            data_dict = {
189                Labels.mz: booster_data[0],
190                Labels.abundance: booster_data[1],
191                Labels.rp: None,
192                Labels.s2n: None,
193            }
194
195            mass_spec = MassSpecProfile(
196                data_dict, d_params, auto_process=self.auto_process
197            )
198
199        return mass_spec
200
201    def run(self):
202        """
203        Run the thread to create the LCMS object.
204        """
205        d_parameters = default_parameters(self.file_location)
206        self.import_mass_spectra(d_parameters)
207
208    def get_lcms_obj(self):
209        """
210        Get the LCMS object.
211
212        """
213        if len(self.lcms) > 0:
214            return self.lcms
215        else:
216            raise Exception("Returning an empty LCMS class")