corems.mass_spectrum.input.massList

View Source

  1__author__ = "Yuri E. Corilo"
  2__date__ = "Jun 12, 2019"
  3
  4import warnings
  5
  6import pandas as pd
  7
  8from corems.encapsulation.constant import Atoms, Labels
  9from corems.mass_spectrum.factory.MassSpectrumClasses import (
 10    MassSpecCentroid,
 11    MassSpecProfile,
 12)
 13from corems.mass_spectrum.input.baseClass import MassListBaseClass
 14from corems.molecular_formula.factory.MolecularFormulaFactory import MolecularFormula
 15
 16
 17class ReadCoremsMasslist(MassListBaseClass):
 18    """
 19    The ReadCoremsMasslist object reads processed mass list data types
 20    and returns the mass spectrum obj with the molecular formula obj
 21
 22    **Only available for centroid mass spectrum type:** it will ignore the parameter **isCentroid**
 23    Please see MassListBaseClass for more details
 24
 25    """
 26
 27    def get_mass_spectrum(
 28        self, loadSettings: bool = True, auto_process: bool = True
 29    ) -> MassSpecCentroid:
 30        """
 31        Get the mass spectrum object from the processed mass list data.
 32
 33        Parameters
 34        ----------
 35        loadSettings : bool, optional
 36            Whether to load the settings for the mass spectrum. Default is True.
 37        auto_process : bool, optional
 38            Whether to automatically process the mass spectrum on instantiation.
 39            When False, molecular formulas are not attached; call
 40            ``process_mass_spec`` and then ``add_molecular_formula`` on the
 41            returned object. Default is True.
 42
 43        Returns
 44        -------
 45        MassSpecCentroid
 46            The mass spectrum object.
 47
 48        Raises
 49        ------
 50        ValueError
 51            If the input file is not a valid CoreMS file.
 52        """
 53
 54        dataframe = self.get_dataframe()
 55
 56        if not set(
 57            ["H/C", "O/C", "Heteroatom Class", "Ion Type", "Is Isotopologue"]
 58        ).issubset(dataframe.columns):
 59            raise ValueError(
 60                "%s it is not a valid CoreMS file" % str(self.file_location)
 61            )
 62
 63        self.check_columns(dataframe.columns)
 64
 65        dataframe.rename(columns=self.parameters.header_translate, inplace=True)
 66
 67        polarity = dataframe["Ion Charge"].values[0]
 68
 69        output_parameters = self.get_output_parameters(polarity)
 70
 71        mass_spec_obj = MassSpecCentroid(
 72            dataframe.to_dict(orient="list"),
 73            output_parameters,
 74            auto_process=auto_process,
 75        )
 76
 77        if loadSettings is True:
 78            self.load_settings(mass_spec_obj, output_parameters)
 79
 80        if auto_process:
 81            self.add_molecular_formula(mass_spec_obj, dataframe)
 82
 83        return mass_spec_obj
 84
 85    def add_molecular_formula(self, mass_spec_obj, dataframe):
 86        """
 87        Add molecular formula information to the mass spectrum object.
 88
 89        Parameters
 90        ----------
 91        mass_spec_obj : MassSpecCentroid
 92            The mass spectrum object to add the molecular formula to.
 93        dataframe : pandas.DataFrame
 94            The processed mass list data.
 95
 96        """
 97
 98        # check if is coreMS file
 99        if "Is Isotopologue" in dataframe:
