corems.mass_spectra.input.mzml

  1from collections import defaultdict
  2from pathlib import Path
  3from typing import Optional, Union, List, Tuple
  4
  5import numpy as np
  6import pandas as pd
  7import pymzml
  8import datetime
  9
 10from corems.encapsulation.constant import Labels
 11from corems.encapsulation.factory.parameters import default_parameters
 12from corems.mass_spectra.factory.lc_class import LCMSBase, MassSpectraBase
 13from corems.mass_spectra.input.parserbase import SpectraParserInterface
 14from corems.mass_spectrum.factory.MassSpectrumClasses import (
 15    MassSpecCentroid,
 16    MassSpecProfile,
 17)
 18
 19
 20class MZMLSpectraParser(SpectraParserInterface):
 21    """A class for parsing mzml spectrometry data files into MassSpectraBase or LCMSBase objects
 22
 23    Parameters
 24    ----------
 25    file_location : str or Path
 26        The path to the RAW file to be parsed.
 27    analyzer : str, optional
 28        The type of mass analyzer used in the instrument. Default is "Unknown".
 29    instrument_label : str, optional
 30        The name of the instrument used to acquire the data. Default is "Unknown".
 31    sample_name : str, optional
 32        The name of the sample being analyzed. If not provided, the stem of the file_location path will be used.
 33
 34    Attributes
 35    ----------
 36    file_location : Path
 37        The path to the RAW file being parsed.
 38    analyzer : str
 39        The type of mass analyzer used in the instrument.
 40    instrument_label : str
 41        The name of the instrument used to acquire the data.
 42    sample_name : str
 43        The name of the sample being analyzed.
 44
 45    Methods
 46    -------
 47    * load().
 48        Load mzML file using pymzml.run.Reader and return the data as a numpy array.
 49    * run(spectra=True).
 50        Parses the mzml file and returns a dictionary of mass spectra dataframes and a scan metadata dataframe.
 51    * get_mass_spectrum_from_scan(scan_number, polarity, auto_process=True)
 52        Parses the mzml file and returns a MassSpecBase object from a single scan.
 53    * get_mass_spectra_obj().
 54        Parses the mzml file and instantiates a MassSpectraBase object.
 55    * get_lcms_obj().
 56        Parses the mzml file and instantiates an LCMSBase object.
 57    * get_instrument_info().
 58        Return instrument information from the mzML file.
 59    * get_creation_time().
 60        Return the creation time of the mzML file as a datetime object.
 61
 62    Inherits from ThermoBaseClass and SpectraParserInterface
 63    """
 64
 65    def __init__(
 66        self,
 67        file_location,
 68        analyzer="Unknown",
 69        instrument_label="Unknown",
 70        sample_name=None,
 71    ):
 72        # implementation details
 73        if isinstance(file_location, str):
 74            # if obj is a string it defaults to create a Path obj, pass the S3Path if needed
 75            file_location = Path(file_location)
 76        if not file_location.exists():
 77            raise FileExistsError("File does not exist: " + str(file_location))
 78        self.file_location = file_location
 79        self.analyzer = analyzer
 80        self.instrument_label = instrument_label
 81
 82        if sample_name:
 83            self.sample_name = sample_name
 84        else:
 85            self.sample_name = file_location.stem
 86
 87    def load(self):
 88        """
 89        Load mzML file using pymzml.run.Reader and return the data as a numpy array.
 90
 91        Returns
 92        -------
 93        numpy.ndarray
 94            The mass spectra data as a numpy array.
 95        """
 96        data = pymzml.run.Reader(self.file_location)
 97        return data
 98
 99    def get_scan_df(
100        self,
101        data=None,
102        time_range: Optional[
103            Union[Tuple[float, float], List[Tuple[float, float]]]
104        ] = None,
105    ):
106        """
107        Return scan data as a pandas DataFrame.
108
109        Parameters
110        ----------
111        data : pymzml.run.Reader, optional
112            The mass spectra data. If None, will load the data.
113        time_range : tuple or list of tuples, optional
114            Retention time range(s) to filter scans. Can be:
115            - Single range: (start_time, end_time) in minutes
116            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
117            If None, returns all scans.
118
119        Returns
120        -------
121        pandas.DataFrame
122            A pandas DataFrame containing metadata for each scan, including scan number, MS level, polarity, and scan time.
123        """
124        if data is None:
125            data = self.load()
126        # Scan dict
127        # instatinate scan dict, with empty lists of size of scans
128        n_scans = data.get_spectrum_count()
129        scan_dict = {
130            "scan": np.empty(n_scans, dtype=np.int32),
131            "scan_time": np.empty(n_scans, dtype=np.float32),
132            "ms_level": [None] * n_scans,
133            "polarity": [None] * n_scans,
134            "precursor_mz": [None] * n_scans,
135            "scan_text": [None] * n_scans,
136            "scan_window_lower": np.empty(n_scans, dtype=np.float32),
137            "scan_window_upper": np.empty(n_scans, dtype=np.float32),
138            "scan_precision": [None] * n_scans,
139            "tic": np.empty(n_scans, dtype=np.float32),
140            "ms_format": [None] * n_scans,
141        }
142
143        # First pass: loop through scans to get scan info
144        for i, spec in enumerate(data):
145            scan_dict["scan"][i] = spec.ID
146            scan_dict["ms_level"][i] = spec.ms_level
147            scan_dict["scan_precision"][i] = spec._measured_precision
148            scan_dict["tic"][i] = spec.TIC
149            if spec.selected_precursors:
150                scan_dict["precursor_mz"][i] = spec.selected_precursors[0].get(
151                    "mz", None
152                )
153            if spec["negative scan"] is not None:
154                scan_dict["polarity"][i] = "negative"
155            if spec["positive scan"] is not None:
156                scan_dict["polarity"][i] = "positive"
157            if spec["negative scan"] is not None and spec["positive scan"] is not None:
158                raise ValueError(
159                    "Error: scan {0} has both negative and positive polarity".format(
160                        spec.ID
161                    )
162                )
163
164            scan_dict["scan_time"][i] = spec.get("MS:1000016")
165            scan_dict["scan_text"][i] = spec.get("MS:1000512")
166            scan_dict["scan_window_lower"][i] = spec.get("MS:1000501")
167            scan_dict["scan_window_upper"][i] = spec.get("MS:1000500")
168            if spec.get("MS:1000128"):
169                scan_dict["ms_format"][i] = "profile"
170            elif spec.get("MS:1000127"):
171                scan_dict["ms_format"][i] = "centroid"
172            else:
173                scan_dict["ms_format"][i] = None
174
175        scan_df = pd.DataFrame(scan_dict)
176
177        # Remove any non-mass spectra scans (e.g., MS level 0 or None)
178        scan_df = scan_df[
179            scan_df.ms_level.notnull() & (scan_df.ms_level > 0)
180        ].reset_index(drop=True)
181
182        # Apply time range filtering if specified
183        if time_range is not None:
184            time_ranges = self._normalize_time_range(time_range)
185            # Create a mask for scans within any of the time ranges
186            mask = np.zeros(len(scan_df), dtype=bool)
187            for start_time, end_time in time_ranges:
188                mask |= (scan_df["scan_time"] >= start_time) & (
189                    scan_df["scan_time"] <= end_time
190                )
191            scan_df = scan_df[mask].reset_index(drop=True)
192
193        return scan_df
194
195    def get_ms_raw(
196        self,
197        spectra,
198        scan_df,
199        data=None,
200        time_range: Optional[
201            Union[Tuple[float, float], List[Tuple[float, float]]]
202        ] = None,
203    ):
204        """Return a dictionary of mass spectra data as a pandas DataFrame.
205
206        Parameters
207        ----------
208        spectra : str
209            Which mass spectra data to include in the output.
210            Options: None, "ms1", "ms2", "all".
211        scan_df : pandas.DataFrame
212            Scan dataframe. Output from get_scan_df().
213        data : pymzml.run.Reader, optional
214            The mass spectra data. If None, will load the data.
215        time_range : tuple or list of tuples, optional
216            Retention time range(s) to filter scans. Can be:
217            - Single range: (start_time, end_time) in minutes
218            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
219            If None, returns all scans. Note: filtering is typically done at scan_df level.
220
221        Returns
222        -------
223        dict
224            A dictionary containing the mass spectra data as pandas DataFrames, with keys corresponding to the MS level.
225
226        """
227        if data is None:
228            data = self.load()
229        if spectra == "all":
230            scan_df_forspec = scan_df
231        elif spectra == "ms1":
232            scan_df_forspec = scan_df[scan_df.ms_level == 1]
233        elif spectra == "ms2":
234            scan_df_forspec = scan_df[scan_df.ms_level == 2]
235        else:
236            raise ValueError("spectra must be 'all', 'ms1', or 'ms2'")
237
238        # Result container
239        res = {}
240
241        # Row count container
242        counter = {}
243
244        # Column name container
245        cols = {}
246
247        # set at float32
248        dtype = np.float32
249
250        # First pass: get nrows
251        N = defaultdict(lambda: 0)
252        for i, spec in enumerate(data):
253            if spec.ID in scan_df_forspec.scan.values:
254                # Get ms level
255                level = "ms{}".format(spec.ms_level)
256
257                # Number of rows
258                N[level] += spec.mz.shape[0]
259
260        # Second pass: parse
261        for i, spec in enumerate(data):
262            if spec.ID in scan_df_forspec.scan.values:
263                # Number of rows
264                n = spec.mz.shape[0]
265
266                # No measurements
267                if n == 0:
268                    continue
269
270                # Dimension check
271                if len(spec.mz) != len(spec.i):
272                    # raise an error if the mz and intensity arrays are not the same length
273                    raise ValueError("m/z and intensity array dimension mismatch")
274
275                # Scan/frame info
276                id_dict = spec.id_dict
277
278                # Get ms level
279                level = "ms{}".format(spec.ms_level)
280
281                # Columns
282                cols[level] = list(id_dict.keys()) + ["mz", "intensity"]
283                m = len(cols[level])
284
285                # Subarray init
286                arr = np.empty((n, m), dtype=dtype)
287                inx = 0
288
289                # Populate scan/frame info
290                for k, v in id_dict.items():
291                    arr[:, inx] = v
292                    inx += 1
293
294                # Populate m/z
295                arr[:, inx] = spec.mz
296                inx += 1
297
298                # Populate intensity
299                arr[:, inx] = spec.i
300                inx += 1
301
302                # Initialize output container
303                if level not in res:
304                    res[level] = np.empty((N[level], m), dtype=dtype)
305                    counter[level] = 0
306
307                # Insert subarray
308                res[level][counter[level] : counter[level] + n, :] = arr
309                counter[level] += n
310
311        # Construct ms1 and ms2 mz dataframes
312        for level in res.keys():
313            res[level] = pd.DataFrame(res[level], columns=cols[level]).drop(
314                columns=["controllerType", "controllerNumber"],
315            )
316
317        return res
318
319    def run(
320        self,
321        spectra="all",
322        scan_df=None,
323        time_range: Optional[
324            Union[Tuple[float, float], List[Tuple[float, float]]]
325        ] = None,
326    ):
327        """Parse the mzML file and return a dictionary of spectra dataframes and a scan metadata dataframe.
328
329        Parameters
330        ----------
331        spectra : str, optional
332            Which mass spectra data to include in the output. Default is "all".
333            Other options: None, "ms1", "ms2".
334        scan_df : pandas.DataFrame, optional
335            Scan dataframe.  If not provided, the scan dataframe is created from the mzML file.
336        time_range : tuple or list of tuples, optional
337            Retention time range(s) to load. Can be:
338            - Single range: (start_time, end_time) in minutes
339            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
340            If None, loads all scans.
341
342        Returns
343        -------
344        tuple
345            A tuple containing two elements:
346            - A dictionary containing the mass spectra data as numpy arrays, with keys corresponding to the MS level.
347            - A pandas DataFrame containing metadata for each scan, including scan number, MS level, polarity, and scan time.
348        """
349
350        # Open file
351        data = self.load()
352
353        if scan_df is None:
354            scan_df = self.get_scan_df(data, time_range=time_range)
355
356        if spectra != "none":
357            res = self.get_ms_raw(spectra, scan_df, data)
358
359        else:
360            res = None
361
362        return res, scan_df
363
364    def get_mass_spectrum_from_scan(
365        self, scan_number, spectrum_mode, auto_process=True
366    ):
367        """Instatiate a mass spectrum object from the mzML file.
368
369        Parameters
370        ----------
371        scan_number : int
372            The scan number to be parsed.
373        spectrum_mode : str
374            The type of spectrum to instantiate.  Must be'profile' or 'centroid'.
375        polarity : int
376            The polarity of the scan.  Must be -1 or 1.
377        auto_process : bool, optional
378            If True, process the mass spectrum. Default is True.
379
380        Returns
381        -------
382        MassSpecProfile | MassSpecCentroid
383            The MassSpecProfile or MassSpecCentroid object containing the parsed mass spectrum.
384        """
385        # Use the batch function and return the first result
386        result_list = self.get_mass_spectra_from_scan_list(
387            [scan_number], spectrum_mode, auto_process
388        )
389        return result_list[0] if result_list else None
390
391    def get_mass_spectra_from_scan_list(
392        self, scan_list, spectrum_mode, auto_process=True
393    ):
394        """Instatiate mass spectrum objects from the mzML file.
395
396        Parameters
397        ----------
398        scan_list : list of int
399            The scan numbers to be parsed.
400        spectrum_mode : str
401            The type of spectrum to instantiate.  Must be'profile' or 'centroid'.
402        auto_process : bool, optional
403            If True, process the mass spectrum. Default is True.
404
405        Returns
406        -------
407        list of MassSpecProfile | MassSpecCentroid
408            List of MassSpecProfile or MassSpecCentroid objects containing the parsed mass spectra.
409        """
410
411        def set_metadata(
412            scan_number: int,
413            polarity: int,
414            file_location: str,
415            label=Labels.thermo_profile,
416        ):
417            """
418            Set the output parameters for creating a MassSpecProfile or MassSpecCentroid object.
419
420            Parameters
421            ----------
422            scan_number : int
423                The scan number.
424            polarity : int
425                The polarity of the data.
426            file_location : str
427                The file location.
428            label : str, optional
429                The label for the mass spectrum. Default is Labels.thermo_profile.
430
431            Returns
432            -------
433            dict
434                The output parameters ready for creating a MassSpecProfile or MassSpecCentroid object.
435            """
436            d_params = default_parameters(file_location)
437            d_params["label"] = label
438            d_params["polarity"] = polarity
439            d_params["filename_path"] = file_location
440            d_params["scan_number"] = scan_number
441
442            return d_params
443
444        # Open file
445        data = self.load()
446
447        mass_spectrum_objects = []
448        scan_set = set(scan_list)
449        # Materialize peak data during iteration. Do not retain live pymzml
450        # spectrum objects, and do not overwrite the first valid MS spectrum for
451        # a scan ID.
452        #
453        # Thermo multi-controller mzML files often reuse scan numbers across
454        # controllers (e.g. controllerType=0 MS data and controllerType=3
455        # auxiliary traces). pymzml exposes only the integer scan number as
456        # spec.ID, so sequential iteration can yield the same ID twice. Keeping
457        # the last yield (or random-access data[scan]) can replace real MS
458        # spectra with non-MS controller spectra and then fail centroid/profile
459        # checks with "spectrum is not centroided".
460        #
461        # Direct random-access via data[scan_number] also uses pymzml's
462        # byte-offset index, which is unreliable on Windows (CRLF vs LF).
463        collected = {}
464
465        for spec in data:
466            scan_id = spec.ID
467            if scan_id not in scan_set or scan_id in collected:
468                continue
469
470            # Skip non-MS / auxiliary-controller spectra that share scan numbers
471            ms_level = spec.ms_level
472            if ms_level is None or ms_level <= 0:
473                continue
474
475            if spec["negative scan"] is not None:
476                polarity = -1
477            elif spec["positive scan"] is not None:
478                polarity = 1
479            else:
480                polarity = None
481
482            mz = np.asarray(spec.mz)
483            abundance = np.asarray(spec.i)
484            collected[scan_id] = {
485                "mz": mz,
486                "abundance": abundance,
487                "polarity": polarity,
488                "is_profile": bool(spec.get("MS:1000128")),
489                "is_centroid": bool(spec.get("MS:1000127")),
490            }
491
492        for scan_number in scan_list:
493            entry = collected.get(scan_number)
494            if entry is None:
495                raise ValueError("Scan number %d not found in mzML file" % scan_number)
496
497            polarity = entry["polarity"]
498            mz = entry["mz"]
499            abundance = entry["abundance"]
500
501            # Get mass spectrum
502            if spectrum_mode == "profile":
503                # Check if profile
504                if not entry["is_profile"]:
505                    raise ValueError("spectrum is not profile")
506                data_dict = {
507                    Labels.mz: mz,
508                    Labels.abundance: abundance,
509                }
510                d_params = set_metadata(
511                    scan_number,
512                    polarity,
513                    self.file_location,
514                    label=Labels.simulated_profile,
515                )
516                mass_spectrum_obj = MassSpecProfile(
517                    data_dict, d_params, auto_process=auto_process
518                )
519            elif spectrum_mode == "centroid":
520                # Check if centroided
521                if not entry["is_centroid"]:
522                    raise ValueError("spectrum is not centroided")
523                data_dict = {
524                    Labels.mz: mz,
525                    Labels.abundance: abundance,
526                    Labels.rp: [np.nan] * len(mz),
527                    Labels.s2n: [np.nan] * len(abundance),
528                }
529                d_params = set_metadata(
530                    scan_number,
531                    polarity,
532                    self.file_location,
533                    label=Labels.corems_centroid,
534                )
535                mass_spectrum_obj = MassSpecCentroid(
536                    data_dict, d_params, auto_process=auto_process
537                )
538            else:
539                raise ValueError(
540                    "spectrum_mode must be 'profile' or 'centroid', got %r"
541                    % spectrum_mode
542                )
543
544            mass_spectrum_objects.append(mass_spectrum_obj)
545
546        return mass_spectrum_objects
547
548    def get_mass_spectra_obj(
549        self,
550        time_range: Optional[
551            Union[Tuple[float, float], List[Tuple[float, float]]]
552        ] = None,
553    ):
554        """Instatiate a MassSpectraBase object from the mzML file.
555
556        Parameters
557        ----------
558        time_range : tuple or list of tuples, optional
559            Retention time range(s) to load. Can be:
560            - Single range: (start_time, end_time) in minutes
561            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
562            If None, loads all scans.
563
564        Returns
565        -------
566        MassSpectraBase
567            The MassSpectra object containing the parsed mass spectra.
568            The object is instatiated with the mzML file, analyzer, instrument, sample name, and scan dataframe.
569        """
570        _, scan_df = self.run(spectra=False, time_range=time_range)
571        mass_spectra_obj = MassSpectraBase(
572            self.file_location,
573            self.analyzer,
574            self.instrument_label,
575            self.sample_name,
576            self,
577        )
578        scan_df = scan_df.set_index("scan", drop=False)
579        mass_spectra_obj.scan_df = scan_df
580
581        return mass_spectra_obj
582
583    def get_lcms_obj(
584        self,
585        spectra="all",
586        time_range: Optional[
587            Union[Tuple[float, float], List[Tuple[float, float]]]
588        ] = None,
589    ):
590        """Instatiates a LCMSBase object from the mzML file.
591
592        Parameters
593        ----------
594        spectra : str, optional
595            Which mass spectra data to include in the output. Default is all.  Other options: none, ms1, ms2.
596        time_range : tuple or list of tuples, optional
597            Retention time range(s) to load. Can be:
598            - Single range: (start_time, end_time) in minutes
599            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
600            If None, loads all scans. Useful for targeted workflows to improve performance.
601
602        Returns
603        -------
604        LCMSBase
605            LCMS object containing mass spectra data.
606            The object is instatiated with the mzML file, analyzer, instrument, sample name, scan dataframe,
607            and mz dataframe(s), as well as lists of scan numbers, retention times, and TICs.
608        """
609        _, scan_df = self.run(
610            spectra="none", time_range=time_range
611        )  # first run it to just get scan info
612        if spectra != "none":
613            res, scan_df = self.run(
614                scan_df=scan_df, spectra=spectra, time_range=time_range
615            )  # second run to parse data
616        lcms_obj = LCMSBase(
617            self.file_location,
618            self.analyzer,
619            self.instrument_label,
620            self.sample_name,
621            self,
622        )
623        if spectra != "none":
624            for key in res:
625                key_int = int(key.replace("ms", ""))
626                res[key] = res[key][res[key].intensity > 0]
627                res[key] = (
628                    res[key].sort_values(by=["scan", "mz"]).reset_index(drop=True)
629                )
630                lcms_obj._ms_unprocessed[key_int] = res[key]
631        lcms_obj.scan_df = scan_df.set_index("scan", drop=False)
632        # Check if polarity is mixed
633        if len(set(scan_df.polarity)) > 1:
634            raise ValueError("Mixed polarities detected in scan data")
635        # Use iloc for positional access (Series.__getitem__ is label-based;
636        # safer across pandas 2.x and 3.x).
637        lcms_obj.polarity = scan_df.polarity.iloc[0]
638        lcms_obj._scans_number_list = list(scan_df.scan)
639        lcms_obj._retention_time_list = list(scan_df.scan_time)
640        lcms_obj._tic_list = list(scan_df.tic)
641
642        return lcms_obj
643
644    def get_scans_in_time_range(
645        self,
646        time_range: Union[Tuple[float, float], List[Tuple[float, float]]],
647        ms_level: Optional[int] = None,
648    ) -> List[int]:
649        """
650        Return scan numbers within specified retention time range(s).
651
652        This method provides efficient filtering of scans by retention time,
653        which is particularly useful for targeted workflows where only specific
654        time windows are of interest.
655
656        Parameters
657        ----------
658        time_range : tuple or list of tuples
659            Retention time range(s) in minutes. Can be:
660            - Single range: (start_time, end_time)
661            - Multiple ranges: [(start1, end1), (start2, end2), ...]
662        ms_level : int, optional
663            If specified, only return scans of this MS level (e.g., 1 for MS1, 2 for MS2).
664            If None, returns scans of all MS levels.
665
666        Returns
667        -------
668        list of int
669            List of scan numbers within the specified time range(s) and MS level.
670
671        Examples
672        --------
673        Get MS1 scans between 1.0 and 2.0 minutes:
674
675        >>> scans = parser.get_scans_in_time_range((1.0, 2.0), ms_level=1)
676
677        Get scans in multiple time windows:
678
679        >>> scans = parser.get_scans_in_time_range([(0.5, 1.5), (3.0, 4.0)])
680        """
681        # Get scan dataframe filtered by time range
682        scan_df = self.get_scan_df(time_range=time_range)
683
684        # Further filter by MS level if specified
685        if ms_level is not None:
686            scan_df = scan_df[scan_df.ms_level == ms_level]
687
688        # Return list of scan numbers
689        return scan_df.scan.tolist()
690
691    def get_instrument_info(self):
692        """
693        Return instrument information.
694
695        Returns
696        -------
697        dict
698            A dictionary with the keys 'model' and 'serial_number'.
699        """
700        # Load the pymzml data
701        data = self.load()
702        instrument_info = data.info.get("referenceable_param_group_list_element")[0]
703        cv_params = instrument_info.findall("{http://psi.hupo.org/ms/mzml}cvParam")
704
705        # Extract details from each cvParam
706        params = []
707        for param in cv_params:
708            accession = param.get("accession")  # Get 'accession' attribute
709            name = param.get("name")  # Get 'name' attribute
710            value = param.get("value")  # Get 'value' attribute
711            params.append({"accession": accession, "name": name, "value": value})
712
713        # Loop through params and try to find the relevant information
714        instrument_dict = {"model": "Unknown", "serial_number": "Unknown"}
715
716        # Assuming there are only two paramters here - one is for the serial number (agnostic to the model) and the other is for the model
717        # If there are more than two, we raise an error
718        if len(params) < 2:
719            raise ValueError(
720                "Not enough parameters found in the instrument info, cannot parse."
721            )
722        if len(params) > 2:
723            raise ValueError(
724                "Too many parameters found in the instrument info, cannot parse."
725            )
726        for param in params:
727            if param["accession"] == "MS:1000529":
728                instrument_dict["serial_number"] = param["value"]
729            else:
730                instrument_dict["model"] = data.OT[param["accession"]]
731
732        return instrument_dict
733
734    def get_creation_time(self) -> datetime.datetime:
735        """
736        Return the creation time of the mzML file.
737        """
738        data = self.load()
739        write_time = data.info.get("start_time")
740        if write_time:
741            # Convert the write time to a datetime object
742            return datetime.datetime.strptime(write_time, "%Y-%m-%dT%H:%M:%SZ")
743        else:
744            raise ValueError(
745                "Creation time is not available in the mzML file. "
746                "Please ensure the file contains the 'start_time' information."
747            )
class MZMLSpectraParser(corems.mass_spectra.input.parserbase.SpectraParserInterface):
 21class MZMLSpectraParser(SpectraParserInterface):
 22    """A class for parsing mzml spectrometry data files into MassSpectraBase or LCMSBase objects
 23
 24    Parameters
 25    ----------
 26    file_location : str or Path
 27        The path to the RAW file to be parsed.
 28    analyzer : str, optional
 29        The type of mass analyzer used in the instrument. Default is "Unknown".
 30    instrument_label : str, optional
 31        The name of the instrument used to acquire the data. Default is "Unknown".
 32    sample_name : str, optional
 33        The name of the sample being analyzed. If not provided, the stem of the file_location path will be used.
 34
 35    Attributes
 36    ----------
 37    file_location : Path
 38        The path to the RAW file being parsed.
 39    analyzer : str
 40        The type of mass analyzer used in the instrument.
 41    instrument_label : str
 42        The name of the instrument used to acquire the data.
 43    sample_name : str
 44        The name of the sample being analyzed.
 45
 46    Methods
 47    -------
 48    * load().
 49        Load mzML file using pymzml.run.Reader and return the data as a numpy array.
 50    * run(spectra=True).
 51        Parses the mzml file and returns a dictionary of mass spectra dataframes and a scan metadata dataframe.
 52    * get_mass_spectrum_from_scan(scan_number, polarity, auto_process=True)
 53        Parses the mzml file and returns a MassSpecBase object from a single scan.
 54    * get_mass_spectra_obj().
 55        Parses the mzml file and instantiates a MassSpectraBase object.
 56    * get_lcms_obj().
 57        Parses the mzml file and instantiates an LCMSBase object.
 58    * get_instrument_info().
 59        Return instrument information from the mzML file.
 60    * get_creation_time().
 61        Return the creation time of the mzML file as a datetime object.
 62
 63    Inherits from ThermoBaseClass and SpectraParserInterface
 64    """
 65
 66    def __init__(
 67        self,
 68        file_location,
 69        analyzer="Unknown",
 70        instrument_label="Unknown",
 71        sample_name=None,
 72    ):
 73        # implementation details
 74        if isinstance(file_location, str):
 75            # if obj is a string it defaults to create a Path obj, pass the S3Path if needed
 76            file_location = Path(file_location)
 77        if not file_location.exists():
 78            raise FileExistsError("File does not exist: " + str(file_location))
 79        self.file_location = file_location
 80        self.analyzer = analyzer
 81        self.instrument_label = instrument_label
 82
 83        if sample_name:
 84            self.sample_name = sample_name
 85        else:
 86            self.sample_name = file_location.stem
 87
 88    def load(self):
 89        """
 90        Load mzML file using pymzml.run.Reader and return the data as a numpy array.
 91
 92        Returns
 93        -------
 94        numpy.ndarray
 95            The mass spectra data as a numpy array.
 96        """
 97        data = pymzml.run.Reader(self.file_location)
 98        return data
 99
