corems.mass_spectrum.input.coremsHDF5 

  1import json
  2
  3from pandas import DataFrame
  4import h5py
  5from io import BytesIO
  6from s3path import S3Path
  7
  8from corems.encapsulation.input.parameter_from_json import _set_dict_data_ms
  9from corems.mass_spectrum.input.massList import ReadCoremsMasslist
 10from corems.mass_spectrum.factory.MassSpectrumClasses import MassSpecCentroid
 11from corems.encapsulation.factory.parameters import default_parameters
 12
 13
 14class ReadCoreMSHDF_MassSpectrum(ReadCoremsMasslist):
 15    """Class for reading mass spectrum data from a CoreMS HDF5 file.
 16
 17    Attributes
 18    ----------
 19    h5pydata : h5py.File
 20        The HDF5 file object.
 21    scans : list
 22        List of scan labels in the HDF5 file.
 23
 24    Parameters
 25    ----------
 26    file_location : str or S3Path
 27        The path to the CoreMS HDF5 file.
 28
 29    Methods
 30    -------
 31    * load_raw_data(scan_index=0). Read a scan's stored profile arrays, if present.
 32    * load_peak_profile_indices(scan_index=0, time_index=-1). Read a scan's stored per-peak profile indices for a given peak-table version, if present.
 33    * get_mass_spectrum(scan_number=0, time_index=-1, auto_process=True, load_settings=True, load_raw=True, load_molecular_formula=True, force_centroid=False). Get a mass spectrum object.
 34    * load_settings(mass_spectrum, scan_index=0, time_index=-1). Load settings into the mass spectrum object.
 35    * get_dataframe(scan_index=0, time_index=-1). Get a pandas DataFrame representing the mass spectrum.
 36    * get_time_index_to_pull(scan_label, time_index). Get the time index to pull from the HDF5 file.
 37    * get_high_level_attr_data(attr_str). Get high-level attribute data from the HDF5 file.
 38    * get_scan_group_attr_data(scan_index, time_index, attr_group, attr_srt=None). Get scan group attribute data from the HDF5 file.
 39    * get_raw_data_attr_data(scan_index, attr_group, attr_str). Get raw data attribute data from the HDF5 file.
 40    * get_output_parameters(polarity, scan_index=0). Get the output parameters for the mass spectrum.
 41    """
 42
 43    def __init__(self, file_location):
 44        super().__init__(file_location)
 45
 46        if isinstance(self.file_location, S3Path):
 47            data = BytesIO(self.file_location.open("rb").read())
 48        else:
 49            data = self.file_location
 50
 51        self.h5pydata = h5py.File(data, "r")
 52
 53        self.scans = list(self.h5pydata.keys())
 54
 55    def close(self):
 56        """Close the underlying HDF5 file handle."""
 57        if self.h5pydata and self.h5pydata.id.valid:
 58            self.h5pydata.close()
 59
 60    def __enter__(self):
 61        return self
 62
 63    def __exit__(self, *args):
 64        self.close()
 65
 66    def load_raw_data(self, scan_index=0):
 67        """
 68        Read the stored profile arrays for a scan, if present.
 69
 70        Parameters
 71        ----------
 72        scan_index : int, optional
 73            The index of the scan to load the raw data from. Default is 0.
 74
 75        Returns
 76        -------
 77        dict or None
 78            ``{"mz": ..., "abundance": ..., "mz_cal": ... or None,
 79            "is_calibrated": bool}`` (the experimental m/z, abundance,
 80            calibrated m/z if stored, and whether the spectrum is actually
 81            calibrated), or None if the scan has no ``raw_ms`` dataset (a
 82            centroid-only file).
 83
 84        Notes
 85        -----
 86        ``raw_ms`` rows are ordered per its ``row_labels`` attribute (JSON
 87        list of ``"mz"``/``"abundance"``/``"mz_cal"``). Files written before
 88        that attribute existed have no ``row_labels`` and are read back in
 89        the legacy ``[abundance, mz]`` order every earlier CoreMS version
 90        wrote, so old files keep loading exactly as they did before.
 91
 92        ``is_calibrated`` is read from the ``raw_ms`` ``is_calibrated``
 93        attribute, trusted over row presence; files written before that
 94        attribute existed fall back to whether an ``"mz_cal"`` row exists.
 95        """
 96        scan_label = self.scans[scan_index]
 97        raw_ms_group = self.h5pydata[scan_label]
 98
 99        if "raw_ms" not in raw_ms_group or raw_ms_group["raw_ms"].shape is None:
