corems.mass_spectra.input.rawFileReader

   1__author__ = "Yuri E. Corilo"
   2__date__ = "Jun 09, 2021"
   3
   4
   5from warnings import warn
   6import warnings
   7from collections import defaultdict
   8
   9from matplotlib import axes
  10from corems.encapsulation.factory.processingSetting import LiquidChromatographSetting
  11
  12import numpy as np
  13import sys
  14import site
  15from pathlib import Path
  16import datetime
  17import importlib.util
  18import os
  19
  20import clr
  21import pandas as pd
  22from s3path import S3Path
  23
  24
  25from typing import Any, Dict, List, Optional, Tuple, Union
  26from corems.encapsulation.constant import Labels
  27from corems.mass_spectra.factory.lc_class import MassSpectraBase, LCMSBase
  28from corems.mass_spectra.factory.chromat_data import EIC_Data, TIC_Data
  29from corems.mass_spectrum.factory.MassSpectrumClasses import (
  30    MassSpecProfile,
  31    MassSpecCentroid,
  32)
  33from corems.encapsulation.factory.parameters import LCMSParameters, default_parameters
  34from corems.mass_spectra.input.parserbase import SpectraParserInterface
  35
  36# Add the path of the Thermo .NET libraries to the system path
  37spec = importlib.util.find_spec("corems")
  38sys.path.append(str(Path(os.path.dirname(spec.origin)).parent) + "/ext_lib/dotnet/")
  39
  40clr.AddReference("ThermoFisher.CommonCore.RawFileReader")
  41clr.AddReference("ThermoFisher.CommonCore.Data")
  42clr.AddReference("ThermoFisher.CommonCore.MassPrecisionEstimator")
  43
  44from System.Collections.Generic import List as DotNetList
  45from ThermoFisher.CommonCore.RawFileReader import RawFileReaderAdapter
  46from ThermoFisher.CommonCore.Data import ToleranceUnits, Extensions
  47from ThermoFisher.CommonCore.Data.Business import (
  48    ChromatogramTraceSettings,
  49    TraceType,
  50    MassOptions,
  51)
  52from ThermoFisher.CommonCore.Data.Business import ChromatogramSignal, Range
  53from ThermoFisher.CommonCore.Data.Business import Device
  54from ThermoFisher.CommonCore.Data.Interfaces import IChromatogramSettings
  55from ThermoFisher.CommonCore.Data.Business import MassOptions, FileHeaderReaderFactory
  56from ThermoFisher.CommonCore.Data.Business import Device
  57from ThermoFisher.CommonCore.Data.FilterEnums import MSOrderType
  58
  59
  60class ThermoBaseClass:
  61    """Class for parsing Thermo Raw files and extracting information from them.
  62
  63    Parameters:
  64    -----------
  65    file_location : str or pathlib.Path or s3path.S3Path
  66        Thermo Raw file path or S3 path.
  67
  68    Attributes:
  69    -----------
  70    file_path : str or pathlib.Path or s3path.S3Path
  71        The file path of the Thermo Raw file.
  72    parameters : LCMSParameters
  73        The LCMS parameters for the Thermo Raw file.
  74    chromatogram_settings : LiquidChromatographSetting
  75        The chromatogram settings for the Thermo Raw file.
  76    scans : list or tuple
  77        The selected scans for the Thermo Raw file.
  78    start_scan : int
  79        The starting scan number for the Thermo Raw file.
  80    end_scan : int
  81        The ending scan number for the Thermo Raw file.
  82
  83    Methods:
  84    --------
  85    * set_msordertype(scanFilter, mstype: str = 'ms1') -> scanFilter
  86        Convert the user-passed MS Type string to a Thermo MSOrderType object.
  87    * get_instrument_info() -> dict
  88        Get the instrument information from the Thermo Raw file.
  89    * get_creation_time() -> datetime.datetime
  90        Extract the creation date stamp from the .RAW file and return it as a formatted datetime object.
  91    * remove_temp_file()
  92        Remove the temporary file if the path is from S3Path.
  93    * get_polarity_mode(scan_number: int) -> int
  94        Get the polarity mode for the given scan number.
  95    * get_filter_for_scan_num(scan_number: int) -> List[str]
  96        Get the filter for the given scan number.
  97    * check_full_scan(scan_number: int) -> bool
  98        Check if the given scan number is a full scan.
  99    * get_all_filters() -> Tuple[Dict[int, str], List[str]]
 100        Get all scan filters for the Thermo Raw file.
 101    * get_scan_header(scan: int) -> Dict[str, Any]
 102        Get the full dictionary of scan header metadata for the given scan number.
 103    * get_rt_time_from_trace(trace) -> Tuple[List[float], List[float], List[int]]
 104        Get the retention time, intensity, and scan number from the given trace.
 105    * get_eics(target_mzs: List[float], tic_data: Dict[str, Any], ms_type: str = 'MS !d',
 106             peak_detection: bool = True, smooth: bool = True, plot: bool = False,
 107             ax: Optional[matplotlib.axes.Axes] = None, legend: bool = False) -> Tuple[Dict[float, EIC_Data], matplotlib.axes.Axes]
 108        Get the extracted ion chromatograms (EICs) for the target m/z values.
 109
 110    """
 111
 112    def __init__(self, file_location):
 113        """file_location: srt pathlib.Path or s3path.S3Path
 114        Thermo Raw file path
 115        """
 116        # Thread.__init__(self)
 117        if isinstance(file_location, str):
 118            file_path = Path(file_location)
 119
 120        elif isinstance(file_location, S3Path):
 121            temp_dir = Path("tmp/")
 122            temp_dir.mkdir(exist_ok=True)
 123
 124            file_path = temp_dir / file_location.name
 125            with open(file_path, "wb") as fh:
 126                fh.write(file_location.read_bytes())
 127
 128        else:
 129            file_path = file_location
 130
 131        self.iRawDataPlus = RawFileReaderAdapter.FileFactory(str(file_path))
 132
 133        if not self.iRawDataPlus.IsOpen:
 134            raise FileNotFoundError(
 135                "Unable to access the RAW file using the RawFileReader class!"
 136            )
 137
 138        # Check for any errors in the RAW file
 139        if self.iRawDataPlus.IsError:
 140            raise IOError(
 141                "Error opening ({}) - {}".format(self.iRawDataPlus.FileError, file_path)
 142            )
 143
 144        self.res = self.iRawDataPlus.SelectInstrument(Device.MS, 1)
 145
 146        self.file_path = file_location
 147        self.iFileHeader = FileHeaderReaderFactory.ReadFile(str(file_path))
 148
 149        # removing tmp file
 150
 151        self._init_settings()
 152
 153    def _init_settings(self):
 154        """
 155        Initialize the LCMSParameters object.
 156        """
 157        self._parameters = LCMSParameters()
 158
 159    @property
 160    def parameters(self) -> LCMSParameters:
 161        """
 162        Get or set the LCMSParameters object.
 163        """
 164        return self._parameters
 165
 166    @parameters.setter
 167    def parameters(self, instance_LCMSParameters: LCMSParameters):
 168        self._parameters = instance_LCMSParameters
 169
 170    @property
 171    def chromatogram_settings(self) -> LiquidChromatographSetting:
 172        """
 173        Get or set the LiquidChromatographSetting object.
 174        """
 175        return self.parameters.lc_ms
 176
 177    @chromatogram_settings.setter
 178    def chromatogram_settings(
 179        self, instance_LiquidChromatographSetting: LiquidChromatographSetting
 180    ):
 181        self.parameters.lc_ms = instance_LiquidChromatographSetting
 182
 183    @property
 184    def scans(self) -> list | tuple:
 185        """scans : list or tuple
 186        If list uses Thermo AverageScansInScanRange for selected scans, ortherwise uses Thermo AverageScans for a scan range
 187        """
 188        return self.chromatogram_settings.scans
 189
 190    @property
 191    def start_scan(self) -> int:
 192        """
 193        Get the starting scan number for the Thermo Raw file.
 194        """
 195        if self.scans[0] == -1:
 196            return self.iRawDataPlus.RunHeaderEx.FirstSpectrum
 197        else:
 198            return self.scans[0]
 199
 200    @property
 201    def end_scan(self) -> int:
 202        """
 203        Get the ending scan number for the Thermo Raw file.
 204        """
 205        if self.scans[-1] == -1:
 206            return self.iRawDataPlus.RunHeaderEx.LastSpectrum
 207        else:
 208            return self.scans[-1]
 209
 210    def set_msordertype(self, scanFilter, mstype: str = "ms1"):
 211        """
 212        Function to convert user passed string MS Type to Thermo MSOrderType object
 213        Limited to MS1 through MS10.
 214
 215        Parameters:
 216        -----------
 217        scanFilter : Thermo.ScanFilter
 218            The scan filter object.
 219        mstype : str, optional
 220            The MS Type string, by default 'ms1'
 221
 222        """
 223        mstype = mstype.upper()
 224        # Check that a valid mstype is passed
 225        if (int(mstype.split("MS")[1]) > 10) or (int(mstype.split("MS")[1]) < 1):
 226            warn("MS Type not valid, must be between MS1 and MS10")
 227
 228        msordertypedict = {
 229            "MS1": MSOrderType.Ms,
 230            "MS2": MSOrderType.Ms2,
 231            "MS3": MSOrderType.Ms3,
 232            "MS4": MSOrderType.Ms4,
 233            "MS5": MSOrderType.Ms5,
 234            "MS6": MSOrderType.Ms6,
 235            "MS7": MSOrderType.Ms7,
 236            "MS8": MSOrderType.Ms8,
 237            "MS9": MSOrderType.Ms9,
 238            "MS10": MSOrderType.Ms10,
 239        }
 240        scanFilter.MSOrder = msordertypedict[mstype]
 241        return scanFilter
 242
 243    def get_instrument_info(self) -> dict:
 244        """
 245        Get the instrument information from the Thermo Raw file.
 246
 247        Returns:
 248        --------
 249        dict
 250            A dictionary with the keys 'model', and 'serial_number'.
 251        """
 252        instrumentData = self.iRawDataPlus.GetInstrumentData()
 253        return {
 254            "model": instrumentData.Model,
 255            "serial_number": instrumentData.SerialNumber
 256        }
 257    
 258    def get_creation_time(self) -> datetime.datetime:
 259        """
 260        Extract the creation date stamp from the .RAW file
 261        Return formatted creation date stamp.
 262
 263        """
 264        credate = self.iRawDataPlus.CreationDate.get_Ticks()
 265        credate = datetime.datetime(1, 1, 1) + datetime.timedelta(
 266            microseconds=credate / 10
 267        )
 268        return credate
 269
 270    def remove_temp_file(self) -> None:
 271        """if the path is from S3Path data cannot be serialized to io.ByteStream and
 272        a temporary copy is stored at the temp dir
 273        use this function only at the end of your execution scrip
 274        some LCMS class methods depend on this file
 275        """
 276
 277        self.file_path.unlink()
 278
 279    def close_file(self) -> None:
 280        """
 281        Close the Thermo Raw file.
 282        """
 283        self.iRawDataPlus.Dispose()
 284
 285    def get_polarity_mode(self, scan_number: int) -> int:
 286        """
 287        Get the polarity mode for the given scan number.
 288
 289        Parameters:
 290        -----------
 291        scan_number : int
 292            The scan number.
 293
 294        Raises:
 295        -------
 296        Exception
 297            If the polarity mode is unknown.
 298
 299        """
 300        polarity_symbol = self.get_filter_for_scan_num(scan_number)[1]
 301
 302        if polarity_symbol == "+":
 303            return 1
 304            # return 'POSITIVE_ION_MODE'
 305
 306        elif polarity_symbol == "-":
 307            return -1
 308
 309        else:
 310            raise Exception("Polarity Mode Unknown, please set it manually")
 311
 312    def get_filter_for_scan_num(self, scan_number: int) -> List[str]:
 313        """
 314        Returns the closest matching run time that corresponds to scan_number for the current
 315        controller. This function is only supported for MS device controllers.
 316        e.g.  ['FTMS', '-', 'p', 'NSI', 'Full', 'ms', '[200.00-1000.00]']
 317
 318        Parameters:
 319        -----------
 320        scan_number : int
 321            The scan number.
 322
 323        """
 324        scan_label = self.iRawDataPlus.GetScanEventStringForScanNumber(scan_number)
 325
 326        return str(scan_label).split()
 327
 328    def get_ms_level_for_scan_num(self, scan_number: int) -> str:
 329        """
 330        Get the MS order for the given scan number.
 331
 332        Parameters:
 333        -----------
 334        scan_number : int
 335            The scan number
 336
 337        Returns:
 338        --------
 339        int
 340            The MS order type (1 for MS, 2 for MS2, etc.)
 341        """
 342        scan_filter = self.iRawDataPlus.GetFilterForScanNumber(scan_number)
 343
 344        msordertype = {
 345            MSOrderType.Ms: 1,
 346            MSOrderType.Ms2: 2,
 347            MSOrderType.Ms3: 3,
 348            MSOrderType.Ms4: 4,
 349            MSOrderType.Ms5: 5,
 350            MSOrderType.Ms6: 6,
 351            MSOrderType.Ms7: 7,
 352            MSOrderType.Ms8: 8,
 353            MSOrderType.Ms9: 9,
 354            MSOrderType.Ms10: 10,
 355        }
 356
 357        if scan_filter.MSOrder in msordertype:
 358            return msordertype[scan_filter.MSOrder]
 359        else:
 360            raise Exception("MS Order Type not found")
 361    
 362    def check_full_scan(self, scan_number: int) -> bool:
 363        # scan_filter.ScanMode 0 = FULL
 364        scan_filter = self.iRawDataPlus.GetFilterForScanNumber(scan_number)
 365
 366        return scan_filter.ScanMode == MSOrderType.Ms
 367
 368    def get_all_filters(self) -> Tuple[Dict[int, str], List[str]]:
 369        """
 370        Get all scan filters.
 371        This function is only supported for MS device controllers.
 372        e.g.  ['FTMS', '-', 'p', 'NSI', 'Full', 'ms', '[200.00-1000.00]']
 373
 374        """
 375
 376        scanrange = range(self.start_scan, self.end_scan + 1)
 377        scanfiltersdic = {}
 378        scanfilterslist = []
 379        for scan_number in scanrange:
 380            scan_label = self.iRawDataPlus.GetScanEventStringForScanNumber(scan_number)
 381            scanfiltersdic[scan_number] = scan_label
 382            scanfilterslist.append(scan_label)
 383        scanfilterset = list(set(scanfilterslist))
 384        return scanfiltersdic, scanfilterset
 385
 386    def get_scan_header(self, scan: int) -> Dict[str, Any]:
 387        """
 388        Get full dictionary of scan header meta data, i.e. AGC status, ion injection time, etc.
 389
 390        Parameters:
 391        -----------
 392        scan : int
 393            The scan number.
 394
 395        """
 396        header = self.iRawDataPlus.GetTrailerExtraInformation(scan)
 397
 398        header_dic = {}
 399        for i in range(header.Length):
 400            header_dic.update({header.Labels[i]: header.Values[i]})
 401        return header_dic
 402
 403    @staticmethod
 404    def get_rt_time_from_trace(trace) -> Tuple[List[float], List[float], List[int]]:
 405        """trace: ThermoFisher.CommonCore.Data.Business.ChromatogramSignal"""
 406        return list(trace.Times), list(trace.Intensities), list(trace.Scans)
 407
 408    def get_eics(
 409        self,
 410        target_mzs: List[float],
 411        tic_data: Dict[str, Any],
 412        ms_type="MS !d",
 413        peak_detection=False,
 414        smooth=False,
 415        plot=False,
 416        ax: Optional[axes.Axes] = None,
 417        legend=False,
 418    ) -> Tuple[Dict[float, EIC_Data], axes.Axes]:
 419        """ms_type: str ('MS', MS2')
 420        start_scan: int default -1 will select the lowest available
 421        end_scan: int default -1 will select the highest available
 422
 423        returns:
 424
 425            chroma: dict{target_mz: EIC_Data(
 426                                        Scans: [int]
 427                                            original thermo scan numbers
 428                                        Time: [floats]
 429                                            list of retention times
 430                                        TIC: [floats]
 431                                            total ion chromatogram
 432                                        Apexes: [int]
 433                                            original thermo apex scan number after peak picking
 434                                        )
 435
 436        """
 437        # If peak_detection or smooth is True, raise exception
 438        if peak_detection or smooth:
 439            raise Exception("Peak detection and smoothing are no longer implemented in this function")
 440
 441        options = MassOptions()
 442        options.ToleranceUnits = ToleranceUnits.ppm
 443        options.Tolerance = self.chromatogram_settings.eic_tolerance_ppm
 444
 445        all_chroma_settings = []
 446
 447        for target_mz in target_mzs:
 448            settings = ChromatogramTraceSettings(TraceType.MassRange)
 449            settings.Filter = ms_type
 450            settings.MassRanges = [Range(target_mz, target_mz)]
 451
 452            chroma_settings = IChromatogramSettings(settings)
 453
 454            all_chroma_settings.append(chroma_settings)
 455
 456        # chroma_settings2 = IChromatogramSettings(settings)
 457        # print(chroma_settings.FragmentMass)
 458        # print(chroma_settings.FragmentMass)
 459        # print(chroma_settings)
 460        # print(chroma_settings)
 461
 462        data = self.iRawDataPlus.GetChromatogramData(
 463            all_chroma_settings, self.start_scan, self.end_scan, options
 464        )
 465
 466        traces = ChromatogramSignal.FromChromatogramData(data)
 467
 468        chroma = {}
 469
 470        if plot:
 471            from matplotlib.transforms import Bbox
 472            import matplotlib.pyplot as plt
 473
 474            if not ax:
 475                # ax = plt.gca()
 476                # ax.clear()
 477                fig, ax = plt.subplots()
 478
 479            else:
 480                fig = plt.gcf()
 481
 482            # plt.show()
 483
 484        for i, trace in enumerate(traces):
 485            if trace.Length > 0:
 486                rt, eic, scans = self.get_rt_time_from_trace(trace)
 487                if smooth:
 488                    eic = self.smooth_tic(eic)
 489
 490                chroma[target_mzs[i]] = EIC_Data(scans=scans, time=rt, eic=eic)
 491                if plot:
 492                    ax.plot(rt, eic, label="{:.5f}".format(target_mzs[i]))
 493
 494        if peak_detection:
 495            # max_eic = self.get_max_eic(chroma)
 496            max_signal = max(tic_data.tic)
 497
 498            for eic_data in chroma.values():
 499                eic = eic_data.eic
 500                time = eic_data.time
 501
 502                if len(eic) != len(tic_data.tic):
 503                    warn(
 504                        "The software assumes same lenth of TIC and EIC, this does not seems to be the case and the results mass spectrum selected by the scan number might not be correct"
 505                    )
 506
 507                if eic.max() > 0:
 508                    centroid_eics = self.eic_centroid_detector(time, eic, max_signal)
 509                    eic_data.apexes = [i for i in centroid_eics]
 510
 511                    if plot:
 512                        for peak_indexes in eic_data.apexes:
 513                            apex_index = peak_indexes[1]
 514                            ax.plot(
 515                                time[apex_index],
 516                                eic[apex_index],
 517                                marker="x",
 518                                linewidth=0,
 519                            )
 520
 521        if plot:
 522            ax.set_xlabel("Time (min)")
 523            ax.set_ylabel("a.u.")
 524            ax.set_title(ms_type + " EIC")
 525            ax.tick_params(axis="both", which="major", labelsize=12)
 526            ax.axes.spines["top"].set_visible(False)
 527            ax.axes.spines["right"].set_visible(False)
 528
 529            if legend:
 530                legend = ax.legend(loc="upper left", bbox_to_anchor=(1.02, 0, 0.07, 1))
 531                fig.subplots_adjust(right=0.76)
 532                # ax.set_prop_cycle(color=plt.cm.gist_rainbow(np.linspace(0, 1, len(traces))))
 533
 534                d = {"down": 30, "up": -30}
 535
 536                def func(evt):
 537                    if legend.contains(evt):
 538                        bbox = legend.get_bbox_to_anchor()
 539                        bbox = Bbox.from_bounds(
 540                            bbox.x0, bbox.y0 + d[evt.button], bbox.width, bbox.height
 541                        )
 542                        tr = legend.axes.transAxes.inverted()
 543                        legend.set_bbox_to_anchor(bbox.transformed(tr))
 544                        fig.canvas.draw_idle()
 545
 546                fig.canvas.mpl_connect("scroll_event", func)
 547            return chroma, ax
 548        else:
 549            return chroma, None
 550            rt = []
 551            tic = []
 552            scans = []
 553            for i in range(traces[0].Length):
 554                # print(trace[0].HasBasePeakData,trace[0].EndTime )
 555
 556                # print("  {} - {}, {}".format( i, trace[0].Times[i], trace[0].Intensities[i] ))
 557                rt.append(traces[0].Times[i])
 558                tic.append(traces[0].Intensities[i])
 559                scans.append(traces[0].Scans[i])
 560
 561            return traces
 562            # plot_chroma(rt, tic)
 563            # plt.show()
 564
 565    def get_tic(
 566        self,
 567        ms_type="MS !d",
 568        peak_detection=False,  # This wont work right now
 569        smooth=False,  # This wont work right now
 570        plot=False,
 571        ax=None,
 572        trace_type="TIC",
 573    ) -> Tuple[TIC_Data, axes.Axes]:
 574        """ms_type: str ('MS !d', 'MS2', None)
 575            if you use None you get all scans.
 576        peak_detection: bool
 577        smooth: bool
 578        plot: bool
 579        ax: matplotlib axis object
 580        trace_type: str ('TIC','BPC')
 581
 582        returns:
 583            chroma: dict
 584            {
 585            Scan: [int]
 586                original thermo scan numberMS
 587            Time: [floats]
 588                list of retention times
 589            TIC: [floats]
 590                total ion chromatogram
 591            Apexes: [int]
 592                original thermo apex scan number after peak picking
 593            }
 594        """
 595        # If peak_detection or smooth is True, raise exception
 596        if peak_detection or smooth:
 597            raise Exception("Peak detection and smoothing are no longer implemented in this function")
 598
 599        if trace_type == "TIC":
 600            settings = ChromatogramTraceSettings(TraceType.TIC)
 601        elif trace_type == "BPC":
 602            settings = ChromatogramTraceSettings(TraceType.BasePeak)
 603        else:
 604            raise ValueError(f"{trace_type} undefined")
 605        if ms_type == "all":
 606            settings.Filter = None
 607        else:
 608            settings.Filter = ms_type
 609
 610        chroma_settings = IChromatogramSettings(settings)
 611
 612        data = self.iRawDataPlus.GetChromatogramData(
 613            [chroma_settings], self.start_scan, self.end_scan
 614        )
 615
 616        trace = ChromatogramSignal.FromChromatogramData(data)
 617
 618        data = TIC_Data(time=[], scans=[], tic=[], bpc=[], apexes=[])
 619
 620        if trace[0].Length > 0:
 621            for i in range(trace[0].Length):
 622                # print(trace[0].HasBasePeakData,trace[0].EndTime )
 623
 624                # print("  {} - {}, {}".format( i, trace[0].Times[i], trace[0].Intensities[i] ))
 625                data.time.append(trace[0].Times[i])
 626                data.tic.append(trace[0].Intensities[i])
 627                data.scans.append(trace[0].Scans[i])
 628
 629                # print(trace[0].Scans[i])
 630            if smooth:
 631                data.tic = self.smooth_tic(data.tic)
 632
 633            else:
 634                data.tic = np.array(data.tic)
 635
 636            if peak_detection:
 637                centroid_peak_indexes = [
 638                    i for i in self.centroid_detector(data.time, data.tic)
 639                ]
 640
 641                data.apexes = centroid_peak_indexes
 642
 643            if plot:
 644                if not ax:
 645                    import matplotlib.pyplot as plt
 646
 647                    ax = plt.gca()
 648                    # fig, ax = plt.subplots(figsize=(6, 3))
 649
 650                ax.plot(data.time, data.tic, label=trace_type)
 651                ax.set_xlabel("Time (min)")
 652                ax.set_ylabel("a.u.")
 653                if peak_detection:
 654                    for peak_indexes in data.apexes:
 655                        apex_index = peak_indexes[1]
 656                        ax.plot(
 657                            data.time[apex_index],
 658                            data.tic[apex_index],
 659                            marker="x",
 660                            linewidth=0,
 661                        )
 662
 663                # plt.show()
 664                if trace_type == "BPC":
 665                    data.bpc = data.tic
 666                    data.tic = []
 667                return data, ax
 668            if trace_type == "BPC":
 669                data.bpc = data.tic
 670                data.tic = []
 671            return data, None
 672
 673        else:
 674            return None, None
 675
 676    def get_average_mass_spectrum(
 677        self,
 678        spectrum_mode: str = "profile",
 679        auto_process: bool = True,
 680        ppm_tolerance: float = 5.0,
 681        ms_type: str = "MS1",
 682    ) -> MassSpecProfile | MassSpecCentroid:
 683        """
 684        Averages mass spectra over a scan range using Thermo's AverageScansInScanRange method
 685        or a scan list using Thermo's AverageScans method
 686        spectrum_mode: str
 687            centroid or profile mass spectrum
 688        auto_process: bool
 689            If true performs peak picking, and noise threshold calculation after creation of mass spectrum object
 690        ms_type: str
 691            String of form 'ms1' or 'ms2' or 'MS3' etc. Valid up to MS10.
 692            Internal function converts to Thermo MSOrderType class.
 693
 694        """
 695
 696        def get_profile_mass_spec(averageScan, d_params: dict, auto_process: bool):
 697            mz_list = list(averageScan.SegmentedScan.Positions)
 698            abund_list = list(averageScan.SegmentedScan.Intensities)
 699
 700            data_dict = {
 701                Labels.mz: mz_list,
 702                Labels.abundance: abund_list,
 703            }
 704
 705            return MassSpecProfile(data_dict, d_params, auto_process=auto_process)
 706
 707        def get_centroid_mass_spec(averageScan, d_params: dict):
 708            noise = list(averageScan.centroidScan.Noises)
 709
 710            baselines = list(averageScan.centroidScan.Baselines)
 711
 712            rp = list(averageScan.centroidScan.Resolutions)
 713
 714            magnitude = list(averageScan.centroidScan.Intensities)
 715
 716            mz = list(averageScan.centroidScan.Masses)
 717
 718            array_noise_std = (np.array(noise) - np.array(baselines)) / 3
 719            l_signal_to_noise = np.array(magnitude) / array_noise_std
 720
 721            d_params["baseline_noise"] = np.average(array_noise_std)
 722
 723            d_params["baseline_noise_std"] = np.std(array_noise_std)
 724
 725            data_dict = {
 726                Labels.mz: mz,
 727                Labels.abundance: magnitude,
 728                Labels.rp: rp,
 729                Labels.s2n: list(l_signal_to_noise),
 730            }
 731
 732            mass_spec = MassSpecCentroid(data_dict, d_params, auto_process=False)
 733
 734            return mass_spec
 735
 736        d_params = self.set_metadata(
 737            firstScanNumber=self.start_scan, lastScanNumber=self.end_scan
 738        )
 739
 740        # Create the mass options object that will be used when averaging the scans
 741        options = MassOptions()
 742        options.ToleranceUnits = ToleranceUnits.ppm
 743        options.Tolerance = ppm_tolerance
 744
 745        # Get the scan filter for the first scan.  This scan filter will be used to located
 746        # scans within the given scan range of the same type
 747        scanFilter = self.iRawDataPlus.GetFilterForScanNumber(self.start_scan)
 748
 749        # force it to only look for the MSType
 750        scanFilter = self.set_msordertype(scanFilter, ms_type)
 751
 752        if isinstance(self.scans, tuple):
 753            averageScan = Extensions.AverageScansInScanRange(
 754                self.iRawDataPlus, self.start_scan, self.end_scan, scanFilter, options
 755            )
 756
 757            if averageScan:
 758                if spectrum_mode == "profile":
 759                    mass_spec = get_profile_mass_spec(
 760                        averageScan, d_params, auto_process
 761                    )
 762
 763                    return mass_spec
 764
 765                elif spectrum_mode == "centroid":
 766                    if averageScan.HasCentroidStream:
 767                        mass_spec = get_centroid_mass_spec(averageScan, d_params)
 768
 769                        return mass_spec
 770
 771                    else:
 772                        raise ValueError(
 773                            "No Centroind data available for the selected scans"
 774                        )
 775                else:
 776                    raise ValueError("spectrum_mode must be 'profile' or centroid")
 777            else:
 778                raise ValueError("No data found for the selected scans")
 779
 780        elif isinstance(self.scans, list):
 781            d_params = self.set_metadata(scans_list=self.scans)
 782
 783            scans = DotNetList[int]()
 784            for scan in self.scans:
 785                scans.Add(scan)
 786
 787            averageScan = Extensions.AverageScans(self.iRawDataPlus, scans, options)
 788
 789            if averageScan:
 790                if spectrum_mode == "profile":
 791                    mass_spec = get_profile_mass_spec(
 792                        averageScan, d_params, auto_process
 793                    )
 794
 795                    return mass_spec
 796
 797                elif spectrum_mode == "centroid":
 798                    if averageScan.HasCentroidStream:
 799                        mass_spec = get_centroid_mass_spec(averageScan, d_params)
 800
 801                        return mass_spec
 802
 803                    else:
 804                        raise ValueError(
 805                            "No Centroind data available for the selected scans"
 806                        )
 807
 808                else:
 809                    raise ValueError("spectrum_mode must be 'profile' or centroid")
 810
 811            else:
 812                raise ValueError("No data found for the selected scans")
 813
 814        else:
 815            raise ValueError("scans must be a list intergers or a tuple if integers")
 816
 817    def set_metadata(
 818        self,
 819        firstScanNumber=0,
 820        lastScanNumber=0,
 821        scans_list=False,
 822        label=Labels.thermo_profile,
 823    ):
 824        """
 825        Collect metadata to be ingested in the mass spectrum object
 826
 827        scans_list: list[int] or false
 828        lastScanNumber: int
 829        firstScanNumber: int
 830        """
 831
 832        d_params = default_parameters(self.file_path)
 833
 834        # assumes scans is full scan or reduced profile scan
 835
 836        d_params["label"] = label
 837
 838        if scans_list:
 839            d_params["scan_number"] = scans_list
 840
 841            d_params["polarity"] = self.get_polarity_mode(scans_list[0])
 842
 843        else:
 844            d_params["scan_number"] = "{}-{}".format(firstScanNumber, lastScanNumber)
 845
 846            d_params["polarity"] = self.get_polarity_mode(firstScanNumber)
 847
 848        d_params["analyzer"] = self.iRawDataPlus.GetInstrumentData().Model
 849
 850        d_params["acquisition_time"] = self.get_creation_time()
 851
 852        d_params["instrument_label"] = self.iRawDataPlus.GetInstrumentData().Name
 853
 854        return d_params
 855
 856    def get_instrument_methods(self, parse_strings: bool = True):
 857        """
 858        This function will extract the instrument methods embedded in the raw file
 859
 860        First it will check if there are any instrument methods, if not returning None
 861        Then it will get the total number of instrument methods.
 862        For each method, it will extract the plaintext string of the method and attempt to parse it into a dictionary
 863        If this fails, it will return just the string object.
 864
 865        This has been tested on data from an Orbitrap ID-X with embedded MS and LC methods, but other instrument types may fail.
 866
 867        Parameters:
 868        -----------
 869        parse_strings: bool
 870            If True, will attempt to parse the instrument methods into a dictionary. If False, will return the raw string.
 871
 872        Returns:
 873        --------
 874        List[Dict[str, Any]] or List
 875            A list of dictionaries containing the instrument methods, or a list of strings if parsing fails.
 876        """
 877
 878        if not self.iRawDataPlus.HasInstrumentMethod:
 879            raise ValueError(
 880                "Raw Data file does not have any instrument methods attached"
 881            )
 882            return None
 883        else:
 884
 885            def parse_instrument_method(data):
 886                lines = data.split("\r\n")
 887                method = {}
 888                current_section = None
 889                sub_section = None
 890
 891                for line in lines:
 892                    if not line.strip():  # Skip empty lines
 893                        continue
 894                    if (
 895                        line.startswith("----")
 896                        or line.endswith("Settings")
 897                        or line.endswith("Summary")
 898                        or line.startswith("Experiment")
 899                        or line.startswith("Scan Event")
 900                    ):
 901                        current_section = line.replace("-", "").strip()
 902                        method[current_section] = {}
 903                        sub_section = None
 904                    elif line.startswith("\t"):
 905                        if "\t\t" in line:
 906                            indent_level = line.count("\t")
 907                            key_value = line.strip()
 908
 909                            if indent_level == 2:
 910                                if sub_section:
 911                                    key, value = (
 912                                        key_value.split("=", 1)
 913                                        if "=" in key_value
 914                                        else (key_value, None)
 915                                    )
 916                                    method[current_section][sub_section][
 917                                        key.strip()
 918                                    ] = value.strip() if value else None
 919                            elif indent_level == 3:
 920                                scan_type, key_value = (
 921                                    key_value.split(" ", 1)
 922                                    if " " in key_value
 923                                    else (key_value, None)
 924                                )
 925                                method.setdefault(current_section, {}).setdefault(
 926                                    sub_section, {}
 927                                ).setdefault(scan_type, {})
 928
 929                                if key_value:
 930                                    key, value = (
 931                                        key_value.split("=", 1)
 932                                        if "=" in key_value
 933                                        else (key_value, None)
 934                                    )
 935                                    method[current_section][sub_section][scan_type][
 936                                        key.strip()
 937                                    ] = value.strip() if value else None
 938                        else:
 939                            key_value = line.strip()
 940                            if "=" in key_value:
 941                                key, value = key_value.split("=", 1)
 942                                method.setdefault(current_section, {})[key.strip()] = (
 943                                    value.strip()
 944                                )
 945                            else:
 946                                sub_section = key_value
 947                    else:
 948                        if ":" in line:
 949                            key, value = line.split(":", 1)
 950                            method[current_section][key.strip()] = value.strip()
 951                        else:
 952                            method[current_section][line] = {}
 953
 954                return method
 955
 956            count_instrument_methods = self.iRawDataPlus.InstrumentMethodsCount
 957            # TODO make this code better...
 958            instrument_methods = []
 959            for i in range(count_instrument_methods):
 960                instrument_method_string = self.iRawDataPlus.GetInstrumentMethod(i)
 961                if parse_strings:
 962                    try:
 963                        instrument_method_dict = parse_instrument_method(
 964                            instrument_method_string
 965                        )
 966                    except:  # if it fails for any reason
 967                        instrument_method_dict = instrument_method_string
 968                else:
 969                    instrument_method_dict = instrument_method_string
 970                instrument_methods.append(instrument_method_dict)
 971            return instrument_methods
 972
 973    def get_tune_method(self):
 974        """
 975        This code will extract the tune method from the raw file
 976        It has been tested on data from a Thermo Orbitrap ID-X, Astral and Q-Exactive, but may fail on other instrument types.
 977        It attempts to parse out section headers and sub-sections, but may not work for all instrument types.
 978        It will also not return Labels (keys) where the value is blank
 979
 980        Returns:
 981        --------
 982        Dict[str, Any]
 983            A dictionary containing the tune method information
 984
 985        Raises:
 986        -------
 987        ValueError
 988            If no tune methods are found in the raw file
 989
 990        """
 991        tunemethodcount = self.iRawDataPlus.GetTuneDataCount()
 992        if tunemethodcount == 0:
 993            raise ValueError("No tune methods found in the raw data file")
 994            return None
 995        elif tunemethodcount > 1:
 996            warnings.warn(
 997                "Multiple tune methods found in the raw data file, returning the 1st"
 998            )
 999