100            # Reindex dataframe to row index to avoid issues with duplicated indexes (e.g. when multiple formula map to single mz_exp)
101            dataframe = dataframe.reset_index(drop=True)
102
103            mz_exp_df = dataframe[Labels.mz].astype(float)
104            formula_df = dataframe[
105                dataframe.columns.intersection(Atoms.atoms_order)
106            ].copy()
107            # Convert to numeric first (pandas 3.x may infer str dtype for
108            # HDF5-sourced atom count columns); coerce handles b"nan" bytes too
109            formula_df = formula_df.apply(pd.to_numeric, errors="coerce").fillna(0)
110
111            ion_type_df = dataframe["Ion Type"]
112            ion_charge_df = dataframe["Ion Charge"]
113            is_isotopologue_df = dataframe["Is Isotopologue"]
114            if "Adduct" in dataframe:
115                adduct_df = dataframe["Adduct"]
116            else:
117                adduct_df = None
118
119        mass_spec_mz_exp_list = mass_spec_obj.mz_exp
120
121        for df_index, mz_exp in enumerate(mz_exp_df):
122            bad_mf = False
123            counts = 0
124
125            ms_peak_index = list(mass_spec_mz_exp_list).index(float(mz_exp))
126
127            if "Is Isotopologue" in dataframe:
128                atoms = list(formula_df.columns.astype(str))
129                counts = list(formula_df.iloc[df_index].astype(int))
130
131                formula_dict = dict(zip(atoms, counts))
132
133                # Drop any atoms with 0 counts
134                formula_dict = {
135                    atom: formula_dict[atom]
136                    for atom in formula_dict
137                    if formula_dict[atom] > 0
138                }
139
140            if sum(counts) > 0:
141                ion_type = str(Labels.ion_type_translate.get(ion_type_df[df_index]))
142                if adduct_df is not None:
143                    adduct_atom = str(adduct_df[df_index])
144                    if adduct_atom == "None":
145                        adduct_atom = None
146                else:
147                    adduct_atom = None
148
149                # If not isotopologue, cast as MolecularFormula
150                if not bool(int(is_isotopologue_df[df_index])):
151                    mfobj = MolecularFormula(
152                        formula_dict,
153                        int(ion_charge_df[df_index]),
154                        mspeak_parent=mass_spec_obj[ms_peak_index],
155                        ion_type=ion_type,
156                        adduct_atom=adduct_atom,
157                    )
158
159                # if is isotopologue, recast as MolecularFormulaIsotopologue
160                if bool(int(is_isotopologue_df[df_index])):
161                    # First make a MolecularFormula object for the parent so we can get probabilities etc
162                    formula_list_parent = {}
163                    for atom in formula_dict:
164                        if atom in Atoms.isotopes.keys():
165                            formula_list_parent[atom] = formula_dict[atom]
166                        else:
167                            # remove any numbers from the atom name to cast as a mono-isotopic atom
168                            atom_mono = atom.strip("0123456789")
169                            if (
170                                atom_mono in Atoms.isotopes.keys()
171                                and atom_mono in formula_list_parent.keys()
172                            ):
173                                formula_list_parent[atom_mono] = (
174                                    formula_list_parent[atom_mono] + formula_dict[atom]
175                                )
176                            elif atom_mono in Atoms.isotopes.keys():
177                                formula_list_parent[atom_mono] = formula_dict[atom]
178                            else:
179                                warnings.warn(f"Atom {atom} not in Atoms.atoms_order")
180                    mono_index = int(dataframe.iloc[df_index]["Mono Isotopic Index"])
181                    mono_mfobj = MolecularFormula(
182                        formula_list_parent,
183                        int(ion_charge_df[df_index]),
184                        mspeak_parent=mass_spec_obj[mono_index],
185                        ion_type=ion_type,
186                        adduct_atom=adduct_atom,
187                    )
188
189                    # Next, generate isotopologues from the parent
190                    isos = list(
191                        mono_mfobj.isotopologues(
192                            min_abundance=mass_spec_obj.abundance.min()*0.01,
193                            current_mono_abundance=mass_spec_obj[mono_index].abundance,
194                            dynamic_range=mass_spec_obj.dynamic_range,
195                        )
196                    )
197
198                    # Finally, find the isotopologue that matches the formula_dict
199                    matched_isos = []
200                    for iso in isos:
201                        # If match was already found, exit the loop
202                        if len(matched_isos) > 0:
203                            break
204                        else:
205                            # Check the atoms match
206                            if set(iso.atoms) == set(formula_dict.keys()):
207                                # Check the values of the atoms match
208                                if all(
209                                    [
210                                        iso[atom] == formula_dict[atom]
211                                        for atom in formula_dict
212                                    ]
213                                ):
214                                    matched_isos.append(iso)
215
216                    if len(matched_isos) == 0:
217                        #FIXME: This should not occur see https://code.emsl.pnl.gov/mass-spectrometry/corems/-/issues/190
218                        warnings.warn(f"No isotopologue matched the formula_dict: {formula_dict}")
219                        bad_mf = True
220                    else:
221                        bad_mf = False                   
222                        mfobj = matched_isos[0]
223
224                        # Add the mono isotopic index, confidence score and isotopologue similarity
225                        mfobj.mspeak_index_mono_isotopic = int(
226                            dataframe.iloc[df_index]["Mono Isotopic Index"]
227                        )
228                if not bad_mf:
229                    # Add the confidence score and isotopologue similarity and average MZ error score
230                    if "m/z Error Score" in dataframe:
231                        mfobj._mass_error_average_score = float(
232                            dataframe.iloc[df_index]["m/z Error Score"]
233                        )
234                    if "Confidence Score" in dataframe:
235                        mfobj._confidence_score = float(
236                            dataframe.iloc[df_index]["Confidence Score"]
237                        )
238                    if "Isotopologue Similarity" in dataframe:
239                        mfobj._isotopologue_similarity = float(
240                            dataframe.iloc[df_index]["Isotopologue Similarity"]
241                        )
242                    mass_spec_obj[ms_peak_index].add_molecular_formula(mfobj)
243
244
245class ReadMassList(MassListBaseClass):
246    """
247    The ReadMassList object reads unprocessed mass list data types
248    and returns the mass spectrum object.
249
250    Parameters
251    ----------
252    MassListBaseClass : class
253        The base class for reading mass list data types.
254
255    Methods
256    -------
257    * get_mass_spectrum(polarity, scan=0, auto_process=True, loadSettings=True). Reads mass list data types and returns the mass spectrum object.
258
259    """
260
261    def get_mass_spectrum(
262        self,
263        polarity: int,
264        scan: int = 0,
265        auto_process: bool = True,
266        loadSettings: bool = True,
267    ):
268        """
269        Reads mass list data types and returns the mass spectrum object.
270
271        Parameters
272        ----------
273        polarity : int
274            The polarity of the mass spectrum (+1 or -1).
275        scan : int, optional
276            The scan number of the mass spectrum (default is 0).
277        auto_process : bool, optional
278            Flag indicating whether to automatically process the mass spectrum (default is True).
279        loadSettings : bool, optional
280            Flag indicating whether to load settings for the mass spectrum (default is True).
281
282        Returns
283        -------
284        mass_spec : MassSpecCentroid or MassSpecProfile
285            The mass spectrum object.
286
287        """
288
289        # delimiter = "  " or " " or  "," or "\t" etc
290
291        if self.isCentroid:
292            dataframe = self.get_dataframe()
293
294            self.check_columns(dataframe.columns)
295
296            self.clean_data_frame(dataframe)
297
298            dataframe.rename(columns=self.parameters.header_translate, inplace=True)
299
300            output_parameters = self.get_output_parameters(polarity)
301
302            mass_spec = MassSpecCentroid(
303                dataframe.to_dict(orient="list"),
304                output_parameters,
305                auto_process=auto_process,
306            )
307
308            if loadSettings:
309                self.load_settings(mass_spec, output_parameters)
310
311            return mass_spec
312
313        else:
314            dataframe = self.get_dataframe()
315
316            self.check_columns(dataframe.columns)
317
318            output_parameters = self.get_output_parameters(polarity)
319
320            self.clean_data_frame(dataframe)
321
322            dataframe.rename(columns=self.parameters.header_translate, inplace=True)
323
324            mass_spec = MassSpecProfile(
325                dataframe.to_dict(orient="list"),
326                output_parameters,
327                auto_process=auto_process,
328            )
329
330            if loadSettings:
331                self.load_settings(mass_spec, output_parameters)
332
333            return mass_spec
334
335
336class ReadBrukerXMLList(MassListBaseClass):
337    """
338    The ReadBrukerXMLList object reads Bruker XML objects
339    and returns the mass spectrum object.
340    See MassListBaseClass for details
341
342    Parameters
343    ----------
344    MassListBaseClass : class
345        The base class for reading mass list data types and returning the mass spectrum object.
346
347    Methods
348    -------
349    * get_mass_spectrum(polarity: bool = None, scan: int = 0, auto_process: bool = True, loadSettings: bool = True). Reads mass list data types and returns the mass spectrum object.
350
351    """
352
353    def get_mass_spectrum(
354        self,
355        polarity: bool = None,
356        scan: int = 0,
357        auto_process: bool = True,
358        loadSettings: bool = True,
359    ):
360        """
361        Reads mass list data types and returns the mass spectrum object.
362
363        Parameters
364        ----------
365        polarity : bool, optional
366            The polarity of the mass spectrum. Can be +1 or -1. If not provided, it will be determined from the XML file.
367        scan : int, optional
368            The scan number of the mass spectrum. Default is 0.
369        auto_process : bool, optional
370            Whether to automatically process the mass spectrum. Default is True.
371        loadSettings : bool, optional
372            Whether to load the settings for the mass spectrum. Default is True.
373
374        Returns
375        -------
376        mass_spec : MassSpecCentroid
377            The mass spectrum object representing the centroided mass spectrum.
378        """
379        # delimiter = "  " or " " or  "," or "\t" etc
380
381        if polarity == None:
382            polarity = self.get_xml_polarity()
383        dataframe = self.get_dataframe()
384
385        self.check_columns(dataframe.columns)
386
387        self.clean_data_frame(dataframe)
388
389        dataframe.rename(columns=self.parameters.header_translate, inplace=True)
390
391        output_parameters = self.get_output_parameters(polarity)
392
393        mass_spec = MassSpecCentroid(
394            dataframe.to_dict(orient="list"),
395            output_parameters,
396            auto_process=auto_process,
397        )
398
399        if loadSettings:
400            self.load_settings(mass_spec, output_parameters)
401
402        return mass_spec