100    def get_scan_df(
101        self,
102        data=None,
103        time_range: Optional[
104            Union[Tuple[float, float], List[Tuple[float, float]]]
105        ] = None,
106    ):
107        """
108        Return scan data as a pandas DataFrame.
109
110        Parameters
111        ----------
112        data : pymzml.run.Reader, optional
113            The mass spectra data. If None, will load the data.
114        time_range : tuple or list of tuples, optional
115            Retention time range(s) to filter scans. Can be:
116            - Single range: (start_time, end_time) in minutes
117            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
118            If None, returns all scans.
119
120        Returns
121        -------
122        pandas.DataFrame
123            A pandas DataFrame containing metadata for each scan, including scan number, MS level, polarity, and scan time.
124        """
125        if data is None:
126            data = self.load()
127        # Scan dict
128        # instatinate scan dict, with empty lists of size of scans
129        n_scans = data.get_spectrum_count()
130        scan_dict = {
131            "scan": np.empty(n_scans, dtype=np.int32),
132            "scan_time": np.empty(n_scans, dtype=np.float32),
133            "ms_level": [None] * n_scans,
134            "polarity": [None] * n_scans,
135            "precursor_mz": [None] * n_scans,
136            "scan_text": [None] * n_scans,
137            "scan_window_lower": np.empty(n_scans, dtype=np.float32),
138            "scan_window_upper": np.empty(n_scans, dtype=np.float32),
139            "scan_precision": [None] * n_scans,
140            "tic": np.empty(n_scans, dtype=np.float32),
141            "ms_format": [None] * n_scans,
142        }
143
144        # First pass: loop through scans to get scan info
145        for i, spec in enumerate(data):
146            scan_dict["scan"][i] = spec.ID
147            scan_dict["ms_level"][i] = spec.ms_level
148            scan_dict["scan_precision"][i] = spec._measured_precision
149            scan_dict["tic"][i] = spec.TIC
150            if spec.selected_precursors:
151                scan_dict["precursor_mz"][i] = spec.selected_precursors[0].get(
152                    "mz", None
153                )
154            if spec["negative scan"] is not None:
155                scan_dict["polarity"][i] = "negative"
156            if spec["positive scan"] is not None:
157                scan_dict["polarity"][i] = "positive"
158            if spec["negative scan"] is not None and spec["positive scan"] is not None:
159                raise ValueError(
160                    "Error: scan {0} has both negative and positive polarity".format(
161                        spec.ID
162                    )
163                )
164
165            scan_dict["scan_time"][i] = spec.get("MS:1000016")
166            scan_dict["scan_text"][i] = spec.get("MS:1000512")
167            scan_dict["scan_window_lower"][i] = spec.get("MS:1000501")
168            scan_dict["scan_window_upper"][i] = spec.get("MS:1000500")
169            if spec.get("MS:1000128"):
170                scan_dict["ms_format"][i] = "profile"
171            elif spec.get("MS:1000127"):
172                scan_dict["ms_format"][i] = "centroid"
173            else:
174                scan_dict["ms_format"][i] = None
175
176        scan_df = pd.DataFrame(scan_dict)
177
178        # Remove any non-mass spectra scans (e.g., MS level 0 or None)
179        scan_df = scan_df[
180            scan_df.ms_level.notnull() & (scan_df.ms_level > 0)
181        ].reset_index(drop=True)
182
183        # Apply time range filtering if specified
184        if time_range is not None:
185            time_ranges = self._normalize_time_range(time_range)
186            # Create a mask for scans within any of the time ranges
187            mask = np.zeros(len(scan_df), dtype=bool)
188            for start_time, end_time in time_ranges:
189                mask |= (scan_df["scan_time"] >= start_time) & (
190                    scan_df["scan_time"] <= end_time
191                )
192            scan_df = scan_df[mask].reset_index(drop=True)
193
194        return scan_df
195
196    def get_ms_raw(
197        self,
198        spectra,
199        scan_df,
200        data=None,
201        time_range: Optional[
202            Union[Tuple[float, float], List[Tuple[float, float]]]
203        ] = None,
204    ):
205        """Return a dictionary of mass spectra data as a pandas DataFrame.
206
207        Parameters
208        ----------
209        spectra : str
210            Which mass spectra data to include in the output.
211            Options: None, "ms1", "ms2", "all".
212        scan_df : pandas.DataFrame
213            Scan dataframe. Output from get_scan_df().
214        data : pymzml.run.Reader, optional
215            The mass spectra data. If None, will load the data.
216        time_range : tuple or list of tuples, optional
217            Retention time range(s) to filter scans. Can be:
218            - Single range: (start_time, end_time) in minutes
219            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
220            If None, returns all scans. Note: filtering is typically done at scan_df level.
221
222        Returns
223        -------
224        dict
225            A dictionary containing the mass spectra data as pandas DataFrames, with keys corresponding to the MS level.
226
227        """
228        if data is None:
229            data = self.load()
230        if spectra == "all":
231            scan_df_forspec = scan_df
232        elif spectra == "ms1":
233            scan_df_forspec = scan_df[scan_df.ms_level == 1]
234        elif spectra == "ms2":
235            scan_df_forspec = scan_df[scan_df.ms_level == 2]
236        else:
237            raise ValueError("spectra must be 'all', 'ms1', or 'ms2'")
238
239        # Result container
240        res = {}
241
242        # Row count container
243        counter = {}
244
245        # Column name container
246        cols = {}
247
248        # set at float32
249        dtype = np.float32
250
251        # First pass: get nrows
252        N = defaultdict(lambda: 0)
253        for i, spec in enumerate(data):
254            if spec.ID in scan_df_forspec.scan.values:
255                # Get ms level
256                level = "ms{}".format(spec.ms_level)
257
258                # Number of rows
259                N[level] += spec.mz.shape[0]
260
261        # Second pass: parse
262        for i, spec in enumerate(data):
263            if spec.ID in scan_df_forspec.scan.values:
264                # Number of rows
265                n = spec.mz.shape[0]
266
267                # No measurements
268                if n == 0:
269                    continue
270
271                # Dimension check
272                if len(spec.mz) != len(spec.i):
273                    # raise an error if the mz and intensity arrays are not the same length
274                    raise ValueError("m/z and intensity array dimension mismatch")
275
276                # Scan/frame info
277                id_dict = spec.id_dict
278
279                # Get ms level
280                level = "ms{}".format(spec.ms_level)
281
282                # Columns
283                cols[level] = list(id_dict.keys()) + ["mz", "intensity"]
284                m = len(cols[level])
285
286                # Subarray init
287                arr = np.empty((n, m), dtype=dtype)
288                inx = 0
289
290                # Populate scan/frame info
291                for k, v in id_dict.items():
292                    arr[:, inx] = v
293                    inx += 1
294
295                # Populate m/z
296                arr[:, inx] = spec.mz
297                inx += 1
298
299                # Populate intensity
300                arr[:, inx] = spec.i
301                inx += 1
302
303                # Initialize output container
304                if level not in res:
305                    res[level] = np.empty((N[level], m), dtype=dtype)
306                    counter[level] = 0
307
308                # Insert subarray
309                res[level][counter[level] : counter[level] + n, :] = arr
310                counter[level] += n
311
312        # Construct ms1 and ms2 mz dataframes
313        for level in res.keys():
314            res[level] = pd.DataFrame(res[level], columns=cols[level]).drop(
315                columns=["controllerType", "controllerNumber"],
316            )
317
318        return res
319
320    def run(
321        self,
322        spectra="all",
323        scan_df=None,
324        time_range: Optional[
325            Union[Tuple[float, float], List[Tuple[float, float]]]
326        ] = None,
327    ):
328        """Parse the mzML file and return a dictionary of spectra dataframes and a scan metadata dataframe.
329
330        Parameters
331        ----------
332        spectra : str, optional
333            Which mass spectra data to include in the output. Default is "all".
334            Other options: None, "ms1", "ms2".
335        scan_df : pandas.DataFrame, optional
336            Scan dataframe.  If not provided, the scan dataframe is created from the mzML file.
337        time_range : tuple or list of tuples, optional
338            Retention time range(s) to load. Can be:
339            - Single range: (start_time, end_time) in minutes
340            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
341            If None, loads all scans.
342
343        Returns
344        -------
345        tuple
346            A tuple containing two elements:
347            - A dictionary containing the mass spectra data as numpy arrays, with keys corresponding to the MS level.
348            - A pandas DataFrame containing metadata for each scan, including scan number, MS level, polarity, and scan time.
349        """
350
351        # Open file
352        data = self.load()
353
354        if scan_df is None:
355            scan_df = self.get_scan_df(data, time_range=time_range)
356
357        if spectra != "none":
358            res = self.get_ms_raw(spectra, scan_df, data)
359
360        else:
361            res = None
362
363        return res, scan_df
364
365    def get_mass_spectrum_from_scan(
366        self, scan_number, spectrum_mode, auto_process=True
367    ):
368        """Instatiate a mass spectrum object from the mzML file.
369
370        Parameters
371        ----------
372        scan_number : int
373            The scan number to be parsed.
374        spectrum_mode : str
375            The type of spectrum to instantiate.  Must be'profile' or 'centroid'.
376        polarity : int
377            The polarity of the scan.  Must be -1 or 1.
378        auto_process : bool, optional
379            If True, process the mass spectrum. Default is True.
380
381        Returns
382        -------
383        MassSpecProfile | MassSpecCentroid
384            The MassSpecProfile or MassSpecCentroid object containing the parsed mass spectrum.
385        """
386        # Use the batch function and return the first result
387        result_list = self.get_mass_spectra_from_scan_list(
388            [scan_number], spectrum_mode, auto_process
389        )
390        return result_list[0] if result_list else None
391
392    def get_mass_spectra_from_scan_list(
393        self, scan_list, spectrum_mode, auto_process=True
394    ):
395        """Instatiate mass spectrum objects from the mzML file.
396
397        Parameters
398        ----------
399        scan_list : list of int
400            The scan numbers to be parsed.
401        spectrum_mode : str
402            The type of spectrum to instantiate.  Must be'profile' or 'centroid'.
403        auto_process : bool, optional
404            If True, process the mass spectrum. Default is True.
405
406        Returns
407        -------
408        list of MassSpecProfile | MassSpecCentroid
409            List of MassSpecProfile or MassSpecCentroid objects containing the parsed mass spectra.
410        """
411
412        def set_metadata(
413            scan_number: int,
414            polarity: int,
415            file_location: str,
416            label=Labels.thermo_profile,
417        ):
418            """
419            Set the output parameters for creating a MassSpecProfile or MassSpecCentroid object.
420
421            Parameters
422            ----------
423            scan_number : int
424                The scan number.
425            polarity : int
426                The polarity of the data.
427            file_location : str
428                The file location.
429            label : str, optional
430                The label for the mass spectrum. Default is Labels.thermo_profile.
431
432            Returns
433            -------
434            dict
435                The output parameters ready for creating a MassSpecProfile or MassSpecCentroid object.
436            """
437            d_params = default_parameters(file_location)
438            d_params["label"] = label
439            d_params["polarity"] = polarity
440            d_params["filename_path"] = file_location
441            d_params["scan_number"] = scan_number
442
443            return d_params
444
445        # Open file
446        data = self.load()
447
448        mass_spectrum_objects = []
449        scan_set = set(scan_list)
450        # Materialize peak data during iteration. Do not retain live pymzml
451        # spectrum objects, and do not overwrite the first valid MS spectrum for
452        # a scan ID.
453        #
454        # Thermo multi-controller mzML files often reuse scan numbers across
455        # controllers (e.g. controllerType=0 MS data and controllerType=3
456        # auxiliary traces). pymzml exposes only the integer scan number as
457        # spec.ID, so sequential iteration can yield the same ID twice. Keeping
458        # the last yield (or random-access data[scan]) can replace real MS
459        # spectra with non-MS controller spectra and then fail centroid/profile
460        # checks with "spectrum is not centroided".
461        #
462        # Direct random-access via data[scan_number] also uses pymzml's
463        # byte-offset index, which is unreliable on Windows (CRLF vs LF).
464        collected = {}
465
466        for spec in data:
467            scan_id = spec.ID
468            if scan_id not in scan_set or scan_id in collected:
469                continue
470
471            # Skip non-MS / auxiliary-controller spectra that share scan numbers
472            ms_level = spec.ms_level
473            if ms_level is None or ms_level <= 0:
474                continue
475
476            if spec["negative scan"] is not None:
477                polarity = -1
478            elif spec["positive scan"] is not None:
479                polarity = 1
480            else:
481                polarity = None
482
483            mz = np.asarray(spec.mz)
484            abundance = np.asarray(spec.i)
485            collected[scan_id] = {
486                "mz": mz,
487                "abundance": abundance,
488                "polarity": polarity,
489                "is_profile": bool(spec.get("MS:1000128")),
490                "is_centroid": bool(spec.get("MS:1000127")),
491            }
492
493        for scan_number in scan_list:
494            entry = collected.get(scan_number)
495            if entry is None:
496                raise ValueError("Scan number %d not found in mzML file" % scan_number)
497
498            polarity = entry["polarity"]
499            mz = entry["mz"]
500            abundance = entry["abundance"]
501
502            # Get mass spectrum
503            if spectrum_mode == "profile":
504                # Check if profile
505                if not entry["is_profile"]:
506                    raise ValueError("spectrum is not profile")
507                data_dict = {
508                    Labels.mz: mz,
509                    Labels.abundance: abundance,
510                }
511                d_params = set_metadata(
512                    scan_number,
513                    polarity,
514                    self.file_location,
515                    label=Labels.simulated_profile,
516                )
517                mass_spectrum_obj = MassSpecProfile(
518                    data_dict, d_params, auto_process=auto_process
519                )
520            elif spectrum_mode == "centroid":
521                # Check if centroided
522                if not entry["is_centroid"]:
523                    raise ValueError("spectrum is not centroided")
524                data_dict = {
525                    Labels.mz: mz,
526                    Labels.abundance: abundance,
527                    Labels.rp: [np.nan] * len(mz),
528                    Labels.s2n: [np.nan] * len(abundance),
529                }
530                d_params = set_metadata(
531                    scan_number,
532                    polarity,
533                    self.file_location,
534                    label=Labels.corems_centroid,
535                )
536                mass_spectrum_obj = MassSpecCentroid(
537                    data_dict, d_params, auto_process=auto_process
538                )
539            else:
540                raise ValueError(
541                    "spectrum_mode must be 'profile' or 'centroid', got %r"
542                    % spectrum_mode
543                )
544
545            mass_spectrum_objects.append(mass_spectrum_obj)
546
547        return mass_spectrum_objects
548
549    def get_mass_spectra_obj(
550        self,
551        time_range: Optional[
552            Union[Tuple[float, float], List[Tuple[float, float]]]
553        ] = None,
554    ):
555        """Instatiate a MassSpectraBase object from the mzML file.
556
557        Parameters
558        ----------
559        time_range : tuple or list of tuples, optional
560            Retention time range(s) to load. Can be:
561            - Single range: (start_time, end_time) in minutes
562            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
563            If None, loads all scans.
564
565        Returns
566        -------
567        MassSpectraBase
568            The MassSpectra object containing the parsed mass spectra.
569            The object is instatiated with the mzML file, analyzer, instrument, sample name, and scan dataframe.
570        """
571        _, scan_df = self.run(spectra=False, time_range=time_range)
572        mass_spectra_obj = MassSpectraBase(
573            self.file_location,
574            self.analyzer,
575            self.instrument_label,
576            self.sample_name,
577            self,
578        )
579        scan_df = scan_df.set_index("scan", drop=False)
580        mass_spectra_obj.scan_df = scan_df
581
582        return mass_spectra_obj
583
584    def get_lcms_obj(
585        self,
586        spectra="all",
587        time_range: Optional[
588            Union[Tuple[float, float], List[Tuple[float, float]]]
589        ] = None,
590    ):
591        """Instatiates a LCMSBase object from the mzML file.
592
593        Parameters
594        ----------
595        spectra : str, optional
596            Which mass spectra data to include in the output. Default is all.  Other options: none, ms1, ms2.
597        time_range : tuple or list of tuples, optional
598            Retention time range(s) to load. Can be:
599            - Single range: (start_time, end_time) in minutes
600            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
601            If None, loads all scans. Useful for targeted workflows to improve performance.
602
603        Returns
604        -------
605        LCMSBase
606            LCMS object containing mass spectra data.
607            The object is instatiated with the mzML file, analyzer, instrument, sample name, scan dataframe,
608            and mz dataframe(s), as well as lists of scan numbers, retention times, and TICs.
609        """
610        _, scan_df = self.run(
611            spectra="none", time_range=time_range
612        )  # first run it to just get scan info
613        if spectra != "none":
614            res, scan_df = self.run(
615                scan_df=scan_df, spectra=spectra, time_range=time_range
616            )  # second run to parse data
617        lcms_obj = LCMSBase(
618            self.file_location,
619            self.analyzer,
620            self.instrument_label,
621            self.sample_name,
622            self,
623        )
624        if spectra != "none":
625            for key in res:
626                key_int = int(key.replace("ms", ""))
627                res[key] = res[key][res[key].intensity > 0]
628                res[key] = (
629                    res[key].sort_values(by=["scan", "mz"]).reset_index(drop=True)
630                )
631                lcms_obj._ms_unprocessed[key_int] = res[key]
632        lcms_obj.scan_df = scan_df.set_index("scan", drop=False)
633        # Check if polarity is mixed
634        if len(set(scan_df.polarity)) > 1:
635            raise ValueError("Mixed polarities detected in scan data")
636        # Use iloc for positional access (Series.__getitem__ is label-based;
637        # safer across pandas 2.x and 3.x).
638        lcms_obj.polarity = scan_df.polarity.iloc[0]
639        lcms_obj._scans_number_list = list(scan_df.scan)
640        lcms_obj._retention_time_list = list(scan_df.scan_time)
641        lcms_obj._tic_list = list(scan_df.tic)
642
643        return lcms_obj
644
645    def get_scans_in_time_range(
646        self,
647        time_range: Union[Tuple[float, float], List[Tuple[float, float]]],
648        ms_level: Optional[int] = None,
649    ) -> List[int]:
650        """
651        Return scan numbers within specified retention time range(s).
652
653        This method provides efficient filtering of scans by retention time,
654        which is particularly useful for targeted workflows where only specific
655        time windows are of interest.
656
657        Parameters
658        ----------
659        time_range : tuple or list of tuples
660            Retention time range(s) in minutes. Can be:
661            - Single range: (start_time, end_time)
662            - Multiple ranges: [(start1, end1), (start2, end2), ...]
663        ms_level : int, optional
664            If specified, only return scans of this MS level (e.g., 1 for MS1, 2 for MS2).
665            If None, returns scans of all MS levels.
666
667        Returns
668        -------
669        list of int
670            List of scan numbers within the specified time range(s) and MS level.
671
672        Examples
673        --------
674        Get MS1 scans between 1.0 and 2.0 minutes:
675
676        >>> scans = parser.get_scans_in_time_range((1.0, 2.0), ms_level=1)
677
678        Get scans in multiple time windows:
679
680        >>> scans = parser.get_scans_in_time_range([(0.5, 1.5), (3.0, 4.0)])
681        """
682        # Get scan dataframe filtered by time range
683        scan_df = self.get_scan_df(time_range=time_range)
684
685        # Further filter by MS level if specified
686        if ms_level is not None:
687            scan_df = scan_df[scan_df.ms_level == ms_level]
688
689        # Return list of scan numbers
690        return scan_df.scan.tolist()
691
692    def get_instrument_info(self):
693        """
694        Return instrument information.
695
696        Returns
697        -------
698        dict
699            A dictionary with the keys 'model' and 'serial_number'.
700        """
701        # Load the pymzml data
702        data = self.load()
703        instrument_info = data.info.get("referenceable_param_group_list_element")[0]
704        cv_params = instrument_info.findall("{http://psi.hupo.org/ms/mzml}cvParam")
705
706        # Extract details from each cvParam
707        params = []
708        for param in cv_params:
709            accession = param.get("accession")  # Get 'accession' attribute
710            name = param.get("name")  # Get 'name' attribute
711            value = param.get("value")  # Get 'value' attribute
712            params.append({"accession": accession, "name": name, "value": value})
713
714        # Loop through params and try to find the relevant information
715        instrument_dict = {"model": "Unknown", "serial_number": "Unknown"}
716
717        # Assuming there are only two paramters here - one is for the serial number (agnostic to the model) and the other is for the model
718        # If there are more than two, we raise an error
719        if len(params) < 2:
720            raise ValueError(
721                "Not enough parameters found in the instrument info, cannot parse."
722            )
723        if len(params) > 2:
724            raise ValueError(
725                "Too many parameters found in the instrument info, cannot parse."
726            )
727        for param in params:
728            if param["accession"] == "MS:1000529":
729                instrument_dict["serial_number"] = param["value"]
730            else:
731                instrument_dict["model"] = data.OT[param["accession"]]
732
733        return instrument_dict
734
735    def get_creation_time(self) -> datetime.datetime:
736        """
737        Return the creation time of the mzML file.
738        """
739        data = self.load()
740        write_time = data.info.get("start_time")
741        if write_time:
742            # Convert the write time to a datetime object
743            return datetime.datetime.strptime(write_time, "%Y-%m-%dT%H:%M:%SZ")
744        else:
745            raise ValueError(
746                "Creation time is not available in the mzML file. "
747                "Please ensure the file contains the 'start_time' information."
748            )

