corems.mass_spectra.input.andiNetCDF

  1__author__ = "Yuri E. Corilo"
  2__date__ = "Feb 12, 2020"
  3
  4from pathlib import Path
  5from threading import Thread
  6# from io import BytesIO
  7
  8from netCDF4 import Dataset
  9from s3path import S3Path
 10
 11from corems.encapsulation.constant import Labels
 12from corems.encapsulation.factory.parameters import default_parameters
 13from corems.mass_spectra.factory.GC_Class import GCMSBase
 14from corems.mass_spectrum.factory.MassSpectrumClasses import MassSpecCentroidLowRes
 15
 16
 17class ReadAndiNetCDF(Thread):
 18    """
 19    A class for reading AndiNetCDF files and extracting mass spectra data.
 20
 21    Parameters
 22    -----------
 23    file_location : str or Path
 24            The location of the AndiNetCDF file.
 25    analyzer : str, optional
 26            The type of analyzer used (default is 'Quadruple').
 27    instrument_label : str, optional
 28            The label of the instrument (default is 'GCMS-Agilent').
 29    auto_process : bool, optional
 30            Whether to automatically process the data (default is True).
 31
 32    Attributes
 33    -----------
 34    file_location : Path
 35            The path to the AndiNetCDF file.
 36    net_cdf_obj : Dataset
 37            The NetCDF dataset object.
 38    ionization_type : str
 39            The ionization type used in the experiment.
 40    experiment_type : str
 41            The type of experiment.
 42    list_scans : range
 43            The range of scan numbers in the dataset.
 44    initial_scan_number : int
 45            The number of the initial scan.
 46    final_scan_number : int
 47            The number of the final scan.
 48    analyzer : str
 49            The type of analyzer used.
 50    instrument_label : str
 51            The label of the instrument.
 52    gcms : GCMSBase
 53            The GCMSBase object for storing mass spectra data.
 54
 55    Methods
 56    --------
 57    * polarity().
 58            Get the polarity of the ionization.
 59    * get_mass_spectrum(mz, abun, rp, d_params).
 60            Add a mass spectrum to the GCMSBase object.
 61    * run().
 62            Populate the GCMSBase object with mass spectra data.
 63    * import_mass_spectra(d_params).
 64            Import mass spectra data from the AndiNetCDF file.
 65    * get_gcms_obj().
 66            Get the GCMSBase object.
 67
 68    """
 69
 70    def __init__(
 71        self,
 72        file_location: str | Path,
 73        analyzer="Quadruple",
 74        instrument_label="GCMS-Agilent",
 75        auto_process=True,
 76    ):
 77        Thread.__init__(self)
 78
 79        if isinstance(file_location, str):
 80            self.file_location = Path(file_location)
 81        else:
 82            self.file_location = file_location
 83
 84        if not self.file_location.exists():
 85            raise FileNotFoundError("File does not exist at %s", file_location)
 86
 87        if isinstance(file_location, S3Path):
 88            bytes_io = self.file_location.open("rb").read()
 89            self.net_cdf_obj = Dataset(
 90                self.file_location.name,
 91                "r",
 92                diskless=True,
 93                memory=bytes_io,
 94                format="NETCDF3_CLASSIC",
 95            )
 96        else:
 97            self.net_cdf_obj = Dataset(
 98                self.file_location, "r", format="NETCDF3_CLASSIC"
 99            )
100
101        self.ionization_type = self.net_cdf_obj.test_ionization_mode
102        self.experiment_type = self.net_cdf_obj.experiment_type
103        self.list_scans = range(
104            len(self.net_cdf_obj.variables.get("actual_scan_number")[:])
105        )
106        self.initial_scan_number = self.list_scans[0]
107        self.final_scan_number = self.list_scans[-1]
108        self.analyzer = analyzer
109        self.instrument_label = instrument_label
110        self.gcms = GCMSBase(self.file_location, analyzer, instrument_label)
111
112    @property
113    def polarity(self):
114        """
115        Get the polarity of the ionization.
116
117        """
118        polarity = str(self.net_cdf_obj.test_ionization_polarity)
119        if polarity == "Positive Polarity":
120            return +1
121        else:
122            return -1
123
124    def get_mass_spectrum(self, mz, abun, rp, d_params):
125        """
126        Add a mass spectrum to the GCMSBase object.
127
128        Parameters
129        -----------
130        mz : array-like
131                The m/z values of the mass spectrum.
132        abun : array-like
133                The abundance values of the mass spectrum.
134        rp : array-like
135                The resolution values of the mass spectrum.
136        d_params : dict
137                Additional parameters for the mass spectrum.
138
139        """
140        data_dict = {
141            Labels.mz: mz,
142            Labels.abundance: abun,
143            Labels.rp: rp,
144            Labels.s2n: None,
145        }
146        mass_spec = MassSpecCentroidLowRes(data_dict, d_params)
147        self.gcms.add_mass_spectrum(mass_spec)
148
149    def run(self):
150        """
151        Populate the GCMSBase object with mass spectra data.
152        """
153        d_parameters = default_parameters(self.file_location)
154        self.import_mass_spectra(d_parameters)
155
156    def import_mass_spectra(self, d_params):
157        """
158        Import mass spectra data from the AndiNetCDF file.
159
160        Parameters
161        -----------
162        d_params : dict
163                Additional parameters for the mass spectra.
164
165        """
166        ms_datapoints_per_scans = self.net_cdf_obj.variables.get("point_count")[:]
167        list_tic = self.net_cdf_obj.variables.get("total_intensity")[:]
168        list_rt = self.net_cdf_obj.variables.get("scan_acquisition_time")[:] / 60
169        mass_values = self.net_cdf_obj.variables.get("mass_values")[:]
170        intensity_values = self.net_cdf_obj.variables.get("intensity_values")[:]
171        resolution = self.net_cdf_obj.variables.get("resolution")[:]
172        individual_rp = len(mass_values) == len(resolution)
173        finish_location = -1
174        for scan_index in self.list_scans:
175            datapoints = ms_datapoints_per_scans[scan_index]
176            finish_location += datapoints
177            start_location = finish_location - datapoints + 1
178            d_params["rt"] = list_rt[scan_index]
179            d_params["scan_number"] = scan_index
180            d_params["label"] = Labels.gcms_centroid
181            d_params["polarity"] = self.polarity
182            d_params["analyzer"] = self.analyzer
183            d_params["instrument_label"] = self.instrument_label
184            mz = mass_values[start_location:finish_location]
185            abun = intensity_values[start_location:finish_location]
186            if individual_rp:
187                rp = resolution[start_location:finish_location]
188            else:
189                rp = [resolution[scan_index]] * datapoints
190            self.get_mass_spectrum(mz, abun, rp, d_params)
191        self.gcms.retention_time = list_rt
192        self.gcms.tic = list_tic
193        self.gcms.scans_number = self.list_scans
194
195    def get_gcms_obj(self):
196        """
197        Get the GCMSBase object.
198
199        """
200        return self.gcms