class ReadCoremsMasslist(corems.mass_spectrum.input.baseClass.MassListBaseClass): View Source

 18class ReadCoremsMasslist(MassListBaseClass):
 19    """
 20    The ReadCoremsMasslist object reads processed mass list data types
 21    and returns the mass spectrum obj with the molecular formula obj
 22
 23    **Only available for centroid mass spectrum type:** it will ignore the parameter **isCentroid**
 24    Please see MassListBaseClass for more details
 25
 26    """
 27
 28    def get_mass_spectrum(
 29        self, loadSettings: bool = True, auto_process: bool = True
 30    ) -> MassSpecCentroid:
 31        """
 32        Get the mass spectrum object from the processed mass list data.
 33
 34        Parameters
 35        ----------
 36        loadSettings : bool, optional
 37            Whether to load the settings for the mass spectrum. Default is True.
 38        auto_process : bool, optional
 39            Whether to automatically process the mass spectrum on instantiation.
 40            When False, molecular formulas are not attached; call
 41            ``process_mass_spec`` and then ``add_molecular_formula`` on the
 42            returned object. Default is True.
 43
 44        Returns
 45        -------
 46        MassSpecCentroid
 47            The mass spectrum object.
 48
 49        Raises
 50        ------
 51        ValueError
 52            If the input file is not a valid CoreMS file.
 53        """
 54
 55        dataframe = self.get_dataframe()
 56
 57        if not set(
 58            ["H/C", "O/C", "Heteroatom Class", "Ion Type", "Is Isotopologue"]
 59        ).issubset(dataframe.columns):
 60            raise ValueError(
 61                "%s it is not a valid CoreMS file" % str(self.file_location)
 62            )
 63
 64        self.check_columns(dataframe.columns)
 65
 66        dataframe.rename(columns=self.parameters.header_translate, inplace=True)
 67
 68        polarity = dataframe["Ion Charge"].values[0]
 69
 70        output_parameters = self.get_output_parameters(polarity)
 71
 72        mass_spec_obj = MassSpecCentroid(
 73            dataframe.to_dict(orient="list"),
 74            output_parameters,
 75            auto_process=auto_process,
 76        )
 77
 78        if loadSettings is True:
 79            self.load_settings(mass_spec_obj, output_parameters)
 80
 81        if auto_process:
 82            self.add_molecular_formula(mass_spec_obj, dataframe)
 83
 84        return mass_spec_obj
 85
 86    def add_molecular_formula(self, mass_spec_obj, dataframe):
 87        """
 88        Add molecular formula information to the mass spectrum object.
 89
 90        Parameters
 91        ----------
 92        mass_spec_obj : MassSpecCentroid
 93            The mass spectrum object to add the molecular formula to.
 94        dataframe : pandas.DataFrame
 95            The processed mass list data.
 96
 97        """
 98
 99        # check if is coreMS file