A class for parsing mzml spectrometry data files into MassSpectraBase or LCMSBase objects

Parameters
  • file_location (str or Path): The path to the RAW file to be parsed.
  • analyzer (str, optional): The type of mass analyzer used in the instrument. Default is "Unknown".
  • instrument_label (str, optional): The name of the instrument used to acquire the data. Default is "Unknown".
  • sample_name (str, optional): The name of the sample being analyzed. If not provided, the stem of the file_location path will be used.
Attributes
  • file_location (Path): The path to the RAW file being parsed.
  • analyzer (str): The type of mass analyzer used in the instrument.
  • instrument_label (str): The name of the instrument used to acquire the data.
  • sample_name (str): The name of the sample being analyzed.
Methods
  • load(). Load mzML file using pymzml.run.Reader and return the data as a numpy array.
  • run(spectra=True). Parses the mzml file and returns a dictionary of mass spectra dataframes and a scan metadata dataframe.
  • get_mass_spectrum_from_scan(scan_number, polarity, auto_process=True) Parses the mzml file and returns a MassSpecBase object from a single scan.
  • get_mass_spectra_obj(). Parses the mzml file and instantiates a MassSpectraBase object.
  • get_lcms_obj(). Parses the mzml file and instantiates an LCMSBase object.
  • get_instrument_info(). Return instrument information from the mzML file.
  • get_creation_time(). Return the creation time of the mzML file as a datetime object.