100            return None
101
102        raw_ms_dataset = raw_ms_group["raw_ms"]
103        row_labels_json = raw_ms_dataset.attrs.get("row_labels")
104        row_labels = (
105            json.loads(row_labels_json) if row_labels_json else ["abundance", "mz"]
106        )
107
108        data = raw_ms_dataset[()]
109        mz_profile = data[row_labels.index("mz")]
110        abundance_profile = data[row_labels.index("abundance")]
111        mz_cal_profile = (
112            data[row_labels.index("mz_cal")] if "mz_cal" in row_labels else None
113        )
114        is_calibrated = bool(
115            raw_ms_dataset.attrs.get("is_calibrated", "mz_cal" in row_labels)
116        )
117
118        return {
119            "mz": mz_profile,
120            "abundance": abundance_profile,
121            "mz_cal": mz_cal_profile,
122            "is_calibrated": is_calibrated,
123        }
124
125    def load_peak_profile_indices(self, scan_index=0, time_index=-1):
126        """
127        Read the stored per-peak profile indices for a scan, if present.
128
129        Parameters
130        ----------
131        scan_index : int, optional
132            The index of the scan to load the indices from. Default is 0.
133        time_index : int, optional
134            Which peak-table version to load the indices for (same meaning
135            as elsewhere: resolved via ``get_time_index_to_pull``). Peak
136            tables are versioned ("0", "1", ...) as a scan is reprocessed,
137            and the indices are stored per version alongside each one, since
138            the picked peaks (unlike the profile itself) differ by version.
139            Default is -1 (the latest).
140
141        Returns
142        -------
143        numpy.ndarray or None
144            Array of shape ``(n_peaks, 3)`` of ``(left, apex, right)``
145            indices into the scan's profile arrays, in the same order as the
146            peak-table version requested, or None if not stored (file
147            predates this feature, or this version predates it).
148        """
149        scan_label = self.scans[scan_index]
150        scan_group = self.h5pydata[scan_label]
151        index_to_pull = self.get_time_index_to_pull(scan_label, time_index)
152        key = f"peak_profile_indices_{index_to_pull}"
153        if key not in scan_group:
154            return None
155        return scan_group[key][()]
156
157    def get_mass_spectrum(
158        self,
159        scan_number=0,
160        time_index=-1,
161        auto_process=True,
162        load_settings=True,
163        load_raw=True,
164        load_molecular_formula=True,
165        force_centroid=False,
166    ):
167        """
168        Instantiate a mass spectrum object from the CoreMS HDF5 file.
169
170        Parameters
171        ----------
172        scan_number : int, optional
173            The index of the scan to retrieve the mass spectrum from. Default is 0.
174        time_index : int, optional
175            The index of the time point to retrieve the mass spectrum from. Default is -1.
176        auto_process : bool, optional
177            Whether to automatically process the mass spectrum. Default is True.
178        load_settings : bool, optional
179            Whether to load the settings into the mass spectrum object. Default is True.
180        load_raw : bool, optional
181            Whether to load the raw profile data into the mass spectrum object,
182            when the scan has one. Default is True.
183        load_molecular_formula : bool, optional
184            Whether to load the molecular formula into the mass spectrum object.
185            Default is True.
186        force_centroid : bool, optional
187            If True, always return a :class:`MassSpecCentroid`, even when the
188            scan has a stored profile. Has no effect on a file with no stored
189            profile for this scan, which already returns a
190            :class:`MassSpecCentroid`. Default is False.
191
192        Returns
193        -------
194        MassSpecCentroid or MassSpecProfile
195            A :class:`MassSpecProfile` sharing the loaded peaks when the scan
196            has a stored profile, ``load_raw`` is True, and ``force_centroid``
197            is False; otherwise a :class:`MassSpecCentroid`.
198
199        Raises
200        ------
201        ValueError
202            If the CoreMS file is not valid.
203            If the mass spectrum has not been processed and load_molecular_formula is True.
204        """
205        if "mass_spectra" in self.scans[0]:
206            scan_index = self.scans.index("mass_spectra/" + str(scan_number))
207        else:
208            scan_index = self.scans.index(str(scan_number))
209        dataframe = self.get_dataframe(scan_index, time_index=time_index)
210        if dataframe["Molecular Formula"].any() and not dataframe["C"].any():
211            cols = dataframe.columns.tolist()
212            cols = cols[cols.index("Molecular Formula") + 1 :]
213            for index, row in dataframe.iterrows():
214                if row["Molecular Formula"] is not None:
215                    og_formula = row["Molecular Formula"]
216                    for col in cols:
217                        if "col" in og_formula:
218                            # get the digit after the element ("col") in the molecular formula and set it to the dataframe
219                            row[col] = int(og_formula.split(col)[1].split(" ")[0])
220
221        if not set(
222            ["H/C", "O/C", "Heteroatom Class", "Ion Type", "Is Isotopologue"]
223        ).issubset(dataframe.columns):
224            raise ValueError(
225                "%s it is not a valid CoreMS file" % str(self.file_location)
226            )
227
228        dataframe.rename(columns=self.parameters.header_translate, inplace=True)
229
230        # Cast m/z, and 'Peak Height' to float
231        dataframe["m/z"] = dataframe["m/z"].astype(float)
232        dataframe["Peak Height"] = dataframe["Peak Height"].astype(float)
233
234        polarity = dataframe["Ion Charge"].iloc[0]
235
236        output_parameters = self.get_output_parameters(polarity, scan_index=scan_index)
237
238        mass_spec_obj = MassSpecCentroid(
239            dataframe.to_dict(orient="list"), output_parameters, auto_process=False
240        )
241
242        if auto_process:
243            # Set the settings on the mass spectrum object to relative abuncance of 0 so all peaks get added
244            mass_spec_obj.settings.noise_threshold_method = "absolute_abundance"
245            mass_spec_obj.settings.noise_threshold_absolute_abundance = 0
246            mass_spec_obj.process_mass_spec()
247            # MassSpecCentroid.process_mass_spec() recomputes baseline_noise/
248            # _std from this scan's own (small) peak table unless label ==
249            # Labels.thermo_centroid, which it never is here -- that estimate
250            # is unrelated to, and much noisier than, the original spectrum's
251            # stored noise stats. Put the stored values back so a reloaded
252            # spectrum's noise stats match what was exported, same as every
253            # other MassSpecAttrs field already restored via output_parameters.
254            # Fall back to the just-recomputed value if the stored one is
255            # somehow missing, rather than erroring or overwriting with None.
256            mass_spec_obj._baseline_noise = output_parameters.get(
257                "baseline_noise", mass_spec_obj._baseline_noise
258            )
259            mass_spec_obj._baseline_noise_std = output_parameters.get(
260                "baseline_noise_std", mass_spec_obj._baseline_noise_std
261            )
262
263        if load_settings:
264            # Load settings into the mass spectrum object
265            self.load_settings(
266                mass_spec_obj, scan_index=scan_index, time_index=time_index
267            )
268
269        if load_molecular_formula:
270            if not auto_process:
271                raise ValueError(
272                    "Can only add molecular formula if the mass spectrum has been processed"
273                )
274            else:
275                # Attach formulas to the centroid's peaks before any profile
276                # conversion below, so to_profile() carries them over for free
277                # (it reuses these same peak objects, it doesn't rebuild them).
278                # Also before calibration restore: this matches the uncalibrated
279                # "m/z" column, and MSPeak.mz_exp returns mz_cal once that is set.
280                self.add_molecular_formula(mass_spec_obj, dataframe)
281
282        if auto_process:
283            # After formula attachment, before to_profile(). The fallback
284            # peak-index search needs a calibrated peak's mz_cal already set.
285            self.restore_peak_calibration(mass_spec_obj, dataframe)
286
287        if load_raw and not force_centroid:
288            raw_arrays = self.load_raw_data(scan_index=scan_index)
289            if raw_arrays is not None:
290                peak_profile_indices = self.load_peak_profile_indices(
291                    scan_index=scan_index, time_index=time_index
292                )
293                mass_spec_obj = mass_spec_obj.to_profile(
294                    raw_arrays["mz"],
295                    raw_arrays["abundance"],
296                    mz_cal_profile=raw_arrays["mz_cal"],
297                    peak_profile_indices=peak_profile_indices,
298                    is_calibrated=raw_arrays["is_calibrated"],
299                )
300
301        return mass_spec_obj
302
303    def load_settings(self, mass_spectrum, scan_index=0, time_index=-1):
304        """
305        Load settings into the mass spectrum object.
306
307        Parameters
308        ----------
309        mass_spectrum : MassSpecCentroid
310            The mass spectrum object to load the settings into.
311        scan_index : int, optional
312            The index of the scan to load the settings from. Default is 0.
313        time_index : int, optional
314            The index of the time point to load the settings from. Default is -1.
315        """
316
317        loaded_settings = {}
318        loaded_settings["MoleculaSearch"] = self.get_scan_group_attr_data(
319            scan_index, time_index, "MoleculaSearchSetting"
320        )
321        loaded_settings["MassSpecPeak"] = self.get_scan_group_attr_data(
322            scan_index, time_index, "MassSpecPeakSetting"
323        )
324        loaded_settings["MassSpectrum"] = self.get_scan_group_attr_data(
325            scan_index, time_index, "MassSpectrumSetting"
326        )
327        loaded_settings["Transient"] = self.get_scan_group_attr_data(
328            scan_index, time_index, "TransientSetting"
329        )
330
331        _set_dict_data_ms(loaded_settings, mass_spectrum)
332
333    def get_dataframe(self, scan_index=0, time_index=-1):
334        """
335        Get a pandas DataFrame representing the mass spectrum.
336
337        Parameters
338        ----------
339        scan_index : int, optional
340            The index of the scan to retrieve the DataFrame from. Default is 0.
341        time_index : int, optional
342            The index of the time point to retrieve the DataFrame from. Default is -1.
343
344        Returns
345        -------
346        DataFrame
347            The pandas DataFrame representing the mass spectrum.
348        """
349
350        columnsLabels = self.get_scan_group_attr_data(
351            scan_index, time_index, "ColumnsLabels"
352        )
353
354        scan_label = self.scans[scan_index]
355
356        index_to_pull = self.get_time_index_to_pull(scan_label, time_index)
357
358        corems_table_data = self.h5pydata[scan_label][index_to_pull]
359
360        list_dict = []
361        for row in corems_table_data:
362            data_dict = {}
363            for data_index, data in enumerate(row):
364                label = columnsLabels[data_index]
365                # if data starts with a b' it is a byte string, so decode it
366                if isinstance(data, bytes):
367                    data = data.decode("utf-8")
368                if data == "nan":
369                    data = None
370                data_dict[label] = data
371
372            list_dict.append(data_dict)
373
374        # Reorder the columns from low to high "Index" to match the order of the dataframe
375        df = DataFrame(list_dict)
376        # set the "Index" column to int so it sorts correctly
377        df["Index"] = df["Index"].astype(int)
378        df = df.sort_values(by="Index")
379        # Reset index to match the "Index" column
380        df = df.set_index("Index", drop=False)
381
382        return df
383
384    def get_time_index_to_pull(self, scan_label, time_index):
385        """
386        Get the time index to pull from the HDF5 file.
387
388        Parameters
389        ----------
390        scan_label : str
391            The label of the scan.
392        time_index : int
393            The index of the time point.
394
395        Returns
396        -------
397        str
398            The time index to pull.
399        """
400
401        # Only numeric keys are processed-data indices; "raw_ms" and
402        # "peak_profile_indices" (when present) are not.
403        time_data = sorted(
404            [(i, int(i)) for i in self.h5pydata[scan_label].keys() if i.isdigit()],
405            key=lambda m: m[1],
406        )
407
408        index_to_pull = time_data[time_index][0]
409
410        return index_to_pull
411
412    def get_high_level_attr_data(self, attr_str):
413        """
414        Get high-level attribute data from the HDF5 file.
415
416        Parameters
417        ----------
418        attr_str : str
419            The attribute string.
420
421        Returns
422        -------
423        dict
424            The attribute data.
425
426        Raises
427        ------
428        KeyError
429            If the attribute string is not found in the HDF5 file.
430        """
431
432        return self.h5pydata.attrs[attr_str]
433
434    def get_scan_group_attr_data(
435        self, scan_index, time_index, attr_group, attr_srt=None
436    ):
437        """
438        Get scan group attribute data from the HDF5 file.
439
440        Parameters
441        ----------
442        scan_index : int
443            The index of the scan.
444        time_index : int
445            The index of the time point.
446        attr_group : str
447            The attribute group.
448        attr_srt : str, optional
449            The attribute string. Default is None.
450
451        Returns
452        -------
453        dict
454            The attribute data.
455
456        Notes
457        -----
458        This method retrieves attribute data from the HDF5 file for a specific scan and time point.
459        The attribute data is stored in the specified attribute group.
460        If an attribute string is provided, only the corresponding attribute value is returned.
461        If no attribute string is provided, all attribute data in the group is returned as a dictionary.
462        """
463        # Get index of self.scans where scan_index_str is found
464        scan_label = self.scans[scan_index]
465
466        index_to_pull = self.get_time_index_to_pull(scan_label, time_index)
467
468        if attr_srt:
469            return json.loads(
470                self.h5pydata[scan_label][index_to_pull].attrs[attr_group]
471            )[attr_srt]
472
473        else:
474            data = self.h5pydata[scan_label][index_to_pull].attrs.get(attr_group)
475            if data:
476                return json.loads(data)
477            else:
478                return {}
479
480    def get_raw_data_attr_data(self, scan_index, attr_group, attr_str):
481        """
482        Get raw data attribute data from the HDF5 file.
483
484        Parameters
485        ----------
486        scan_index : int
487            The index of the scan.
488        attr_group : str
489            The attribute group.
490        attr_str : str
491            The attribute string.
492
493        Returns
494        -------
495        dict
496            The attribute data.
497
498        Raises
499        ------
500        KeyError
501            If the attribute string is not found in the attribute group.
502
503        Notes
504        -----
505        This method retrieves the attribute data associated with a specific scan, attribute group, and attribute string
506        from the HDF5 file. It returns the attribute data as a dictionary.
507
508        Example usage:
509        >>> data = get_raw_data_attr_data(0, "group1", "attribute1")
510        >>> print(data)
511        {'key1': 'value1', 'key2': 'value2'}
512        """
513        scan_label = self.scans[scan_index]
514
515        # First try to get from raw_ms dataset attributes (backward compatibility)
516        if (
517            "raw_ms" in self.h5pydata[scan_label]
518            and attr_group in self.h5pydata[scan_label]["raw_ms"].attrs
519        ):
520            try:
521                return json.loads(
522                    self.h5pydata[scan_label]["raw_ms"].attrs[attr_group]
523                )[attr_str]
524            except KeyError:
525                attr_str = attr_str.replace("baseline", "baselise")
526                return json.loads(
527                    self.h5pydata[scan_label]["raw_ms"].attrs[attr_group]
528                )[attr_str]
529
530        # If not found in raw_ms, try to get from scan group attributes (new format)
531        elif attr_group in self.h5pydata[scan_label].attrs:
532            try:
533                return json.loads(self.h5pydata[scan_label].attrs[attr_group])[attr_str]
534            except KeyError:
535                attr_str = attr_str.replace("baseline", "baselise")
536                return json.loads(self.h5pydata[scan_label].attrs[attr_group])[attr_str]
537
538        else:
539            raise KeyError(
540                f"Attribute group '{attr_group}' not found in either raw_ms dataset or scan group for scan {scan_label}"
541            )
542
543    def get_output_parameters(self, polarity, scan_index=0):
544        """
545        Get the output parameters for the mass spectrum.
546
547        Parameters
548        ----------
549        polarity : str
550            The polarity of the mass spectrum.
551        scan_index : int, optional
552            The index of the scan. Default is 0.
553
554        Returns
555        -------
556        dict
557            The output parameters.
558        """
559
560        d_params = default_parameters(self.file_location)
561        d_params["filename_path"] = self.file_location
562        d_params["polarity"] = self.get_raw_data_attr_data(
563            scan_index, "MassSpecAttrs", "polarity"
564        )
565        d_params["rt"] = self.get_raw_data_attr_data(scan_index, "MassSpecAttrs", "rt")
566
567        d_params["tic"] = self.get_raw_data_attr_data(
568            scan_index, "MassSpecAttrs", "tic"
569        )
570
571        d_params["mobility_scan"] = self.get_raw_data_attr_data(
572            scan_index, "MassSpecAttrs", "mobility_scan"
573        )
574        d_params["mobility_rt"] = self.get_raw_data_attr_data(
575            scan_index, "MassSpecAttrs", "mobility_rt"
576        )
577        d_params["Aterm"] = self.get_raw_data_attr_data(
578            scan_index, "MassSpecAttrs", "Aterm"
579        )
580        d_params["Bterm"] = self.get_raw_data_attr_data(
581            scan_index, "MassSpecAttrs", "Bterm"
582        )
583        d_params["Cterm"] = self.get_raw_data_attr_data(
584            scan_index, "MassSpecAttrs", "Cterm"
585        )
586        d_params["baseline_noise"] = self.get_raw_data_attr_data(
587            scan_index, "MassSpecAttrs", "baseline_noise"
588        )
589        d_params["baseline_noise_std"] = self.get_raw_data_attr_data(
590            scan_index, "MassSpecAttrs", "baseline_noise_std"
591        )
592
593        d_params["analyzer"] = self.get_high_level_attr_data("analyzer")
594        d_params["instrument_label"] = self.get_high_level_attr_data("instrument_label")
595        d_params["sample_name"] = self.get_high_level_attr_data("sample_name")
596
597        return d_params
class ReadCoreMSHDF_MassSpectrum(corems.mass_spectrum.input.massList.ReadCoremsMasslist): 
 15class ReadCoreMSHDF_MassSpectrum(ReadCoremsMasslist):
 16    """Class for reading mass spectrum data from a CoreMS HDF5 file.
 17
 18    Attributes
 19    ----------
 20    h5pydata : h5py.File
 21        The HDF5 file object.
 22    scans : list
 23        List of scan labels in the HDF5 file.
 24
 25    Parameters
 26    ----------
 27    file_location : str or S3Path
 28        The path to the CoreMS HDF5 file.
 29
 30    Methods
 31    -------
 32    * load_raw_data(scan_index=0). Read a scan's stored profile arrays, if present.
 33    * load_peak_profile_indices(scan_index=0, time_index=-1). Read a scan's stored per-peak profile indices for a given peak-table version, if present.
 34    * get_mass_spectrum(scan_number=0, time_index=-1, auto_process=True, load_settings=True, load_raw=True, load_molecular_formula=True, force_centroid=False). Get a mass spectrum object.
 35    * load_settings(mass_spectrum, scan_index=0, time_index=-1). Load settings into the mass spectrum object.
 36    * get_dataframe(scan_index=0, time_index=-1). Get a pandas DataFrame representing the mass spectrum.
 37    * get_time_index_to_pull(scan_label, time_index). Get the time index to pull from the HDF5 file.
 38    * get_high_level_attr_data(attr_str). Get high-level attribute data from the HDF5 file.
 39    * get_scan_group_attr_data(scan_index, time_index, attr_group, attr_srt=None). Get scan group attribute data from the HDF5 file.
 40    * get_raw_data_attr_data(scan_index, attr_group, attr_str). Get raw data attribute data from the HDF5 file.
 41    * get_output_parameters(polarity, scan_index=0). Get the output parameters for the mass spectrum.
 42    """
 43
 44    def __init__(self, file_location):
 45        super().__init__(file_location)
 46
 47        if isinstance(self.file_location, S3Path):
 48            data = BytesIO(self.file_location.open("rb").read())
 49        else:
 50            data = self.file_location
 51
 52        self.h5pydata = h5py.File(data, "r")
 53
 54        self.scans = list(self.h5pydata.keys())
 55
 56    def close(self):
 57        """Close the underlying HDF5 file handle."""
 58        if self.h5pydata and self.h5pydata.id.valid:
 59            self.h5pydata.close()
 60
 61    def __enter__(self):
 62        return self
 63
 64    def __exit__(self, *args):
 65        self.close()
 66
 67    def load_raw_data(self, scan_index=0):
 68        """
 69        Read the stored profile arrays for a scan, if present.
 70
 71        Parameters
 72        ----------
 73        scan_index : int, optional
 74            The index of the scan to load the raw data from. Default is 0.
 75
 76        Returns
 77        -------
 78        dict or None
 79            ``{"mz": ..., "abundance": ..., "mz_cal": ... or None,
 80            "is_calibrated": bool}`` (the experimental m/z, abundance,
 81            calibrated m/z if stored, and whether the spectrum is actually
 82            calibrated), or None if the scan has no ``raw_ms`` dataset (a
 83            centroid-only file).
 84
 85        Notes
 86        -----
 87        ``raw_ms`` rows are ordered per its ``row_labels`` attribute (JSON
 88        list of ``"mz"``/``"abundance"``/``"mz_cal"``). Files written before
 89        that attribute existed have no ``row_labels`` and are read back in
 90        the legacy ``[abundance, mz]`` order every earlier CoreMS version
 91        wrote, so old files keep loading exactly as they did before.
 92
 93        ``is_calibrated`` is read from the ``raw_ms`` ``is_calibrated``
 94        attribute, trusted over row presence; files written before that
 95        attribute existed fall back to whether an ``"mz_cal"`` row exists.
 96        """
 97        scan_label = self.scans[scan_index]
 98        raw_ms_group = self.h5pydata[scan_label]
 99