100        if "Is Isotopologue" in dataframe:
101            # Reindex dataframe to row index to avoid issues with duplicated indexes (e.g. when multiple formula map to single mz_exp)
102            dataframe = dataframe.reset_index(drop=True)
103
104            mz_exp_df = dataframe[Labels.mz].astype(float)
105            formula_df = dataframe[
106                dataframe.columns.intersection(Atoms.atoms_order)
107            ].copy()
108            # Convert to numeric first (pandas 3.x may infer str dtype for
109            # HDF5-sourced atom count columns); coerce handles b"nan" bytes too
110            formula_df = formula_df.apply(pd.to_numeric, errors="coerce").fillna(0)
111
112            ion_type_df = dataframe["Ion Type"]
113            ion_charge_df = dataframe["Ion Charge"]
114            is_isotopologue_df = dataframe["Is Isotopologue"]
115            if "Adduct" in dataframe:
116                adduct_df = dataframe["Adduct"]
117            else:
118                adduct_df = None
119
120        mass_spec_mz_exp_list = mass_spec_obj.mz_exp
121
122        for df_index, mz_exp in enumerate(mz_exp_df):
123            bad_mf = False
124            counts = 0
125
126            ms_peak_index = list(mass_spec_mz_exp_list).index(float(mz_exp))
127
128            if "Is Isotopologue" in dataframe:
129                atoms = list(formula_df.columns.astype(str))
130                counts = list(formula_df.iloc[df_index].astype(int))
131
132                formula_dict = dict(zip(atoms, counts))
133
134                # Drop any atoms with 0 counts
135                formula_dict = {
136                    atom: formula_dict[atom]
137                    for atom in formula_dict
138                    if formula_dict[atom] > 0
139                }
140
141            if sum(counts) > 0:
142                ion_type = str(Labels.ion_type_translate.get(ion_type_df[df_index]))
143                if adduct_df is not None:
144                    adduct_atom = str(adduct_df[df_index])
145                    if adduct_atom == "None":
146                        adduct_atom = None
147                else:
148                    adduct_atom = None
149
150                # If not isotopologue, cast as MolecularFormula
151                if not bool(int(is_isotopologue_df[df_index])):
152                    mfobj = MolecularFormula(
153                        formula_dict,
154                        int(ion_charge_df[df_index]),
155                        mspeak_parent=mass_spec_obj[ms_peak_index],
156                        ion_type=ion_type,
157                        adduct_atom=adduct_atom,
158                    )
159
160                # if is isotopologue, recast as MolecularFormulaIsotopologue
161                if bool(int(is_isotopologue_df[df_index])):
162                    # First make a MolecularFormula object for the parent so we can get probabilities etc
163                    formula_list_parent = {}
164                    for atom in formula_dict:
165                        if atom in Atoms.isotopes.keys():
166                            formula_list_parent[atom] = formula_dict[atom]
167                        else:
168                            # remove any numbers from the atom name to cast as a mono-isotopic atom
169                            atom_mono = atom.strip("0123456789")
170                            if (
171                                atom_mono in Atoms.isotopes.keys()
172                                and atom_mono in formula_list_parent.keys()
173                            ):
174                                formula_list_parent[atom_mono] = (
175                                    formula_list_parent[atom_mono] + formula_dict[atom]
176                                )
177                            elif atom_mono in Atoms.isotopes.keys():
178                                formula_list_parent[atom_mono] = formula_dict[atom]
179                            else:
180                                warnings.warn(f"Atom {atom} not in Atoms.atoms_order")
181                    mono_index = int(dataframe.iloc[df_index]["Mono Isotopic Index"])
182                    mono_mfobj = MolecularFormula(
183                        formula_list_parent,
184                        int(ion_charge_df[df_index]),
185                        mspeak_parent=mass_spec_obj[mono_index],
186                        ion_type=ion_type,
187                        adduct_atom=adduct_atom,
188                    )
189
190                    # Next, generate isotopologues from the parent
191                    isos = list(
192                        mono_mfobj.isotopologues(
193                            min_abundance=mass_spec_obj.abundance.min()*0.01,
194                            current_mono_abundance=mass_spec_obj[mono_index].abundance,
195                            dynamic_range=mass_spec_obj.dynamic_range,
196                        )
197                    )
198
199                    # Finally, find the isotopologue that matches the formula_dict
200                    matched_isos = []
201                    for iso in isos:
202                        # If match was already found, exit the loop
203                        if len(matched_isos) > 0:
204                            break
205                        else:
206                            # Check the atoms match
207                            if set(iso.atoms) == set(formula_dict.keys()):
208                                # Check the values of the atoms match
209                                if all(
210                                    [
211                                        iso[atom] == formula_dict[atom]
212                                        for atom in formula_dict
213                                    ]
214                                ):
215                                    matched_isos.append(iso)
216
217                    if len(matched_isos) == 0:
218                        #FIXME: This should not occur see https://code.emsl.pnl.gov/mass-spectrometry/corems/-/issues/190
219                        warnings.warn(f"No isotopologue matched the formula_dict: {formula_dict}")
220                        bad_mf = True
221                    else:
222                        bad_mf = False                   
223                        mfobj = matched_isos[0]
224
225                        # Add the mono isotopic index, confidence score and isotopologue similarity
226                        mfobj.mspeak_index_mono_isotopic = int(
227                            dataframe.iloc[df_index]["Mono Isotopic Index"]
228                        )
229                if not bad_mf:
230                    # Add the confidence score and isotopologue similarity and average MZ error score
231                    if "m/z Error Score" in dataframe:
232                        mfobj._mass_error_average_score = float(
233                            dataframe.iloc[df_index]["m/z Error Score"]
234                        )
235                    if "Confidence Score" in dataframe:
236                        mfobj._confidence_score = float(
237                            dataframe.iloc[df_index]["Confidence Score"]
238                        )
239                    if "Isotopologue Similarity" in dataframe:
240                        mfobj._isotopologue_similarity = float(
241                            dataframe.iloc[df_index]["Isotopologue Similarity"]
242                        )
243                    mass_spec_obj[ms_peak_index].add_molecular_formula(mfobj)