Inherits from ThermoBaseClass and SpectraParserInterface

MZMLSpectraParser( file_location, analyzer='Unknown', instrument_label='Unknown', sample_name=None)
66    def __init__(
67        self,
68        file_location,
69        analyzer="Unknown",
70        instrument_label="Unknown",
71        sample_name=None,
72    ):
73        # implementation details
74        if isinstance(file_location, str):
75            # if obj is a string it defaults to create a Path obj, pass the S3Path if needed
76            file_location = Path(file_location)
77        if not file_location.exists():
78            raise FileExistsError("File does not exist: " + str(file_location))
79        self.file_location = file_location
80        self.analyzer = analyzer
81        self.instrument_label = instrument_label
82
83        if sample_name:
84            self.sample_name = sample_name
85        else:
86            self.sample_name = file_location.stem
file_location
analyzer
instrument_label
def load(self):
88    def load(self):
89        """
90        Load mzML file using pymzml.run.Reader and return the data as a numpy array.
91
92        Returns
93        -------
94        numpy.ndarray
95            The mass spectra data as a numpy array.
96        """
97        data = pymzml.run.Reader(self.file_location)
98        return data

Load mzML file using pymzml.run.Reader and return the data as a numpy array.

Returns
  • numpy.ndarray: The mass spectra data as a numpy array.