100        if "raw_ms" not in raw_ms_group or raw_ms_group["raw_ms"].shape is None:
101            return None
102
103        raw_ms_dataset = raw_ms_group["raw_ms"]
104        row_labels_json = raw_ms_dataset.attrs.get("row_labels")
105        row_labels = (
106            json.loads(row_labels_json) if row_labels_json else ["abundance", "mz"]
107        )
108
109        data = raw_ms_dataset[()]
110        mz_profile = data[row_labels.index("mz")]
111        abundance_profile = data[row_labels.index("abundance")]
112        mz_cal_profile = (
113            data[row_labels.index("mz_cal")] if "mz_cal" in row_labels else None
114        )
115        is_calibrated = bool(
116            raw_ms_dataset.attrs.get("is_calibrated", "mz_cal" in row_labels)
117        )
118
119        return {
120            "mz": mz_profile,
121            "abundance": abundance_profile,
122            "mz_cal": mz_cal_profile,
123            "is_calibrated": is_calibrated,
124        }
125
126    def load_peak_profile_indices(self, scan_index=0, time_index=-1):
127        """
128        Read the stored per-peak profile indices for a scan, if present.
129
130        Parameters
131        ----------
132        scan_index : int, optional
133            The index of the scan to load the indices from. Default is 0.
134        time_index : int, optional
135            Which peak-table version to load the indices for (same meaning
136            as elsewhere: resolved via ``get_time_index_to_pull``). Peak
137            tables are versioned ("0", "1", ...) as a scan is reprocessed,
138            and the indices are stored per version alongside each one, since
139            the picked peaks (unlike the profile itself) differ by version.
140            Default is -1 (the latest).
141
142        Returns
143        -------
144        numpy.ndarray or None
145            Array of shape ``(n_peaks, 3)`` of ``(left, apex, right)``
146            indices into the scan's profile arrays, in the same order as the
147            peak-table version requested, or None if not stored (file
148            predates this feature, or this version predates it).
149        """
150        scan_label = self.scans[scan_index]
151        scan_group = self.h5pydata[scan_label]
152        index_to_pull = self.get_time_index_to_pull(scan_label, time_index)
153        key = f"peak_profile_indices_{index_to_pull}"
154        if key not in scan_group:
155            return None
156        return scan_group[key][()]
157
158    def get_mass_spectrum(
159        self,
160        scan_number=0,
161        time_index=-1,
162        auto_process=True,
163        load_settings=True,
164        load_raw=True,
165        load_molecular_formula=True,
166        force_centroid=False,
167    ):
168        """
169        Instantiate a mass spectrum object from the CoreMS HDF5 file.
170
171        Parameters
172        ----------
173        scan_number : int, optional
174            The index of the scan to retrieve the mass spectrum from. Default is 0.
175        time_index : int, optional
176            The index of the time point to retrieve the mass spectrum from. Default is -1.
177        auto_process : bool, optional
178            Whether to automatically process the mass spectrum. Default is True.
179        load_settings : bool, optional
180            Whether to load the settings into the mass spectrum object. Default is True.
181        load_raw : bool, optional
182            Whether to load the raw profile data into the mass spectrum object,
183            when the scan has one. Default is True.
184        load_molecular_formula : bool, optional
185            Whether to load the molecular formula into the mass spectrum object.
186            Default is True.
187        force_centroid : bool, optional
188            If True, always return a :class:`MassSpecCentroid`, even when the
189            scan has a stored profile. Has no effect on a file with no stored
190            profile for this scan, which already returns a
191            :class:`MassSpecCentroid`. Default is False.
192
193        Returns
194        -------
195        MassSpecCentroid or MassSpecProfile
196            A :class:`MassSpecProfile` sharing the loaded peaks when the scan
197            has a stored profile, ``load_raw`` is True, and ``force_centroid``
198            is False; otherwise a :class:`MassSpecCentroid`.
199
200        Raises
201        ------
202        ValueError
203            If the CoreMS file is not valid.
204            If the mass spectrum has not been processed and load_molecular_formula is True.
205        """
206        if "mass_spectra" in self.scans[0]:
207            scan_index = self.scans.index("mass_spectra/" + str(scan_number))
208        else:
209            scan_index = self.scans.index(str(scan_number))
210        dataframe = self.get_dataframe(scan_index, time_index=time_index)
211        if dataframe["Molecular Formula"].any() and not dataframe["C"].any():
212            cols = dataframe.columns.tolist()
213            cols = cols[cols.index("Molecular Formula") + 1 :]
214            for index, row in dataframe.iterrows():
215                if row["Molecular Formula"] is not None:
216                    og_formula = row["Molecular Formula"]
217                    for col in cols:
218                        if "col" in og_formula:
219                            # get the digit after the element ("col") in the molecular formula and set it to the dataframe
220                            row[col] = int(og_formula.split(col)[1].split(" ")[0])
221
222        if not set(
223            ["H/C", "O/C", "Heteroatom Class", "Ion Type", "Is Isotopologue"]
224        ).issubset(dataframe.columns):
225            raise ValueError(
226                "%s it is not a valid CoreMS file" % str(self.file_location)
227            )
228
229        dataframe.rename(columns=self.parameters.header_translate, inplace=True)
230
231        # Cast m/z, and 'Peak Height' to float
232        dataframe["m/z"] = dataframe["m/z"].astype(float)
233        dataframe["Peak Height"] = dataframe["Peak Height"].astype(float)
234
235        polarity = dataframe["Ion Charge"].iloc[0]
236
237        output_parameters = self.get_output_parameters(polarity, scan_index=scan_index)
238
239        mass_spec_obj = MassSpecCentroid(
240            dataframe.to_dict(orient="list"), output_parameters, auto_process=False
241        )
242
243        if auto_process:
244            # Set the settings on the mass spectrum object to relative abuncance of 0 so all peaks get added
245            mass_spec_obj.settings.noise_threshold_method = "absolute_abundance"
246            mass_spec_obj.settings.noise_threshold_absolute_abundance = 0
247            mass_spec_obj.process_mass_spec()
248            # MassSpecCentroid.process_mass_spec() recomputes baseline_noise/
249            # _std from this scan's own (small) peak table unless label ==
250            # Labels.thermo_centroid, which it never is here -- that estimate
251            # is unrelated to, and much noisier than, the original spectrum's
252            # stored noise stats. Put the stored values back so a reloaded
253            # spectrum's noise stats match what was exported, same as every
254            # other MassSpecAttrs field already restored via output_parameters.
255            # Fall back to the just-recomputed value if the stored one is
256            # somehow missing, rather than erroring or overwriting with None.
257            mass_spec_obj._baseline_noise = output_parameters.get(
258                "baseline_noise", mass_spec_obj._baseline_noise
259            )
260            mass_spec_obj._baseline_noise_std = output_parameters.get(
261                "baseline_noise_std", mass_spec_obj._baseline_noise_std
262            )
263
264        if load_settings:
265            # Load settings into the mass spectrum object
266            self.load_settings(
267                mass_spec_obj, scan_index=scan_index, time_index=time_index
268            )
269
270        if load_molecular_formula:
271            if not auto_process:
272                raise ValueError(
273                    "Can only add molecular formula if the mass spectrum has been processed"
274                )
275            else:
276                # Attach formulas to the centroid's peaks before any profile
277                # conversion below, so to_profile() carries them over for free
278                # (it reuses these same peak objects, it doesn't rebuild them).
279                # Also before calibration restore: this matches the uncalibrated
280                # "m/z" column, and MSPeak.mz_exp returns mz_cal once that is set.
281                self.add_molecular_formula(mass_spec_obj, dataframe)
282
283        if auto_process:
284            # After formula attachment, before to_profile(). The fallback
285            # peak-index search needs a calibrated peak's mz_cal already set.
286            self.restore_peak_calibration(mass_spec_obj, dataframe)
287
288        if load_raw and not force_centroid:
289            raw_arrays = self.load_raw_data(scan_index=scan_index)
290            if raw_arrays is not None:
291                peak_profile_indices = self.load_peak_profile_indices(
292                    scan_index=scan_index, time_index=time_index
293                )
294                mass_spec_obj = mass_spec_obj.to_profile(
295                    raw_arrays["mz"],
296                    raw_arrays["abundance"],
297                    mz_cal_profile=raw_arrays["mz_cal"],
298                    peak_profile_indices=peak_profile_indices,
299                    is_calibrated=raw_arrays["is_calibrated"],
300                )
301
302        return mass_spec_obj
303
304    def load_settings(self, mass_spectrum, scan_index=0, time_index=-1):
305        """
306        Load settings into the mass spectrum object.
307
308        Parameters
309        ----------
310        mass_spectrum : MassSpecCentroid
311            The mass spectrum object to load the settings into.
312        scan_index : int, optional
313            The index of the scan to load the settings from. Default is 0.
314        time_index : int, optional
315            The index of the time point to load the settings from. Default is -1.
316        """
317
318        loaded_settings = {}
319        loaded_settings["MoleculaSearch"] = self.get_scan_group_attr_data(
320            scan_index, time_index, "MoleculaSearchSetting"
321        )
322        loaded_settings["MassSpecPeak"] = self.get_scan_group_attr_data(
323            scan_index, time_index, "MassSpecPeakSetting"
324        )
325        loaded_settings["MassSpectrum"] = self.get_scan_group_attr_data(
326            scan_index, time_index, "MassSpectrumSetting"
327        )
328        loaded_settings["Transient"] = self.get_scan_group_attr_data(
329            scan_index, time_index, "TransientSetting"
330        )
331
332        _set_dict_data_ms(loaded_settings, mass_spectrum)
333
334    def get_dataframe(self, scan_index=0, time_index=-1):
335        """
336        Get a pandas DataFrame representing the mass spectrum.
337
338        Parameters
339        ----------
340        scan_index : int, optional
341            The index of the scan to retrieve the DataFrame from. Default is 0.
342        time_index : int, optional
343            The index of the time point to retrieve the DataFrame from. Default is -1.
344
345        Returns
346        -------
347        DataFrame
348            The pandas DataFrame representing the mass spectrum.
349        """
350
351        columnsLabels = self.get_scan_group_attr_data(
352            scan_index, time_index, "ColumnsLabels"
353        )
354
355        scan_label = self.scans[scan_index]
356
357        index_to_pull = self.get_time_index_to_pull(scan_label, time_index)
358
359        corems_table_data = self.h5pydata[scan_label][index_to_pull]
360
361        list_dict = []
362        for row in corems_table_data:
363            data_dict = {}
364            for data_index, data in enumerate(row):
365                label = columnsLabels[data_index]
366                # if data starts with a b' it is a byte string, so decode it
367                if isinstance(data, bytes):
368                    data = data.decode("utf-8")
369                if data == "nan":
370                    data = None
371                data_dict[label] = data
372
373            list_dict.append(data_dict)
374
375        # Reorder the columns from low to high "Index" to match the order of the dataframe
376        df = DataFrame(list_dict)
377        # set the "Index" column to int so it sorts correctly
378        df["Index"] = df["Index"].astype(int)
379        df = df.sort_values(by="Index")
380        # Reset index to match the "Index" column
381        df = df.set_index("Index", drop=False)
382
383        return df
384
385    def get_time_index_to_pull(self, scan_label, time_index):
386        """
387        Get the time index to pull from the HDF5 file.
388
389        Parameters
390        ----------
391        scan_label : str
392            The label of the scan.
393        time_index : int
394            The index of the time point.
395
396        Returns
397        -------
398        str
399            The time index to pull.
400        """
401
402        # Only numeric keys are processed-data indices; "raw_ms" and
403        # "peak_profile_indices" (when present) are not.
404        time_data = sorted(
405            [(i, int(i)) for i in self.h5pydata[scan_label].keys() if i.isdigit()],
406            key=lambda m: m[1],
407        )
408
409        index_to_pull = time_data[time_index][0]
410
411        return index_to_pull
412
413    def get_high_level_attr_data(self, attr_str):
414        """
415        Get high-level attribute data from the HDF5 file.
416
417        Parameters
418        ----------
419        attr_str : str
420            The attribute string.
421
422        Returns
423        -------
424        dict
425            The attribute data.
426
427        Raises
428        ------
429        KeyError
430            If the attribute string is not found in the HDF5 file.
431        """
432
433        return self.h5pydata.attrs[attr_str]
434
435    def get_scan_group_attr_data(
436        self, scan_index, time_index, attr_group, attr_srt=None
437    ):
438        """
439        Get scan group attribute data from the HDF5 file.
440
441        Parameters
442        ----------
443        scan_index : int
444            The index of the scan.
445        time_index : int
446            The index of the time point.
447        attr_group : str
448            The attribute group.
449        attr_srt : str, optional
450            The attribute string. Default is None.
451
452        Returns
453        -------
454        dict
455            The attribute data.
456
457        Notes
458        -----
459        This method retrieves attribute data from the HDF5 file for a specific scan and time point.
460        The attribute data is stored in the specified attribute group.
461        If an attribute string is provided, only the corresponding attribute value is returned.
462        If no attribute string is provided, all attribute data in the group is returned as a dictionary.
463        """
464        # Get index of self.scans where scan_index_str is found
465        scan_label = self.scans[scan_index]
466
467        index_to_pull = self.get_time_index_to_pull(scan_label, time_index)
468
469        if attr_srt:
470            return json.loads(
471                self.h5pydata[scan_label][index_to_pull].attrs[attr_group]
472            )[attr_srt]
473
474        else:
475            data = self.h5pydata[scan_label][index_to_pull].attrs.get(attr_group)
476            if data:
477                return json.loads(data)
478            else:
479                return {}
480
481    def get_raw_data_attr_data(self, scan_index, attr_group, attr_str):
482        """
483        Get raw data attribute data from the HDF5 file.
484
485        Parameters
486        ----------
487        scan_index : int
488            The index of the scan.
489        attr_group : str
490            The attribute group.
491        attr_str : str
492            The attribute string.
493
494        Returns
495        -------
496        dict
497            The attribute data.
498
499        Raises
500        ------
501        KeyError
502            If the attribute string is not found in the attribute group.
503
504        Notes
505        -----
506        This method retrieves the attribute data associated with a specific scan, attribute group, and attribute string
507        from the HDF5 file. It returns the attribute data as a dictionary.
508
509        Example usage:
510        >>> data = get_raw_data_attr_data(0, "group1", "attribute1")
511        >>> print(data)
512        {'key1': 'value1', 'key2': 'value2'}
513        """
514        scan_label = self.scans[scan_index]
515
516        # First try to get from raw_ms dataset attributes (backward compatibility)
517        if (
518            "raw_ms" in self.h5pydata[scan_label]
519            and attr_group in self.h5pydata[scan_label]["raw_ms"].attrs
520        ):
521            try:
522                return json.loads(
523                    self.h5pydata[scan_label]["raw_ms"].attrs[attr_group]
524                )[attr_str]
525            except KeyError:
526                attr_str = attr_str.replace("baseline", "baselise")
527                return json.loads(
528                    self.h5pydata[scan_label]["raw_ms"].attrs[attr_group]
529                )[attr_str]
530
531        # If not found in raw_ms, try to get from scan group attributes (new format)
532        elif attr_group in self.h5pydata[scan_label].attrs:
533            try:
534                return json.loads(self.h5pydata[scan_label].attrs[attr_group])[attr_str]
535            except KeyError:
536                attr_str = attr_str.replace("baseline", "baselise")
537                return json.loads(self.h5pydata[scan_label].attrs[attr_group])[attr_str]
538
539        else:
540            raise KeyError(
541                f"Attribute group '{attr_group}' not found in either raw_ms dataset or scan group for scan {scan_label}"
542            )
543
544    def get_output_parameters(self, polarity, scan_index=0):
545        """
546        Get the output parameters for the mass spectrum.
547
548        Parameters
549        ----------
550        polarity : str
551            The polarity of the mass spectrum.
552        scan_index : int, optional
553            The index of the scan. Default is 0.
554
555        Returns
556        -------
557        dict
558            The output parameters.
559        """
560
561        d_params = default_parameters(self.file_location)
562        d_params["filename_path"] = self.file_location
563        d_params["polarity"] = self.get_raw_data_attr_data(
564            scan_index, "MassSpecAttrs", "polarity"
565        )
566        d_params["rt"] = self.get_raw_data_attr_data(scan_index, "MassSpecAttrs", "rt")
567
568        d_params["tic"] = self.get_raw_data_attr_data(
569            scan_index, "MassSpecAttrs", "tic"
570        )
571
572        d_params["mobility_scan"] = self.get_raw_data_attr_data(
573            scan_index, "MassSpecAttrs", "mobility_scan"
574        )
575        d_params["mobility_rt"] = self.get_raw_data_attr_data(
576            scan_index, "MassSpecAttrs", "mobility_rt"
577        )
578        d_params["Aterm"] = self.get_raw_data_attr_data(
579            scan_index, "MassSpecAttrs", "Aterm"
580        )
581        d_params["Bterm"] = self.get_raw_data_attr_data(
582            scan_index, "MassSpecAttrs", "Bterm"
583        )
584        d_params["Cterm"] = self.get_raw_data_attr_data(
585            scan_index, "MassSpecAttrs", "Cterm"
586        )
587        d_params["baseline_noise"] = self.get_raw_data_attr_data(
588            scan_index, "MassSpecAttrs", "baseline_noise"
589        )
590        d_params["baseline_noise_std"] = self.get_raw_data_attr_data(
591            scan_index, "MassSpecAttrs", "baseline_noise_std"
592        )
593
594        d_params["analyzer"] = self.get_high_level_attr_data("analyzer")
595        d_params["instrument_label"] = self.get_high_level_attr_data("instrument_label")
596        d_params["sample_name"] = self.get_high_level_attr_data("sample_name")
597
598        return d_params