The ReadCoremsMasslist object reads processed mass list data types and returns the mass spectrum obj with the molecular formula obj

Only available for centroid mass spectrum type: it will ignore the parameter isCentroid Please see MassListBaseClass for more details

def get_mass_spectrum( self, loadSettings: bool = True, auto_process: bool = True) -> corems.mass_spectrum.factory.MassSpectrumClasses.MassSpecCentroid: View Source

28    def get_mass_spectrum(
29        self, loadSettings: bool = True, auto_process: bool = True
30    ) -> MassSpecCentroid:
31        """
32        Get the mass spectrum object from the processed mass list data.
33
34        Parameters
35        ----------
36        loadSettings : bool, optional
37            Whether to load the settings for the mass spectrum. Default is True.
38        auto_process : bool, optional
39            Whether to automatically process the mass spectrum on instantiation.
40            When False, molecular formulas are not attached; call
41            ``process_mass_spec`` and then ``add_molecular_formula`` on the
42            returned object. Default is True.
43
44        Returns
45        -------
46        MassSpecCentroid
47            The mass spectrum object.
48
49        Raises
50        ------
51        ValueError
52            If the input file is not a valid CoreMS file.
53        """
54
55        dataframe = self.get_dataframe()
56
57        if not set(
58            ["H/C", "O/C", "Heteroatom Class", "Ion Type", "Is Isotopologue"]
59        ).issubset(dataframe.columns):
60            raise ValueError(
61                "%s it is not a valid CoreMS file" % str(self.file_location)
62            )
63
64        self.check_columns(dataframe.columns)
65
66        dataframe.rename(columns=self.parameters.header_translate, inplace=True)
67
68        polarity = dataframe["Ion Charge"].values[0]
69
70        output_parameters = self.get_output_parameters(polarity)
71
72        mass_spec_obj = MassSpecCentroid(
73            dataframe.to_dict(orient="list"),
74            output_parameters,
75            auto_process=auto_process,
76        )
77
78        if loadSettings is True:
79            self.load_settings(mass_spec_obj, output_parameters)
80
81        if auto_process:
82            self.add_molecular_formula(mass_spec_obj, dataframe)
83
84        return mass_spec_obj

Get the mass spectrum object from the processed mass list data.

Parameters

loadSettings (bool, optional): Whether to load the settings for the mass spectrum. Default is True.
auto_process (bool, optional): Whether to automatically process the mass spectrum on instantiation. When False, molecular formulas are not attached; call process_mass_spec and then add_molecular_formula on the returned object. Default is True.

Returns

MassSpecCentroid: The mass spectrum object.

Raises

ValueError: If the input file is not a valid CoreMS file.

def add_molecular_formula(self, mass_spec_obj, dataframe): View Source

 86    def add_molecular_formula(self, mass_spec_obj, dataframe):
 87        """
 88        Add molecular formula information to the mass spectrum object.
 89
 90        Parameters
 91        ----------
 92        mass_spec_obj : MassSpecCentroid
 93            The mass spectrum object to add the molecular formula to.
 94        dataframe : pandas.DataFrame
 95            The processed mass list data.
 96
 97        """
 98
 99        # check if is coreMS file
