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