Class for reading mass spectrum data from a CoreMS HDF5 file.

Attributes
  • h5pydata (h5py.File): The HDF5 file object.
  • scans (list): List of scan labels in the HDF5 file.
Parameters
  • file_location (str or S3Path): The path to the CoreMS HDF5 file.
Methods
  • load_raw_data(scan_index=0). Read a scan's stored profile arrays, if present.
  • load_peak_profile_indices(scan_index=0, time_index=-1). Read a scan's stored per-peak profile indices for a given peak-table version, if present.
  • get_mass_spectrum(scan_number=0, time_index=-1, auto_process=True, load_settings=True, load_raw=True, load_molecular_formula=True, force_centroid=False). Get a mass spectrum object.
  • load_settings(mass_spectrum, scan_index=0, time_index=-1). Load settings into the mass spectrum object.
  • get_dataframe(scan_index=0, time_index=-1). Get a pandas DataFrame representing the mass spectrum.
  • get_time_index_to_pull(scan_label, time_index). Get the time index to pull from the HDF5 file.
  • get_high_level_attr_data(attr_str). Get high-level attribute data from the HDF5 file.
  • get_scan_group_attr_data(scan_index, time_index, attr_group, attr_srt=None). Get scan group attribute data from the HDF5 file.
  • get_raw_data_attr_data(scan_index, attr_group, attr_str). Get raw data attribute data from the HDF5 file.
  • get_output_parameters(polarity, scan_index=0). Get the output parameters for the mass spectrum.