100        if "Is Isotopologue" in dataframe:
101            # Reindex dataframe to row index to avoid issues with duplicated indexes (e.g. when multiple formula map to single mz_exp)
102            dataframe = dataframe.reset_index(drop=True)
103
104            mz_exp_df = dataframe[Labels.mz].astype(float)
105            formula_df = dataframe[
106                dataframe.columns.intersection(Atoms.atoms_order)
107            ].copy()
108            # Convert to numeric first (pandas 3.x may infer str dtype for
109            # HDF5-sourced atom count columns); coerce handles b"nan" bytes too
110            formula_df = formula_df.apply(pd.to_numeric, errors="coerce").fillna(0)
111
112            ion_type_df = dataframe["Ion Type"]
113            ion_charge_df = dataframe["Ion Charge"]
114            is_isotopologue_df = dataframe["Is Isotopologue"]
115            if "Adduct" in dataframe:
116                adduct_df = dataframe["Adduct"]
117            else:
118                adduct_df = None
119
120        mass_spec_mz_exp_list = mass_spec_obj.mz_exp
121
122        for df_index, mz_exp in enumerate(mz_exp_df):
123            bad_mf = False
124            counts = 0
125
126            ms_peak_index = list(mass_spec_mz_exp_list).index(float(mz_exp))
127
128            if "Is Isotopologue" in dataframe:
129                atoms = list(formula_df.columns.astype(str))
130                counts = list(formula_df.iloc[df_index].astype(int))
131
132                formula_dict = dict(zip(atoms, counts))
133
134                # Drop any atoms with 0 counts
135                formula_dict = {
136                    atom: formula_dict[atom]
137                    for atom in formula_dict
138                    if formula_dict[atom] > 0
139                }
140
141            if sum(counts) > 0:
142                ion_type = str(Labels.ion_type_translate.get(ion_type_df[df_index]))
143                if adduct_df is not None:
144                    adduct_atom = str(adduct_df[df_index])
145                    if adduct_atom == "None":
146                        adduct_atom = None
147                else:
148                    adduct_atom = None
149
150                # If not isotopologue, cast as MolecularFormula
151                if not bool(int(is_isotopologue_df[df_index])):
152                    mfobj = MolecularFormula(
153                        formula_dict,
154                        int(ion_charge_df[df_index]),
155                        mspeak_parent=mass_spec_obj[ms_peak_index],
156                        ion_type=ion_type,
157                        adduct_atom=adduct_atom,
158                    )
159
160                # if is isotopologue, recast as MolecularFormulaIsotopologue
161                if bool(int(is_isotopologue_df[df_index])):
162                    # First make a MolecularFormula object for the parent so we can get probabilities etc
163                    formula_list_parent = {}
164                    for atom in formula_dict:
165                        if atom in Atoms.isotopes.keys():
166                            formula_list_parent[atom] = formula_dict[atom]
167                        else:
168                            # remove any numbers from the atom name to cast as a mono-isotopic atom
169                            atom_mono = atom.strip("0123456789")
170                            if (
171                                atom_mono in Atoms.isotopes.keys()
172                                and atom_mono in formula_list_parent.keys()
173                            ):
174                                formula_list_parent[atom_mono] = (
175                                    formula_list_parent[atom_mono] + formula_dict[atom]
176                                )
177                            elif atom_mono in Atoms.isotopes.keys():
178                                formula_list_parent[atom_mono] = formula_dict[atom]
179                            else:
180                                warnings.warn(f"Atom {atom} not in Atoms.atoms_order")
181                    mono_index = int(dataframe.iloc[df_index]["Mono Isotopic Index"])
182                    mono_mfobj = MolecularFormula(
183                        formula_list_parent,
184                        int(ion_charge_df[df_index]),
185                        mspeak_parent=mass_spec_obj[mono_index],
186                        ion_type=ion_type,
187                        adduct_atom=adduct_atom,
188                    )
189
190                    # Next, generate isotopologues from the parent
191                    isos = list(
192                        mono_mfobj.isotopologues(
193                            min_abundance=mass_spec_obj.abundance.min()*0.01,
194                            current_mono_abundance=mass_spec_obj[mono_index].abundance,
195                            dynamic_range=mass_spec_obj.dynamic_range,
196                        )
197                    )
198
199                    # Finally, find the isotopologue that matches the formula_dict
200                    matched_isos = []
201                    for iso in isos:
202                        # If match was already found, exit the loop
203                        if len(matched_isos) > 0:
204                            break
205                        else:
206                            # Check the atoms match
207                            if set(iso.atoms) == set(formula_dict.keys()):
208                                # Check the values of the atoms match
209                                if all(
210                                    [
211                                        iso[atom] == formula_dict[atom]
212                                        for atom in formula_dict
213                                    ]
214                                ):
215                                    matched_isos.append(iso)
216
217                    if len(matched_isos) == 0:
218                        #FIXME: This should not occur see https://code.emsl.pnl.gov/mass-spectrometry/corems/-/issues/190
219                        warnings.warn(f"No isotopologue matched the formula_dict: {formula_dict}")
220                        bad_mf = True
221                    else:
222                        bad_mf = False                   
223                        mfobj = matched_isos[0]
224
225                        # Add the mono isotopic index, confidence score and isotopologue similarity
226                        mfobj.mspeak_index_mono_isotopic = int(
227                            dataframe.iloc[df_index]["Mono Isotopic Index"]
228                        )
229                if not bad_mf:
230                    # Add the confidence score and isotopologue similarity and average MZ error score
231                    if "m/z Error Score" in dataframe:
232                        mfobj._mass_error_average_score = float(
233                            dataframe.iloc[df_index]["m/z Error Score"]
234                        )
235                    if "Confidence Score" in dataframe:
236                        mfobj._confidence_score = float(
237                            dataframe.iloc[df_index]["Confidence Score"]
238                        )
239                    if "Isotopologue Similarity" in dataframe:
240                        mfobj._isotopologue_similarity = float(
241                            dataframe.iloc[df_index]["Isotopologue Similarity"]
242                        )
243                    mass_spec_obj[ms_peak_index].add_molecular_formula(mfobj)

Add molecular formula information to the mass spectrum object.

Parameters

mass_spec_obj (MassSpecCentroid): The mass spectrum object to add the molecular formula to.
dataframe (pandas.DataFrame): The processed mass list data.

Inherited Members

corems.mass_spectrum.input.baseClass.MassListBaseClass: MassListBaseClass; file_location; header_lines; isCentroid; isThermoProfile; headerless; analyzer; instrument_label; sample_name; parameters; set_parameter_from_toml; set_parameter_from_json; data_type; delimiter; encoding_detector; set_data_type; get_dataframe; load_settings; get_output_parameters; clean_data_frame; check_columns; read_xml_peaks; get_xml_polarity

class ReadMassList(corems.mass_spectrum.input.baseClass.MassListBaseClass): View Source