def get_scan_df( self, data=None, time_range: Union[Tuple[float, float], List[Tuple[float, float]], NoneType] = None):
100    def get_scan_df(
101        self,
102        data=None,
103        time_range: Optional[
104            Union[Tuple[float, float], List[Tuple[float, float]]]
105        ] = None,
106    ):
107        """
108        Return scan data as a pandas DataFrame.
109
110        Parameters
111        ----------
112        data : pymzml.run.Reader, optional
113            The mass spectra data. If None, will load the data.
114        time_range : tuple or list of tuples, optional
115            Retention time range(s) to filter scans. Can be:
116            - Single range: (start_time, end_time) in minutes
117            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
118            If None, returns all scans.
119
120        Returns
121        -------
122        pandas.DataFrame
123            A pandas DataFrame containing metadata for each scan, including scan number, MS level, polarity, and scan time.
124        """
125        if data is None:
126            data = self.load()
127        # Scan dict
128        # instatinate scan dict, with empty lists of size of scans
129        n_scans = data.get_spectrum_count()
130        scan_dict = {
131            "scan": np.empty(n_scans, dtype=np.int32),
132            "scan_time": np.empty(n_scans, dtype=np.float32),
133            "ms_level": [None] * n_scans,
134            "polarity": [None] * n_scans,
135            "precursor_mz": [None] * n_scans,
136            "scan_text": [None] * n_scans,
137            "scan_window_lower": np.empty(n_scans, dtype=np.float32),
138            "scan_window_upper": np.empty(n_scans, dtype=np.float32),
139            "scan_precision": [None] * n_scans,
140            "tic": np.empty(n_scans, dtype=np.float32),
141            "ms_format": [None] * n_scans,
142        }
143
144        # First pass: loop through scans to get scan info
145        for i, spec in enumerate(data):
146            scan_dict["scan"][i] = spec.ID
147            scan_dict["ms_level"][i] = spec.ms_level
148            scan_dict["scan_precision"][i] = spec._measured_precision
149            scan_dict["tic"][i] = spec.TIC
150            if spec.selected_precursors:
151                scan_dict["precursor_mz"][i] = spec.selected_precursors[0].get(
152                    "mz", None
153                )
154            if spec["negative scan"] is not None:
155                scan_dict["polarity"][i] = "negative"
156            if spec["positive scan"] is not None:
157                scan_dict["polarity"][i] = "positive"
158            if spec["negative scan"] is not None and spec["positive scan"] is not None:
159                raise ValueError(
160                    "Error: scan {0} has both negative and positive polarity".format(
161                        spec.ID
162                    )
163                )
164
165            scan_dict["scan_time"][i] = spec.get("MS:1000016")
166            scan_dict["scan_text"][i] = spec.get("MS:1000512")
167            scan_dict["scan_window_lower"][i] = spec.get("MS:1000501")
168            scan_dict["scan_window_upper"][i] = spec.get("MS:1000500")
169            if spec.get("MS:1000128"):
170                scan_dict["ms_format"][i] = "profile"
171            elif spec.get("MS:1000127"):
172                scan_dict["ms_format"][i] = "centroid"
173            else:
174                scan_dict["ms_format"][i] = None
175
176        scan_df = pd.DataFrame(scan_dict)
177
178        # Remove any non-mass spectra scans (e.g., MS level 0 or None)
179        scan_df = scan_df[
180            scan_df.ms_level.notnull() & (scan_df.ms_level > 0)
181        ].reset_index(drop=True)
182
183        # Apply time range filtering if specified
184        if time_range is not None:
185            time_ranges = self._normalize_time_range(time_range)
186            # Create a mask for scans within any of the time ranges
187            mask = np.zeros(len(scan_df), dtype=bool)
188            for start_time, end_time in time_ranges:
189                mask |= (scan_df["scan_time"] >= start_time) & (
190                    scan_df["scan_time"] <= end_time
191                )
192            scan_df = scan_df[mask].reset_index(drop=True)
193
194        return scan_df