ReadCoreMSHDF_MassSpectrum(file_location) 
44    def __init__(self, file_location):
45        super().__init__(file_location)
46
47        if isinstance(self.file_location, S3Path):
48            data = BytesIO(self.file_location.open("rb").read())
49        else:
50            data = self.file_location
51
52        self.h5pydata = h5py.File(data, "r")
53
54        self.scans = list(self.h5pydata.keys())
h5pydata
scans
def close(self): 
56    def close(self):
57        """Close the underlying HDF5 file handle."""
58        if self.h5pydata and self.h5pydata.id.valid:
59            self.h5pydata.close()

Close the underlying HDF5 file handle.

def load_raw_data(self, scan_index=0): 
 67    def load_raw_data(self, scan_index=0):
 68        """
 69        Read the stored profile arrays for a scan, if present.
 70
 71        Parameters
 72        ----------
 73        scan_index : int, optional
 74            The index of the scan to load the raw data from. Default is 0.
 75
 76        Returns
 77        -------
 78        dict or None
 79            ``{"mz": ..., "abundance": ..., "mz_cal": ... or None,
 80            "is_calibrated": bool}`` (the experimental m/z, abundance,
 81            calibrated m/z if stored, and whether the spectrum is actually
 82            calibrated), or None if the scan has no ``raw_ms`` dataset (a
 83            centroid-only file).
 84
 85        Notes
 86        -----
 87        ``raw_ms`` rows are ordered per its ``row_labels`` attribute (JSON
 88        list of ``"mz"``/``"abundance"``/``"mz_cal"``). Files written before
 89        that attribute existed have no ``row_labels`` and are read back in
 90        the legacy ``[abundance, mz]`` order every earlier CoreMS version
 91        wrote, so old files keep loading exactly as they did before.
 92
 93        ``is_calibrated`` is read from the ``raw_ms`` ``is_calibrated``
 94        attribute, trusted over row presence; files written before that
 95        attribute existed fall back to whether an ``"mz_cal"`` row exists.
 96        """
 97        scan_label = self.scans[scan_index]
 98        raw_ms_group = self.h5pydata[scan_label]
 99
