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