Return scan data as a pandas DataFrame.

Parameters
  • data (pymzml.run.Reader, optional): The mass spectra data. If None, will load the data.
  • time_range (tuple or list of tuples, optional): Retention time range(s) to filter scans. Can be:
    • Single range: (start_time, end_time) in minutes
    • Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes If None, returns all scans.
Returns
  • pandas.DataFrame: A pandas DataFrame containing metadata for each scan, including scan number, MS level, polarity, and scan time.
def get_ms_raw( self, spectra, scan_df, data=None, time_range: Union[Tuple[float, float], List[Tuple[float, float]], NoneType] = None):
196    def get_ms_raw(
197        self,
198        spectra,
199        scan_df,
200        data=None,
201        time_range: Optional[
202            Union[Tuple[float, float], List[Tuple[float, float]]]
203        ] = None,
204    ):
205        """Return a dictionary of mass spectra data as a pandas DataFrame.
206
207        Parameters
208        ----------
209        spectra : str
210            Which mass spectra data to include in the output.
211            Options: None, "ms1", "ms2", "all".
212        scan_df : pandas.DataFrame
213            Scan dataframe. Output from get_scan_df().
214        data : pymzml.run.Reader, optional
215            The mass spectra data. If None, will load the data.
216        time_range : tuple or list of tuples, optional
217            Retention time range(s) to filter scans. Can be:
218            - Single range: (start_time, end_time) in minutes
219            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
220            If None, returns all scans. Note: filtering is typically done at scan_df level.
221
222        Returns
223        -------
224        dict
225            A dictionary containing the mass spectra data as pandas DataFrames, with keys corresponding to the MS level.
226
227        """
228        if data is None:
229            data = self.load()
230        if spectra == "all":
231            scan_df_forspec = scan_df
232        elif spectra == "ms1":
233            scan_df_forspec = scan_df[scan_df.ms_level == 1]
234        elif spectra == "ms2":
235            scan_df_forspec = scan_df[scan_df.ms_level == 2]
236        else:
237            raise ValueError("spectra must be 'all', 'ms1', or 'ms2'")
238
239        # Result container
240        res = {}
241
242        # Row count container
243        counter = {}
244
245        # Column name container
246        cols = {}
247
248        # set at float32
249        dtype = np.float32
250
251        # First pass: get nrows
252        N = defaultdict(lambda: 0)
253        for i, spec in enumerate(data):
254            if spec.ID in scan_df_forspec.scan.values:
255                # Get ms level
256                level = "ms{}".format(spec.ms_level)
257
258                # Number of rows
259                N[level] += spec.mz.shape[0]
260
261        # Second pass: parse
262        for i, spec in enumerate(data):
263            if spec.ID in scan_df_forspec.scan.values:
264                # Number of rows
265                n = spec.mz.shape[0]
266
267                # No measurements
268                if n == 0:
269                    continue
270
271                # Dimension check
272                if len(spec.mz) != len(spec.i):
273                    # raise an error if the mz and intensity arrays are not the same length
274                    raise ValueError("m/z and intensity array dimension mismatch")
275
276                # Scan/frame info
277                id_dict = spec.id_dict
278
279                # Get ms level
280                level = "ms{}".format(spec.ms_level)
281
282                # Columns
283                cols[level] = list(id_dict.keys()) + ["mz", "intensity"]
284                m = len(cols[level])
285
286                # Subarray init
287                arr = np.empty((n, m), dtype=dtype)
288                inx = 0
289
290                # Populate scan/frame info
291                for k, v in id_dict.items():
292                    arr[:, inx] = v
293                    inx += 1
294
295                # Populate m/z
296                arr[:, inx] = spec.mz
297                inx += 1
298
299                # Populate intensity
300                arr[:, inx] = spec.i
301                inx += 1
302
303                # Initialize output container
304                if level not in res:
305                    res[level] = np.empty((N[level], m), dtype=dtype)
306                    counter[level] = 0
307
308                # Insert subarray
309                res[level][counter[level] : counter[level] + n, :] = arr
310                counter[level] += n
311
312        # Construct ms1 and ms2 mz dataframes
313        for level in res.keys():
314            res[level] = pd.DataFrame(res[level], columns=cols[level]).drop(
315                columns=["controllerType", "controllerNumber"],
316            )
317
318        return res