100        if "raw_ms" not in raw_ms_group or raw_ms_group["raw_ms"].shape is None:
101            return None
102
103        raw_ms_dataset = raw_ms_group["raw_ms"]
104        row_labels_json = raw_ms_dataset.attrs.get("row_labels")
105        row_labels = (
106            json.loads(row_labels_json) if row_labels_json else ["abundance", "mz"]
107        )
108
109        data = raw_ms_dataset[()]
110        mz_profile = data[row_labels.index("mz")]
111        abundance_profile = data[row_labels.index("abundance")]
112        mz_cal_profile = (
113            data[row_labels.index("mz_cal")] if "mz_cal" in row_labels else None
114        )
115        is_calibrated = bool(
116            raw_ms_dataset.attrs.get("is_calibrated", "mz_cal" in row_labels)
117        )
118
119        return {
120            "mz": mz_profile,
121            "abundance": abundance_profile,
122            "mz_cal": mz_cal_profile,
123            "is_calibrated": is_calibrated,
124        }

Read the stored profile arrays for a scan, if present.

Parameters
  • scan_index (int, optional): The index of the scan to load the raw data from. Default is 0.
Returns
  • dict or None: {"mz": ..., "abundance": ..., "mz_cal": ... or None, "is_calibrated": bool} (the experimental m/z, abundance, calibrated m/z if stored, and whether the spectrum is actually calibrated), or None if the scan has no raw_ms dataset (a centroid-only file).
Notes

raw_ms rows are ordered per its row_labels attribute (JSON list of "mz"/"abundance"/"mz_cal"). Files written before that attribute existed have no row_labels and are read back in the legacy [abundance, mz] order every earlier CoreMS version wrote, so old files keep loading exactly as they did before.

is_calibrated is read from the raw_ms is_calibrated attribute, trusted over row presence; files written before that attribute existed fall back to whether an "mz_cal" row exists.

def load_peak_profile_indices(self, scan_index=0, time_index=-1): 
126    def load_peak_profile_indices(self, scan_index=0, time_index=-1):
127        """
128        Read the stored per-peak profile indices for a scan, if present.
129
130        Parameters
131        ----------
132        scan_index : int, optional
133            The index of the scan to load the indices from. Default is 0.
134        time_index : int, optional
135            Which peak-table version to load the indices for (same meaning
136            as elsewhere: resolved via ``get_time_index_to_pull``). Peak
137            tables are versioned ("0", "1", ...) as a scan is reprocessed,
138            and the indices are stored per version alongside each one, since
139            the picked peaks (unlike the profile itself) differ by version.
140            Default is -1 (the latest).
141
142        Returns
143        -------
144        numpy.ndarray or None
145            Array of shape ``(n_peaks, 3)`` of ``(left, apex, right)``
146            indices into the scan's profile arrays, in the same order as the
147            peak-table version requested, or None if not stored (file
148            predates this feature, or this version predates it).
149        """
150        scan_label = self.scans[scan_index]
151        scan_group = self.h5pydata[scan_label]
152        index_to_pull = self.get_time_index_to_pull(scan_label, time_index)
153        key = f"peak_profile_indices_{index_to_pull}"
154        if key not in scan_group:
155            return None
156        return scan_group[key][()]

Read the stored per-peak profile indices for a scan, if present.

Parameters
  • scan_index (int, optional): The index of the scan to load the indices from. Default is 0.
  • time_index (int, optional): Which peak-table version to load the indices for (same meaning as elsewhere: resolved via get_time_index_to_pull). Peak tables are versioned ("0", "1", ...) as a scan is reprocessed, and the indices are stored per version alongside each one, since the picked peaks (unlike the profile itself) differ by version. Default is -1 (the latest).
Returns
  • numpy.ndarray or None: Array of shape (n_peaks, 3) of (left, apex, right) indices into the scan's profile arrays, in the same order as the peak-table version requested, or None if not stored (file predates this feature, or this version predates it).