246class ReadMassList(MassListBaseClass):
247    """
248    The ReadMassList object reads unprocessed mass list data types
249    and returns the mass spectrum object.
250
251    Parameters
252    ----------
253    MassListBaseClass : class
254        The base class for reading mass list data types.
255
256    Methods
257    -------
258    * get_mass_spectrum(polarity, scan=0, auto_process=True, loadSettings=True). Reads mass list data types and returns the mass spectrum object.
259
260    """
261
262    def get_mass_spectrum(
263        self,
264        polarity: int,
265        scan: int = 0,
266        auto_process: bool = True,
267        loadSettings: bool = True,
268    ):
269        """
270        Reads mass list data types and returns the mass spectrum object.
271
272        Parameters
273        ----------
274        polarity : int
275            The polarity of the mass spectrum (+1 or -1).
276        scan : int, optional
277            The scan number of the mass spectrum (default is 0).
278        auto_process : bool, optional
279            Flag indicating whether to automatically process the mass spectrum (default is True).
280        loadSettings : bool, optional
281            Flag indicating whether to load settings for the mass spectrum (default is True).
282
283        Returns
284        -------
285        mass_spec : MassSpecCentroid or MassSpecProfile
286            The mass spectrum object.
287
288        """
289
290        # delimiter = "  " or " " or  "," or "\t" etc
291
292        if self.isCentroid:
293            dataframe = self.get_dataframe()
294
295            self.check_columns(dataframe.columns)
296
297            self.clean_data_frame(dataframe)
298
299            dataframe.rename(columns=self.parameters.header_translate, inplace=True)
300
301            output_parameters = self.get_output_parameters(polarity)
302
303            mass_spec = MassSpecCentroid(
304                dataframe.to_dict(orient="list"),
305                output_parameters,
306                auto_process=auto_process,
307            )
308
309            if loadSettings:
310                self.load_settings(mass_spec, output_parameters)
311
312            return mass_spec
313
314        else:
315            dataframe = self.get_dataframe()
316
317            self.check_columns(dataframe.columns)
318
319            output_parameters = self.get_output_parameters(polarity)
320
321            self.clean_data_frame(dataframe)
322
323            dataframe.rename(columns=self.parameters.header_translate, inplace=True)
324
325            mass_spec = MassSpecProfile(
326                dataframe.to_dict(orient="list"),
327                output_parameters,
328                auto_process=auto_process,
329            )
330
331            if loadSettings:
332                self.load_settings(mass_spec, output_parameters)
333
334            return mass_spec

The ReadMassList object reads unprocessed mass list data types and returns the mass spectrum object.

Parameters

MassListBaseClass (class): The base class for reading mass list data types.

Methods

get_mass_spectrum(polarity, scan=0, auto_process=True, loadSettings=True). Reads mass list data types and returns the mass spectrum object.

def get_mass_spectrum( self, polarity: int, scan: int = 0, auto_process: bool = True, loadSettings: bool = True): View Source

262    def get_mass_spectrum(
263        self,
264        polarity: int,
265        scan: int = 0,
266        auto_process: bool = True,
267        loadSettings: bool = True,
268    ):
269        """
270        Reads mass list data types and returns the mass spectrum object.
271
272        Parameters
273        ----------
274        polarity : int
275            The polarity of the mass spectrum (+1 or -1).
276        scan : int, optional
277            The scan number of the mass spectrum (default is 0).
278        auto_process : bool, optional
279            Flag indicating whether to automatically process the mass spectrum (default is True).
280        loadSettings : bool, optional
281            Flag indicating whether to load settings for the mass spectrum (default is True).
282
283        Returns
284        -------
285        mass_spec : MassSpecCentroid or MassSpecProfile
286            The mass spectrum object.
287
288        """
289
290        # delimiter = "  " or " " or  "," or "\t" etc
291
292        if self.isCentroid:
293            dataframe = self.get_dataframe()
294
295            self.check_columns(dataframe.columns)
296
297            self.clean_data_frame(dataframe)
298
299            dataframe.rename(columns=self.parameters.header_translate, inplace=True)
300
301            output_parameters = self.get_output_parameters(polarity)
302
303            mass_spec = MassSpecCentroid(
304                dataframe.to_dict(orient="list"),
305                output_parameters,
306                auto_process=auto_process,
307            )
308
309            if loadSettings:
310                self.load_settings(mass_spec, output_parameters)
311
312            return mass_spec
313
314        else:
315            dataframe = self.get_dataframe()
316
317            self.check_columns(dataframe.columns)
318
319            output_parameters = self.get_output_parameters(polarity)
320
321            self.clean_data_frame(dataframe)
322
323            dataframe.rename(columns=self.parameters.header_translate, inplace=True)
324
325            mass_spec = MassSpecProfile(
326                dataframe.to_dict(orient="list"),
327                output_parameters,
328                auto_process=auto_process,
329            )
330
331            if loadSettings:
332                self.load_settings(mass_spec, output_parameters)
333
334            return mass_spec

Reads mass list data types and returns the mass spectrum object.

Parameters

polarity (int): The polarity of the mass spectrum (+1 or -1).
scan (int, optional): The scan number of the mass spectrum (default is 0).
auto_process (bool, optional): Flag indicating whether to automatically process the mass spectrum (default is True).
loadSettings (bool, optional): Flag indicating whether to load settings for the mass spectrum (default is True).

Returns

mass_spec (MassSpecCentroid or MassSpecProfile): The mass spectrum object.

Inherited Members