Return a dictionary of mass spectra data as a pandas DataFrame.

Parameters
  • spectra (str): Which mass spectra data to include in the output. Options: None, "ms1", "ms2", "all".
  • scan_df (pandas.DataFrame): Scan dataframe. Output from get_scan_df().
  • data (pymzml.run.Reader, optional): The mass spectra data. If None, will load the data.
  • time_range (tuple or list of tuples, optional): Retention time range(s) to filter scans. Can be:
    • Single range: (start_time, end_time) in minutes
    • Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes If None, returns all scans. Note: filtering is typically done at scan_df level.
Returns
  • dict: A dictionary containing the mass spectra data as pandas DataFrames, with keys corresponding to the MS level.
def run( self, spectra='all', scan_df=None, time_range: Union[Tuple[float, float], List[Tuple[float, float]], NoneType] = None):
320    def run(
321        self,
322        spectra="all",
323        scan_df=None,
324        time_range: Optional[
325            Union[Tuple[float, float], List[Tuple[float, float]]]
326        ] = None,
327    ):
328        """Parse the mzML file and return a dictionary of spectra dataframes and a scan metadata dataframe.
329
330        Parameters
331        ----------
332        spectra : str, optional
333            Which mass spectra data to include in the output. Default is "all".
334            Other options: None, "ms1", "ms2".
335        scan_df : pandas.DataFrame, optional
336            Scan dataframe.  If not provided, the scan dataframe is created from the mzML file.
337        time_range : tuple or list of tuples, optional
338            Retention time range(s) to load. Can be:
339            - Single range: (start_time, end_time) in minutes
340            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
341            If None, loads all scans.
342
343        Returns
344        -------
345        tuple
346            A tuple containing two elements:
347            - A dictionary containing the mass spectra data as numpy arrays, with keys corresponding to the MS level.
348            - A pandas DataFrame containing metadata for each scan, including scan number, MS level, polarity, and scan time.
349        """
350
351        # Open file
352        data = self.load()
353
354        if scan_df is None:
355            scan_df = self.get_scan_df(data, time_range=time_range)
356
357        if spectra != "none":
358            res = self.get_ms_raw(spectra, scan_df, data)
359
360        else:
361            res = None
362
363        return res, scan_df

Parse the mzML file and return a dictionary of spectra dataframes and a scan metadata dataframe.

Parameters
  • spectra (str, optional): Which mass spectra data to include in the output. Default is "all". Other options: None, "ms1", "ms2".
  • scan_df (pandas.DataFrame, optional): Scan dataframe. If not provided, the scan dataframe is created from the mzML file.
  • time_range (tuple or list of tuples, optional): Retention time range(s) to load. Can be:
    • Single range: (start_time, end_time) in minutes
    • Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes If None, loads all scans.
Returns
  • tuple: A tuple containing two elements:
    • A dictionary containing the mass spectra data as numpy arrays, with keys corresponding to the MS level.
    • A pandas DataFrame containing metadata for each scan, including scan number, MS level, polarity, and scan time.
def get_mass_spectrum_from_scan(self, scan_number, spectrum_mode, auto_process=True):
365    def get_mass_spectrum_from_scan(
366        self, scan_number, spectrum_mode, auto_process=True
367    ):
368        """Instatiate a mass spectrum object from the mzML file.
369
370        Parameters
371        ----------
372        scan_number : int
373            The scan number to be parsed.
374        spectrum_mode : str
375            The type of spectrum to instantiate.  Must be'profile' or 'centroid'.
376        polarity : int
377            The polarity of the scan.  Must be -1 or 1.
378        auto_process : bool, optional
379            If True, process the mass spectrum. Default is True.
380
381        Returns
382        -------
383        MassSpecProfile | MassSpecCentroid
384            The MassSpecProfile or MassSpecCentroid object containing the parsed mass spectrum.
385        """
386        # Use the batch function and return the first result
387        result_list = self.get_mass_spectra_from_scan_list(
388            [scan_number], spectrum_mode, auto_process
389        )
390        return result_list[0] if result_list else None

Instatiate a mass spectrum object from the mzML file.

Parameters
  • scan_number (int): The scan number to be parsed.
  • spectrum_mode (str): The type of spectrum to instantiate. Must be'profile' or 'centroid'.
  • polarity (int): The polarity of the scan. Must be -1 or 1.
  • auto_process (bool, optional): If True, process the mass spectrum. Default is True.
Returns
  • MassSpecProfile | MassSpecCentroid: The MassSpecProfile or MassSpecCentroid object containing the parsed mass spectrum.
def get_mass_spectra_from_scan_list(self, scan_list, spectrum_mode, auto_process=True):
392    def get_mass_spectra_from_scan_list(
393        self, scan_list, spectrum_mode, auto_process=True
394    ):
395        """Instatiate mass spectrum objects from the mzML file.
396
397        Parameters
398        ----------
399        scan_list : list of int
400            The scan numbers to be parsed.
401        spectrum_mode : str
402            The type of spectrum to instantiate.  Must be'profile' or 'centroid'.
403        auto_process : bool, optional
404            If True, process the mass spectrum. Default is True.
405
406        Returns
407        -------
408        list of MassSpecProfile | MassSpecCentroid
409            List of MassSpecProfile or MassSpecCentroid objects containing the parsed mass spectra.
410        """
411
412        def set_metadata(
413            scan_number: int,
414            polarity: int,
415            file_location: str,
416            label=Labels.thermo_profile,
417        ):
418            """
419            Set the output parameters for creating a MassSpecProfile or MassSpecCentroid object.
420
421            Parameters
422            ----------
423            scan_number : int
424                The scan number.
425            polarity : int
426                The polarity of the data.
427            file_location : str
428                The file location.
429            label : str, optional
430                The label for the mass spectrum. Default is Labels.thermo_profile.
431
432            Returns
433            -------
434            dict
435                The output parameters ready for creating a MassSpecProfile or MassSpecCentroid object.
436            """
437            d_params = default_parameters(file_location)
438            d_params["label"] = label
439            d_params["polarity"] = polarity
440            d_params["filename_path"] = file_location
441            d_params["scan_number"] = scan_number
442
443            return d_params
444
445        # Open file
446        data = self.load()
447
448        mass_spectrum_objects = []
449        scan_set = set(scan_list)
450        # Materialize peak data during iteration. Do not retain live pymzml
451        # spectrum objects, and do not overwrite the first valid MS spectrum for
452        # a scan ID.
453        #
454        # Thermo multi-controller mzML files often reuse scan numbers across
455        # controllers (e.g. controllerType=0 MS data and controllerType=3
456        # auxiliary traces). pymzml exposes only the integer scan number as
457        # spec.ID, so sequential iteration can yield the same ID twice. Keeping
458        # the last yield (or random-access data[scan]) can replace real MS
459        # spectra with non-MS controller spectra and then fail centroid/profile
460        # checks with "spectrum is not centroided".
461        #
462        # Direct random-access via data[scan_number] also uses pymzml's
463        # byte-offset index, which is unreliable on Windows (CRLF vs LF).
464        collected = {}
465
466        for spec in data:
467            scan_id = spec.ID
468            if scan_id not in scan_set or scan_id in collected:
469                continue
470
471            # Skip non-MS / auxiliary-controller spectra that share scan numbers
472            ms_level = spec.ms_level
473            if ms_level is None or ms_level <= 0:
474                continue
475
476            if spec["negative scan"] is not None:
477                polarity = -1
478            elif spec["positive scan"] is not None:
479                polarity = 1
480            else:
481                polarity = None
482
483            mz = np.asarray(spec.mz)
484            abundance = np.asarray(spec.i)
485            collected[scan_id] = {
486                "mz": mz,
487                "abundance": abundance,
488                "polarity": polarity,
489                "is_profile": bool(spec.get("MS:1000128")),
490                "is_centroid": bool(spec.get("MS:1000127")),
491            }
492
493        for scan_number in scan_list:
494            entry = collected.get(scan_number)
495            if entry is None:
496                raise ValueError("Scan number %d not found in mzML file" % scan_number)
497
498            polarity = entry["polarity"]
499            mz = entry["mz"]
500            abundance = entry["abundance"]
501
502            # Get mass spectrum
503            if spectrum_mode == "profile":
504                # Check if profile
505                if not entry["is_profile"]:
506                    raise ValueError("spectrum is not profile")
507                data_dict = {
508                    Labels.mz: mz,
509                    Labels.abundance: abundance,
510                }
511                d_params = set_metadata(
512                    scan_number,
513                    polarity,
514                    self.file_location,
515                    label=Labels.simulated_profile,
516                )
517                mass_spectrum_obj = MassSpecProfile(
518                    data_dict, d_params, auto_process=auto_process
519                )
520            elif spectrum_mode == "centroid":
521                # Check if centroided
522                if not entry["is_centroid"]:
523                    raise ValueError("spectrum is not centroided")
524                data_dict = {
525                    Labels.mz: mz,
526                    Labels.abundance: abundance,
527                    Labels.rp: [np.nan] * len(mz),
528                    Labels.s2n: [np.nan] * len(abundance),
529                }
530                d_params = set_metadata(
531                    scan_number,
532                    polarity,
533                    self.file_location,
534                    label=Labels.corems_centroid,
535                )
536                mass_spectrum_obj = MassSpecCentroid(
537                    data_dict, d_params, auto_process=auto_process
538                )
539            else:
540                raise ValueError(
541                    "spectrum_mode must be 'profile' or 'centroid', got %r"
542                    % spectrum_mode
543                )
544
545            mass_spectrum_objects.append(mass_spectrum_obj)
546
547        return mass_spectrum_objects