def get_mass_spectrum( self, scan_number=0, time_index=-1, auto_process=True, load_settings=True, load_raw=True, load_molecular_formula=True, force_centroid=False): 
158    def get_mass_spectrum(
159        self,
160        scan_number=0,
161        time_index=-1,
162        auto_process=True,
163        load_settings=True,
164        load_raw=True,
165        load_molecular_formula=True,
166        force_centroid=False,
167    ):
168        """
169        Instantiate a mass spectrum object from the CoreMS HDF5 file.
170
171        Parameters
172        ----------
173        scan_number : int, optional
174            The index of the scan to retrieve the mass spectrum from. Default is 0.
175        time_index : int, optional
176            The index of the time point to retrieve the mass spectrum from. Default is -1.
177        auto_process : bool, optional
178            Whether to automatically process the mass spectrum. Default is True.
179        load_settings : bool, optional
180            Whether to load the settings into the mass spectrum object. Default is True.
181        load_raw : bool, optional
182            Whether to load the raw profile data into the mass spectrum object,
183            when the scan has one. Default is True.
184        load_molecular_formula : bool, optional
185            Whether to load the molecular formula into the mass spectrum object.
186            Default is True.
187        force_centroid : bool, optional
188            If True, always return a :class:`MassSpecCentroid`, even when the
189            scan has a stored profile. Has no effect on a file with no stored
190            profile for this scan, which already returns a
191            :class:`MassSpecCentroid`. Default is False.
192
193        Returns
194        -------
195        MassSpecCentroid or MassSpecProfile
196            A :class:`MassSpecProfile` sharing the loaded peaks when the scan
197            has a stored profile, ``load_raw`` is True, and ``force_centroid``
198            is False; otherwise a :class:`MassSpecCentroid`.
199
200        Raises
201        ------
202        ValueError
203            If the CoreMS file is not valid.
204            If the mass spectrum has not been processed and load_molecular_formula is True.
205        """
206        if "mass_spectra" in self.scans[0]:
207            scan_index = self.scans.index("mass_spectra/" + str(scan_number))
208        else:
209            scan_index = self.scans.index(str(scan_number))
210        dataframe = self.get_dataframe(scan_index, time_index=time_index)
211        if dataframe["Molecular Formula"].any() and not dataframe["C"].any():
212            cols = dataframe.columns.tolist()
213            cols = cols[cols.index("Molecular Formula") + 1 :]
214            for index, row in dataframe.iterrows():
215                if row["Molecular Formula"] is not None:
216                    og_formula = row["Molecular Formula"]
217                    for col in cols:
218                        if "col" in og_formula:
219                            # get the digit after the element ("col") in the molecular formula and set it to the dataframe
220                            row[col] = int(og_formula.split(col)[1].split(" ")[0])
221
222        if not set(
223            ["H/C", "O/C", "Heteroatom Class", "Ion Type", "Is Isotopologue"]
224        ).issubset(dataframe.columns):
225            raise ValueError(
226                "%s it is not a valid CoreMS file" % str(self.file_location)
227            )
228
229        dataframe.rename(columns=self.parameters.header_translate, inplace=True)
230
231        # Cast m/z, and 'Peak Height' to float
232        dataframe["m/z"] = dataframe["m/z"].astype(float)
233        dataframe["Peak Height"] = dataframe["Peak Height"].astype(float)
234
235        polarity = dataframe["Ion Charge"].iloc[0]
236
237        output_parameters = self.get_output_parameters(polarity, scan_index=scan_index)
238
239        mass_spec_obj = MassSpecCentroid(
240            dataframe.to_dict(orient="list"), output_parameters, auto_process=False
241        )
242
243        if auto_process:
244            # Set the settings on the mass spectrum object to relative abuncance of 0 so all peaks get added
245            mass_spec_obj.settings.noise_threshold_method = "absolute_abundance"
246            mass_spec_obj.settings.noise_threshold_absolute_abundance = 0
247            mass_spec_obj.process_mass_spec()
248            # MassSpecCentroid.process_mass_spec() recomputes baseline_noise/
249            # _std from this scan's own (small) peak table unless label ==
250            # Labels.thermo_centroid, which it never is here -- that estimate
251            # is unrelated to, and much noisier than, the original spectrum's
252            # stored noise stats. Put the stored values back so a reloaded
253            # spectrum's noise stats match what was exported, same as every
254            # other MassSpecAttrs field already restored via output_parameters.
255            # Fall back to the just-recomputed value if the stored one is
256            # somehow missing, rather than erroring or overwriting with None.
257            mass_spec_obj._baseline_noise = output_parameters.get(
258                "baseline_noise", mass_spec_obj._baseline_noise
259            )
260            mass_spec_obj._baseline_noise_std = output_parameters.get(
261                "baseline_noise_std", mass_spec_obj._baseline_noise_std
262            )
263
264        if load_settings:
265            # Load settings into the mass spectrum object
266            self.load_settings(
267                mass_spec_obj, scan_index=scan_index, time_index=time_index
268            )
269
270        if load_molecular_formula:
271            if not auto_process:
272                raise ValueError(
273                    "Can only add molecular formula if the mass spectrum has been processed"
274                )
275            else:
276                # Attach formulas to the centroid's peaks before any profile
277                # conversion below, so to_profile() carries them over for free
278                # (it reuses these same peak objects, it doesn't rebuild them).
279                # Also before calibration restore: this matches the uncalibrated
280                # "m/z" column, and MSPeak.mz_exp returns mz_cal once that is set.
281                self.add_molecular_formula(mass_spec_obj, dataframe)
282
283        if auto_process:
284            # After formula attachment, before to_profile(). The fallback
285            # peak-index search needs a calibrated peak's mz_cal already set.
286            self.restore_peak_calibration(mass_spec_obj, dataframe)
287
288        if load_raw and not force_centroid:
289            raw_arrays = self.load_raw_data(scan_index=scan_index)
290            if raw_arrays is not None:
291                peak_profile_indices = self.load_peak_profile_indices(
292                    scan_index=scan_index, time_index=time_index
293                )
294                mass_spec_obj = mass_spec_obj.to_profile(
295                    raw_arrays["mz"],
296                    raw_arrays["abundance"],
297                    mz_cal_profile=raw_arrays["mz_cal"],
298                    peak_profile_indices=peak_profile_indices,
299                    is_calibrated=raw_arrays["is_calibrated"],
300                )
301
302        return mass_spec_obj

Instantiate a mass spectrum object from the CoreMS HDF5 file.

Parameters
  • scan_number (int, optional): The index of the scan to retrieve the mass spectrum from. Default is 0.
  • time_index (int, optional): The index of the time point to retrieve the mass spectrum from. Default is -1.
  • auto_process (bool, optional): Whether to automatically process the mass spectrum. Default is True.
  • load_settings (bool, optional): Whether to load the settings into the mass spectrum object. Default is True.
  • load_raw (bool, optional): Whether to load the raw profile data into the mass spectrum object, when the scan has one. Default is True.
  • load_molecular_formula (bool, optional): Whether to load the molecular formula into the mass spectrum object. Default is True.
  • force_centroid (bool, optional): If True, always return a MassSpecCentroid, even when the scan has a stored profile. Has no effect on a file with no stored profile for this scan, which already returns a MassSpecCentroid. Default is False.
Returns
  • MassSpecCentroid or MassSpecProfile: A MassSpecProfile sharing the loaded peaks when the scan has a stored profile, load_raw is True, and force_centroid is False; otherwise a MassSpecCentroid.
Raises
  • ValueError: If the CoreMS file is not valid. If the mass spectrum has not been processed and load_molecular_formula is True.
def load_settings(self, mass_spectrum, scan_index=0, time_index=-1): 
304    def load_settings(self, mass_spectrum, scan_index=0, time_index=-1):
305        """
306        Load settings into the mass spectrum object.
307
308        Parameters
309        ----------
310        mass_spectrum : MassSpecCentroid
311            The mass spectrum object to load the settings into.
312        scan_index : int, optional
313            The index of the scan to load the settings from. Default is 0.
314        time_index : int, optional
315            The index of the time point to load the settings from. Default is -1.
316        """
317
318        loaded_settings = {}
319        loaded_settings["MoleculaSearch"] = self.get_scan_group_attr_data(
320            scan_index, time_index, "MoleculaSearchSetting"
321        )
322        loaded_settings["MassSpecPeak"] = self.get_scan_group_attr_data(
323            scan_index, time_index, "MassSpecPeakSetting"
324        )
325        loaded_settings["MassSpectrum"] = self.get_scan_group_attr_data(
326            scan_index, time_index, "MassSpectrumSetting"
327        )
328        loaded_settings["Transient"] = self.get_scan_group_attr_data(
329            scan_index, time_index, "TransientSetting"
330        )
331
332        _set_dict_data_ms(loaded_settings, mass_spectrum)

Load settings into the mass spectrum object.

Parameters
  • mass_spectrum (MassSpecCentroid): The mass spectrum object to load the settings into.
  • scan_index (int, optional): The index of the scan to load the settings from. Default is 0.
  • time_index (int, optional): The index of the time point to load the settings from. Default is -1.
def get_dataframe(self, scan_index=0, time_index=-1): 
334    def get_dataframe(self, scan_index=0, time_index=-1):
335        """
336        Get a pandas DataFrame representing the mass spectrum.
337
338        Parameters
339        ----------
340        scan_index : int, optional
341            The index of the scan to retrieve the DataFrame from. Default is 0.
342        time_index : int, optional
343            The index of the time point to retrieve the DataFrame from. Default is -1.
344
345        Returns
346        -------
347        DataFrame
348            The pandas DataFrame representing the mass spectrum.
349        """
350
351        columnsLabels = self.get_scan_group_attr_data(
352            scan_index, time_index, "ColumnsLabels"
353        )
354
355        scan_label = self.scans[scan_index]
356
357        index_to_pull = self.get_time_index_to_pull(scan_label, time_index)
358
359        corems_table_data = self.h5pydata[scan_label][index_to_pull]
360
361        list_dict = []
362        for row in corems_table_data:
363            data_dict = {}
364            for data_index, data in enumerate(row):
365                label = columnsLabels[data_index]
366                # if data starts with a b' it is a byte string, so decode it
367                if isinstance(data, bytes):
368                    data = data.decode("utf-8")
369                if data == "nan":
370                    data = None
371                data_dict[label] = data
372
373            list_dict.append(data_dict)
374
375        # Reorder the columns from low to high "Index" to match the order of the dataframe
376        df = DataFrame(list_dict)
377        # set the "Index" column to int so it sorts correctly
378        df["Index"] = df["Index"].astype(int)
379        df = df.sort_values(by="Index")
380        # Reset index to match the "Index" column
381        df = df.set_index("Index", drop=False)
382
383        return df

Get a pandas DataFrame representing the mass spectrum.

Parameters
  • scan_index (int, optional): The index of the scan to retrieve the DataFrame from. Default is 0.
  • time_index (int, optional): The index of the time point to retrieve the DataFrame from. Default is -1.
Returns
  • DataFrame: The pandas DataFrame representing the mass spectrum.
