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