Instatiate mass spectrum objects from the mzML file.

Parameters
  • scan_list (list of int): The scan numbers to be parsed.
  • spectrum_mode (str): The type of spectrum to instantiate. Must be'profile' or 'centroid'.
  • auto_process (bool, optional): If True, process the mass spectrum. Default is True.
Returns
  • list of MassSpecProfile | MassSpecCentroid: List of MassSpecProfile or MassSpecCentroid objects containing the parsed mass spectra.
def get_mass_spectra_obj( self, time_range: Union[Tuple[float, float], List[Tuple[float, float]], NoneType] = None):
549    def get_mass_spectra_obj(
550        self,
551        time_range: Optional[
552            Union[Tuple[float, float], List[Tuple[float, float]]]
553        ] = None,
554    ):
555        """Instatiate a MassSpectraBase object from the mzML file.
556
557        Parameters
558        ----------
559        time_range : tuple or list of tuples, optional
560            Retention time range(s) to load. Can be:
561            - Single range: (start_time, end_time) in minutes
562            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
563            If None, loads all scans.
564
565        Returns
566        -------
567        MassSpectraBase
568            The MassSpectra object containing the parsed mass spectra.
569            The object is instatiated with the mzML file, analyzer, instrument, sample name, and scan dataframe.
570        """
571        _, scan_df = self.run(spectra=False, time_range=time_range)
572        mass_spectra_obj = MassSpectraBase(
573            self.file_location,
574            self.analyzer,
575            self.instrument_label,
576            self.sample_name,
577            self,
578        )
579        scan_df = scan_df.set_index("scan", drop=False)
580        mass_spectra_obj.scan_df = scan_df
581
582        return mass_spectra_obj

Instatiate a MassSpectraBase object from the mzML file.

Parameters
  • time_range (tuple or list of tuples, optional): Retention time range(s) to load. Can be:
    • Single range: (start_time, end_time) in minutes
    • Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes If None, loads all scans.
Returns
  • MassSpectraBase: The MassSpectra object containing the parsed mass spectra. The object is instatiated with the mzML file, analyzer, instrument, sample name, and scan dataframe.
def get_lcms_obj( self, spectra='all', time_range: Union[Tuple[float, float], List[Tuple[float, float]], NoneType] = None):
584    def get_lcms_obj(
585        self,
586        spectra="all",
587        time_range: Optional[
588            Union[Tuple[float, float], List[Tuple[float, float]]]
589        ] = None,
590    ):
591        """Instatiates a LCMSBase object from the mzML file.
592
593        Parameters
594        ----------
595        spectra : str, optional
596            Which mass spectra data to include in the output. Default is all.  Other options: none, ms1, ms2.
597        time_range : tuple or list of tuples, optional
598            Retention time range(s) to load. Can be:
599            - Single range: (start_time, end_time) in minutes
600            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
601            If None, loads all scans. Useful for targeted workflows to improve performance.
602
603        Returns
604        -------
605        LCMSBase
606            LCMS object containing mass spectra data.
607            The object is instatiated with the mzML file, analyzer, instrument, sample name, scan dataframe,
608            and mz dataframe(s), as well as lists of scan numbers, retention times, and TICs.
609        """
610        _, scan_df = self.run(
611            spectra="none", time_range=time_range
612        )  # first run it to just get scan info
613        if spectra != "none":
614            res, scan_df = self.run(
615                scan_df=scan_df, spectra=spectra, time_range=time_range
616            )  # second run to parse data
617        lcms_obj = LCMSBase(
618            self.file_location,
619            self.analyzer,
620            self.instrument_label,
621            self.sample_name,
622            self,
623        )
624        if spectra != "none":
625            for key in res:
626                key_int = int(key.replace("ms", ""))
627                res[key] = res[key][res[key].intensity > 0]
628                res[key] = (
629                    res[key].sort_values(by=["scan", "mz"]).reset_index(drop=True)
630                )
631                lcms_obj._ms_unprocessed[key_int] = res[key]
632        lcms_obj.scan_df = scan_df.set_index("scan", drop=False)
633        # Check if polarity is mixed
634        if len(set(scan_df.polarity)) > 1:
635            raise ValueError("Mixed polarities detected in scan data")
636        # Use iloc for positional access (Series.__getitem__ is label-based;
637        # safer across pandas 2.x and 3.x).
638        lcms_obj.polarity = scan_df.polarity.iloc[0]
639        lcms_obj._scans_number_list = list(scan_df.scan)
640        lcms_obj._retention_time_list = list(scan_df.scan_time)
641        lcms_obj._tic_list = list(scan_df.tic)
642
643        return lcms_obj

Instatiates a LCMSBase object from the mzML file.

Parameters
  • spectra (str, optional): Which mass spectra data to include in the output. Default is all. Other options: none, ms1, ms2.
  • time_range (tuple or list of tuples, optional): Retention time range(s) to load. Can be:
    • Single range: (start_time, end_time) in minutes
    • Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes If None, loads all scans. Useful for targeted workflows to improve performance.
Returns
  • LCMSBase: LCMS object containing mass spectra data. The object is instatiated with the mzML file, analyzer, instrument, sample name, scan dataframe, and mz dataframe(s), as well as lists of scan numbers, retention times, and TICs.
def get_scans_in_time_range( self, time_range: Union[Tuple[float, float], List[Tuple[float, float]]], ms_level: Optional[int] = None) -> List[int]:
645    def get_scans_in_time_range(
646        self,
647        time_range: Union[Tuple[float, float], List[Tuple[float, float]]],
648        ms_level: Optional[int] = None,
649    ) -> List[int]:
650        """
651        Return scan numbers within specified retention time range(s).
652
653        This method provides efficient filtering of scans by retention time,
654        which is particularly useful for targeted workflows where only specific
655        time windows are of interest.
656
657        Parameters
658        ----------
659        time_range : tuple or list of tuples
660            Retention time range(s) in minutes. Can be:
661            - Single range: (start_time, end_time)
662            - Multiple ranges: [(start1, end1), (start2, end2), ...]
663        ms_level : int, optional
664            If specified, only return scans of this MS level (e.g., 1 for MS1, 2 for MS2).
665            If None, returns scans of all MS levels.
666
667        Returns
668        -------
669        list of int
670            List of scan numbers within the specified time range(s) and MS level.
671
672        Examples
673        --------
674        Get MS1 scans between 1.0 and 2.0 minutes:
675
676        >>> scans = parser.get_scans_in_time_range((1.0, 2.0), ms_level=1)
677
678        Get scans in multiple time windows:
679
680        >>> scans = parser.get_scans_in_time_range([(0.5, 1.5), (3.0, 4.0)])
681        """
682        # Get scan dataframe filtered by time range
683        scan_df = self.get_scan_df(time_range=time_range)
684
685        # Further filter by MS level if specified
686        if ms_level is not None:
687            scan_df = scan_df[scan_df.ms_level == ms_level]
688
689        # Return list of scan numbers
690        return scan_df.scan.tolist()

Return scan numbers within specified retention time range(s).

This method provides efficient filtering of scans by retention time, which is particularly useful for targeted workflows where only specific time windows are of interest.

Parameters
  • time_range (tuple or list of tuples): Retention time range(s) in minutes. Can be:
    • Single range: (start_time, end_time)
    • Multiple ranges: [(start1, end1), (start2, end2), ...]
  • ms_level (int, optional): If specified, only return scans of this MS level (e.g., 1 for MS1, 2 for MS2). If None, returns scans of all MS levels.
Returns
  • list of int: List of scan numbers within the specified time range(s) and MS level.
Examples

Get MS1 scans between 1.0 and 2.0 minutes:

>>> scans = parser.get_scans_in_time_range((1.0, 2.0), ms_level=1)

Get scans in multiple time windows:

>>> scans = parser.get_scans_in_time_range([(0.5, 1.5), (3.0, 4.0)])
def get_instrument_info(self):
692    def get_instrument_info(self):
693        """
694        Return instrument information.
695
696        Returns
697        -------
698        dict
699            A dictionary with the keys 'model' and 'serial_number'.
700        """
701        # Load the pymzml data
702        data = self.load()
703        instrument_info = data.info.get("referenceable_param_group_list_element")[0]
704        cv_params = instrument_info.findall("{http://psi.hupo.org/ms/mzml}cvParam")
705
706        # Extract details from each cvParam
707        params = []
708        for param in cv_params:
709            accession = param.get("accession")  # Get 'accession' attribute
710            name = param.get("name")  # Get 'name' attribute
711            value = param.get("value")  # Get 'value' attribute
712            params.append({"accession": accession, "name": name, "value": value})
713
714        # Loop through params and try to find the relevant information
715        instrument_dict = {"model": "Unknown", "serial_number": "Unknown"}
716
717        # Assuming there are only two paramters here - one is for the serial number (agnostic to the model) and the other is for the model
718        # If there are more than two, we raise an error
719        if len(params) < 2:
720            raise ValueError(
721                "Not enough parameters found in the instrument info, cannot parse."
722            )
723        if len(params) > 2:
724            raise ValueError(
725                "Too many parameters found in the instrument info, cannot parse."
726            )
727        for param in params:
728            if param["accession"] == "MS:1000529":
729                instrument_dict["serial_number"] = param["value"]
730            else:
731                instrument_dict["model"] = data.OT[param["accession"]]
732
733        return instrument_dict

Return instrument information.

Returns
  • dict: A dictionary with the keys 'model' and 'serial_number'.
def get_creation_time(self) -> datetime.datetime:
735    def get_creation_time(self) -> datetime.datetime:
736        """
737        Return the creation time of the mzML file.
738        """
739        data = self.load()
740        write_time = data.info.get("start_time")
741        if write_time:
742            # Convert the write time to a datetime object
743            return datetime.datetime.strptime(write_time, "%Y-%m-%dT%H:%M:%SZ")
744        else:
745            raise ValueError(
746                "Creation time is not available in the mzML file. "
747                "Please ensure the file contains the 'start_time' information."
748            )

Return the creation time of the mzML file.