def get_time_index_to_pull(self, scan_label, time_index): 
385    def get_time_index_to_pull(self, scan_label, time_index):
386        """
387        Get the time index to pull from the HDF5 file.
388
389        Parameters
390        ----------
391        scan_label : str
392            The label of the scan.
393        time_index : int
394            The index of the time point.
395
396        Returns
397        -------
398        str
399            The time index to pull.
400        """
401
402        # Only numeric keys are processed-data indices; "raw_ms" and
403        # "peak_profile_indices" (when present) are not.
404        time_data = sorted(
405            [(i, int(i)) for i in self.h5pydata[scan_label].keys() if i.isdigit()],
406            key=lambda m: m[1],
407        )
408
409        index_to_pull = time_data[time_index][0]
410
411        return index_to_pull

Get the time index to pull from the HDF5 file.

Parameters
  • scan_label (str): The label of the scan.
  • time_index (int): The index of the time point.
Returns
  • str: The time index to pull.
def get_high_level_attr_data(self, attr_str): 
413    def get_high_level_attr_data(self, attr_str):
414        """
415        Get high-level attribute data from the HDF5 file.
416
417        Parameters
418        ----------
419        attr_str : str
420            The attribute string.
421
422        Returns
423        -------
424        dict
425            The attribute data.
426
427        Raises
428        ------
429        KeyError
430            If the attribute string is not found in the HDF5 file.
431        """
432
433        return self.h5pydata.attrs[attr_str]

Get high-level attribute data from the HDF5 file.

Parameters
  • attr_str (str): The attribute string.
Returns
  • dict: The attribute data.
Raises
  • KeyError: If the attribute string is not found in the HDF5 file.
def get_scan_group_attr_data(self, scan_index, time_index, attr_group, attr_srt=None): 
435    def get_scan_group_attr_data(
436        self, scan_index, time_index, attr_group, attr_srt=None
437    ):
438        """
439        Get scan group attribute data from the HDF5 file.
440
441        Parameters
442        ----------
443        scan_index : int
444            The index of the scan.
445        time_index : int
446            The index of the time point.
447        attr_group : str
448            The attribute group.
449        attr_srt : str, optional
450            The attribute string. Default is None.
451
452        Returns
453        -------
454        dict
455            The attribute data.
456
457        Notes
458        -----
459        This method retrieves attribute data from the HDF5 file for a specific scan and time point.
460        The attribute data is stored in the specified attribute group.
461        If an attribute string is provided, only the corresponding attribute value is returned.
462        If no attribute string is provided, all attribute data in the group is returned as a dictionary.
463        """
464        # Get index of self.scans where scan_index_str is found
465        scan_label = self.scans[scan_index]
466
467        index_to_pull = self.get_time_index_to_pull(scan_label, time_index)
468
469        if attr_srt:
470            return json.loads(
471                self.h5pydata[scan_label][index_to_pull].attrs[attr_group]
472            )[attr_srt]
473
474        else:
475            data = self.h5pydata[scan_label][index_to_pull].attrs.get(attr_group)
476            if data:
477                return json.loads(data)
478            else:
479                return {}

Get scan group attribute data from the HDF5 file.

Parameters
  • scan_index (int): The index of the scan.
  • time_index (int): The index of the time point.
  • attr_group (str): The attribute group.
  • attr_srt (str, optional): The attribute string. Default is None.
Returns
  • dict: The attribute data.
Notes

This method retrieves attribute data from the HDF5 file for a specific scan and time point. The attribute data is stored in the specified attribute group. If an attribute string is provided, only the corresponding attribute value is returned. If no attribute string is provided, all attribute data in the group is returned as a dictionary.

def get_raw_data_attr_data(self, scan_index, attr_group, attr_str): 
481    def get_raw_data_attr_data(self, scan_index, attr_group, attr_str):
482        """
483        Get raw data attribute data from the HDF5 file.
484
485        Parameters
486        ----------
487        scan_index : int
488            The index of the scan.
489        attr_group : str
490            The attribute group.
491        attr_str : str
492            The attribute string.
493
494        Returns
495        -------
496        dict
497            The attribute data.
498
499        Raises
500        ------
501        KeyError
502            If the attribute string is not found in the attribute group.
503
504        Notes
505        -----
506        This method retrieves the attribute data associated with a specific scan, attribute group, and attribute string
507        from the HDF5 file. It returns the attribute data as a dictionary.
508
509        Example usage:
510        >>> data = get_raw_data_attr_data(0, "group1", "attribute1")
511        >>> print(data)
512        {'key1': 'value1', 'key2': 'value2'}
513        """
514        scan_label = self.scans[scan_index]
515
516        # First try to get from raw_ms dataset attributes (backward compatibility)
517        if (
518            "raw_ms" in self.h5pydata[scan_label]
519            and attr_group in self.h5pydata[scan_label]["raw_ms"].attrs
520        ):
521            try:
522                return json.loads(
523                    self.h5pydata[scan_label]["raw_ms"].attrs[attr_group]
524                )[attr_str]
525            except KeyError:
526                attr_str = attr_str.replace("baseline", "baselise")
527                return json.loads(
528                    self.h5pydata[scan_label]["raw_ms"].attrs[attr_group]
529                )[attr_str]
530
531        # If not found in raw_ms, try to get from scan group attributes (new format)
532        elif attr_group in self.h5pydata[scan_label].attrs:
533            try:
534                return json.loads(self.h5pydata[scan_label].attrs[attr_group])[attr_str]
535            except KeyError:
536                attr_str = attr_str.replace("baseline", "baselise")
537                return json.loads(self.h5pydata[scan_label].attrs[attr_group])[attr_str]
538
539        else:
540            raise KeyError(
541                f"Attribute group '{attr_group}' not found in either raw_ms dataset or scan group for scan {scan_label}"
542            )

Get raw data attribute data from the HDF5 file.

Parameters
  • scan_index (int): The index of the scan.
  • attr_group (str): The attribute group.
  • attr_str (str): The attribute string.
Returns
  • dict: The attribute data.
Raises
  • KeyError: If the attribute string is not found in the attribute group.
Notes

This method retrieves the attribute data associated with a specific scan, attribute group, and attribute string from the HDF5 file. It returns the attribute data as a dictionary.

Example usage:

>>> data = get_raw_data_attr_data(0, "group1", "attribute1")
>>> print(data)
{'key1': 'value1', 'key2': 'value2'}
def get_output_parameters(self, polarity, scan_index=0): 
544    def get_output_parameters(self, polarity, scan_index=0):
545        """
546        Get the output parameters for the mass spectrum.
547
548        Parameters
549        ----------
550        polarity : str
551            The polarity of the mass spectrum.
552        scan_index : int, optional
553            The index of the scan. Default is 0.
554
555        Returns
556        -------
557        dict
558            The output parameters.
559        """
560
561        d_params = default_parameters(self.file_location)
562        d_params["filename_path"] = self.file_location
563        d_params["polarity"] = self.get_raw_data_attr_data(
564            scan_index, "MassSpecAttrs", "polarity"
565        )
566        d_params["rt"] = self.get_raw_data_attr_data(scan_index, "MassSpecAttrs", "rt")
567
568        d_params["tic"] = self.get_raw_data_attr_data(
569            scan_index, "MassSpecAttrs", "tic"
570        )
571
572        d_params["mobility_scan"] = self.get_raw_data_attr_data(
573            scan_index, "MassSpecAttrs", "mobility_scan"
574        )
575        d_params["mobility_rt"] = self.get_raw_data_attr_data(
576            scan_index, "MassSpecAttrs", "mobility_rt"
577        )
578        d_params["Aterm"] = self.get_raw_data_attr_data(
579            scan_index, "MassSpecAttrs", "Aterm"
580        )
581        d_params["Bterm"] = self.get_raw_data_attr_data(
582            scan_index, "MassSpecAttrs", "Bterm"
583        )
584        d_params["Cterm"] = self.get_raw_data_attr_data(
585            scan_index, "MassSpecAttrs", "Cterm"
586        )
587        d_params["baseline_noise"] = self.get_raw_data_attr_data(
588            scan_index, "MassSpecAttrs", "baseline_noise"
589        )
590        d_params["baseline_noise_std"] = self.get_raw_data_attr_data(
591            scan_index, "MassSpecAttrs", "baseline_noise_std"
592        )
593
594        d_params["analyzer"] = self.get_high_level_attr_data("analyzer")
595        d_params["instrument_label"] = self.get_high_level_attr_data("instrument_label")
596        d_params["sample_name"] = self.get_high_level_attr_data("sample_name")
597
598        return d_params

Get the output parameters for the mass spectrum.

Parameters
  • polarity (str): The polarity of the mass spectrum.
  • scan_index (int, optional): The index of the scan. Default is 0.
Returns
  • dict: The output parameters.
Inherited Members
corems.mass_spectrum.input.massList.ReadCoremsMasslist
restore_peak_calibration
add_molecular_formula
corems.mass_spectrum.input.baseClass.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
clean_data_frame
check_columns
read_xml_peaks
get_xml_polarity