1000        header = self.iRawDataPlus.GetTuneData(0)
1001
1002        header_dic = {}
1003        current_section = None
1004
1005        for i in range(header.Length):
1006            label = header.Labels[i]
1007            value = header.Values[i]
1008
1009            # Check for section headers
1010            if "===" in label or (
1011                (value == "" or value is None) and not label.endswith(":")
1012            ):
1013                # This is a section header
1014                section_name = (
1015                    label.replace("=", "").replace(":", "").strip()
1016                )  # Clean the label if it contains '='
1017                header_dic[section_name] = {}
1018                current_section = section_name
1019            else:
1020                if current_section:
1021                    header_dic[current_section][label] = value
1022                else:
1023                    header_dic[label] = value
1024        return header_dic
1025
1026    def get_status_log(self, retention_time: float = 0):
1027        """
1028        This code will extract the status logs from the raw file
1029        It has been tested on data from a Thermo Orbitrap ID-X, Astral and Q-Exactive, but may fail on other instrument types.
1030        It attempts to parse out section headers and sub-sections, but may not work for all instrument types.
1031        It will also not return Labels (keys) where the value is blank
1032
1033        Parameters:
1034        -----------
1035        retention_time: float
1036            The retention time in minutes to extract the status log data from.
1037            Will use the closest retention time found. Default 0.
1038
1039        Returns:
1040        --------
1041        Dict[str, Any]
1042            A dictionary containing the status log information
1043
1044        Raises:
1045        -------
1046        ValueError
1047            If no status logs are found in the raw file
1048
1049        """
1050        tunemethodcount = self.iRawDataPlus.GetStatusLogEntriesCount()
1051        if tunemethodcount == 0:
1052            raise ValueError("No status logs found in the raw data file")
1053            return None
1054
1055        header = self.iRawDataPlus.GetStatusLogForRetentionTime(retention_time)
1056
1057        header_dic = {}
1058        current_section = None
1059
1060        for i in range(header.Length):
1061            label = header.Labels[i]
1062            value = header.Values[i]
1063
1064            # Check for section headers
1065            if "===" in label or (
1066                (value == "" or value is None) and not label.endswith(":")
1067            ):
1068                # This is a section header
1069                section_name = (
1070                    label.replace("=", "").replace(":", "").strip()
1071                )  # Clean the label if it contains '='
1072                header_dic[section_name] = {}
1073                current_section = section_name
1074            else:
1075                if current_section:
1076                    header_dic[current_section][label] = value
1077                else:
1078                    header_dic[label] = value
1079        return header_dic
1080
1081    def get_error_logs(self):
1082        """
1083        This code will extract the error logs from the raw file
1084
1085        Returns:
1086        --------
1087        Dict[float, str]
1088            A dictionary containing the error log information with the retention time as the key
1089
1090        Raises:
1091        -------
1092        ValueError
1093            If no error logs are found in the raw file
1094        """
1095
1096        error_log_count = self.iRawDataPlus.RunHeaderEx.ErrorLogCount
1097        if error_log_count == 0:
1098            raise ValueError("No error logs found in the raw data file")
1099            return None
1100
1101        error_logs = {}
1102
1103        for i in range(error_log_count):
1104            error_log_item = self.iRawDataPlus.GetErrorLogItem(i)
1105            rt = error_log_item.RetentionTime
1106            message = error_log_item.Message
1107            # Use the index `i` as the unique ID key
1108            error_logs[i] = {"rt": rt, "message": message}
1109        return error_logs
1110
1111    def get_sample_information(self):
1112        """
1113        This code will extract the sample information from the raw file
1114
1115        Returns:
1116        --------
1117        Dict[str, Any]
1118            A dictionary containing the sample information
1119            Note that UserText field may not be handled properly and may need further processing
1120        """
1121        sminfo = self.iRawDataPlus.SampleInformation
1122        smdict = {}
1123        smdict["Comment"] = sminfo.Comment
1124        smdict["SampleId"] = sminfo.SampleId
1125        smdict["SampleName"] = sminfo.SampleName
1126        smdict["Vial"] = sminfo.Vial
1127        smdict["InjectionVolume"] = sminfo.InjectionVolume
1128        smdict["Barcode"] = sminfo.Barcode
1129        smdict["BarcodeStatus"] = str(sminfo.BarcodeStatus)
1130        smdict["CalibrationLevel"] = sminfo.CalibrationLevel
1131        smdict["DilutionFactor"] = sminfo.DilutionFactor
1132        smdict["InstrumentMethodFile"] = sminfo.InstrumentMethodFile
1133        smdict["RawFileName"] = sminfo.RawFileName
1134        smdict["CalibrationFile"] = sminfo.CalibrationFile
1135        smdict["IstdAmount"] = sminfo.IstdAmount
1136        smdict["RowNumber"] = sminfo.RowNumber
1137        smdict["Path"] = sminfo.Path
1138        smdict["ProcessingMethodFile"] = sminfo.ProcessingMethodFile
1139        smdict["SampleType"] = str(sminfo.SampleType)
1140        smdict["SampleWeight"] = sminfo.SampleWeight
1141        smdict["UserText"] = {
1142            "UserText": [x for x in sminfo.UserText]
1143        }  # [0] #This may not work - needs debugging with
1144        return smdict
1145
1146    def get_instrument_data(self):
1147        """
1148        This code will extract the instrument data from the raw file
1149
1150        Returns:
1151        --------
1152        Dict[str, Any]
1153            A dictionary containing the instrument data
1154        """
1155        instrument_data = self.iRawDataPlus.GetInstrumentData()
1156        id_dict = {}
1157        id_dict["Name"] = instrument_data.Name
1158        id_dict["Model"] = instrument_data.Model
1159        id_dict["SerialNumber"] = instrument_data.SerialNumber
1160        id_dict["SoftwareVersion"] = instrument_data.SoftwareVersion
1161        id_dict["HardwareVersion"] = instrument_data.HardwareVersion
1162        id_dict["ChannelLabels"] = {
1163            "ChannelLabels": [x for x in instrument_data.ChannelLabels]
1164        }
1165        id_dict["Flags"] = instrument_data.Flags
1166        id_dict["AxisLabelY"] = instrument_data.AxisLabelY
1167        id_dict["AxisLabelX"] = instrument_data.AxisLabelX
1168        return id_dict
1169
1170    def get_centroid_msms_data(self, scan):
1171        """
1172        .. deprecated:: 2.0
1173            This function will be removed in CoreMS 2.0. Please use `get_average_mass_spectrum()` instead for similar functionality.
1174        """
1175
1176        warnings.warn(
1177            "The `get_centroid_msms_data()` is deprecated as of CoreMS 2.0 and will be removed in a future version. "
1178            "Please use `get_average_mass_spectrum()` instead.",
1179            DeprecationWarning,
1180        )
1181
1182        d_params = self.set_metadata(scans_list=[scan], label=Labels.thermo_centroid)
1183
1184        centroidStream = self.iRawDataPlus.GetCentroidStream(scan, False)
1185
1186        noise = list(centroidStream.Noises)
1187
1188        baselines = list(centroidStream.Baselines)
1189
1190        rp = list(centroidStream.Resolutions)
1191
1192        magnitude = list(centroidStream.Intensities)
1193
1194        mz = list(centroidStream.Masses)
1195
1196        # charge = scans_labels[5]
1197        array_noise_std = (np.array(noise) - np.array(baselines)) / 3
1198        l_signal_to_noise = np.array(magnitude) / array_noise_std
1199
1200        d_params["baseline_noise"] = np.average(array_noise_std)
1201
1202        d_params["baseline_noise_std"] = np.std(array_noise_std)
1203
1204        data_dict = {
1205            Labels.mz: mz,
1206            Labels.abundance: magnitude,
1207            Labels.rp: rp,
1208            Labels.s2n: list(l_signal_to_noise),
1209        }
1210
1211        mass_spec = MassSpecCentroid(data_dict, d_params, auto_process=False)
1212        mass_spec.settings.noise_threshold_method = "relative_abundance"
1213        mass_spec.settings.noise_threshold_min_relative_abundance = 1
1214        mass_spec.process_mass_spec()
1215        return mass_spec
1216
1217    def get_average_mass_spectrum_by_scanlist(
1218        self,
1219        scans_list: List[int],
1220        auto_process: bool = True,
1221        ppm_tolerance: float = 5.0,
1222    ) -> MassSpecProfile:
1223        """
1224        Averages selected scans mass spectra using Thermo's AverageScans method
1225        scans_list: list[int]
1226        auto_process: bool
1227            If true performs peak picking, and noise threshold calculation after creation of mass spectrum object
1228        Returns:
1229            MassSpecProfile
1230
1231         .. deprecated:: 2.0
1232        This function will be removed in CoreMS 2.0. Please use `get_average_mass_spectrum()` instead for similar functionality.
1233        """
1234
1235        warnings.warn(
1236            "The `get_average_mass_spectrum_by_scanlist()` is deprecated as of CoreMS 2.0 and will be removed in a future version. "
1237            "Please use `get_average_mass_spectrum()` instead.",
1238            DeprecationWarning,
1239        )
1240
1241        d_params = self.set_metadata(scans_list=scans_list)
1242
1243        # assumes scans is full scan or reduced profile scan
1244
1245        scans = DotNetList[int]()
1246        for scan in scans_list:
1247            scans.Add(scan)
1248
1249        # Create the mass options object that will be used when averaging the scans
1250        options = MassOptions()
1251        options.ToleranceUnits = ToleranceUnits.ppm
1252        options.Tolerance = ppm_tolerance
1253
1254        # Get the scan filter for the first scan.  This scan filter will be used to located
1255        # scans within the given scan range of the same type
1256
1257        averageScan = Extensions.AverageScans(self.iRawDataPlus, scans, options)
1258
1259        len_data = averageScan.SegmentedScan.Positions.Length
1260
1261        mz_list = list(averageScan.SegmentedScan.Positions)
1262        abund_list = list(averageScan.SegmentedScan.Intensities)
1263
1264        data_dict = {
1265            Labels.mz: mz_list,
1266            Labels.abundance: abund_list,
1267        }
1268
1269        mass_spec = MassSpecProfile(data_dict, d_params, auto_process=auto_process)
1270
1271        return mass_spec
1272
1273
1274class ImportMassSpectraThermoMSFileReader(ThermoBaseClass, SpectraParserInterface):
1275    """A class for parsing Thermo RAW mass spectrometry data files and instatiating MassSpectraBase or LCMSBase objects
1276
1277    Parameters
1278    ----------
1279    file_location : str or Path
1280        The path to the RAW file to be parsed.
1281    analyzer : str, optional
1282        The type of mass analyzer used in the instrument. Default is "Unknown".
1283    instrument_label : str, optional
1284        The name of the instrument used to acquire the data. Default is "Unknown".
1285    sample_name : str, optional
1286        The name of the sample being analyzed. If not provided, the stem of the file_location path will be used.
1287
1288    Attributes
1289    ----------
1290    file_location : Path
1291        The path to the RAW file being parsed.
1292    analyzer : str
1293        The type of mass analyzer used in the instrument.
1294    instrument_label : str
1295        The name of the instrument used to acquire the data.
1296    sample_name : str
1297        The name of the sample being analyzed.
1298
1299    Methods
1300    -------
1301    * run(spectra=True).
1302        Parses the RAW file and returns a dictionary of mass spectra dataframes and a scan metadata dataframe.
1303    * get_mass_spectrum_from_scan(scan_number, polarity, auto_process=True)
1304        Parses the RAW file and returns a MassSpecBase object from a single scan.
1305    * get_mass_spectra_obj().
1306        Parses the RAW file and instantiates a MassSpectraBase object.
1307    * get_lcms_obj().
1308        Parses the RAW file and instantiates an LCMSBase object.
1309    * get_icr_transient_times().
1310        Return a list for transient time targets for all scans, or selected scans range
1311
1312    Inherits from ThermoBaseClass and SpectraParserInterface
1313    """
1314
1315    def __init__(
1316        self,
1317        file_location,
1318        analyzer="Unknown",
1319        instrument_label="Unknown",
1320        sample_name=None,
1321    ):
1322        super().__init__(file_location)
1323        if isinstance(file_location, str):
1324            # if obj is a string it defaults to create a Path obj, pass the S3Path if needed
1325            file_location = Path(file_location)
1326        if not file_location.exists():
1327            raise FileExistsError("File does not exist: " + str(file_location))
1328
1329        self.file_location = file_location
1330        self.analyzer = analyzer
1331        self.instrument_label = instrument_label
1332
1333        if sample_name:
1334            self.sample_name = sample_name
1335        else:
1336            self.sample_name = file_location.stem
1337
1338    def load(self):
1339        pass
1340
1341    def get_scans_in_time_range(
1342        self, 
1343        time_range: Union[Tuple[float, float], List[Tuple[float, float]]],
1344        ms_level: Optional[int] = None
1345    ) -> List[int]:
1346        """Return scan numbers within specified retention time range(s).
1347        
1348        Parameters
1349        ----------
1350        time_range : tuple or list of tuples
1351            Retention time range(s) in minutes. Can be:
1352            - Single range: (start_time, end_time)
1353            - Multiple ranges: [(start1, end1), (start2, end2), ...]
1354        ms_level : int, optional
1355            If specified, only return scans of this MS level (e.g., 1 for MS1, 2 for MS2).
1356            If None, returns scans of all MS levels.
1357        
1358        Returns
1359        -------
1360        list of int
1361            List of scan numbers within the specified time range(s) and MS level.
1362        """
1363        # Normalize time range to list of tuples
1364        time_ranges = self._normalize_time_range(time_range)
1365        
1366        # Get all scan data
1367        scan_df = self.get_scan_df()
1368        
1369        # Filter by time range
1370        mask = pd.Series([False] * len(scan_df), index=scan_df.index)
1371        for start_time, end_time in time_ranges:
1372            mask |= (scan_df.scan_time >= start_time) & (scan_df.scan_time <= end_time)
1373        
1374        filtered_df = scan_df[mask]
1375        
1376        # Filter by MS level if specified
1377        if ms_level is not None:
1378            filtered_df = filtered_df[filtered_df.ms_level == ms_level]
1379        
1380        return filtered_df.scan.tolist()
1381
1382    def get_scan_df(self, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None):
1383        """Return scan data as a pandas DataFrame.
1384        
1385        Parameters
1386        ----------
1387        time_range : tuple or list of tuples, optional
1388            Retention time range(s) to filter scans. Can be:
1389            - Single range: (start_time, end_time) in minutes
1390            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
1391            If None, returns all scans.
1392        
1393        Returns
1394        -------
1395        pd.DataFrame
1396            DataFrame containing scan information, optionally filtered by time range.
1397        """
1398        # This automatically brings in all the data
1399        self.chromatogram_settings.scans = (-1, -1)
1400
1401        # Get scan df info; starting with TIC data
1402        tic_data, _ = self.get_tic(ms_type="all", peak_detection=False, smooth=False)
1403        tic_data = {
1404            "scan": tic_data.scans,
1405            "scan_time": tic_data.time,
1406            "tic": tic_data.tic,
1407        }
1408        scan_df = pd.DataFrame.from_dict(tic_data)
1409        scan_df["ms_level"] = None
1410        
1411        # get scan text
1412        scan_filter_df = pd.DataFrame.from_dict(
1413            self.get_all_filters()[0], orient="index"
1414        )
1415        scan_filter_df.reset_index(inplace=True)
1416        scan_filter_df.rename(columns={"index": "scan", 0: "scan_text"}, inplace=True)
1417
1418        scan_df = scan_df.merge(scan_filter_df, on="scan", how="left")
1419        scan_df["scan_window_lower"] = scan_df.scan_text.str.extract(
1420            r"\[(\d+\.\d+)-\d+\.\d+\]"
1421        )
1422        scan_df["scan_window_upper"] = scan_df.scan_text.str.extract(
1423            r"\[\d+\.\d+-(\d+\.\d+)\]"
1424        )
1425        scan_df["polarity"] = np.where(
1426            scan_df.scan_text.str.contains(" - "), "negative", "positive"
1427        )
1428        scan_df["precursor_mz"] = scan_df.scan_text.str.extract(r"(\d+\.\d+)@")
1429        scan_df["precursor_mz"] = scan_df["precursor_mz"].astype(float)
1430
1431        # Assign each scan as centroid or profile and add ms_level
1432        scan_df["ms_format"] = None
1433        for i in scan_df.scan.to_list():
1434            scan_df.loc[scan_df.scan == i, "ms_level"] = self.get_ms_level_for_scan_num(i)
1435            if self.iRawDataPlus.IsCentroidScanFromScanNumber(i):
1436                scan_df.loc[scan_df.scan == i, "ms_format"] = "centroid"
1437            else:
1438                scan_df.loc[scan_df.scan == i, "ms_format"] = "profile"
1439        
1440        # Remove any non-mass spectra scans (e.g., MS level 0 or None)
1441        scan_df = scan_df[scan_df.ms_level.notnull() & (scan_df.ms_level > 0)].reset_index(drop=True)
1442        
1443        # Filter by time range if specified
1444        if time_range is not None:
1445            time_ranges = self._normalize_time_range(time_range)
1446            mask = pd.Series([False] * len(scan_df), index=scan_df.index)
1447            for start_time, end_time in time_ranges:
1448                mask |= (scan_df.scan_time >= start_time) & (scan_df.scan_time <= end_time)
1449            scan_df = scan_df[mask].reset_index(drop=True)
1450
1451        return scan_df
1452
1453    def get_ms_raw(self, spectra, scan_df, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None):
1454        """Return a dictionary of mass spectra data as pandas DataFrames.
1455        
1456        Parameters
1457        ----------
1458        spectra : str
1459            Specifies which spectra to load (e.g., 'all', 'ms1', 'ms2')
1460        scan_df : pd.DataFrame
1461            Scan information DataFrame
1462        time_range : tuple or list of tuples, optional
1463            Retention time range(s) to filter scans. Can be:
1464            - Single range: (start_time, end_time) in minutes
1465            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
1466            If None, returns all scans. Note: filtering is typically done at scan_df level.
1467        
1468        Returns
1469        -------
1470        dict
1471            Dictionary of raw mass spectra data, optionally filtered by time range.
1472        """
1473        # Note: time_range filtering is handled at the scan_df level before calling this method
1474        # The parameter is here for interface consistency with SpectraParserInterface
1475        
1476        if spectra == "all":
1477            scan_df_forspec = scan_df
1478        elif spectra == "ms1":
1479            scan_df_forspec = scan_df[scan_df.ms_level == 1]
1480        elif spectra == "ms2":
1481            scan_df_forspec = scan_df[scan_df.ms_level == 2]
1482        else:
1483            raise ValueError("spectra must be 'none', 'all', 'ms1', or 'ms2'")
1484
1485        # Result container
1486        res = {}
1487
1488        # Row count container
1489        counter = {}
1490
1491        # Column name container
1492        cols = {}
1493
1494        # set at float32
1495        dtype = np.float32
1496
1497        # First pass: get nrows
1498        N = defaultdict(lambda: 0)
1499        for i in scan_df_forspec.scan.to_list():
1500            level = scan_df_forspec.loc[scan_df_forspec.scan == i, "ms_level"].values[0]
1501            scanStatistics = self.iRawDataPlus.GetScanStatsForScanNumber(i)
1502            profileStream = self.iRawDataPlus.GetSegmentedScanFromScanNumber(
1503                i, scanStatistics
1504            )
1505            abun = list(profileStream.Intensities)
1506            abun = np.array(abun)[np.where(np.array(abun) > 0)[0]]
1507
1508            N[level] += len(abun)
1509
1510        # Second pass: parse
1511        for i in scan_df_forspec.scan.to_list():
1512            scanStatistics = self.iRawDataPlus.GetScanStatsForScanNumber(i)
1513            profileStream = self.iRawDataPlus.GetSegmentedScanFromScanNumber(
1514                i, scanStatistics
1515            )
1516            abun = list(profileStream.Intensities)
1517            mz = list(profileStream.Positions)
1518
1519            # Get index of abun that are > 0
1520            inx = np.where(np.array(abun) > 0)[0]
1521            mz = np.array(mz)[inx]
1522            mz = np.float32(mz)
1523            abun = np.array(abun)[inx]
1524            abun = np.float32(abun)
1525
1526            level = scan_df_forspec.loc[scan_df_forspec.scan == i, "ms_level"].values[0]
1527
1528            # Number of rows
1529            n = len(mz)
1530
1531            # No measurements
1532            if n == 0:
1533                continue
1534
1535            # Dimension check
1536            if len(mz) != len(abun):
1537                warnings.warn("m/z and intensity array dimension mismatch")
1538                continue
1539
1540            # Scan/frame info
1541            id_dict = i
1542
1543            # Columns
1544            cols[level] = ["scan", "mz", "intensity"]
1545            m = len(cols[level])
1546
1547            # Subarray init
1548            arr = np.empty((n, m), dtype=dtype)
1549            inx = 0
1550
1551            # Populate scan/frame info
1552            arr[:, inx] = i
1553            inx += 1
1554
1555            # Populate m/z
1556            arr[:, inx] = mz
1557            inx += 1
1558
1559            # Populate intensity
1560            arr[:, inx] = abun
1561            inx += 1
1562
1563            # Initialize output container
1564            if level not in res:
1565                res[level] = np.empty((N[level], m), dtype=dtype)
1566                counter[level] = 0
1567
1568            # Insert subarray
1569            res[level][counter[level] : counter[level] + n, :] = arr
1570            counter[level] += n
1571
1572        # Construct ms1 and ms2 mz dataframes
1573        for level in res.keys():
1574            res[level] = pd.DataFrame(res[level])
1575            res[level].columns = cols[level]
1576        # rename keys in res to add 'ms' prefix
1577        res = {f"ms{key}": value for key, value in res.items()}
1578
1579        return res
1580
1581    def run(self, spectra="all", scan_df=None, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None):
1582        """
1583        Extracts mass spectra data from a raw file.
1584
1585        Parameters
1586        ----------
1587        spectra : str, optional
1588            Which mass spectra data to include in the output. Default is all.  Other options: none, ms1, ms2.
1589        scan_df : pandas.DataFrame, optional
1590            Scan dataframe.  If not provided, the scan dataframe is created from the mzML file.
1591        time_range : tuple or list of tuples, optional
1592            Retention time range(s) to filter scans. Can be:
1593            - Single range: (start_time, end_time) in minutes
1594            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
1595            If None, returns all scans.
1596
1597        Returns
1598        -------
1599        tuple
1600            A tuple containing two elements:
1601            - A dictionary containing mass spectra data, separated by MS level.
1602            - A pandas DataFrame containing scan information, including scan number, scan time, TIC, MS level,
1603                scan text, scan window lower and upper bounds, polarity, and precursor m/z (if applicable).
1604        """
1605        # Prepare scan_df
1606        if scan_df is None:
1607            scan_df = self.get_scan_df(time_range=time_range)
1608
1609        # Prepare mass spectra data
1610        if spectra != "none":
1611            res = self.get_ms_raw(spectra=spectra, scan_df=scan_df, time_range=time_range)
1612        else:
1613            res = None
1614
1615        return res, scan_df
1616
1617    def get_mass_spectra_from_scan_list(
1618        self, scan_list, spectrum_mode, auto_process=True
1619    ):
1620        """Instantiate multiple MassSpecBase objects from a list of scan numbers from the binary file.
1621        
1622        Parameters
1623        ----------
1624        scan_list : list[int]
1625            A list of scan numbers to extract the mass spectra from.
1626        spectrum_mode : str
1627            The type of mass spectrum to extract.  Must be 'profile' or 'centroid'.
1628            All scans in the list must be of the same type.
1629        auto_process : bool, optional
1630            If True, perform peak picking and noise threshold calculation after creating the mass spectrum object. Default is True.
1631        """
1632        mass_spectra = []
1633        for scan in scan_list:
1634            mass_spectrum = self.get_mass_spectrum_from_scan(
1635                scan, spectrum_mode, auto_process=auto_process
1636            )
1637            mass_spectra.append(mass_spectrum)
1638
1639        return mass_spectra
1640    
1641    def get_mass_spectrum_from_scan(
1642        self, scan_number, spectrum_mode, auto_process=True
1643    ):
1644        """Instantiate a MassSpecBase object from a single scan number from the binary file.
1645
1646        Parameters
1647        ----------
1648        scan_number : int
1649            The scan number to extract the mass spectrum from.
1650        polarity : int
1651            The polarity of the scan.  1 for positive mode, -1 for negative mode.
1652        spectrum_mode : str
1653            The type of mass spectrum to extract.  Must be 'profile' or 'centroid'.
1654        auto_process : bool, optional
1655            If True, perform peak picking and noise threshold calculation after creating the mass spectrum object. Default is True.
1656
1657        Returns
1658        -------
1659        MassSpecProfile | MassSpecCentroid
1660            The MassSpecProfile or MassSpecCentroid object containing the parsed mass spectrum.
1661        """
1662
1663        if spectrum_mode == "profile":
1664            scanStatistics = self.iRawDataPlus.GetScanStatsForScanNumber(scan_number)
1665            profileStream = self.iRawDataPlus.GetSegmentedScanFromScanNumber(
1666                scan_number, scanStatistics
1667            )
1668            abun = list(profileStream.Intensities)
1669            mz = list(profileStream.Positions)
1670            data_dict = {
1671                Labels.mz: mz,
1672                Labels.abundance: abun,
1673            }
1674            d_params = self.set_metadata(
1675                firstScanNumber=scan_number,
1676                lastScanNumber=scan_number,
1677                scans_list=False,
1678                label=Labels.thermo_profile,
1679            )
1680            mass_spectrum_obj = MassSpecProfile(
1681                data_dict, d_params, auto_process=auto_process
1682            )
1683
1684        elif spectrum_mode == "centroid":
1685            centroid_scan = self.iRawDataPlus.GetCentroidStream(scan_number, False)
1686            if centroid_scan.Masses is not None:
1687                mz = list(centroid_scan.Masses)
1688                abun = list(centroid_scan.Intensities)
1689                rp = list(centroid_scan.Resolutions)
1690                magnitude = list(centroid_scan.Intensities)
1691                noise = list(centroid_scan.Noises)
1692                baselines = list(centroid_scan.Baselines)
1693                array_noise_std = (np.array(noise) - np.array(baselines)) / 3
1694                l_signal_to_noise = np.array(magnitude) / array_noise_std
1695                data_dict = {
1696                    Labels.mz: mz,
1697                    Labels.abundance: abun,
1698                    Labels.rp: rp,
1699                    Labels.s2n: list(l_signal_to_noise),
1700                }
1701            else:  # For CID MS2, the centroid data are stored in the profile data location, they do not have any associated rp or baseline data, but they should be treated as centroid data
1702                scanStatistics = self.iRawDataPlus.GetScanStatsForScanNumber(
1703                    scan_number
1704                )
1705                profileStream = self.iRawDataPlus.GetSegmentedScanFromScanNumber(
1706                    scan_number, scanStatistics
1707                )
1708                abun = list(profileStream.Intensities)
1709                mz = list(profileStream.Positions)
1710                data_dict = {
1711                    Labels.mz: mz,
1712                    Labels.abundance: abun,
1713                    Labels.rp: [np.nan] * len(mz),
1714                    Labels.s2n: [np.nan] * len(mz),
1715                }
1716            d_params = self.set_metadata(
1717                firstScanNumber=scan_number,
1718                lastScanNumber=scan_number,
1719                scans_list=False,
1720                label=Labels.thermo_centroid,
1721            )
1722            mass_spectrum_obj = MassSpecCentroid(
1723                data_dict, d_params, auto_process=auto_process
1724            )
1725
1726        return mass_spectrum_obj
1727
1728    def get_mass_spectra_obj(self, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None):
1729        """Instatiate a MassSpectraBase object from the binary data file file.
1730
1731        Parameters
1732        ----------
1733        time_range : tuple or list of tuples, optional
1734            Retention time range(s) to load. Can be:
1735            - Single range: (start_time, end_time) in minutes
1736            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
1737            If None, loads all scans. Useful for targeted workflows to improve performance.
1738
1739        Returns
1740        -------
1741        MassSpectraBase
1742            The MassSpectra object containing the parsed mass spectra.  The object is instatiated with the mzML file, analyzer, instrument, sample name, and scan dataframe.
1743        """
1744        _, scan_df = self.run(spectra="none", time_range=time_range)
1745        mass_spectra_obj = MassSpectraBase(
1746            self.file_location,
1747            self.analyzer,
1748            self.instrument_label,
1749            self.sample_name,
1750            self,
1751        )
1752        scan_df = scan_df.set_index("scan", drop=False)
1753        mass_spectra_obj.scan_df = scan_df
1754
1755        return mass_spectra_obj
1756
1757    def get_lcms_obj(self, spectra="all", time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None):
1758        """Instatiates a LCMSBase object from the mzML file.
1759
1760        Parameters
1761        ----------
1762        spectra : str, optional
1763            Which mass spectra data to include in the output. Default is "all".  Other options: "none", "ms1", "ms2".
1764        time_range : tuple or list of tuples, optional
1765            Retention time range(s) to load. Can be:
1766            - Single range: (start_time, end_time) in minutes
1767            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
1768            If None, loads all scans. Useful for targeted workflows to improve performance.
1769
1770        Returns
1771        -------
1772        LCMSBase
1773            LCMS object containing mass spectra data. The object is instatiated with the file location, analyzer, instrument, sample name, scan info, mz dataframe (as specifified), polarity, as well as the attributes holding the scans, retention times, and tics.
1774        """
1775        _, scan_df = self.run(spectra="none", time_range=time_range)  # first run it to just get scan info
1776        res, scan_df = self.run(
1777            scan_df=scan_df, spectra=spectra, time_range=time_range
1778        )  # second run to parse data
1779        lcms_obj = LCMSBase(
1780            self.file_location,
1781            self.analyzer,
1782            self.instrument_label,
1783            self.sample_name,
1784            self,
1785        )
1786        if spectra != "none":
1787            for key in res:
1788                key_int = int(key.replace("ms", ""))
1789                res[key] = res[key][res[key].intensity > 0]
1790                res[key] = (
1791                    res[key].sort_values(by=["scan", "mz"]).reset_index(drop=True)
1792                )
1793                lcms_obj._ms_unprocessed[key_int] = res[key]
1794        lcms_obj.scan_df = scan_df.set_index("scan", drop=False)
1795        # Check if polarity is mixed
1796        if len(set(scan_df.polarity)) > 1:
1797            raise ValueError("Mixed polarities detected in scan data")
1798        lcms_obj.polarity = scan_df.polarity.iloc[0]
1799        lcms_obj._scans_number_list = list(scan_df.scan)
1800        lcms_obj._retention_time_list = list(scan_df.scan_time)
1801        lcms_obj._tic_list = list(scan_df.tic)
1802
1803        return lcms_obj
1804
1805    def get_icr_transient_times(self):
1806        """Return a list for transient time targets for all scans, or selected scans range
1807
1808        Notes
1809        --------
1810        Resolving Power and Transient time targets based on 7T FT-ICR MS system
1811        """
1812
1813        res_trans_time = {
1814            "50": 0.384,
1815            "100000": 0.768,
1816            "200000": 1.536,
1817            "400000": 3.072,
1818            "750000": 6.144,
1819            "1000000": 12.288,
1820        }
1821
1822        firstScanNumber = self.start_scan
1823
1824        lastScanNumber = self.end_scan
1825
1826        transient_time_list = []
1827
1828        for scan in range(firstScanNumber, lastScanNumber):
1829            scan_header = self.get_scan_header(scan)
1830
1831            rp_target = scan_header["FT Resolution:"]
1832
1833            transient_time = res_trans_time.get(rp_target)
1834
1835            transient_time_list.append(transient_time)
1836
1837            # print(transient_time, rp_target)
1838
1839        return transient_time_list