corems.mass_spectrum.input.baseClass.MassListBaseClass: MassListBaseClass; file_location; header_lines; isCentroid; isThermoProfile; headerless; analyzer; instrument_label; sample_name; parameters; set_parameter_from_toml; set_parameter_from_json; data_type; delimiter; encoding_detector; set_data_type; get_dataframe; load_settings; get_output_parameters; clean_data_frame; check_columns; read_xml_peaks; get_xml_polarity

class ReadBrukerXMLList(corems.mass_spectrum.input.baseClass.MassListBaseClass): View Source

337class ReadBrukerXMLList(MassListBaseClass):
338    """
339    The ReadBrukerXMLList object reads Bruker XML objects
340    and returns the mass spectrum object.
341    See MassListBaseClass for details
342
343    Parameters
344    ----------
345    MassListBaseClass : class
346        The base class for reading mass list data types and returning the mass spectrum object.
347
348    Methods
349    -------
350    * get_mass_spectrum(polarity: bool = None, scan: int = 0, auto_process: bool = True, loadSettings: bool = True). Reads mass list data types and returns the mass spectrum object.
351
352    """
353
354    def get_mass_spectrum(
355        self,
356        polarity: bool = None,
357        scan: int = 0,
358        auto_process: bool = True,
359        loadSettings: bool = True,
360    ):
361        """
362        Reads mass list data types and returns the mass spectrum object.
363
364        Parameters
365        ----------
366        polarity : bool, optional
367            The polarity of the mass spectrum. Can be +1 or -1. If not provided, it will be determined from the XML file.
368        scan : int, optional
369            The scan number of the mass spectrum. Default is 0.
370        auto_process : bool, optional
371            Whether to automatically process the mass spectrum. Default is True.
372        loadSettings : bool, optional
373            Whether to load the settings for the mass spectrum. Default is True.
374
375        Returns
376        -------
377        mass_spec : MassSpecCentroid
378            The mass spectrum object representing the centroided mass spectrum.
379        """
380        # delimiter = "  " or " " or  "," or "\t" etc
381
382        if polarity == None:
383            polarity = self.get_xml_polarity()
384        dataframe = self.get_dataframe()
385
386        self.check_columns(dataframe.columns)
387
388        self.clean_data_frame(dataframe)
389
390        dataframe.rename(columns=self.parameters.header_translate, inplace=True)
391
392        output_parameters = self.get_output_parameters(polarity)
393
394        mass_spec = MassSpecCentroid(
395            dataframe.to_dict(orient="list"),
396            output_parameters,
397            auto_process=auto_process,
398        )
399
400        if loadSettings:
401            self.load_settings(mass_spec, output_parameters)
402
403        return mass_spec

The ReadBrukerXMLList object reads Bruker XML objects and returns the mass spectrum object. See MassListBaseClass for details

Parameters

MassListBaseClass (class): The base class for reading mass list data types and returning the mass spectrum object.

Methods

get_mass_spectrum(polarity: bool = None, scan: int = 0, auto_process: bool = True, loadSettings: bool = True). Reads mass list data types and returns the mass spectrum object.

def get_mass_spectrum( self, polarity: bool = None, scan: int = 0, auto_process: bool = True, loadSettings: bool = True): View Source

354    def get_mass_spectrum(
355        self,
356        polarity: bool = None,
357        scan: int = 0,
358        auto_process: bool = True,
359        loadSettings: bool = True,
360    ):
361        """
362        Reads mass list data types and returns the mass spectrum object.
363
364        Parameters
365        ----------
366        polarity : bool, optional
367            The polarity of the mass spectrum. Can be +1 or -1. If not provided, it will be determined from the XML file.
368        scan : int, optional
369            The scan number of the mass spectrum. Default is 0.
370        auto_process : bool, optional
371            Whether to automatically process the mass spectrum. Default is True.
372        loadSettings : bool, optional
373            Whether to load the settings for the mass spectrum. Default is True.
374
375        Returns
376        -------
377        mass_spec : MassSpecCentroid
378            The mass spectrum object representing the centroided mass spectrum.
379        """
380        # delimiter = "  " or " " or  "," or "\t" etc
381
382        if polarity == None:
383            polarity = self.get_xml_polarity()
384        dataframe = self.get_dataframe()
385
386        self.check_columns(dataframe.columns)
387
388        self.clean_data_frame(dataframe)
389
390        dataframe.rename(columns=self.parameters.header_translate, inplace=True)
391
392        output_parameters = self.get_output_parameters(polarity)
393
394        mass_spec = MassSpecCentroid(
395            dataframe.to_dict(orient="list"),
396            output_parameters,
397            auto_process=auto_process,
398        )
399
400        if loadSettings:
401            self.load_settings(mass_spec, output_parameters)
402
403        return mass_spec

Reads mass list data types and returns the mass spectrum object.

Parameters

polarity (bool, optional): The polarity of the mass spectrum. Can be +1 or -1. If not provided, it will be determined from the XML file.
scan (int, optional): The scan number of the mass spectrum. Default is 0.
auto_process (bool, optional): Whether to automatically process the mass spectrum. Default is True.
loadSettings (bool, optional): Whether to load the settings for the mass spectrum. Default is True.

Returns

mass_spec (MassSpecCentroid): The mass spectrum object representing the centroided mass spectrum.

Inherited Members

corems.mass_spectrum.input.baseClass.MassListBaseClass: MassListBaseClass; file_location; header_lines; isCentroid; isThermoProfile; headerless; analyzer; instrument_label; sample_name; parameters; set_parameter_from_toml; set_parameter_from_json; data_type; delimiter; encoding_detector; set_data_type; get_dataframe; load_settings; get_output_parameters; clean_data_frame; check_columns; read_xml_peaks; get_xml_polarity