corems.mass_spectra.input.corems_hdf5
1__author__ = "Yuri E. Corilo" 2__date__ = "Oct 29, 2019" 3 4 5from threading import Thread 6import h5py 7import toml 8import json 9import multiprocessing 10from pathlib import Path 11import datetime 12from typing import Union, Tuple, List, Optional 13 14import numpy as np 15import pandas as pd 16import warnings 17 18from corems.chroma_peak.factory.chroma_peak_classes import LCMSMassFeature 19from corems.encapsulation.input.parameter_from_json import ( 20 load_and_set_json_parameters_lcms, 21 load_and_set_toml_parameters_lcms, 22) 23from corems.mass_spectra.factory.lc_class import LCMSBase, MassSpectraBase, LCMSCollection 24from corems.mass_spectra.factory.chromat_data import EIC_Data 25from corems.mass_spectra.input.parserbase import SpectraParserInterface 26from corems.mass_spectrum.input.coremsHDF5 import ReadCoreMSHDF_MassSpectrum 27from corems.molecular_id.factory.spectrum_search_results import SpectrumSearchResults 28from corems.mass_spectra.input.rawFileReader import ImportMassSpectraThermoMSFileReader 29from corems.mass_spectra.input.mzml import MZMLSpectraParser 30 31 32def create_manifest_from_folder( 33 folder_path: Path, 34 output_path: Path = None, 35 batch_time_threshold_hours: float = 12.0, 36 center_name: str = None, 37 overwrite: bool = False 38) -> Path: 39 """ 40 Create a manifest CSV file for ReadCoreMSHDFMassSpectraCollection from CoreMS HDF5 files. 41 42 Scans a folder for .corems subdirectories and generates a manifest with columns: 43 sample_name, batch, order, center, time. Files are batched by creation time, and 44 one sample is designated as the retention time alignment center. 45 46 Parameters 47 ---------- 48 folder_path : Path 49 Path to folder containing .corems subdirectories with HDF5 files. 50 output_path : Path, optional 51 Output manifest CSV path. Default: folder_path/manifest.csv. 52 batch_time_threshold_hours : float, optional 53 Time gap in hours for batch separation. Default: 12.0. 54 center_name : str, optional 55 Sample name to designate as RT alignment center (must exist in samples). 56 If None, the middle sample (by creation time) is used. 57 overwrite : bool, optional 58 Whether to overwrite existing manifest. Default: False. 59 60 Returns 61 ------- 62 Path 63 Path to created manifest file. 64 65 Raises 66 ------ 67 FileNotFoundError 68 If folder_path doesn't exist or contains no .corems subdirectories. 69 FileExistsError 70 If output file exists and overwrite is False. 71 ValueError 72 If no HDF5 files found, or center_name doesn't match any sample. 73 """ 74 if not folder_path.exists(): 75 raise FileNotFoundError(f"Folder {folder_path} does not exist.") 76 77 # Set default output path if not provided 78 if output_path is None: 79 output_path = folder_path / "manifest.csv" 80 81 # Check if output file exists 82 if output_path.exists() and not overwrite: 83 raise FileExistsError( 84 f"Manifest file {output_path} already exists. " 85 "Set overwrite=True to replace it." 86 ) 87 88 # Find all .corems subdirectories 89 corems_dirs = sorted([d for d in folder_path.iterdir() if d.is_dir() and d.suffix == ".corems"]) 90 91 if not corems_dirs: 92 raise FileNotFoundError( 93 f"No .corems subdirectories found in {folder_path}. " 94 "Ensure the folder contains processed CoreMS data." 95 ) 96 97 # Collect sample information 98 sample_data = [] 99 100 for corems_dir in corems_dirs: 101 sample_name = corems_dir.stem # Remove .corems extension 102 hdf5_file = corems_dir / f"{sample_name}.hdf5" 103 104 if not hdf5_file.exists(): 105 print(f"Warning: HDF5 file not found for {sample_name}, skipping.") 106 continue 107 108 # Get creation time using the ReadCoreMSHDFMassSpectra method 109 try: 110 # Use context manager to ensure file is properly closed 111 with ReadCoreMSHDFMassSpectra(str(hdf5_file)) as parser: 112 # Use the get_original_creation_time() method which checks HDF5 attrs first, 113 # then falls back to original parser if needed 114 creation_time = parser.get_original_creation_time() 115 116 # Skip sample if creation time unavailable 117 if creation_time is None: 118 print(f"Warning: Could not get original creation time for {sample_name}, skipping.") 119 continue 120 121 except Exception as e: 122 print(f"Warning: Error getting creation time for {sample_name}: {e}, skipping.") 123 continue 124 125 sample_data.append({ 126 'sample_name': sample_name, 127 'creation_time': creation_time, 128 'hdf5_path': hdf5_file 129 }) 130 131 if not sample_data: 132 raise ValueError( 133 f"No valid HDF5 files found in {folder_path}. " 134 "Ensure .corems subdirectories contain .hdf5 files." 135 ) 136 137 # Sort by creation time 138 sample_data.sort(key=lambda x: x['creation_time']) 139 140 # Assign batches based on time threshold 141 batch_assignments = [] 142 current_batch = 1 143 144 for i, sample in enumerate(sample_data): 145 if i == 0: 146 batch_assignments.append(current_batch) 147 else: 148 time_diff = sample['creation_time'] - sample_data[i-1]['creation_time'] 149 time_diff_hours = time_diff.total_seconds() / 3600 150 151 if time_diff_hours > batch_time_threshold_hours: 152 current_batch += 1 153 154 batch_assignments.append(current_batch) 155 156 # Determine which sample should be the center for retention time alignment 157 sample_names = [s['sample_name'] for s in sample_data] 158 159 if center_name is not None: 160 # Validate that center_name is in the discovered samples 161 if center_name not in sample_names: 162 raise ValueError( 163 f"Specified center_name '{center_name}' not found in discovered samples. " 164 f"Available samples: {', '.join(sample_names)}" 165 ) 166 center_sample = center_name 167 else: 168 # Use the middle sample (by creation time) as center 169 middle_idx = len(sample_data) // 2 170 center_sample = sample_data[middle_idx]['sample_name'] 171 print(f"Auto-selected center sample: {center_sample} (index {middle_idx} of {len(sample_data)}, middle by creation time)") 172 173 # Create manifest dataframe with center column as TRUE/FALSE 174 manifest_df = pd.DataFrame({ 175 'sample_name': sample_names, 176 'batch': batch_assignments, 177 'order': list(range(1, len(sample_data) + 1)), 178 'center': ['TRUE' if name == center_sample else 'FALSE' for name in sample_names], 179 'time': [s['creation_time'].strftime('%Y-%m-%dT%H:%M:%SZ') for s in sample_data] 180 }) 181 182 # Sort manifest by time before saving to ensure proper order 183 manifest_df = manifest_df.sort_values('time').reset_index(drop=True) 184 # Update order column to reflect sorted order 185 manifest_df['order'] = list(range(1, len(manifest_df) + 1)) 186 187 # Save manifest 188 manifest_df.to_csv(output_path, index=False) 189 190 print(f"Manifest created successfully at {output_path}") 191 print(f"Total samples: {len(sample_data)}") 192 print(f"Number of batches: {current_batch}") 193 print(f"Batch assignments: {dict(zip(range(1, current_batch + 1), [batch_assignments.count(b) for b in range(1, current_batch + 1)]))}") 194 195 return output_path 196 197 198class ReadCoreMSHDFMassSpectra( 199 SpectraParserInterface, ReadCoreMSHDF_MassSpectrum, Thread 200): 201 """Class to read CoreMS HDF5 files and populate a LCMS or MassSpectraBase object. 202 203 Parameters 204 ---------- 205 file_location : str 206 The location of the HDF5 file to read, including the suffix. 207 208 Attributes 209 ---------- 210 file_location : str 211 The location of the HDF5 file to read. 212 h5pydata : h5py.File 213 The HDF5 file object. 214 scans : list 215 A list of the location of individual mass spectra within the HDF5 file. 216 scan_number_list : list 217 A list of the scan numbers of the mass spectra within the HDF5 file. 218 parameters_location : str 219 The location of the parameters file (json or toml). 220 221 Methods 222 ------- 223 * import_mass_spectra(mass_spectra). 224 Imports all mass spectra from the HDF5 file onto the LCMS or MassSpectraBase object. 225 * get_mass_spectrum_from_scan(scan_number). 226 Return mass spectrum data object from scan number. 227 * load(). 228 Placeholder method to meet the requirements of the SpectraParserInterface. 229 * run(mass_spectra). 230 Runs the importer functions to populate a LCMS or MassSpectraBase object. 231 * import_scan_info(mass_spectra). 232 Imports the scan info from the HDF5 file to populate the _scan_info attribute 233 on the LCMS or MassSpectraBase object 234 * import_ms_unprocessed(mass_spectra). 235 Imports the unprocessed mass spectra from the HDF5 file to populate the 236 _ms_unprocessed attribute on the LCMS or MassSpectraBase object 237 * import_parameters(mass_spectra). 238 Imports the parameters from the HDF5 file to populate the parameters 239 attribute on the LCMS or MassSpectraBase object 240 * import_mass_features(mass_spectra). 241 Imports the mass features from the HDF5 file to populate the mass_features 242 attribute on the LCMS or MassSpectraBase object 243 * import_eics(mass_spectra). 244 Imports the extracted ion chromatograms from the HDF5 file to populate the 245 eics attribute on the LCMS or MassSpectraBase object 246 * import_spectral_search_results(mass_spectra). 247 Imports the spectral search results from the HDF5 file to populate the 248 spectral_search_results attribute on the LCMS or MassSpectraBase object 249 * get_mass_spectra_obj(). 250 Return mass spectra data object, populating the _ms list on the LCMS or 251 MassSpectraBase object from the HDF5 file 252 * get_lcms_obj(). 253 Return LCMSBase object, populating the majority of the attributes on the 254 LCMS object from the HDF5 file 255 256 """ 257 258 def __init__(self, file_location: str): 259 Thread.__init__(self) 260 ReadCoreMSHDF_MassSpectrum.__init__(self, file_location) 261 262 # override the scans attribute on ReadCoreMSHDF_MassSpectrum class to expect a nested location within the HDF5 file 263 self.scans = [ 264 "mass_spectra/" + x for x in list(self.h5pydata["mass_spectra"].keys()) 265 ] 266 self.scan_number_list = sorted( 267 [int(float(i)) for i in list(self.h5pydata["mass_spectra"].keys())] 268 ) 269 270 # set the location of the parameters file (json or toml) 271 add_files = [ 272 x 273 for x in self.file_location.parent.glob( 274 self.file_location.name.replace(".hdf5", ".*") 275 ) 276 if x.suffix != ".hdf5" 277 ] 278 if len([x for x in add_files if x.suffix == ".json"]) > 0: 279 self.parameters_location = [x for x in add_files if x.suffix == ".json"][0] 280 elif len([x for x in add_files if x.suffix == ".toml"]) > 0: 281 self.parameters_location = [x for x in add_files if x.suffix == ".toml"][0] 282 else: 283 self.parameters_location = None 284 285 def __enter__(self): 286 """Context manager entry.""" 287 return self 288 289 def __exit__(self, exc_type, exc_val, exc_tb): 290 """Context manager exit - closes the HDF5 file.""" 291 if hasattr(self, 'h5pydata') and self.h5pydata is not None: 292 self.h5pydata.close() 293 return False 294 295 def close(self): 296 """Explicitly close the HDF5 file.""" 297 if hasattr(self, 'h5pydata') and self.h5pydata is not None: 298 self.h5pydata.close() 299 300 def get_mass_spectrum_from_scan(self, scan_number): 301 """Return mass spectrum data object from scan number.""" 302 if scan_number in self.scan_number_list: 303 mass_spec = self.get_mass_spectrum(scan_number) 304 return mass_spec 305 else: 306 raise Exception("Scan number not found in HDF5 file.") 307 308 def get_mass_spectra_from_scan_list( 309 self, scan_list, spectrum_mode, auto_process=True 310 ): 311 """Return a list of mass spectrum data objects from a list of scan numbers. 312 313 Parameters 314 ---------- 315 scan_list : list 316 A list of scan numbers to retrieve mass spectra for. 317 spectrum_mode : str 318 The spectrum mode to use when retrieving the mass spectra. 319 Note that this parameter is not used for CoreMS HDF5 files, as the spectra are already processed and only 320 centroided spectra are saved. 321 auto_process : bool 322 If True, automatically process the mass spectra when retrieving them. 323 Note that this parameter is not used for CoreMS HDF5 files, as the spectra are already processed and only 324 centroided spectra are saved. 325 326 Returns 327 ------- 328 list 329 A list of mass spectrum data objects corresponding to the provided scan numbers. 330 """ 331 mass_spectra_list = [] 332 for scan_number in scan_list: 333 if scan_number in self.scan_number_list: 334 mass_spec = self.get_mass_spectrum_from_scan(scan_number) 335 mass_spectra_list.append(mass_spec) 336 else: 337 warnings.warn(f"Scan number {scan_number} not found in HDF5 file.") 338 return mass_spectra_list 339 340 def load(self) -> None: 341 """ """ 342 pass 343 344 def get_ms_raw(self, spectra=None, scan_df=None, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None) -> dict: 345 """ """ 346 # Warn if spectra or scan_df are not None that they are not used for CoreMS HDF5 files and should be rerun after instantiation 347 if spectra is not None or scan_df is not None: 348 SyntaxWarning( 349 "get_ms_raw method for CoreMS HDF5 files can only access saved data, consider rerunning after instantiation." 350 ) 351 ms_unprocessed = {} 352 dict_group_load = self.h5pydata["ms_unprocessed"] 353 dict_group_keys = dict_group_load.keys() 354 for k in dict_group_keys: 355 ms_up_int = dict_group_load[k][:] 356 ms_unprocessed[int(k)] = pd.DataFrame( 357 ms_up_int, columns=["scan", "mz", "intensity"] 358 ) 359 return ms_unprocessed 360 361 def get_scan_df(self, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None) -> pd.DataFrame: 362 scan_info = {} 363 dict_group_load = self.h5pydata["scan_info"] 364 dict_group_keys = dict_group_load.keys() 365 for k in dict_group_keys: 366 scan_info[k] = dict_group_load[k][:] 367 scan_df = pd.DataFrame(scan_info) 368 scan_df.set_index("scan", inplace=True, drop=False) 369 str_df = scan_df.select_dtypes([object]) 370 str_df = str_df.stack().str.decode("utf-8").unstack() 371 for col in str_df: 372 scan_df[col] = str_df[col] 373 374 # Apply time range filtering if specified 375 if time_range is not None: 376 time_ranges = self._normalize_time_range(time_range) 377 mask = pd.Series([False] * len(scan_df), index=scan_df.index) 378 for start_time, end_time in time_ranges: 379 mask |= (scan_df.scan_time >= start_time) & (scan_df.scan_time <= end_time) 380 scan_df = scan_df[mask] 381 382 return scan_df 383 384 def run(self, mass_spectra, load_raw=True, load_light=False) -> None: 385 """Runs the importer functions to populate a LCMS or MassSpectraBase object. 386 387 Notes 388 ----- 389 The following functions are run in order, if the HDF5 file contains the necessary data: 390 1. import_parameters(), which populates the parameters attribute on the LCMS or MassSpectraBase object. 391 2. import_mass_spectra(), which populates the _ms list on the LCMS or MassSpectraBase object. 392 3. import_scan_info(), which populates the _scan_info on the LCMS or MassSpectraBase object. 393 4. import_ms_unprocessed(), which populates the _ms_unprocessed attribute on the LCMS or MassSpectraBase object. 394 5. import_mass_features(), which populates the mass_features attribute on the LCMS or MassSpectraBase object. 395 6. import_eics(), which populates the eics attribute on the LCMS or MassSpectraBase object. 396 7. import_spectral_search_results(), which populates the spectral_search_results attribute on the LCMS or MassSpectraBase object. 397 398 Parameters 399 ---------- 400 mass_spectra : LCMSBase or MassSpectraBase 401 The LCMS or MassSpectraBase object to populate with mass spectra, generally instantiated with only the file_location, analyzer, and instrument_label attributes. 402 load_raw : bool 403 If True, load raw data (unprocessed) from HDF5 files for overall lcms object and individual mass spectra. Default is True. 404 load_light : bool 405 If True, only load the parameters, mass features, and scan info. Default is False. 406 407 Returns 408 ------- 409 None, but populates several attributes on the LCMS or MassSpectraBase object. 410 411 """ 412 if self.parameters_location is not None: 413 # Populate the parameters attribute on the LCMS object 414 self.import_parameters(mass_spectra) 415 416 if "mass_spectra" in self.h5pydata and not load_light: 417 # Populate the _ms list on the LCMS object 418 self.import_mass_spectra(mass_spectra, load_raw=load_raw) 419 420 if "scan_info" in self.h5pydata: 421 # Populate the _scan_info attribute on the LCMS object 422 self.import_scan_info(mass_spectra) 423 424 if "ms_unprocessed" in self.h5pydata and load_raw and not load_light: 425 # Populate the _ms_unprocessed attribute on the LCMS object 426 self.import_ms_unprocessed(mass_spectra) 427 428 if "mass_features" in self.h5pydata: 429 # Populate the mass_features attribute on the LCMS object 430 self.import_mass_features(mass_spectra) 431 432 if "eics" in self.h5pydata and not load_light: 433 # Populate the eics attribute on the LCMS object 434 self.import_eics(mass_spectra) 435 436 if "spectral_search_results" in self.h5pydata and not load_light: 437 # Populate the spectral_search_results attribute on the LCMS object 438 self.import_spectral_search_results(mass_spectra) 439 440 def import_mass_spectra(self, mass_spectra, load_raw=True) -> None: 441 """Imports all mass spectra from the HDF5 file. 442 443 Parameters 444 ---------- 445 mass_spectra : LCMSBase | MassSpectraBase 446 The MassSpectraBase or LCMSBase object to populate with mass spectra. 447 load_raw : bool 448 If True, load raw data (unprocessed) from HDF5 files for overall lcms object and individual mass spectra. Default 449 450 Returns 451 ------- 452 None, but populates the '_ms' list on the LCMSBase or MassSpectraBase 453 object with mass spectra from the HDF5 file. 454 """ 455 for scan_number in self.scan_number_list: 456 mass_spec = self.get_mass_spectrum(scan_number, load_raw=load_raw) 457 mass_spec.scan_number = scan_number 458 mass_spectra.add_mass_spectrum(mass_spec) 459 460 def import_scan_info(self, mass_spectra) -> None: 461 """Imports the scan info from the HDF5 file. 462 463 Parameters 464 ---------- 465 lcms : LCMSBase | MassSpectraBase 466 The MassSpectraBase or LCMSBase objects 467 468 Returns 469 ------- 470 None, but populates the 'scan_df' attribute on the LCMSBase or MassSpectraBase 471 object with a pandas DataFrame of the 'scan_info' from the HDF5 file. 472 473 """ 474 scan_df = self.get_scan_df() 475 mass_spectra.scan_df = scan_df 476 477 def import_ms_unprocessed(self, mass_spectra) -> None: 478 """Imports the unprocessed mass spectra from the HDF5 file. 479 480 Parameters 481 ---------- 482 lcms : LCMSBase | MassSpectraBase 483 The MassSpectraBase or LCMSBase objects 484 485 Returns 486 ------- 487 None, but populates the '_ms_unprocessed' attribute on the LCMSBase or MassSpectraBase 488 object with a dictionary of the 'ms_unprocessed' from the HDF5 file. 489 490 """ 491 ms_unprocessed = self.get_ms_raw() 492 mass_spectra._ms_unprocessed = ms_unprocessed 493 494 def import_parameters(self, mass_spectra) -> None: 495 """Imports the parameters from the HDF5 file. 496 497 Parameters 498 ---------- 499 mass_spectra : LCMSBase | MassSpectraBase 500 The MassSpectraBase or LCMSBase object to populate with parameters. 501 502 Returns 503 ------- 504 None, but populates the 'parameters' attribute on the LCMS or MassSpectraBase 505 object with a dictionary of the 'parameters' from the HDF5 file. 506 507 """ 508 if ".json" == self.parameters_location.suffix: 509 load_and_set_json_parameters_lcms(mass_spectra, self.parameters_location) 510 if ".toml" == self.parameters_location.suffix: 511 load_and_set_toml_parameters_lcms(mass_spectra, self.parameters_location) 512 else: 513 raise Exception( 514 "Parameters file must be in JSON format, TOML format is not yet supported." 515 ) 516 517 def import_mass_features(self, mass_spectra, mf_ids=None) -> None: 518 """Imports the mass features from the HDF5 file. 519 520 Parameters 521 ---------- 522 mass_spectra : LCMSBase | MassSpectraBase 523 The MassSpectraBase or LCMSBase object to populate with mass features. 524 mf_ids : list, optional 525 A list of mass feature IDs to import. If None, all mass features are imported. 526 527 Returns 528 ------- 529 None, but populates the 'mass_features' attribute on the LCMSBase or MassSpectraBase 530 object with a dictionary of the 'mass_features' from the HDF5 file. 531 532 """ 533 dict_group_load = self.h5pydata["mass_features"] 534 dict_group_keys = dict_group_load.keys() 535 for k in dict_group_keys: 536 if mf_ids is not None and int(k) not in mf_ids: 537 continue 538 # Instantiate the MassFeature object 539 mass_feature = LCMSMassFeature( 540 mass_spectra, 541 mz=dict_group_load[k].attrs["_mz_exp"], 542 retention_time=dict_group_load[k].attrs["_retention_time"], 543 intensity=dict_group_load[k].attrs["_intensity"], 544 apex_scan=dict_group_load[k].attrs["_apex_scan"], 545 persistence=dict_group_load[k].attrs["_persistence"], 546 id=int(k), 547 ) 548 549 # Populate additional attributes on the MassFeature object 550 for key in dict_group_load[k].attrs.keys() - { 551 "_mz_exp", 552 "_mz_cal", 553 "_retention_time", 554 "_intensity", 555 "_apex_scan", 556 "_persistence", 557 }: 558 setattr(mass_feature, key, dict_group_load[k].attrs[key]) 559 560 # Populate attributes on MassFeature object that are lists 561 for key in dict_group_load[k].keys(): 562 setattr(mass_feature, key, dict_group_load[k][key][:]) 563 # Convert _noise_score from array to tuple 564 if key == "_noise_score": 565 mass_feature._noise_score = tuple(mass_feature._noise_score) 566 mass_spectra.mass_features[int(k)] = mass_feature 567 568 # Associate mass features with ms1 and ms2 spectra, if available 569 for mf_id in mass_spectra.mass_features.keys(): 570 if mass_spectra.mass_features[mf_id].apex_scan in mass_spectra._ms.keys(): 571 mass_spectra.mass_features[mf_id].mass_spectrum = mass_spectra._ms[ 572 mass_spectra.mass_features[mf_id].apex_scan 573 ] 574 if mass_spectra.mass_features[mf_id].ms2_scan_numbers is not None: 575 for ms2_scan in mass_spectra.mass_features[mf_id].ms2_scan_numbers: 576 if ms2_scan in mass_spectra._ms.keys(): 577 mass_spectra.mass_features[mf_id].ms2_mass_spectra[ms2_scan] = ( 578 mass_spectra._ms[ms2_scan] 579 ) 580 581 def import_eics(self, mass_spectra, mz_list=None, mz_tolerance=0.0001): 582 """Imports the extracted ion chromatograms from the HDF5 file. 583 584 Parameters 585 ---------- 586 mass_spectra : LCMSBase | MassSpectraBase 587 The MassSpectraBase or LCMSBase object to populate with extracted ion chromatograms. 588 mz_list : list of float, optional 589 List of m/z values to load EICs for. If None, loads all EICs. Default is None. 590 mz_tolerance : float, optional 591 Tolerance in Daltons for matching m/z values when mz_list is provided. 592 Default is 0.0001 Da. 593 594 Returns 595 ------- 596 None, but populates the 'eics' attribute on the LCMSBase or MassSpectraBase 597 object with a dictionary of the 'eics' from the HDF5 file. 598 599 """ 600 dict_group_load = self.h5pydata["eics"] 601 dict_group_keys = dict_group_load.keys() 602 603 # Prefilter dict_group_keys if mz_list is provided to EICs within tolerance 604 if mz_list is not None: 605 target_mz_array = np.array(sorted(mz_list)) 606 mzs = [float(k) for k in dict_group_keys if np.abs(float(k)-target_mz_array).min() < mz_tolerance] 607 dict_group_keys = [str(mz) for mz in mzs] 608 609 for k in dict_group_keys: 610 # Check if we should load this EIC (filter by m/z if list provided) 611 eic_mz = dict_group_load[k].attrs["mz"] 612 613 my_eic = EIC_Data( 614 scans=dict_group_load[k]["scans"][:], 615 time=dict_group_load[k]["time"][:], 616 eic=dict_group_load[k]["eic"][:], 617 ) 618 for key in dict_group_load[k].keys(): 619 if key not in ["scans", "time", "eic"]: 620 setattr(my_eic, key, dict_group_load[k][key][:]) 621 # if key is apexes, convert to a tuple of a list 622 if key == "apexes" and len(my_eic.apexes) > 0: 623 my_eic.apexes = [tuple(x) for x in my_eic.apexes] 624 # Add to mass_spectra object 625 mass_spectra.eics[eic_mz] = my_eic 626 627 # Associate EICs with mass features using tolerance-based matching 628 mass_spectra.associate_eics_with_mass_features() 629 630 @staticmethod 631 def _load_eics_from_hdf5_group(eics_group, lcms_obj, mz_filter=None): 632 """Load EICs from an HDF5 group. 633 634 This is a static helper method that can be reused to load EIC data 635 from any HDF5 group in a consistent format. 636 637 Parameters 638 ---------- 639 eics_group : h5py.Group 640 The HDF5 group containing EIC data. 641 lcms_obj : LCMSBase 642 The LCMS object to associate EICs with (for reference, not modified). 643 mz_filter : list, optional 644 List of m/z values to load. If None, loads all EICs. Default is None. 645 Uses tolerance-based matching (0.0001). 646 647 Returns 648 ------- 649 dict 650 Dictionary of EIC_Data objects keyed by m/z value. 651 """ 652 from corems.mass_spectra.factory.chromat_data import EIC_Data 653 654 loaded_eics = {} 655 tolerance = 0.0001 656 657 for eic_key_str in eics_group.keys(): 658 eic_mz = float(eic_key_str) if eic_key_str.replace('.', '', 1).replace('-', '', 1).isdigit() else eics_group[eic_key_str].attrs.get("mz") 659 660 # If mz_filter is provided, check if this EIC matches any requested m/z 661 if mz_filter is not None: 662 if not any(abs(eic_mz - mz) < tolerance for mz in mz_filter): 663 continue 664 665 eic_data = eics_group[eic_key_str] 666 667 # Create EIC_Data object from datasets 668 eic = EIC_Data( 669 scans=list(eic_data["scans"][:]) if "scans" in eic_data else [], 670 time=list(eic_data["time"][:]) if "time" in eic_data else [], 671 eic=list(eic_data["eic"][:]) if "eic" in eic_data else [], 672 apexes=list(eic_data["apexes"][:]) if "apexes" in eic_data else [], 673 ) 674 675 # Load any additional datasets 676 for key in eic_data.keys(): 677 if key not in ["scans", "time", "eic", "apexes"]: 678 setattr(eic, key, eic_data[key][:]) 679 680 loaded_eics[eic_mz] = eic 681 682 return loaded_eics 683 684 def import_spectral_search_results(self, mass_spectra): 685 """Imports the spectral search results from the HDF5 file. 686 687 Parameters 688 ---------- 689 mass_spectra : LCMSBase | MassSpectraBase 690 The MassSpectraBase or LCMSBase object to populate with spectral search results. 691 692 Returns 693 ------- 694 None, but populates the 'spectral_search_results' attribute on the LCMSBase or MassSpectraBase 695 object with a dictionary of the 'spectral_search_results' from the HDF5 file. 696 697 """ 698 overall_results_dict = {} 699 ms2_results_load = self.h5pydata["spectral_search_results"] 700 for k in ms2_results_load.keys(): 701 overall_results_dict[int(k)] = {} 702 for k2 in ms2_results_load[k].keys(): 703 ms2_search_res = SpectrumSearchResults( 704 query_spectrum=mass_spectra._ms[int(k)], 705 precursor_mz=ms2_results_load[k][k2].attrs["precursor_mz"], 706 spectral_similarity_search_results={}, 707 ) 708 709 for key in ms2_results_load[k][k2].keys() - {"precursor_mz"}: 710 data = list(ms2_results_load[k][k2][key][:]) 711 if data and isinstance(data[0], bytes): 712 data = [x.decode("utf-8") for x in data] 713 setattr(ms2_search_res, key, data) 714 715 overall_results_dict[int(k)][ 716 ms2_results_load[k][k2].attrs["precursor_mz"] 717 ] = ms2_search_res 718 719 # add to mass_spectra 720 mass_spectra.spectral_search_results.update(overall_results_dict) 721 722 # If there are mass features, associate the results with each mass feature 723 if len(mass_spectra.mass_features) > 0: 724 for mass_feature_id, mass_feature in mass_spectra.mass_features.items(): 725 scan_ids = mass_feature.ms2_scan_numbers 726 for ms2_scan_id in scan_ids: 727 precursor_mz = mass_feature.mz 728 try: 729 mass_spectra.spectral_search_results[ms2_scan_id][precursor_mz] 730 except KeyError: 731 pass 732 else: 733 mass_spectra.mass_features[ 734 mass_feature_id 735 ].ms2_similarity_results.append( 736 mass_spectra.spectral_search_results[ms2_scan_id][ 737 precursor_mz 738 ] 739 ) 740 741 def get_mass_spectra_obj(self, load_raw=True, load_light=False, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None) -> MassSpectraBase: 742 """ 743 Return mass spectra data object, populating the _ms list on MassSpectraBase object from the HDF5 file. 744 745 Parameters 746 ---------- 747 load_raw : bool 748 If True, load raw data (unprocessed) from HDF5 files for overall spectra object and individual mass spectra. Default is True. 749 load_light : bool 750 If True, only load the parameters, mass features, and scan info. Default is False. 751 time_range : tuple or list of tuples, optional 752 Retention time range(s) to load. Can be: 753 - Single range: (start_time, end_time) in minutes 754 - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes 755 If None, loads all scans. Note: For HDF5 files, this parameter is accepted for 756 interface consistency but not currently used in filtering. 757 758 """ 759 # Instantiate the LCMS object 760 spectra_obj = MassSpectraBase( 761 file_location=self.file_location, 762 analyzer=self.analyzer, 763 instrument_label=self.instrument_label, 764 sample_name=self.sample_name, 765 ) 766 767 # This will populate the _ms list on the LCMS or MassSpectraBase object 768 self.run(spectra_obj, load_raw=load_raw, load_light=load_light) 769 770 return spectra_obj 771 772 def get_lcms_obj( 773 self, load_raw=True, load_light=False, use_original_parser=True, raw_file_path=None, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None 774 ) -> LCMSBase: 775 """ 776 Return LCMSBase object, populating attributes on the LCMSBase object from the HDF5 file. 777 778 Parameters 779 ---------- 780 load_raw : bool 781 If True, load raw data (unprocessed) from HDF5 files for overall lcms object and individual mass spectra. Default is True. 782 load_light : bool 783 If True, only load the parameters, mass features, and scan info. Default is False. 784 use_original_parser : bool 785 If True, use the original parser to populate the LCMS object. Default is True. 786 raw_file_path : str 787 The location of the raw file to parse if attempting to use original parser. 788 Default is None, which attempts to get the raw file path from the HDF5 file. 789 If the original file path has moved, this parameter can be used to specify the new location. 790 time_range : tuple or list of tuples, optional 791 Retention time range(s) to load. Can be: 792 - Single range: (start_time, end_time) in minutes 793 - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes 794 If None, loads all scans. Note: For HDF5 files, this parameter is accepted for 795 interface consistency. If use_original_parser=True, time_range can be passed to the 796 original parser for filtering. 797 """ 798 # Instantiate the LCMS object 799 lcms_obj = LCMSBase( 800 file_location=self.file_location, 801 analyzer=self.analyzer, 802 instrument_label=self.instrument_label, 803 sample_name=self.sample_name, 804 ) 805 806 # This will populate the majority of the attributes on the LCMS object 807 self.run(lcms_obj, load_raw=load_raw, load_light=load_light) 808 809 # Set final attributes of the LCMS object 810 lcms_obj.polarity = self.h5pydata.attrs["polarity"] 811 lcms_obj._scans_number_list = list(lcms_obj.scan_df.scan) 812 lcms_obj._retention_time_list = list(lcms_obj.scan_df.scan_time) 813 lcms_obj._tic_list = list(lcms_obj.scan_df.tic) 814 815 # If use_original_parser is True, instantiate the original parser and populate the LCMS object 816 if use_original_parser: 817 lcms_obj = self.add_original_parser(lcms_obj, raw_file_path=raw_file_path) 818 else: 819 lcms_obj.spectra_parser_class = self.__class__ 820 821 return lcms_obj 822 823 def get_raw_file_location(self): 824 """ 825 Get the raw file location from the HDF5 file attributes. 826 827 Returns 828 ------- 829 str 830 The raw file location. 831 """ 832 if "original_file_location" in self.h5pydata.attrs: 833 return self.h5pydata.attrs["original_file_location"] 834 else: 835 return None 836 837 def add_original_parser(self, mass_spectra, raw_file_path=None): 838 """ 839 Add the original parser to the mass spectra object. 840 841 Parameters 842 ---------- 843 mass_spectra : MassSpectraBase | LCMSBase 844 The MassSpectraBase or LCMSBase object to add the original parser to. 845 raw_file_path : str 846 The location of the raw file to parse. Default is None, which attempts to get the raw file path from the HDF5 file. 847 """ 848 # Get the original parser type 849 og_parser_type = self.h5pydata.attrs["parser_type"] 850 851 # If raw_file_path is None, get it from the HDF5 file attributes 852 if raw_file_path is None: 853 raw_file_path = self.get_raw_file_location() 854 if raw_file_path is None: 855 raise ValueError( 856 "Raw file path not found in HDF5 file attributes, cannot instantiate original parser." 857 ) 858 859 # Set the raw file path on the mass_spectra object so the parser knows where to find the raw file 860 mass_spectra.raw_file_location = raw_file_path 861 862 if og_parser_type == "ImportMassSpectraThermoMSFileReader": 863 # Check that the parser can be instantiated with the raw file path 864 parser_class = ImportMassSpectraThermoMSFileReader 865 elif og_parser_type == "MZMLSpectraParser": 866 # Check that the parser can be instantiated with the raw file path 867 parser_class = MZMLSpectraParser 868 869 # Set the spectra parser class on the mass_spectra object so the spectra_parser property can be used with the original parser 870 mass_spectra.spectra_parser_class = parser_class 871 872 return mass_spectra 873 874 def get_original_creation_time(self): 875 """ 876 Get the creation time of the original raw data file. 877 878 First checks if creation_time is saved in the HDF5 file attributes. 879 If not found, attempts to instantiate the original parser and get the creation time. 880 881 Returns 882 ------- 883 datetime 884 The creation time of the original raw data file, or None if not available. 885 """ 886 # Check if creation_time is saved in HDF5 attributes 887 if "creation_time" in self.h5pydata.attrs: 888 from datetime import datetime 889 return datetime.fromisoformat(self.h5pydata.attrs["creation_time"]) 890 891 # Fall back to using original parser to get creation time 892 try: 893 # Get the original parser type and raw file path 894 og_parser_type = self.h5pydata.attrs.get("parser_type") 895 raw_file_path = self.get_raw_file_location() 896 897 if og_parser_type is None or raw_file_path is None: 898 warnings.warn( 899 "Cannot retrieve creation time: parser_type or original_file_location not found in HDF5 attributes." 900 ) 901 return None 902 903 # Check if raw file exists 904 from pathlib import Path 905 if not Path(raw_file_path).exists(): 906 warnings.warn( 907 f"Cannot retrieve creation time: original raw file not found at {raw_file_path}" 908 ) 909 return None 910 911 # Instantiate the original parser 912 if og_parser_type == "ImportMassSpectraThermoMSFileReader": 913 parser = ImportMassSpectraThermoMSFileReader(raw_file_path) 914 elif og_parser_type == "MZMLSpectraParser": 915 parser = MZMLSpectraParser(raw_file_path) 916 else: 917 warnings.warn( 918 f"Unknown parser type: {og_parser_type}, cannot retrieve creation time." 919 ) 920 return None 921 922 # Get creation time from parser 923 return parser.get_creation_time() 924 925 except Exception as e: 926 warnings.warn( 927 f"Failed to retrieve creation time from original parser: {e}" 928 ) 929 return None 930 931 def get_creation_time(self): 932 """ 933 Get the creation time of the original raw data file. 934 935 This is an alias for get_original_creation_time() for backward compatibility. 936 937 Returns 938 ------- 939 datetime 940 The creation time of the original raw data file, or None if not available. 941 """ 942 return self.get_original_creation_time() 943 944 def get_instrument_info(self): 945 """ 946 Raise a NotImplemented Warning, as instrument info is not available in CoreMS HDF5 files and returning None. 947 """ 948 warnings.warn( 949 "Instrument info is not available in CoreMS HDF5 files, returning None." 950 "This should be accessed through the original parser.", 951 ) 952 return None 953 954 def get_scans_in_time_range( 955 self, 956 time_range: Union[Tuple[float, float], List[Tuple[float, float]]], 957 ms_level: Optional[int] = None 958 ) -> List[int]: 959 """Return scan numbers within specified retention time range(s). 960 961 Parameters 962 ---------- 963 time_range : tuple or list of tuples 964 Retention time range(s) in minutes. Can be: 965 - Single range: (start_time, end_time) 966 - Multiple ranges: [(start1, end1), (start2, end2), ...] 967 ms_level : int, optional 968 If specified, only return scans of this MS level (e.g., 1 for MS1, 2 for MS2). 969 If None, returns scans of all MS levels. 970 971 Returns 972 ------- 973 list of int 974 List of scan numbers within the specified time range(s) and MS level. 975 """ 976 # Normalize time range to list of tuples 977 time_ranges = self._normalize_time_range(time_range) 978 979 # Get all scan data 980 scan_df = self.get_scan_df() 981 982 # Filter by time range 983 mask = pd.Series([False] * len(scan_df), index=scan_df.index) 984 for start_time, end_time in time_ranges: 985 mask |= (scan_df.scan_time >= start_time) & (scan_df.scan_time <= end_time) 986 987 filtered_df = scan_df[mask] 988 989 # Filter by MS level if specified 990 if ms_level is not None: 991 filtered_df = filtered_df[filtered_df.ms_level == ms_level] 992 993 return filtered_df.scan.tolist() 994 995 996class ReadCoreMSHDFMassSpectraCollection: 997 """Read a collection of CoreMS HDF5 files and populate an LCMSCollection object. 998 999 Parameters 1000 ---------- 1001 folder_location : Path 1002 Folder containing .corems subdirectories with HDF5 files. 1003 manifest_file : Path, optional 1004 Manifest CSV with columns: sample_name, order, batch, center, time. 1005 One sample must have center='TRUE' for RT alignment. 1006 If None, checks if auto-generated manifest_auto.csv exists in the folder. If not, 1007 auto-generates from folder contents. Default: None. 1008 cores : int, optional 1009 Number of cores for multiprocessing. Default: 1. 1010 auto_manifest_batch_threshold_hours : float, optional 1011 Time gap (hours) for auto-generated batch separation. Default: 12.0. 1012 auto_manifest_center_name : str, optional 1013 Sample name for RT alignment center when auto-generating. 1014 Must match a discovered sample. If None, uses middle sample. Default: None. 1015 1016 Attributes 1017 ---------- 1018 folder_location : Path 1019 Folder containing CoreMS HDF5 files. 1020 manifest_filepath : Path 1021 Path to manifest file. 1022 manifest : dict 1023 Manifest data indexed by sample_name. 1024 """ 1025 def __init__( 1026 self, 1027 folder_location: Path, 1028 manifest_file: Path = None, 1029 cores: int = 1, 1030 auto_manifest_batch_threshold_hours: float = 12.0, 1031 auto_manifest_center_name: str = None 1032 ): 1033 # Check for folder location 1034 folder_location = Path(folder_location) 1035 if not folder_location.exists(): 1036 raise FileNotFoundError(f"Folder location {folder_location} not found.") 1037 1038 # Auto-generate manifest if not provided 1039 if manifest_file is None: 1040 # Check if manifest_auto.csv already exists 1041 auto_manifest_path = folder_location / "manifest_auto.csv" 1042 if auto_manifest_path.exists(): 1043 print(f"No manifest file provided. Using existing manifest_auto.csv from {folder_location}") 1044 manifest_file = auto_manifest_path 1045 else: 1046 print(f"No manifest file provided. Auto-generating manifest from {folder_location}") 1047 manifest_file = create_manifest_from_folder( 1048 folder_path=folder_location, 1049 output_path=auto_manifest_path, 1050 batch_time_threshold_hours=auto_manifest_batch_threshold_hours, 1051 center_name=auto_manifest_center_name, 1052 overwrite=True 1053 ) 1054 else: 1055 manifest_file = Path(manifest_file) 1056 if not manifest_file.exists(): 1057 raise FileNotFoundError(f"Manifest file {manifest_file} not found.") 1058 1059 # Check if the manifest file is a CSV 1060 if manifest_file.suffix != ".csv": 1061 raise ValueError("Manifest file must be a CSV.") 1062 1063 self.folder_location = folder_location 1064 self._manifest_dict = None 1065 self._parse_manifest(manifest_file) 1066 self._validate_manifest() 1067 self._validate_parameters() 1068 self._validate_cores(cores) 1069 1070 def _validate_cores(self, cores): 1071 # Check if the cores parameter is an integer greater than 0 and less than the number of cores available 1072 if not isinstance(cores, int) or cores < 1: 1073 raise ValueError("Cores must be an integer greater than 0.") 1074 if cores > multiprocessing.cpu_count(): 1075 raise ValueError( 1076 f"Cores must be less than or equal to the number of cores available ({multiprocessing.cpu_count()})." 1077 ) 1078 self._cores = cores 1079 1080 def _parse_manifest(self, manifest_file): 1081 """Parse the manifest file and set the manifest dictionary.""" 1082 self.manifest_filepath = manifest_file 1083 manifest = pd.read_csv(manifest_file) 1084 # Check if the following columns exisit in the manifest file 1085 if not all( 1086 col in manifest.columns for col in ["sample_name", "order", "batch"] 1087 ): 1088 raise ValueError( 1089 "Manifest file must contain the following columns: 'sample_name', 'order', 'batch'." 1090 ) 1091 # Set index to the 'sample_name' column 1092 manifest.set_index("sample_name", inplace=True) 1093 self._manifest_dict = manifest.to_dict(orient="index") 1094 1095 def _validate_manifest(self): 1096 """Validate the manifest dictionary against the CoreMS folder location.""" 1097 # Check if the folder location contains HDF5 files for each sample 1098 for sample_name in self._manifest_dict.keys(): 1099 corems_dir = self.folder_location / f"{sample_name}.corems" 1100 if not corems_dir.exists(): 1101 raise FileNotFoundError(f"CoreMS folder for {sample_name} not found.") 1102 hdf5_file = corems_dir / f"{sample_name}.hdf5" 1103 if not hdf5_file.exists(): 1104 raise FileNotFoundError(f"HDF5 file for {sample_name} not found.") 1105 1106 # Check that at least one sample has center='TRUE' for retention time alignment 1107 center_values = [sample_data.get('center') for sample_data in self._manifest_dict.values()] 1108 if not any(center_val == 'TRUE' or center_val == True for center_val in center_values): 1109 raise ValueError( 1110 "Manifest must contain at least one sample with center='TRUE' for retention time alignment. " 1111 "None of the samples in the manifest have center='TRUE'." 1112 ) 1113 1114 def _validate_parameters(self): 1115 """Validate that the parameters used for all samples within a batch are the same.""" 1116 # Check if parameters files are saved as JSON or TOML 1117 if self.parameters_files[0].suffix == ".json": 1118 importer = json 1119 suffix = ".json" 1120 1121 elif self.parameters_files[0].suffix == ".toml": 1122 importer = toml 1123 suffix = ".toml" 1124 1125 manfiest_df = self.manifest_dataframe 1126 1127 # Split up samples by batch 1128 batches = manfiest_df["batch"].unique() 1129 1130 for batch in batches: 1131 samples = manfiest_df[manfiest_df["batch"] == batch].index 1132 # check if self.parameters_files end with .json or .toml 1133 batch_param_files = [ 1134 self.folder_location / f"{sample_name}.corems/{sample_name}{suffix}" 1135 for sample_name in self._manifest_dict.keys() 1136 if sample_name in samples 1137 ] 1138 with open( 1139 batch_param_files[0], 1140 "r", 1141 encoding="utf8", 1142 ) as stream: 1143 first_parameters = importer.load(stream) 1144 for parameters_file in batch_param_files[1:]: 1145 with open( 1146 parameters_file, 1147 "r", 1148 encoding="utf8", 1149 ) as stream: 1150 parameters = importer.load(stream) 1151 if parameters != first_parameters: 1152 raise ValueError( 1153 f"Parameters files for samples in batch {batch} are not equal." 1154 ) 1155 1156 def get_lcms_obj(self, sample_name: str, load_raw=False, load_light=True, use_original_parser=True, raw_file_path=None) -> LCMSBase: 1157 """Return a LCMSBase object for a given sample name within the collection. 1158 1159 Parameters 1160 ---------- 1161 sample_name : str 1162 The sample name to retrieve the LCMS object for. 1163 load_raw : bool 1164 If True, load raw data from HDF5 files. Default is False. 1165 load_light : bool 1166 If True, only load the parameters, mass features, and scan info are initially loaded for each lcms object. Default is True. 1167 """ 1168 hdf5_file = self.folder_location / f"{sample_name}.corems/{sample_name}.hdf5" 1169 with ReadCoreMSHDFMassSpectra(hdf5_file) as parser: 1170 lcms_obj = parser.get_lcms_obj(load_raw=load_raw, load_light=load_light, use_original_parser=use_original_parser, raw_file_path=raw_file_path) 1171 if load_light: 1172 mf_df = lcms_obj.mass_features_to_df() 1173 # Add ._eic_mz to mf_df for each mass_feature 1174 eic_mz_list = [] 1175 for mf_id, mf in lcms_obj.mass_features.items(): 1176 if hasattr(mf, "_eic_mz"): 1177 eic_mz_list.append(mf._eic_mz) 1178 else: 1179 eic_mz_list.append(None) 1180 mf_df["_eic_mz"] = eic_mz_list 1181 lcms_obj.mass_features = {} 1182 lcms_obj.light_mf_df = mf_df 1183 return lcms_obj 1184 1185 def get_lcms_collection(self, load_raw = False, load_light = True, use_original_parser = True) -> LCMSCollection: 1186 """Return a LCMSCollection object 1187 1188 Parameters 1189 ---------- 1190 load_raw : bool 1191 If True, load raw data from HDF5 files. Default is False. 1192 load_light : bool 1193 If True, only load the parameters, mass features, and scan info are initially loaded for each lcms object. 1194 After concatenating the mass_features, remove the mass_features attribute from the individual LCMS objects for memory efficiency. Default is True. 1195 Default is True. 1196 """ 1197 # Instantiate the LCMSCollection object 1198 lcms_coll = LCMSCollection( 1199 collection_location=self.folder_location, 1200 manifest=self.manifest, 1201 collection_parser=self 1202 ) 1203 1204 # Set the number of cores on the LCMSCollection object from the ReadCoreMSHDFMassSpectraCollection object 1205 lcms_coll.parameters.lcms_collection.cores = self._cores 1206 1207 # Add LCMS objects to the collection 1208 samples = self._manifest_dict.keys() 1209 1210 # Initialize the LCMS object dictionary 1211 if self._cores > 1: 1212 if self._cores > len(samples): 1213 ncores = len(samples) 1214 else: 1215 ncores = self._cores 1216 # Create a pool of workers (one for each core or sample, whichever is smaller) 1217 pool = multiprocessing.Pool(ncores) 1218 args = [(sample, load_raw, load_light, use_original_parser) for sample in samples] 1219 lcms_objs = pool.starmap(self.get_lcms_obj, args) 1220 for sample_name, lcms_obj in zip(samples, lcms_objs): 1221 lcms_coll._lcms[sample_name] = lcms_obj 1222 1223 elif self._cores == 1: 1224 # Load the LCMS objects sequentially 1225 for sample_name in samples: 1226 lcms_coll._lcms[sample_name] = self.get_lcms_obj(sample_name, load_raw=load_raw, load_light=load_light, use_original_parser=use_original_parser) 1227 1228 else: 1229 raise ValueError("Number of cores must be greater than 0 and set on the ReadCoreMSHDFMassSpectraCollection object.") 1230 1231 # Check that all LCMS objects have the same polarity 1232 if len(set([x.polarity for k, x in lcms_coll._lcms.items()])) != 1: 1233 raise ValueError("All samples must have the same polarity.") 1234 1235 # Set ids on the LCMS objects in the manifest 1236 i = 0 1237 for sample in lcms_coll.samples: 1238 lcms_coll._manifest_dict[sample]["collection_id"] = i 1239 i += 1 1240 1241 # Reorder the LCMS objects 1242 lcms_coll._reorder_lcms_objects() 1243 1244 # Collect the mass features from the LCMS objects and combine them into a single dataframe for the collection 1245 lcms_coll._combine_mass_features() 1246 1247 # If load_light, remove the mass_feature attribute from the individual LCMS objects 1248 if load_light: 1249 for sample_name in lcms_coll.samples: 1250 lcms_coll._lcms[sample_name].mass_features = {} 1251 # Remove the light_mf_df attribute from the individual LCMS objects 1252 del lcms_coll._lcms[sample_name].light_mf_df 1253 1254 1255 return lcms_coll 1256 1257 @property 1258 def manifest(self): 1259 return self._manifest_dict 1260 1261 @property 1262 def manifest_dataframe(self): 1263 return pd.DataFrame(self._manifest_dict).T 1264 1265 @property 1266 def hdf5_files(self): 1267 return [ 1268 self.folder_location / f"{sample_name}.corems/{sample_name}.hdf5" 1269 for sample_name in self._manifest_dict.keys() 1270 ] 1271 1272 @property 1273 def parameters_files(self): 1274 # Check if parameters files are saved as JSON or TOML 1275 json_files = [ 1276 self.folder_location / f"{sample_name}.corems/{sample_name}.json" 1277 for sample_name in self._manifest_dict.keys() 1278 ] 1279 toml_files = [ 1280 self.folder_location / f"{sample_name}.corems/{sample_name}.toml" 1281 for sample_name in self._manifest_dict.keys() 1282 ] 1283 if all([x.exists() for x in json_files]): 1284 return json_files 1285 elif all([x.exists() for x in toml_files]): 1286 return toml_files 1287 else: 1288 raise ValueError("Parameters files are not saved for all samples.") 1289 1290class ReadSavedLCMSCollection(ReadCoreMSHDFMassSpectraCollection): 1291 """ 1292 Subclass to read and re-instantiate a LCMSCollection from a saved HDF5 file. 1293 1294 1295 Parameters 1296 ---------- 1297 collection_hdf5_path : str or Path 1298 Path to the saved LCMSCollection HDF5 file. 1299 cores : int, optional 1300 Number of cores for processing. Default is 1. 1301 """ 1302 1303 def __init__( 1304 self, 1305 collection_hdf5_path: str, 1306 cores: int = 1 1307 ): 1308 # Convert to Path objects 1309 self.collection_hdf5_path = Path(collection_hdf5_path) 1310 1311 # Validate the collection file exists 1312 if not self.collection_hdf5_path.exists(): 1313 raise FileNotFoundError(f"Collection HDF5 file {self.collection_hdf5_path} not found.") 1314 1315 # Validate cores 1316 self._validate_cores(cores) 1317 1318 # Load metadata from saved collection 1319 self._load_collection_metadata() 1320 1321 if not self.folder_location.exists(): 1322 raise FileNotFoundError(f"Folder location {self.folder_location} not found.") 1323 1324 # Load the mass spectra data 1325 self._validate_manifest() 1326 1327 # Set the parameters file location 1328 self.parameters_location = self._get_parameters_location() 1329 1330 def _get_parameters_location(self): 1331 """Find the parameters file (JSON or TOML) associated with the collection HDF5 file.""" 1332 # Check for TOML file first (preferred) 1333 toml_path = self.collection_hdf5_path.with_suffix('.toml') 1334 if toml_path.exists(): 1335 return toml_path 1336 1337 # Check for JSON file 1338 json_path = self.collection_hdf5_path.with_suffix('.json') 1339 if json_path.exists(): 1340 return json_path 1341 1342 # No parameters file found 1343 return None 1344 1345 def _load_collection_metadata(self): 1346 """Load metadata and manifest from the saved collection HDF5 file.""" 1347 with h5py.File(self.collection_hdf5_path, 'r') as f: 1348 self.folder_location = Path(f.attrs.get('lcms_objects_folder', '')) 1349 self.missing_mass_features_searched = f.attrs.get('missing_mass_features_searched', False) 1350 1351 # Call the _load_manifest function to process the manifest 1352 self._manifest_dict = self._load_manifest(f) 1353 1354 def _load_manifest(self, hdf_handle): 1355 """Load and clean the manifest from the HDF5 file.""" 1356 manifest_json = hdf_handle.attrs.get('manifest', '{}') 1357 if isinstance(manifest_json, bytes): 1358 manifest_json = manifest_json.decode('utf-8') 1359 loaded_manifest = json.loads(manifest_json) 1360 1361 # Convert integer values for 'use_rt_alignment' back to booleans 1362 def convert_back_to_bool(data): 1363 if isinstance(data, dict): 1364 # Process each key-value pair recursively 1365 return {k: (bool(v) if k == 'use_rt_alignment' and isinstance(v, int) else convert_back_to_bool(v)) for k, v in data.items()} 1366 elif isinstance(data, list): 1367 # Recursively process lists 1368 return [convert_back_to_bool(item) for item in data] 1369 else: 1370 # Return non-dict/list types unchanged 1371 return data 1372 1373 # Clean the loaded manifest 1374 return convert_back_to_bool(loaded_manifest) 1375 1376 def _load_rt_alignments(self, lcms_collection): 1377 """Load retention time alignments from the saved collection HDF5 file.""" 1378 # First Set the rt_aligned flag from the collection-level attribute saved directly 1379 with h5py.File(self.collection_hdf5_path, 'r') as f: 1380 lcms_collection.rt_aligned = f.attrs.get('rt_aligned', False) 1381 lcms_collection.rt_alignment_attempted = f.attrs.get('rt_alignment_attempted', False) 1382 1383 if lcms_collection.rt_aligned: 1384 with h5py.File(self.collection_hdf5_path, 'r') as f: 1385 if "rt_alignments" in f: 1386 # Iterate over the group `rt_alignments` containing datasets and add to the corresponding lcms object 1387 rt_alignments_group = f["rt_alignments"] 1388 for sample_idx, lcms_obj in zip(rt_alignments_group.keys(), lcms_collection): 1389 alignment_data = rt_alignments_group[sample_idx][:] 1390 scan_df = lcms_obj.scan_df 1391 scan_df["scan_time_aligned"] = alignment_data 1392 lcms_obj.scan_df = scan_df 1393 elif lcms_collection.rt_alignment_attempted: 1394 # This means it was attempted and not used, so we populate the "scan_time_aligned" 1395 for lcms_obj in lcms_collection: 1396 scan_df = lcms_obj.scan_df 1397 scan_df["scan_time_aligned"] = scan_df["scan_time"] 1398 lcms_obj.scan_df = scan_df 1399 1400 def _load_cluster_assignments(self, lcms_collection): 1401 """Load cluster assignments from the saved collection HDF5 file.""" 1402 with h5py.File(self.collection_hdf5_path, 'r') as f: 1403 if "cluster_assignments" in f: 1404 # Access the group containing cluster assignments 1405 cluster_grp = f["cluster_assignments"] 1406 1407 # Reload index and cluster data 1408 index = cluster_grp["index"][:] # Extract index 1409 index = [idx.decode('utf-8') for idx in index] # Convert byte strings back to regular strings 1410 cluster_data = cluster_grp["cluster"][:] # Extract cluster column 1411 1412 # Reassemble the DataFrame 1413 cluster_df = pd.DataFrame({"cluster": cluster_data}, index=index) 1414 1415 # Assign cluster data back to lcms_collection.mass_features_dataframe 1416 lcms_collection.mass_features_dataframe = lcms_collection.mass_features_dataframe.join(cluster_df, how='left') 1417 1418 # Drop rows with NaN cluster values 1419 lcms_collection.mass_features_dataframe.dropna(subset=['cluster'], inplace=True) 1420 1421 def get_lcms_collection(self, load_raw=False, load_light=False, load_representatives=False, load_eics=False, load_ms1=False, load_ms2=False): 1422 """Get the LCMS collection from the saved HDF5 file. 1423 1424 Parameters 1425 ---------- 1426 load_raw : bool, optional 1427 If True, load raw data. Default is False. 1428 load_light : bool, optional 1429 If True, load light data (minimal). Default is False. 1430 load_representatives : bool, optional 1431 If True, load representative mass features from clusters. Default is False. 1432 load_eics : bool, optional 1433 If True, load EIC data for clustered mass features. Default is False. 1434 load_ms1 : bool, optional 1435 If True, load MS1 spectra for loaded mass features. Default is False. 1436 load_ms2 : bool, optional 1437 If True, load MS2 spectra for loaded mass features. Default is False. 1438 1439 Returns 1440 ------- 1441 LCMSCollection 1442 The loaded LCMS collection object. 1443 """ 1444 # First load the LCMSCollection object exactly as in the parent class 1445 lcms_collection = super().get_lcms_collection(load_raw=load_raw, load_light=load_light) 1446 1447 # Set the missing_mass_features_searched flag from saved metadata 1448 lcms_collection.missing_mass_features_searched = self.missing_mass_features_searched 1449 1450 # Load parameters if a parameters file exists 1451 if self.parameters_location: 1452 self._load_parameters(lcms_collection) 1453 1454 # Add retention time alignments if they exist 1455 self._load_rt_alignments(lcms_collection) 1456 1457 # Add cluster assignments if they exist 1458 self._load_cluster_assignments(lcms_collection) 1459 1460 # Load induced mass features if they exist 1461 self._load_induced_mass_features(lcms_collection) 1462 1463 # Load EICs for induced mass features from collection HDF5 1464 if lcms_collection.missing_mass_features_searched and load_eics: 1465 self._load_induced_eics_from_collection(lcms_collection) 1466 1467 # Combine induced mass features into the collection-level dataframe if any were loaded 1468 if lcms_collection.missing_mass_features_searched: 1469 lcms_collection._combine_mass_features(induced_features=True) 1470 1471 # Load representative mass features if requested 1472 if load_representatives: 1473 self._load_representative_mass_features(lcms_collection) 1474 1475 # Load MS1 and/or MS2 spectra for loaded mass features if requested 1476 if load_ms1 or load_ms2: 1477 # Reuse the existing ReloadFeaturesOperation from the pipeline system 1478 from corems.mass_spectra.calc.lc_calc_operations import ReloadFeaturesOperation 1479 1480 operations = [ReloadFeaturesOperation('reload_spectra', add_ms1=load_ms1, add_ms2=load_ms2)] 1481 lcms_collection.process_samples_pipeline(operations, keep_raw_data=False, show_progress=False) 1482 1483 # Load EICs for clustered features if requested 1484 if load_eics: 1485 # Reuse the existing LoadEICsOperation from the pipeline system 1486 from corems.mass_spectra.calc.lc_calc_operations import LoadEICsOperation 1487 1488 operations = [LoadEICsOperation('load_eics')] 1489 lcms_collection.process_samples_pipeline(operations, keep_raw_data=False, show_progress=False) 1490 1491 # Associate EICs with mass features (same as in process_consensus_features) 1492 for sample_id in range(len(lcms_collection.samples)): 1493 sample = lcms_collection[sample_id] 1494 if sample.eics: # Only if EICs were loaded 1495 # Associate EICs with regular mass features 1496 sample.associate_eics_with_mass_features(induced=False) 1497 # Associate EICs with induced mass features 1498 sample.associate_eics_with_mass_features(induced=True) 1499 1500 return lcms_collection 1501 1502 def _load_parameters(self, lcms_collection): 1503 """Load collection-level parameters from the saved parameters file.""" 1504 from corems.encapsulation.input.parameter_from_json import ( 1505 load_and_set_json_parameters_lcms_collection, 1506 load_and_set_toml_parameters_lcms_collection, 1507 ) 1508 1509 if self.parameters_location.suffix == ".json": 1510 load_and_set_json_parameters_lcms_collection(lcms_collection, self.parameters_location) 1511 elif self.parameters_location.suffix == ".toml": 1512 load_and_set_toml_parameters_lcms_collection(lcms_collection, self.parameters_location) 1513 else: 1514 warnings.warn(f"Unknown parameter file format: {self.parameters_location.suffix}. Skipping parameter loading.") 1515 1516 def _load_induced_mass_features(self, lcms_collection): 1517 """Load induced mass features from the saved collection HDF5 file. 1518 1519 Induced mass features are gap-filled features that exist at the collection level. 1520 This method loads them from the collection HDF5 file with all their attributes 1521 and datasets, and distributes them to individual LCMS objects. 1522 1523 Parameters 1524 ---------- 1525 lcms_collection : LCMSCollection 1526 The LCMS collection object to populate with induced mass features. 1527 """ 1528 with h5py.File(self.collection_hdf5_path, 'r') as f: 1529 if "induced_mass_features" not in f: 1530 return 1531 1532 # Access the top-level induced mass features group 1533 imf_group = f["induced_mass_features"] 1534 1535 # Iterate through each sample's induced mass features 1536 for sample_idx in imf_group.keys(): 1537 lcms_obj = lcms_collection[int(sample_idx)] 1538 sample_group = imf_group[sample_idx] 1539 1540 # Load each mass feature for this sample 1541 for mf_id_str in sample_group.keys(): 1542 mf_group = sample_group[mf_id_str] 1543 1544 # Note: Induced mass feature IDs are strings like 'c2923_0_i', not integers 1545 # Keep them as strings since that's how they're stored 1546 mf_id = mf_id_str 1547 1548 # Instantiate the LCMSMassFeature object with required attributes 1549 mass_feature = LCMSMassFeature( 1550 lcms_obj, 1551 mz=mf_group.attrs["_mz_exp"], 1552 retention_time=mf_group.attrs["_retention_time"], 1553 intensity=mf_group.attrs["_intensity"], 1554 apex_scan=mf_group.attrs["_apex_scan"], 1555 persistence=mf_group.attrs.get("_persistence", 0), 1556 id=mf_id, 1557 ) 1558 1559 # Populate additional attributes from HDF5 attributes 1560 for key in mf_group.attrs.keys() - { 1561 "_mz_exp", 1562 "_mz_cal", 1563 "_retention_time", 1564 "_intensity", 1565 "_apex_scan", 1566 "_persistence", 1567 }: 1568 setattr(mass_feature, key, mf_group.attrs[key]) 1569 1570 # Populate attributes from HDF5 datasets (arrays) 1571 for key in mf_group.keys(): 1572 setattr(mass_feature, key, mf_group[key][:]) 1573 # Convert _noise_score from array to tuple 1574 if key == "_noise_score": 1575 mass_feature._noise_score = tuple(mass_feature._noise_score) 1576 1577 # Add to the LCMS object's induced_mass_features dictionary 1578 lcms_obj.induced_mass_features[mf_id] = mass_feature 1579 1580 def _load_induced_eics_from_collection(self, lcms_collection): 1581 """Load EICs for induced mass features from the collection HDF5 file. 1582 1583 Induced mass features are gap-filled features. Their EICs are saved at the 1584 collection level and need to be loaded and associated with the induced mass features. 1585 1586 Parameters 1587 ---------- 1588 lcms_collection : LCMSCollection 1589 The LCMS collection object with induced mass features to associate EICs with. 1590 """ 1591 with h5py.File(self.collection_hdf5_path, 'r') as f: 1592 if "induced_eics" not in f: 1593 return 1594 1595 # Access the top-level induced EICs group 1596 induced_eics_group = f["induced_eics"] 1597 1598 # Iterate through each sample's induced EICs 1599 for sample_idx in induced_eics_group.keys(): 1600 lcms_obj = lcms_collection[int(sample_idx)] 1601 sample_group = induced_eics_group[sample_idx] 1602 1603 # Use the static helper to load EICs 1604 loaded_eics = ReadCoreMSHDFMassSpectra._load_eics_from_hdf5_group(sample_group, lcms_obj) 1605 1606 # Ensure eics dictionary exists (should already be initialized in __init__) 1607 if not hasattr(lcms_obj, 'eics') or lcms_obj.eics is None: 1608 lcms_obj.eics = {} 1609 1610 # Add to lcms_obj.eics dictionary 1611 for eic_mz, eic in loaded_eics.items(): 1612 lcms_obj.eics[eic_mz] = eic 1613 1614 # Associate EICs with induced mass features after all samples processed 1615 # This is done outside the loop to handle all samples at once 1616 for lcms_obj in lcms_collection: 1617 if len(lcms_obj.induced_mass_features) > 0: 1618 lcms_obj.associate_eics_with_mass_features(induced=True) 1619 1620 def _load_representative_mass_features(self, lcms_collection): 1621 """Load representative mass features for all clusters from HDF5 files. 1622 1623 This method uses the same logic as process_consensus_features() when loading 1624 representatives, calling get_sample_mf_map_for_representatives() (DRY helper) 1625 to determine which features to load. 1626 1627 Parameters 1628 ---------- 1629 lcms_collection : LCMSCollection 1630 The LCMS collection object to populate with representative mass features. 1631 """ 1632 # Get cluster assignments from the mass_features_dataframe 1633 if "cluster" not in lcms_collection.mass_features_dataframe.columns: 1634 return 1635 1636 # Use DRY helper method to build sample_mf_map with cluster IDs 1637 sample_mf_map = lcms_collection.get_sample_mf_map_for_representatives(include_cluster_id=True) 1638 1639 # Load mass features for each sample 1640 for sample_id, mf_list in sample_mf_map.items(): 1641 lcms_obj = lcms_collection[sample_id] 1642 1643 # Load each mass feature 1644 for mf_id, cluster_id in mf_list: 1645 self._load_single_mass_feature(lcms_obj, mf_id, cluster_id) 1646 1647 def _load_single_mass_feature(self, lcms_obj, feature_id, cluster_index=None): 1648 """Load a single mass feature from an LCMS object's HDF5 file. 1649 1650 Parameters 1651 ---------- 1652 lcms_obj : LCMSBase 1653 The LCMS object to add the mass feature to. 1654 feature_id : int 1655 The ID of the mass feature to load. 1656 cluster_index : int, optional 1657 The cluster index to assign to the loaded mass feature. 1658 """ 1659 hdf5_path = lcms_obj.file_location.with_suffix('.hdf5') 1660 1661 if not hdf5_path.exists(): 1662 return 1663 1664 with h5py.File(hdf5_path, 'r') as f: 1665 if 'mass_features' not in f: 1666 return 1667 1668 mf_group = f['mass_features'] 1669 feature_id_str = str(feature_id) 1670 1671 if feature_id_str not in mf_group: 1672 return 1673 1674 mf_data = mf_group[feature_id_str] 1675 1676 # Create LCMSMassFeature object 1677 mass_feature = LCMSMassFeature( 1678 lcms_obj, 1679 mz=mf_data.attrs["_mz_exp"], 1680 retention_time=mf_data.attrs["_retention_time"], 1681 intensity=mf_data.attrs["_intensity"], 1682 apex_scan=mf_data.attrs["_apex_scan"], 1683 persistence=mf_data.attrs.get("_persistence", 0), 1684 id=feature_id, 1685 ) 1686 1687 # Set cluster_index if provided 1688 if cluster_index is not None: 1689 mass_feature.cluster_index = cluster_index 1690 1691 # Populate additional attributes from HDF5 attributes 1692 for key in mf_data.attrs.keys() - { 1693 "_mz_exp", 1694 "_mz_cal", 1695 "_retention_time", 1696 "_intensity", 1697 "_apex_scan", 1698 "_persistence", 1699 }: 1700 setattr(mass_feature, key, mf_data.attrs[key]) 1701 1702 # Populate attributes from HDF5 datasets (arrays) 1703 for key in mf_data.keys(): 1704 setattr(mass_feature, key, mf_data[key][:]) 1705 # Convert _noise_score from array to tuple 1706 if key == "_noise_score": 1707 mass_feature._noise_score = tuple(mass_feature._noise_score) 1708 1709 # Add to the LCMS object's mass_features dictionary 1710 lcms_obj.mass_features[feature_id] = mass_feature
def
create_manifest_from_folder( folder_path: pathlib._local.Path, output_path: pathlib._local.Path = None, batch_time_threshold_hours: float = 12.0, center_name: str = None, overwrite: bool = False) -> pathlib._local.Path:
33def create_manifest_from_folder( 34 folder_path: Path, 35 output_path: Path = None, 36 batch_time_threshold_hours: float = 12.0, 37 center_name: str = None, 38 overwrite: bool = False 39) -> Path: 40 """ 41 Create a manifest CSV file for ReadCoreMSHDFMassSpectraCollection from CoreMS HDF5 files. 42 43 Scans a folder for .corems subdirectories and generates a manifest with columns: 44 sample_name, batch, order, center, time. Files are batched by creation time, and 45 one sample is designated as the retention time alignment center. 46 47 Parameters 48 ---------- 49 folder_path : Path 50 Path to folder containing .corems subdirectories with HDF5 files. 51 output_path : Path, optional 52 Output manifest CSV path. Default: folder_path/manifest.csv. 53 batch_time_threshold_hours : float, optional 54 Time gap in hours for batch separation. Default: 12.0. 55 center_name : str, optional 56 Sample name to designate as RT alignment center (must exist in samples). 57 If None, the middle sample (by creation time) is used. 58 overwrite : bool, optional 59 Whether to overwrite existing manifest. Default: False. 60 61 Returns 62 ------- 63 Path 64 Path to created manifest file. 65 66 Raises 67 ------ 68 FileNotFoundError 69 If folder_path doesn't exist or contains no .corems subdirectories. 70 FileExistsError 71 If output file exists and overwrite is False. 72 ValueError 73 If no HDF5 files found, or center_name doesn't match any sample. 74 """ 75 if not folder_path.exists(): 76 raise FileNotFoundError(f"Folder {folder_path} does not exist.") 77 78 # Set default output path if not provided 79 if output_path is None: 80 output_path = folder_path / "manifest.csv" 81 82 # Check if output file exists 83 if output_path.exists() and not overwrite: 84 raise FileExistsError( 85 f"Manifest file {output_path} already exists. " 86 "Set overwrite=True to replace it." 87 ) 88 89 # Find all .corems subdirectories 90 corems_dirs = sorted([d for d in folder_path.iterdir() if d.is_dir() and d.suffix == ".corems"]) 91 92 if not corems_dirs: 93 raise FileNotFoundError( 94 f"No .corems subdirectories found in {folder_path}. " 95 "Ensure the folder contains processed CoreMS data." 96 ) 97 98 # Collect sample information 99 sample_data = [] 100 101 for corems_dir in corems_dirs: 102 sample_name = corems_dir.stem # Remove .corems extension 103 hdf5_file = corems_dir / f"{sample_name}.hdf5" 104 105 if not hdf5_file.exists(): 106 print(f"Warning: HDF5 file not found for {sample_name}, skipping.") 107 continue 108 109 # Get creation time using the ReadCoreMSHDFMassSpectra method 110 try: 111 # Use context manager to ensure file is properly closed 112 with ReadCoreMSHDFMassSpectra(str(hdf5_file)) as parser: 113 # Use the get_original_creation_time() method which checks HDF5 attrs first, 114 # then falls back to original parser if needed 115 creation_time = parser.get_original_creation_time() 116 117 # Skip sample if creation time unavailable 118 if creation_time is None: 119 print(f"Warning: Could not get original creation time for {sample_name}, skipping.") 120 continue 121 122 except Exception as e: 123 print(f"Warning: Error getting creation time for {sample_name}: {e}, skipping.") 124 continue 125 126 sample_data.append({ 127 'sample_name': sample_name, 128 'creation_time': creation_time, 129 'hdf5_path': hdf5_file 130 }) 131 132 if not sample_data: 133 raise ValueError( 134 f"No valid HDF5 files found in {folder_path}. " 135 "Ensure .corems subdirectories contain .hdf5 files." 136 ) 137 138 # Sort by creation time 139 sample_data.sort(key=lambda x: x['creation_time']) 140 141 # Assign batches based on time threshold 142 batch_assignments = [] 143 current_batch = 1 144 145 for i, sample in enumerate(sample_data): 146 if i == 0: 147 batch_assignments.append(current_batch) 148 else: 149 time_diff = sample['creation_time'] - sample_data[i-1]['creation_time'] 150 time_diff_hours = time_diff.total_seconds() / 3600 151 152 if time_diff_hours > batch_time_threshold_hours: 153 current_batch += 1 154 155 batch_assignments.append(current_batch) 156 157 # Determine which sample should be the center for retention time alignment 158 sample_names = [s['sample_name'] for s in sample_data] 159 160 if center_name is not None: 161 # Validate that center_name is in the discovered samples 162 if center_name not in sample_names: 163 raise ValueError( 164 f"Specified center_name '{center_name}' not found in discovered samples. " 165 f"Available samples: {', '.join(sample_names)}" 166 ) 167 center_sample = center_name 168 else: 169 # Use the middle sample (by creation time) as center 170 middle_idx = len(sample_data) // 2 171 center_sample = sample_data[middle_idx]['sample_name'] 172 print(f"Auto-selected center sample: {center_sample} (index {middle_idx} of {len(sample_data)}, middle by creation time)") 173 174 # Create manifest dataframe with center column as TRUE/FALSE 175 manifest_df = pd.DataFrame({ 176 'sample_name': sample_names, 177 'batch': batch_assignments, 178 'order': list(range(1, len(sample_data) + 1)), 179 'center': ['TRUE' if name == center_sample else 'FALSE' for name in sample_names], 180 'time': [s['creation_time'].strftime('%Y-%m-%dT%H:%M:%SZ') for s in sample_data] 181 }) 182 183 # Sort manifest by time before saving to ensure proper order 184 manifest_df = manifest_df.sort_values('time').reset_index(drop=True) 185 # Update order column to reflect sorted order 186 manifest_df['order'] = list(range(1, len(manifest_df) + 1)) 187 188 # Save manifest 189 manifest_df.to_csv(output_path, index=False) 190 191 print(f"Manifest created successfully at {output_path}") 192 print(f"Total samples: {len(sample_data)}") 193 print(f"Number of batches: {current_batch}") 194 print(f"Batch assignments: {dict(zip(range(1, current_batch + 1), [batch_assignments.count(b) for b in range(1, current_batch + 1)]))}") 195 196 return output_path
Create a manifest CSV file for ReadCoreMSHDFMassSpectraCollection from CoreMS HDF5 files.
Scans a folder for .corems subdirectories and generates a manifest with columns: sample_name, batch, order, center, time. Files are batched by creation time, and one sample is designated as the retention time alignment center.
Parameters
- folder_path (Path): Path to folder containing .corems subdirectories with HDF5 files.
- output_path (Path, optional): Output manifest CSV path. Default: folder_path/manifest.csv.
- batch_time_threshold_hours (float, optional): Time gap in hours for batch separation. Default: 12.0.
- center_name (str, optional): Sample name to designate as RT alignment center (must exist in samples). If None, the middle sample (by creation time) is used.
- overwrite (bool, optional): Whether to overwrite existing manifest. Default: False.
Returns
- Path: Path to created manifest file.
Raises
- FileNotFoundError: If folder_path doesn't exist or contains no .corems subdirectories.
- FileExistsError: If output file exists and overwrite is False.
- ValueError: If no HDF5 files found, or center_name doesn't match any sample.
class
ReadCoreMSHDFMassSpectra(corems.mass_spectra.input.parserbase.SpectraParserInterface, corems.mass_spectrum.input.coremsHDF5.ReadCoreMSHDF_MassSpectrum, threading.Thread):
199class ReadCoreMSHDFMassSpectra( 200 SpectraParserInterface, ReadCoreMSHDF_MassSpectrum, Thread 201): 202 """Class to read CoreMS HDF5 files and populate a LCMS or MassSpectraBase object. 203 204 Parameters 205 ---------- 206 file_location : str 207 The location of the HDF5 file to read, including the suffix. 208 209 Attributes 210 ---------- 211 file_location : str 212 The location of the HDF5 file to read. 213 h5pydata : h5py.File 214 The HDF5 file object. 215 scans : list 216 A list of the location of individual mass spectra within the HDF5 file. 217 scan_number_list : list 218 A list of the scan numbers of the mass spectra within the HDF5 file. 219 parameters_location : str 220 The location of the parameters file (json or toml). 221 222 Methods 223 ------- 224 * import_mass_spectra(mass_spectra). 225 Imports all mass spectra from the HDF5 file onto the LCMS or MassSpectraBase object. 226 * get_mass_spectrum_from_scan(scan_number). 227 Return mass spectrum data object from scan number. 228 * load(). 229 Placeholder method to meet the requirements of the SpectraParserInterface. 230 * run(mass_spectra). 231 Runs the importer functions to populate a LCMS or MassSpectraBase object. 232 * import_scan_info(mass_spectra). 233 Imports the scan info from the HDF5 file to populate the _scan_info attribute 234 on the LCMS or MassSpectraBase object 235 * import_ms_unprocessed(mass_spectra). 236 Imports the unprocessed mass spectra from the HDF5 file to populate the 237 _ms_unprocessed attribute on the LCMS or MassSpectraBase object 238 * import_parameters(mass_spectra). 239 Imports the parameters from the HDF5 file to populate the parameters 240 attribute on the LCMS or MassSpectraBase object 241 * import_mass_features(mass_spectra). 242 Imports the mass features from the HDF5 file to populate the mass_features 243 attribute on the LCMS or MassSpectraBase object 244 * import_eics(mass_spectra). 245 Imports the extracted ion chromatograms from the HDF5 file to populate the 246 eics attribute on the LCMS or MassSpectraBase object 247 * import_spectral_search_results(mass_spectra). 248 Imports the spectral search results from the HDF5 file to populate the 249 spectral_search_results attribute on the LCMS or MassSpectraBase object 250 * get_mass_spectra_obj(). 251 Return mass spectra data object, populating the _ms list on the LCMS or 252 MassSpectraBase object from the HDF5 file 253 * get_lcms_obj(). 254 Return LCMSBase object, populating the majority of the attributes on the 255 LCMS object from the HDF5 file 256 257 """ 258 259 def __init__(self, file_location: str): 260 Thread.__init__(self) 261 ReadCoreMSHDF_MassSpectrum.__init__(self, file_location) 262 263 # override the scans attribute on ReadCoreMSHDF_MassSpectrum class to expect a nested location within the HDF5 file 264 self.scans = [ 265 "mass_spectra/" + x for x in list(self.h5pydata["mass_spectra"].keys()) 266 ] 267 self.scan_number_list = sorted( 268 [int(float(i)) for i in list(self.h5pydata["mass_spectra"].keys())] 269 ) 270 271 # set the location of the parameters file (json or toml) 272 add_files = [ 273 x 274 for x in self.file_location.parent.glob( 275 self.file_location.name.replace(".hdf5", ".*") 276 ) 277 if x.suffix != ".hdf5" 278 ] 279 if len([x for x in add_files if x.suffix == ".json"]) > 0: 280 self.parameters_location = [x for x in add_files if x.suffix == ".json"][0] 281 elif len([x for x in add_files if x.suffix == ".toml"]) > 0: 282 self.parameters_location = [x for x in add_files if x.suffix == ".toml"][0] 283 else: 284 self.parameters_location = None 285 286 def __enter__(self): 287 """Context manager entry.""" 288 return self 289 290 def __exit__(self, exc_type, exc_val, exc_tb): 291 """Context manager exit - closes the HDF5 file.""" 292 if hasattr(self, 'h5pydata') and self.h5pydata is not None: 293 self.h5pydata.close() 294 return False 295 296 def close(self): 297 """Explicitly close the HDF5 file.""" 298 if hasattr(self, 'h5pydata') and self.h5pydata is not None: 299 self.h5pydata.close() 300 301 def get_mass_spectrum_from_scan(self, scan_number): 302 """Return mass spectrum data object from scan number.""" 303 if scan_number in self.scan_number_list: 304 mass_spec = self.get_mass_spectrum(scan_number) 305 return mass_spec 306 else: 307 raise Exception("Scan number not found in HDF5 file.") 308 309 def get_mass_spectra_from_scan_list( 310 self, scan_list, spectrum_mode, auto_process=True 311 ): 312 """Return a list of mass spectrum data objects from a list of scan numbers. 313 314 Parameters 315 ---------- 316 scan_list : list 317 A list of scan numbers to retrieve mass spectra for. 318 spectrum_mode : str 319 The spectrum mode to use when retrieving the mass spectra. 320 Note that this parameter is not used for CoreMS HDF5 files, as the spectra are already processed and only 321 centroided spectra are saved. 322 auto_process : bool 323 If True, automatically process the mass spectra when retrieving them. 324 Note that this parameter is not used for CoreMS HDF5 files, as the spectra are already processed and only 325 centroided spectra are saved. 326 327 Returns 328 ------- 329 list 330 A list of mass spectrum data objects corresponding to the provided scan numbers. 331 """ 332 mass_spectra_list = [] 333 for scan_number in scan_list: 334 if scan_number in self.scan_number_list: 335 mass_spec = self.get_mass_spectrum_from_scan(scan_number) 336 mass_spectra_list.append(mass_spec) 337 else: 338 warnings.warn(f"Scan number {scan_number} not found in HDF5 file.") 339 return mass_spectra_list 340 341 def load(self) -> None: 342 """ """ 343 pass 344 345 def get_ms_raw(self, spectra=None, scan_df=None, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None) -> dict: 346 """ """ 347 # Warn if spectra or scan_df are not None that they are not used for CoreMS HDF5 files and should be rerun after instantiation 348 if spectra is not None or scan_df is not None: 349 SyntaxWarning( 350 "get_ms_raw method for CoreMS HDF5 files can only access saved data, consider rerunning after instantiation." 351 ) 352 ms_unprocessed = {} 353 dict_group_load = self.h5pydata["ms_unprocessed"] 354 dict_group_keys = dict_group_load.keys() 355 for k in dict_group_keys: 356 ms_up_int = dict_group_load[k][:] 357 ms_unprocessed[int(k)] = pd.DataFrame( 358 ms_up_int, columns=["scan", "mz", "intensity"] 359 ) 360 return ms_unprocessed 361 362 def get_scan_df(self, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None) -> pd.DataFrame: 363 scan_info = {} 364 dict_group_load = self.h5pydata["scan_info"] 365 dict_group_keys = dict_group_load.keys() 366 for k in dict_group_keys: 367 scan_info[k] = dict_group_load[k][:] 368 scan_df = pd.DataFrame(scan_info) 369 scan_df.set_index("scan", inplace=True, drop=False) 370 str_df = scan_df.select_dtypes([object]) 371 str_df = str_df.stack().str.decode("utf-8").unstack() 372 for col in str_df: 373 scan_df[col] = str_df[col] 374 375 # Apply time range filtering if specified 376 if time_range is not None: 377 time_ranges = self._normalize_time_range(time_range) 378 mask = pd.Series([False] * len(scan_df), index=scan_df.index) 379 for start_time, end_time in time_ranges: 380 mask |= (scan_df.scan_time >= start_time) & (scan_df.scan_time <= end_time) 381 scan_df = scan_df[mask] 382 383 return scan_df 384 385 def run(self, mass_spectra, load_raw=True, load_light=False) -> None: 386 """Runs the importer functions to populate a LCMS or MassSpectraBase object. 387 388 Notes 389 ----- 390 The following functions are run in order, if the HDF5 file contains the necessary data: 391 1. import_parameters(), which populates the parameters attribute on the LCMS or MassSpectraBase object. 392 2. import_mass_spectra(), which populates the _ms list on the LCMS or MassSpectraBase object. 393 3. import_scan_info(), which populates the _scan_info on the LCMS or MassSpectraBase object. 394 4. import_ms_unprocessed(), which populates the _ms_unprocessed attribute on the LCMS or MassSpectraBase object. 395 5. import_mass_features(), which populates the mass_features attribute on the LCMS or MassSpectraBase object. 396 6. import_eics(), which populates the eics attribute on the LCMS or MassSpectraBase object. 397 7. import_spectral_search_results(), which populates the spectral_search_results attribute on the LCMS or MassSpectraBase object. 398 399 Parameters 400 ---------- 401 mass_spectra : LCMSBase or MassSpectraBase 402 The LCMS or MassSpectraBase object to populate with mass spectra, generally instantiated with only the file_location, analyzer, and instrument_label attributes. 403 load_raw : bool 404 If True, load raw data (unprocessed) from HDF5 files for overall lcms object and individual mass spectra. Default is True. 405 load_light : bool 406 If True, only load the parameters, mass features, and scan info. Default is False. 407 408 Returns 409 ------- 410 None, but populates several attributes on the LCMS or MassSpectraBase object. 411 412 """ 413 if self.parameters_location is not None: 414 # Populate the parameters attribute on the LCMS object 415 self.import_parameters(mass_spectra) 416 417 if "mass_spectra" in self.h5pydata and not load_light: 418 # Populate the _ms list on the LCMS object 419 self.import_mass_spectra(mass_spectra, load_raw=load_raw) 420 421 if "scan_info" in self.h5pydata: 422 # Populate the _scan_info attribute on the LCMS object 423 self.import_scan_info(mass_spectra) 424 425 if "ms_unprocessed" in self.h5pydata and load_raw and not load_light: 426 # Populate the _ms_unprocessed attribute on the LCMS object 427 self.import_ms_unprocessed(mass_spectra) 428 429 if "mass_features" in self.h5pydata: 430 # Populate the mass_features attribute on the LCMS object 431 self.import_mass_features(mass_spectra) 432 433 if "eics" in self.h5pydata and not load_light: 434 # Populate the eics attribute on the LCMS object 435 self.import_eics(mass_spectra) 436 437 if "spectral_search_results" in self.h5pydata and not load_light: 438 # Populate the spectral_search_results attribute on the LCMS object 439 self.import_spectral_search_results(mass_spectra) 440 441 def import_mass_spectra(self, mass_spectra, load_raw=True) -> None: 442 """Imports all mass spectra from the HDF5 file. 443 444 Parameters 445 ---------- 446 mass_spectra : LCMSBase | MassSpectraBase 447 The MassSpectraBase or LCMSBase object to populate with mass spectra. 448 load_raw : bool 449 If True, load raw data (unprocessed) from HDF5 files for overall lcms object and individual mass spectra. Default 450 451 Returns 452 ------- 453 None, but populates the '_ms' list on the LCMSBase or MassSpectraBase 454 object with mass spectra from the HDF5 file. 455 """ 456 for scan_number in self.scan_number_list: 457 mass_spec = self.get_mass_spectrum(scan_number, load_raw=load_raw) 458 mass_spec.scan_number = scan_number 459 mass_spectra.add_mass_spectrum(mass_spec) 460 461 def import_scan_info(self, mass_spectra) -> None: 462 """Imports the scan info from the HDF5 file. 463 464 Parameters 465 ---------- 466 lcms : LCMSBase | MassSpectraBase 467 The MassSpectraBase or LCMSBase objects 468 469 Returns 470 ------- 471 None, but populates the 'scan_df' attribute on the LCMSBase or MassSpectraBase 472 object with a pandas DataFrame of the 'scan_info' from the HDF5 file. 473 474 """ 475 scan_df = self.get_scan_df() 476 mass_spectra.scan_df = scan_df 477 478 def import_ms_unprocessed(self, mass_spectra) -> None: 479 """Imports the unprocessed mass spectra from the HDF5 file. 480 481 Parameters 482 ---------- 483 lcms : LCMSBase | MassSpectraBase 484 The MassSpectraBase or LCMSBase objects 485 486 Returns 487 ------- 488 None, but populates the '_ms_unprocessed' attribute on the LCMSBase or MassSpectraBase 489 object with a dictionary of the 'ms_unprocessed' from the HDF5 file. 490 491 """ 492 ms_unprocessed = self.get_ms_raw() 493 mass_spectra._ms_unprocessed = ms_unprocessed 494 495 def import_parameters(self, mass_spectra) -> None: 496 """Imports the parameters from the HDF5 file. 497 498 Parameters 499 ---------- 500 mass_spectra : LCMSBase | MassSpectraBase 501 The MassSpectraBase or LCMSBase object to populate with parameters. 502 503 Returns 504 ------- 505 None, but populates the 'parameters' attribute on the LCMS or MassSpectraBase 506 object with a dictionary of the 'parameters' from the HDF5 file. 507 508 """ 509 if ".json" == self.parameters_location.suffix: 510 load_and_set_json_parameters_lcms(mass_spectra, self.parameters_location) 511 if ".toml" == self.parameters_location.suffix: 512 load_and_set_toml_parameters_lcms(mass_spectra, self.parameters_location) 513 else: 514 raise Exception( 515 "Parameters file must be in JSON format, TOML format is not yet supported." 516 ) 517 518 def import_mass_features(self, mass_spectra, mf_ids=None) -> None: 519 """Imports the mass features from the HDF5 file. 520 521 Parameters 522 ---------- 523 mass_spectra : LCMSBase | MassSpectraBase 524 The MassSpectraBase or LCMSBase object to populate with mass features. 525 mf_ids : list, optional 526 A list of mass feature IDs to import. If None, all mass features are imported. 527 528 Returns 529 ------- 530 None, but populates the 'mass_features' attribute on the LCMSBase or MassSpectraBase 531 object with a dictionary of the 'mass_features' from the HDF5 file. 532 533 """ 534 dict_group_load = self.h5pydata["mass_features"] 535 dict_group_keys = dict_group_load.keys() 536 for k in dict_group_keys: 537 if mf_ids is not None and int(k) not in mf_ids: 538 continue 539 # Instantiate the MassFeature object 540 mass_feature = LCMSMassFeature( 541 mass_spectra, 542 mz=dict_group_load[k].attrs["_mz_exp"], 543 retention_time=dict_group_load[k].attrs["_retention_time"], 544 intensity=dict_group_load[k].attrs["_intensity"], 545 apex_scan=dict_group_load[k].attrs["_apex_scan"], 546 persistence=dict_group_load[k].attrs["_persistence"], 547 id=int(k), 548 ) 549 550 # Populate additional attributes on the MassFeature object 551 for key in dict_group_load[k].attrs.keys() - { 552 "_mz_exp", 553 "_mz_cal", 554 "_retention_time", 555 "_intensity", 556 "_apex_scan", 557 "_persistence", 558 }: 559 setattr(mass_feature, key, dict_group_load[k].attrs[key]) 560 561 # Populate attributes on MassFeature object that are lists 562 for key in dict_group_load[k].keys(): 563 setattr(mass_feature, key, dict_group_load[k][key][:]) 564 # Convert _noise_score from array to tuple 565 if key == "_noise_score": 566 mass_feature._noise_score = tuple(mass_feature._noise_score) 567 mass_spectra.mass_features[int(k)] = mass_feature 568 569 # Associate mass features with ms1 and ms2 spectra, if available 570 for mf_id in mass_spectra.mass_features.keys(): 571 if mass_spectra.mass_features[mf_id].apex_scan in mass_spectra._ms.keys(): 572 mass_spectra.mass_features[mf_id].mass_spectrum = mass_spectra._ms[ 573 mass_spectra.mass_features[mf_id].apex_scan 574 ] 575 if mass_spectra.mass_features[mf_id].ms2_scan_numbers is not None: 576 for ms2_scan in mass_spectra.mass_features[mf_id].ms2_scan_numbers: 577 if ms2_scan in mass_spectra._ms.keys(): 578 mass_spectra.mass_features[mf_id].ms2_mass_spectra[ms2_scan] = ( 579 mass_spectra._ms[ms2_scan] 580 ) 581 582 def import_eics(self, mass_spectra, mz_list=None, mz_tolerance=0.0001): 583 """Imports the extracted ion chromatograms from the HDF5 file. 584 585 Parameters 586 ---------- 587 mass_spectra : LCMSBase | MassSpectraBase 588 The MassSpectraBase or LCMSBase object to populate with extracted ion chromatograms. 589 mz_list : list of float, optional 590 List of m/z values to load EICs for. If None, loads all EICs. Default is None. 591 mz_tolerance : float, optional 592 Tolerance in Daltons for matching m/z values when mz_list is provided. 593 Default is 0.0001 Da. 594 595 Returns 596 ------- 597 None, but populates the 'eics' attribute on the LCMSBase or MassSpectraBase 598 object with a dictionary of the 'eics' from the HDF5 file. 599 600 """ 601 dict_group_load = self.h5pydata["eics"] 602 dict_group_keys = dict_group_load.keys() 603 604 # Prefilter dict_group_keys if mz_list is provided to EICs within tolerance 605 if mz_list is not None: 606 target_mz_array = np.array(sorted(mz_list)) 607 mzs = [float(k) for k in dict_group_keys if np.abs(float(k)-target_mz_array).min() < mz_tolerance] 608 dict_group_keys = [str(mz) for mz in mzs] 609 610 for k in dict_group_keys: 611 # Check if we should load this EIC (filter by m/z if list provided) 612 eic_mz = dict_group_load[k].attrs["mz"] 613 614 my_eic = EIC_Data( 615 scans=dict_group_load[k]["scans"][:], 616 time=dict_group_load[k]["time"][:], 617 eic=dict_group_load[k]["eic"][:], 618 ) 619 for key in dict_group_load[k].keys(): 620 if key not in ["scans", "time", "eic"]: 621 setattr(my_eic, key, dict_group_load[k][key][:]) 622 # if key is apexes, convert to a tuple of a list 623 if key == "apexes" and len(my_eic.apexes) > 0: 624 my_eic.apexes = [tuple(x) for x in my_eic.apexes] 625 # Add to mass_spectra object 626 mass_spectra.eics[eic_mz] = my_eic 627 628 # Associate EICs with mass features using tolerance-based matching 629 mass_spectra.associate_eics_with_mass_features() 630 631 @staticmethod 632 def _load_eics_from_hdf5_group(eics_group, lcms_obj, mz_filter=None): 633 """Load EICs from an HDF5 group. 634 635 This is a static helper method that can be reused to load EIC data 636 from any HDF5 group in a consistent format. 637 638 Parameters 639 ---------- 640 eics_group : h5py.Group 641 The HDF5 group containing EIC data. 642 lcms_obj : LCMSBase 643 The LCMS object to associate EICs with (for reference, not modified). 644 mz_filter : list, optional 645 List of m/z values to load. If None, loads all EICs. Default is None. 646 Uses tolerance-based matching (0.0001). 647 648 Returns 649 ------- 650 dict 651 Dictionary of EIC_Data objects keyed by m/z value. 652 """ 653 from corems.mass_spectra.factory.chromat_data import EIC_Data 654 655 loaded_eics = {} 656 tolerance = 0.0001 657 658 for eic_key_str in eics_group.keys(): 659 eic_mz = float(eic_key_str) if eic_key_str.replace('.', '', 1).replace('-', '', 1).isdigit() else eics_group[eic_key_str].attrs.get("mz") 660 661 # If mz_filter is provided, check if this EIC matches any requested m/z 662 if mz_filter is not None: 663 if not any(abs(eic_mz - mz) < tolerance for mz in mz_filter): 664 continue 665 666 eic_data = eics_group[eic_key_str] 667 668 # Create EIC_Data object from datasets 669 eic = EIC_Data( 670 scans=list(eic_data["scans"][:]) if "scans" in eic_data else [], 671 time=list(eic_data["time"][:]) if "time" in eic_data else [], 672 eic=list(eic_data["eic"][:]) if "eic" in eic_data else [], 673 apexes=list(eic_data["apexes"][:]) if "apexes" in eic_data else [], 674 ) 675 676 # Load any additional datasets 677 for key in eic_data.keys(): 678 if key not in ["scans", "time", "eic", "apexes"]: 679 setattr(eic, key, eic_data[key][:]) 680 681 loaded_eics[eic_mz] = eic 682 683 return loaded_eics 684 685 def import_spectral_search_results(self, mass_spectra): 686 """Imports the spectral search results from the HDF5 file. 687 688 Parameters 689 ---------- 690 mass_spectra : LCMSBase | MassSpectraBase 691 The MassSpectraBase or LCMSBase object to populate with spectral search results. 692 693 Returns 694 ------- 695 None, but populates the 'spectral_search_results' attribute on the LCMSBase or MassSpectraBase 696 object with a dictionary of the 'spectral_search_results' from the HDF5 file. 697 698 """ 699 overall_results_dict = {} 700 ms2_results_load = self.h5pydata["spectral_search_results"] 701 for k in ms2_results_load.keys(): 702 overall_results_dict[int(k)] = {} 703 for k2 in ms2_results_load[k].keys(): 704 ms2_search_res = SpectrumSearchResults( 705 query_spectrum=mass_spectra._ms[int(k)], 706 precursor_mz=ms2_results_load[k][k2].attrs["precursor_mz"], 707 spectral_similarity_search_results={}, 708 ) 709 710 for key in ms2_results_load[k][k2].keys() - {"precursor_mz"}: 711 data = list(ms2_results_load[k][k2][key][:]) 712 if data and isinstance(data[0], bytes): 713 data = [x.decode("utf-8") for x in data] 714 setattr(ms2_search_res, key, data) 715 716 overall_results_dict[int(k)][ 717 ms2_results_load[k][k2].attrs["precursor_mz"] 718 ] = ms2_search_res 719 720 # add to mass_spectra 721 mass_spectra.spectral_search_results.update(overall_results_dict) 722 723 # If there are mass features, associate the results with each mass feature 724 if len(mass_spectra.mass_features) > 0: 725 for mass_feature_id, mass_feature in mass_spectra.mass_features.items(): 726 scan_ids = mass_feature.ms2_scan_numbers 727 for ms2_scan_id in scan_ids: 728 precursor_mz = mass_feature.mz 729 try: 730 mass_spectra.spectral_search_results[ms2_scan_id][precursor_mz] 731 except KeyError: 732 pass 733 else: 734 mass_spectra.mass_features[ 735 mass_feature_id 736 ].ms2_similarity_results.append( 737 mass_spectra.spectral_search_results[ms2_scan_id][ 738 precursor_mz 739 ] 740 ) 741 742 def get_mass_spectra_obj(self, load_raw=True, load_light=False, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None) -> MassSpectraBase: 743 """ 744 Return mass spectra data object, populating the _ms list on MassSpectraBase object from the HDF5 file. 745 746 Parameters 747 ---------- 748 load_raw : bool 749 If True, load raw data (unprocessed) from HDF5 files for overall spectra object and individual mass spectra. Default is True. 750 load_light : bool 751 If True, only load the parameters, mass features, and scan info. Default is False. 752 time_range : tuple or list of tuples, optional 753 Retention time range(s) to load. Can be: 754 - Single range: (start_time, end_time) in minutes 755 - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes 756 If None, loads all scans. Note: For HDF5 files, this parameter is accepted for 757 interface consistency but not currently used in filtering. 758 759 """ 760 # Instantiate the LCMS object 761 spectra_obj = MassSpectraBase( 762 file_location=self.file_location, 763 analyzer=self.analyzer, 764 instrument_label=self.instrument_label, 765 sample_name=self.sample_name, 766 ) 767 768 # This will populate the _ms list on the LCMS or MassSpectraBase object 769 self.run(spectra_obj, load_raw=load_raw, load_light=load_light) 770 771 return spectra_obj 772 773 def get_lcms_obj( 774 self, load_raw=True, load_light=False, use_original_parser=True, raw_file_path=None, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None 775 ) -> LCMSBase: 776 """ 777 Return LCMSBase object, populating attributes on the LCMSBase object from the HDF5 file. 778 779 Parameters 780 ---------- 781 load_raw : bool 782 If True, load raw data (unprocessed) from HDF5 files for overall lcms object and individual mass spectra. Default is True. 783 load_light : bool 784 If True, only load the parameters, mass features, and scan info. Default is False. 785 use_original_parser : bool 786 If True, use the original parser to populate the LCMS object. Default is True. 787 raw_file_path : str 788 The location of the raw file to parse if attempting to use original parser. 789 Default is None, which attempts to get the raw file path from the HDF5 file. 790 If the original file path has moved, this parameter can be used to specify the new location. 791 time_range : tuple or list of tuples, optional 792 Retention time range(s) to load. Can be: 793 - Single range: (start_time, end_time) in minutes 794 - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes 795 If None, loads all scans. Note: For HDF5 files, this parameter is accepted for 796 interface consistency. If use_original_parser=True, time_range can be passed to the 797 original parser for filtering. 798 """ 799 # Instantiate the LCMS object 800 lcms_obj = LCMSBase( 801 file_location=self.file_location, 802 analyzer=self.analyzer, 803 instrument_label=self.instrument_label, 804 sample_name=self.sample_name, 805 ) 806 807 # This will populate the majority of the attributes on the LCMS object 808 self.run(lcms_obj, load_raw=load_raw, load_light=load_light) 809 810 # Set final attributes of the LCMS object 811 lcms_obj.polarity = self.h5pydata.attrs["polarity"] 812 lcms_obj._scans_number_list = list(lcms_obj.scan_df.scan) 813 lcms_obj._retention_time_list = list(lcms_obj.scan_df.scan_time) 814 lcms_obj._tic_list = list(lcms_obj.scan_df.tic) 815 816 # If use_original_parser is True, instantiate the original parser and populate the LCMS object 817 if use_original_parser: 818 lcms_obj = self.add_original_parser(lcms_obj, raw_file_path=raw_file_path) 819 else: 820 lcms_obj.spectra_parser_class = self.__class__ 821 822 return lcms_obj 823 824 def get_raw_file_location(self): 825 """ 826 Get the raw file location from the HDF5 file attributes. 827 828 Returns 829 ------- 830 str 831 The raw file location. 832 """ 833 if "original_file_location" in self.h5pydata.attrs: 834 return self.h5pydata.attrs["original_file_location"] 835 else: 836 return None 837 838 def add_original_parser(self, mass_spectra, raw_file_path=None): 839 """ 840 Add the original parser to the mass spectra object. 841 842 Parameters 843 ---------- 844 mass_spectra : MassSpectraBase | LCMSBase 845 The MassSpectraBase or LCMSBase object to add the original parser to. 846 raw_file_path : str 847 The location of the raw file to parse. Default is None, which attempts to get the raw file path from the HDF5 file. 848 """ 849 # Get the original parser type 850 og_parser_type = self.h5pydata.attrs["parser_type"] 851 852 # If raw_file_path is None, get it from the HDF5 file attributes 853 if raw_file_path is None: 854 raw_file_path = self.get_raw_file_location() 855 if raw_file_path is None: 856 raise ValueError( 857 "Raw file path not found in HDF5 file attributes, cannot instantiate original parser." 858 ) 859 860 # Set the raw file path on the mass_spectra object so the parser knows where to find the raw file 861 mass_spectra.raw_file_location = raw_file_path 862 863 if og_parser_type == "ImportMassSpectraThermoMSFileReader": 864 # Check that the parser can be instantiated with the raw file path 865 parser_class = ImportMassSpectraThermoMSFileReader 866 elif og_parser_type == "MZMLSpectraParser": 867 # Check that the parser can be instantiated with the raw file path 868 parser_class = MZMLSpectraParser 869 870 # Set the spectra parser class on the mass_spectra object so the spectra_parser property can be used with the original parser 871 mass_spectra.spectra_parser_class = parser_class 872 873 return mass_spectra 874 875 def get_original_creation_time(self): 876 """ 877 Get the creation time of the original raw data file. 878 879 First checks if creation_time is saved in the HDF5 file attributes. 880 If not found, attempts to instantiate the original parser and get the creation time. 881 882 Returns 883 ------- 884 datetime 885 The creation time of the original raw data file, or None if not available. 886 """ 887 # Check if creation_time is saved in HDF5 attributes 888 if "creation_time" in self.h5pydata.attrs: 889 from datetime import datetime 890 return datetime.fromisoformat(self.h5pydata.attrs["creation_time"]) 891 892 # Fall back to using original parser to get creation time 893 try: 894 # Get the original parser type and raw file path 895 og_parser_type = self.h5pydata.attrs.get("parser_type") 896 raw_file_path = self.get_raw_file_location() 897 898 if og_parser_type is None or raw_file_path is None: 899 warnings.warn( 900 "Cannot retrieve creation time: parser_type or original_file_location not found in HDF5 attributes." 901 ) 902 return None 903 904 # Check if raw file exists 905 from pathlib import Path 906 if not Path(raw_file_path).exists(): 907 warnings.warn( 908 f"Cannot retrieve creation time: original raw file not found at {raw_file_path}" 909 ) 910 return None 911 912 # Instantiate the original parser 913 if og_parser_type == "ImportMassSpectraThermoMSFileReader": 914 parser = ImportMassSpectraThermoMSFileReader(raw_file_path) 915 elif og_parser_type == "MZMLSpectraParser": 916 parser = MZMLSpectraParser(raw_file_path) 917 else: 918 warnings.warn( 919 f"Unknown parser type: {og_parser_type}, cannot retrieve creation time." 920 ) 921 return None 922 923 # Get creation time from parser 924 return parser.get_creation_time() 925 926 except Exception as e: 927 warnings.warn( 928 f"Failed to retrieve creation time from original parser: {e}" 929 ) 930 return None 931 932 def get_creation_time(self): 933 """ 934 Get the creation time of the original raw data file. 935 936 This is an alias for get_original_creation_time() for backward compatibility. 937 938 Returns 939 ------- 940 datetime 941 The creation time of the original raw data file, or None if not available. 942 """ 943 return self.get_original_creation_time() 944 945 def get_instrument_info(self): 946 """ 947 Raise a NotImplemented Warning, as instrument info is not available in CoreMS HDF5 files and returning None. 948 """ 949 warnings.warn( 950 "Instrument info is not available in CoreMS HDF5 files, returning None." 951 "This should be accessed through the original parser.", 952 ) 953 return None 954 955 def get_scans_in_time_range( 956 self, 957 time_range: Union[Tuple[float, float], List[Tuple[float, float]]], 958 ms_level: Optional[int] = None 959 ) -> List[int]: 960 """Return scan numbers within specified retention time range(s). 961 962 Parameters 963 ---------- 964 time_range : tuple or list of tuples 965 Retention time range(s) in minutes. Can be: 966 - Single range: (start_time, end_time) 967 - Multiple ranges: [(start1, end1), (start2, end2), ...] 968 ms_level : int, optional 969 If specified, only return scans of this MS level (e.g., 1 for MS1, 2 for MS2). 970 If None, returns scans of all MS levels. 971 972 Returns 973 ------- 974 list of int 975 List of scan numbers within the specified time range(s) and MS level. 976 """ 977 # Normalize time range to list of tuples 978 time_ranges = self._normalize_time_range(time_range) 979 980 # Get all scan data 981 scan_df = self.get_scan_df() 982 983 # Filter by time range 984 mask = pd.Series([False] * len(scan_df), index=scan_df.index) 985 for start_time, end_time in time_ranges: 986 mask |= (scan_df.scan_time >= start_time) & (scan_df.scan_time <= end_time) 987 988 filtered_df = scan_df[mask] 989 990 # Filter by MS level if specified 991 if ms_level is not None: 992 filtered_df = filtered_df[filtered_df.ms_level == ms_level] 993 994 return filtered_df.scan.tolist()
Class to read CoreMS HDF5 files and populate a LCMS or MassSpectraBase object.
Parameters
- file_location (str): The location of the HDF5 file to read, including the suffix.
Attributes
- file_location (str): The location of the HDF5 file to read.
- h5pydata (h5py.File): The HDF5 file object.
- scans (list): A list of the location of individual mass spectra within the HDF5 file.
- scan_number_list (list): A list of the scan numbers of the mass spectra within the HDF5 file.
- parameters_location (str): The location of the parameters file (json or toml).
Methods
- import_mass_spectra(mass_spectra). Imports all mass spectra from the HDF5 file onto the LCMS or MassSpectraBase object.
- get_mass_spectrum_from_scan(scan_number). Return mass spectrum data object from scan number.
- load(). Placeholder method to meet the requirements of the SpectraParserInterface.
- run(mass_spectra). Runs the importer functions to populate a LCMS or MassSpectraBase object.
- import_scan_info(mass_spectra). Imports the scan info from the HDF5 file to populate the _scan_info attribute on the LCMS or MassSpectraBase object
- import_ms_unprocessed(mass_spectra). Imports the unprocessed mass spectra from the HDF5 file to populate the _ms_unprocessed attribute on the LCMS or MassSpectraBase object
- import_parameters(mass_spectra). Imports the parameters from the HDF5 file to populate the parameters attribute on the LCMS or MassSpectraBase object
- import_mass_features(mass_spectra). Imports the mass features from the HDF5 file to populate the mass_features attribute on the LCMS or MassSpectraBase object
- import_eics(mass_spectra). Imports the extracted ion chromatograms from the HDF5 file to populate the eics attribute on the LCMS or MassSpectraBase object
- import_spectral_search_results(mass_spectra). Imports the spectral search results from the HDF5 file to populate the spectral_search_results attribute on the LCMS or MassSpectraBase object
- get_mass_spectra_obj(). Return mass spectra data object, populating the _ms list on the LCMS or MassSpectraBase object from the HDF5 file
- get_lcms_obj(). Return LCMSBase object, populating the majority of the attributes on the LCMS object from the HDF5 file
ReadCoreMSHDFMassSpectra(file_location: str)
259 def __init__(self, file_location: str): 260 Thread.__init__(self) 261 ReadCoreMSHDF_MassSpectrum.__init__(self, file_location) 262 263 # override the scans attribute on ReadCoreMSHDF_MassSpectrum class to expect a nested location within the HDF5 file 264 self.scans = [ 265 "mass_spectra/" + x for x in list(self.h5pydata["mass_spectra"].keys()) 266 ] 267 self.scan_number_list = sorted( 268 [int(float(i)) for i in list(self.h5pydata["mass_spectra"].keys())] 269 ) 270 271 # set the location of the parameters file (json or toml) 272 add_files = [ 273 x 274 for x in self.file_location.parent.glob( 275 self.file_location.name.replace(".hdf5", ".*") 276 ) 277 if x.suffix != ".hdf5" 278 ] 279 if len([x for x in add_files if x.suffix == ".json"]) > 0: 280 self.parameters_location = [x for x in add_files if x.suffix == ".json"][0] 281 elif len([x for x in add_files if x.suffix == ".toml"]) > 0: 282 self.parameters_location = [x for x in add_files if x.suffix == ".toml"][0] 283 else: 284 self.parameters_location = None
This constructor should always be called with keyword arguments. Arguments are:
group should be None; reserved for future extension when a ThreadGroup class is implemented.
target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.
name is the thread name. By default, a unique name is constructed of the form "Thread-N" where N is a small decimal number.
args is a list or tuple of arguments for the target invocation. Defaults to ().
kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.
If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.
def
close(self):
296 def close(self): 297 """Explicitly close the HDF5 file.""" 298 if hasattr(self, 'h5pydata') and self.h5pydata is not None: 299 self.h5pydata.close()
Explicitly close the HDF5 file.
def
get_mass_spectrum_from_scan(self, scan_number):
301 def get_mass_spectrum_from_scan(self, scan_number): 302 """Return mass spectrum data object from scan number.""" 303 if scan_number in self.scan_number_list: 304 mass_spec = self.get_mass_spectrum(scan_number) 305 return mass_spec 306 else: 307 raise Exception("Scan number not found in HDF5 file.")
Return mass spectrum data object from scan number.
def
get_mass_spectra_from_scan_list(self, scan_list, spectrum_mode, auto_process=True):
309 def get_mass_spectra_from_scan_list( 310 self, scan_list, spectrum_mode, auto_process=True 311 ): 312 """Return a list of mass spectrum data objects from a list of scan numbers. 313 314 Parameters 315 ---------- 316 scan_list : list 317 A list of scan numbers to retrieve mass spectra for. 318 spectrum_mode : str 319 The spectrum mode to use when retrieving the mass spectra. 320 Note that this parameter is not used for CoreMS HDF5 files, as the spectra are already processed and only 321 centroided spectra are saved. 322 auto_process : bool 323 If True, automatically process the mass spectra when retrieving them. 324 Note that this parameter is not used for CoreMS HDF5 files, as the spectra are already processed and only 325 centroided spectra are saved. 326 327 Returns 328 ------- 329 list 330 A list of mass spectrum data objects corresponding to the provided scan numbers. 331 """ 332 mass_spectra_list = [] 333 for scan_number in scan_list: 334 if scan_number in self.scan_number_list: 335 mass_spec = self.get_mass_spectrum_from_scan(scan_number) 336 mass_spectra_list.append(mass_spec) 337 else: 338 warnings.warn(f"Scan number {scan_number} not found in HDF5 file.") 339 return mass_spectra_list
Return a list of mass spectrum data objects from a list of scan numbers.
Parameters
- scan_list (list): A list of scan numbers to retrieve mass spectra for.
- spectrum_mode (str): The spectrum mode to use when retrieving the mass spectra. Note that this parameter is not used for CoreMS HDF5 files, as the spectra are already processed and only centroided spectra are saved.
- auto_process (bool): If True, automatically process the mass spectra when retrieving them. Note that this parameter is not used for CoreMS HDF5 files, as the spectra are already processed and only centroided spectra are saved.
Returns
- list: A list of mass spectrum data objects corresponding to the provided scan numbers.
def
get_ms_raw( self, spectra=None, scan_df=None, time_range: Union[Tuple[float, float], List[Tuple[float, float]], NoneType] = None) -> dict:
345 def get_ms_raw(self, spectra=None, scan_df=None, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None) -> dict: 346 """ """ 347 # Warn if spectra or scan_df are not None that they are not used for CoreMS HDF5 files and should be rerun after instantiation 348 if spectra is not None or scan_df is not None: 349 SyntaxWarning( 350 "get_ms_raw method for CoreMS HDF5 files can only access saved data, consider rerunning after instantiation." 351 ) 352 ms_unprocessed = {} 353 dict_group_load = self.h5pydata["ms_unprocessed"] 354 dict_group_keys = dict_group_load.keys() 355 for k in dict_group_keys: 356 ms_up_int = dict_group_load[k][:] 357 ms_unprocessed[int(k)] = pd.DataFrame( 358 ms_up_int, columns=["scan", "mz", "intensity"] 359 ) 360 return ms_unprocessed
def
get_scan_df( self, time_range: Union[Tuple[float, float], List[Tuple[float, float]], NoneType] = None) -> pandas.DataFrame:
362 def get_scan_df(self, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None) -> pd.DataFrame: 363 scan_info = {} 364 dict_group_load = self.h5pydata["scan_info"] 365 dict_group_keys = dict_group_load.keys() 366 for k in dict_group_keys: 367 scan_info[k] = dict_group_load[k][:] 368 scan_df = pd.DataFrame(scan_info) 369 scan_df.set_index("scan", inplace=True, drop=False) 370 str_df = scan_df.select_dtypes([object]) 371 str_df = str_df.stack().str.decode("utf-8").unstack() 372 for col in str_df: 373 scan_df[col] = str_df[col] 374 375 # Apply time range filtering if specified 376 if time_range is not None: 377 time_ranges = self._normalize_time_range(time_range) 378 mask = pd.Series([False] * len(scan_df), index=scan_df.index) 379 for start_time, end_time in time_ranges: 380 mask |= (scan_df.scan_time >= start_time) & (scan_df.scan_time <= end_time) 381 scan_df = scan_df[mask] 382 383 return scan_df
Return scan data as a pandas DataFrame.
Parameters
- 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
- pd.DataFrame: DataFrame containing scan information, optionally filtered by time range.
def
run(self, mass_spectra, load_raw=True, load_light=False) -> None:
385 def run(self, mass_spectra, load_raw=True, load_light=False) -> None: 386 """Runs the importer functions to populate a LCMS or MassSpectraBase object. 387 388 Notes 389 ----- 390 The following functions are run in order, if the HDF5 file contains the necessary data: 391 1. import_parameters(), which populates the parameters attribute on the LCMS or MassSpectraBase object. 392 2. import_mass_spectra(), which populates the _ms list on the LCMS or MassSpectraBase object. 393 3. import_scan_info(), which populates the _scan_info on the LCMS or MassSpectraBase object. 394 4. import_ms_unprocessed(), which populates the _ms_unprocessed attribute on the LCMS or MassSpectraBase object. 395 5. import_mass_features(), which populates the mass_features attribute on the LCMS or MassSpectraBase object. 396 6. import_eics(), which populates the eics attribute on the LCMS or MassSpectraBase object. 397 7. import_spectral_search_results(), which populates the spectral_search_results attribute on the LCMS or MassSpectraBase object. 398 399 Parameters 400 ---------- 401 mass_spectra : LCMSBase or MassSpectraBase 402 The LCMS or MassSpectraBase object to populate with mass spectra, generally instantiated with only the file_location, analyzer, and instrument_label attributes. 403 load_raw : bool 404 If True, load raw data (unprocessed) from HDF5 files for overall lcms object and individual mass spectra. Default is True. 405 load_light : bool 406 If True, only load the parameters, mass features, and scan info. Default is False. 407 408 Returns 409 ------- 410 None, but populates several attributes on the LCMS or MassSpectraBase object. 411 412 """ 413 if self.parameters_location is not None: 414 # Populate the parameters attribute on the LCMS object 415 self.import_parameters(mass_spectra) 416 417 if "mass_spectra" in self.h5pydata and not load_light: 418 # Populate the _ms list on the LCMS object 419 self.import_mass_spectra(mass_spectra, load_raw=load_raw) 420 421 if "scan_info" in self.h5pydata: 422 # Populate the _scan_info attribute on the LCMS object 423 self.import_scan_info(mass_spectra) 424 425 if "ms_unprocessed" in self.h5pydata and load_raw and not load_light: 426 # Populate the _ms_unprocessed attribute on the LCMS object 427 self.import_ms_unprocessed(mass_spectra) 428 429 if "mass_features" in self.h5pydata: 430 # Populate the mass_features attribute on the LCMS object 431 self.import_mass_features(mass_spectra) 432 433 if "eics" in self.h5pydata and not load_light: 434 # Populate the eics attribute on the LCMS object 435 self.import_eics(mass_spectra) 436 437 if "spectral_search_results" in self.h5pydata and not load_light: 438 # Populate the spectral_search_results attribute on the LCMS object 439 self.import_spectral_search_results(mass_spectra)
Runs the importer functions to populate a LCMS or MassSpectraBase object.
Notes
The following functions are run in order, if the HDF5 file contains the necessary data:
- import_parameters(), which populates the parameters attribute on the LCMS or MassSpectraBase object.
- import_mass_spectra(), which populates the _ms list on the LCMS or MassSpectraBase object.
- import_scan_info(), which populates the _scan_info on the LCMS or MassSpectraBase object.
- import_ms_unprocessed(), which populates the _ms_unprocessed attribute on the LCMS or MassSpectraBase object.
- import_mass_features(), which populates the mass_features attribute on the LCMS or MassSpectraBase object.
- import_eics(), which populates the eics attribute on the LCMS or MassSpectraBase object.
- import_spectral_search_results(), which populates the spectral_search_results attribute on the LCMS or MassSpectraBase object.
Parameters
- mass_spectra (LCMSBase or MassSpectraBase): The LCMS or MassSpectraBase object to populate with mass spectra, generally instantiated with only the file_location, analyzer, and instrument_label attributes.
- load_raw (bool): If True, load raw data (unprocessed) from HDF5 files for overall lcms object and individual mass spectra. Default is True.
- load_light (bool): If True, only load the parameters, mass features, and scan info. Default is False.
Returns
- None, but populates several attributes on the LCMS or MassSpectraBase object.
def
import_mass_spectra(self, mass_spectra, load_raw=True) -> None:
441 def import_mass_spectra(self, mass_spectra, load_raw=True) -> None: 442 """Imports all mass spectra from the HDF5 file. 443 444 Parameters 445 ---------- 446 mass_spectra : LCMSBase | MassSpectraBase 447 The MassSpectraBase or LCMSBase object to populate with mass spectra. 448 load_raw : bool 449 If True, load raw data (unprocessed) from HDF5 files for overall lcms object and individual mass spectra. Default 450 451 Returns 452 ------- 453 None, but populates the '_ms' list on the LCMSBase or MassSpectraBase 454 object with mass spectra from the HDF5 file. 455 """ 456 for scan_number in self.scan_number_list: 457 mass_spec = self.get_mass_spectrum(scan_number, load_raw=load_raw) 458 mass_spec.scan_number = scan_number 459 mass_spectra.add_mass_spectrum(mass_spec)
Imports all mass spectra from the HDF5 file.
Parameters
- mass_spectra (LCMSBase | MassSpectraBase): The MassSpectraBase or LCMSBase object to populate with mass spectra.
- load_raw (bool): If True, load raw data (unprocessed) from HDF5 files for overall lcms object and individual mass spectra. Default
Returns
- None, but populates the '_ms' list on the LCMSBase or MassSpectraBase
- object with mass spectra from the HDF5 file.
def
import_scan_info(self, mass_spectra) -> None:
461 def import_scan_info(self, mass_spectra) -> None: 462 """Imports the scan info from the HDF5 file. 463 464 Parameters 465 ---------- 466 lcms : LCMSBase | MassSpectraBase 467 The MassSpectraBase or LCMSBase objects 468 469 Returns 470 ------- 471 None, but populates the 'scan_df' attribute on the LCMSBase or MassSpectraBase 472 object with a pandas DataFrame of the 'scan_info' from the HDF5 file. 473 474 """ 475 scan_df = self.get_scan_df() 476 mass_spectra.scan_df = scan_df
Imports the scan info from the HDF5 file.
Parameters
- lcms (LCMSBase | MassSpectraBase): The MassSpectraBase or LCMSBase objects
Returns
- None, but populates the 'scan_df' attribute on the LCMSBase or MassSpectraBase
- object with a pandas DataFrame of the 'scan_info' from the HDF5 file.
def
import_ms_unprocessed(self, mass_spectra) -> None:
478 def import_ms_unprocessed(self, mass_spectra) -> None: 479 """Imports the unprocessed mass spectra from the HDF5 file. 480 481 Parameters 482 ---------- 483 lcms : LCMSBase | MassSpectraBase 484 The MassSpectraBase or LCMSBase objects 485 486 Returns 487 ------- 488 None, but populates the '_ms_unprocessed' attribute on the LCMSBase or MassSpectraBase 489 object with a dictionary of the 'ms_unprocessed' from the HDF5 file. 490 491 """ 492 ms_unprocessed = self.get_ms_raw() 493 mass_spectra._ms_unprocessed = ms_unprocessed
Imports the unprocessed mass spectra from the HDF5 file.
Parameters
- lcms (LCMSBase | MassSpectraBase): The MassSpectraBase or LCMSBase objects
Returns
- None, but populates the '_ms_unprocessed' attribute on the LCMSBase or MassSpectraBase
- object with a dictionary of the 'ms_unprocessed' from the HDF5 file.
def
import_parameters(self, mass_spectra) -> None:
495 def import_parameters(self, mass_spectra) -> None: 496 """Imports the parameters from the HDF5 file. 497 498 Parameters 499 ---------- 500 mass_spectra : LCMSBase | MassSpectraBase 501 The MassSpectraBase or LCMSBase object to populate with parameters. 502 503 Returns 504 ------- 505 None, but populates the 'parameters' attribute on the LCMS or MassSpectraBase 506 object with a dictionary of the 'parameters' from the HDF5 file. 507 508 """ 509 if ".json" == self.parameters_location.suffix: 510 load_and_set_json_parameters_lcms(mass_spectra, self.parameters_location) 511 if ".toml" == self.parameters_location.suffix: 512 load_and_set_toml_parameters_lcms(mass_spectra, self.parameters_location) 513 else: 514 raise Exception( 515 "Parameters file must be in JSON format, TOML format is not yet supported." 516 )
Imports the parameters from the HDF5 file.
Parameters
- mass_spectra (LCMSBase | MassSpectraBase): The MassSpectraBase or LCMSBase object to populate with parameters.
Returns
- None, but populates the 'parameters' attribute on the LCMS or MassSpectraBase
- object with a dictionary of the 'parameters' from the HDF5 file.
def
import_mass_features(self, mass_spectra, mf_ids=None) -> None:
518 def import_mass_features(self, mass_spectra, mf_ids=None) -> None: 519 """Imports the mass features from the HDF5 file. 520 521 Parameters 522 ---------- 523 mass_spectra : LCMSBase | MassSpectraBase 524 The MassSpectraBase or LCMSBase object to populate with mass features. 525 mf_ids : list, optional 526 A list of mass feature IDs to import. If None, all mass features are imported. 527 528 Returns 529 ------- 530 None, but populates the 'mass_features' attribute on the LCMSBase or MassSpectraBase 531 object with a dictionary of the 'mass_features' from the HDF5 file. 532 533 """ 534 dict_group_load = self.h5pydata["mass_features"] 535 dict_group_keys = dict_group_load.keys() 536 for k in dict_group_keys: 537 if mf_ids is not None and int(k) not in mf_ids: 538 continue 539 # Instantiate the MassFeature object 540 mass_feature = LCMSMassFeature( 541 mass_spectra, 542 mz=dict_group_load[k].attrs["_mz_exp"], 543 retention_time=dict_group_load[k].attrs["_retention_time"], 544 intensity=dict_group_load[k].attrs["_intensity"], 545 apex_scan=dict_group_load[k].attrs["_apex_scan"], 546 persistence=dict_group_load[k].attrs["_persistence"], 547 id=int(k), 548 ) 549 550 # Populate additional attributes on the MassFeature object 551 for key in dict_group_load[k].attrs.keys() - { 552 "_mz_exp", 553 "_mz_cal", 554 "_retention_time", 555 "_intensity", 556 "_apex_scan", 557 "_persistence", 558 }: 559 setattr(mass_feature, key, dict_group_load[k].attrs[key]) 560 561 # Populate attributes on MassFeature object that are lists 562 for key in dict_group_load[k].keys(): 563 setattr(mass_feature, key, dict_group_load[k][key][:]) 564 # Convert _noise_score from array to tuple 565 if key == "_noise_score": 566 mass_feature._noise_score = tuple(mass_feature._noise_score) 567 mass_spectra.mass_features[int(k)] = mass_feature 568 569 # Associate mass features with ms1 and ms2 spectra, if available 570 for mf_id in mass_spectra.mass_features.keys(): 571 if mass_spectra.mass_features[mf_id].apex_scan in mass_spectra._ms.keys(): 572 mass_spectra.mass_features[mf_id].mass_spectrum = mass_spectra._ms[ 573 mass_spectra.mass_features[mf_id].apex_scan 574 ] 575 if mass_spectra.mass_features[mf_id].ms2_scan_numbers is not None: 576 for ms2_scan in mass_spectra.mass_features[mf_id].ms2_scan_numbers: 577 if ms2_scan in mass_spectra._ms.keys(): 578 mass_spectra.mass_features[mf_id].ms2_mass_spectra[ms2_scan] = ( 579 mass_spectra._ms[ms2_scan] 580 )
Imports the mass features from the HDF5 file.
Parameters
- mass_spectra (LCMSBase | MassSpectraBase): The MassSpectraBase or LCMSBase object to populate with mass features.
- mf_ids (list, optional): A list of mass feature IDs to import. If None, all mass features are imported.
Returns
- None, but populates the 'mass_features' attribute on the LCMSBase or MassSpectraBase
- object with a dictionary of the 'mass_features' from the HDF5 file.
def
import_eics(self, mass_spectra, mz_list=None, mz_tolerance=0.0001):
582 def import_eics(self, mass_spectra, mz_list=None, mz_tolerance=0.0001): 583 """Imports the extracted ion chromatograms from the HDF5 file. 584 585 Parameters 586 ---------- 587 mass_spectra : LCMSBase | MassSpectraBase 588 The MassSpectraBase or LCMSBase object to populate with extracted ion chromatograms. 589 mz_list : list of float, optional 590 List of m/z values to load EICs for. If None, loads all EICs. Default is None. 591 mz_tolerance : float, optional 592 Tolerance in Daltons for matching m/z values when mz_list is provided. 593 Default is 0.0001 Da. 594 595 Returns 596 ------- 597 None, but populates the 'eics' attribute on the LCMSBase or MassSpectraBase 598 object with a dictionary of the 'eics' from the HDF5 file. 599 600 """ 601 dict_group_load = self.h5pydata["eics"] 602 dict_group_keys = dict_group_load.keys() 603 604 # Prefilter dict_group_keys if mz_list is provided to EICs within tolerance 605 if mz_list is not None: 606 target_mz_array = np.array(sorted(mz_list)) 607 mzs = [float(k) for k in dict_group_keys if np.abs(float(k)-target_mz_array).min() < mz_tolerance] 608 dict_group_keys = [str(mz) for mz in mzs] 609 610 for k in dict_group_keys: 611 # Check if we should load this EIC (filter by m/z if list provided) 612 eic_mz = dict_group_load[k].attrs["mz"] 613 614 my_eic = EIC_Data( 615 scans=dict_group_load[k]["scans"][:], 616 time=dict_group_load[k]["time"][:], 617 eic=dict_group_load[k]["eic"][:], 618 ) 619 for key in dict_group_load[k].keys(): 620 if key not in ["scans", "time", "eic"]: 621 setattr(my_eic, key, dict_group_load[k][key][:]) 622 # if key is apexes, convert to a tuple of a list 623 if key == "apexes" and len(my_eic.apexes) > 0: 624 my_eic.apexes = [tuple(x) for x in my_eic.apexes] 625 # Add to mass_spectra object 626 mass_spectra.eics[eic_mz] = my_eic 627 628 # Associate EICs with mass features using tolerance-based matching 629 mass_spectra.associate_eics_with_mass_features()
Imports the extracted ion chromatograms from the HDF5 file.
Parameters
- mass_spectra (LCMSBase | MassSpectraBase): The MassSpectraBase or LCMSBase object to populate with extracted ion chromatograms.
- mz_list (list of float, optional): List of m/z values to load EICs for. If None, loads all EICs. Default is None.
- mz_tolerance (float, optional): Tolerance in Daltons for matching m/z values when mz_list is provided. Default is 0.0001 Da.
Returns
- None, but populates the 'eics' attribute on the LCMSBase or MassSpectraBase
- object with a dictionary of the 'eics' from the HDF5 file.
def
import_spectral_search_results(self, mass_spectra):
685 def import_spectral_search_results(self, mass_spectra): 686 """Imports the spectral search results from the HDF5 file. 687 688 Parameters 689 ---------- 690 mass_spectra : LCMSBase | MassSpectraBase 691 The MassSpectraBase or LCMSBase object to populate with spectral search results. 692 693 Returns 694 ------- 695 None, but populates the 'spectral_search_results' attribute on the LCMSBase or MassSpectraBase 696 object with a dictionary of the 'spectral_search_results' from the HDF5 file. 697 698 """ 699 overall_results_dict = {} 700 ms2_results_load = self.h5pydata["spectral_search_results"] 701 for k in ms2_results_load.keys(): 702 overall_results_dict[int(k)] = {} 703 for k2 in ms2_results_load[k].keys(): 704 ms2_search_res = SpectrumSearchResults( 705 query_spectrum=mass_spectra._ms[int(k)], 706 precursor_mz=ms2_results_load[k][k2].attrs["precursor_mz"], 707 spectral_similarity_search_results={}, 708 ) 709 710 for key in ms2_results_load[k][k2].keys() - {"precursor_mz"}: 711 data = list(ms2_results_load[k][k2][key][:]) 712 if data and isinstance(data[0], bytes): 713 data = [x.decode("utf-8") for x in data] 714 setattr(ms2_search_res, key, data) 715 716 overall_results_dict[int(k)][ 717 ms2_results_load[k][k2].attrs["precursor_mz"] 718 ] = ms2_search_res 719 720 # add to mass_spectra 721 mass_spectra.spectral_search_results.update(overall_results_dict) 722 723 # If there are mass features, associate the results with each mass feature 724 if len(mass_spectra.mass_features) > 0: 725 for mass_feature_id, mass_feature in mass_spectra.mass_features.items(): 726 scan_ids = mass_feature.ms2_scan_numbers 727 for ms2_scan_id in scan_ids: 728 precursor_mz = mass_feature.mz 729 try: 730 mass_spectra.spectral_search_results[ms2_scan_id][precursor_mz] 731 except KeyError: 732 pass 733 else: 734 mass_spectra.mass_features[ 735 mass_feature_id 736 ].ms2_similarity_results.append( 737 mass_spectra.spectral_search_results[ms2_scan_id][ 738 precursor_mz 739 ] 740 )
Imports the spectral search results from the HDF5 file.
Parameters
- mass_spectra (LCMSBase | MassSpectraBase): The MassSpectraBase or LCMSBase object to populate with spectral search results.
Returns
- None, but populates the 'spectral_search_results' attribute on the LCMSBase or MassSpectraBase
- object with a dictionary of the 'spectral_search_results' from the HDF5 file.
def
get_mass_spectra_obj( self, load_raw=True, load_light=False, time_range: Union[Tuple[float, float], List[Tuple[float, float]], NoneType] = None) -> corems.mass_spectra.factory.lc_class.MassSpectraBase:
742 def get_mass_spectra_obj(self, load_raw=True, load_light=False, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None) -> MassSpectraBase: 743 """ 744 Return mass spectra data object, populating the _ms list on MassSpectraBase object from the HDF5 file. 745 746 Parameters 747 ---------- 748 load_raw : bool 749 If True, load raw data (unprocessed) from HDF5 files for overall spectra object and individual mass spectra. Default is True. 750 load_light : bool 751 If True, only load the parameters, mass features, and scan info. Default is False. 752 time_range : tuple or list of tuples, optional 753 Retention time range(s) to load. Can be: 754 - Single range: (start_time, end_time) in minutes 755 - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes 756 If None, loads all scans. Note: For HDF5 files, this parameter is accepted for 757 interface consistency but not currently used in filtering. 758 759 """ 760 # Instantiate the LCMS object 761 spectra_obj = MassSpectraBase( 762 file_location=self.file_location, 763 analyzer=self.analyzer, 764 instrument_label=self.instrument_label, 765 sample_name=self.sample_name, 766 ) 767 768 # This will populate the _ms list on the LCMS or MassSpectraBase object 769 self.run(spectra_obj, load_raw=load_raw, load_light=load_light) 770 771 return spectra_obj
Return mass spectra data object, populating the _ms list on MassSpectraBase object from the HDF5 file.
Parameters
- load_raw (bool): If True, load raw data (unprocessed) from HDF5 files for overall spectra object and individual mass spectra. Default is True.
- load_light (bool): If True, only load the parameters, mass features, and scan info. Default is False.
- 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. Note: For HDF5 files, this parameter is accepted for interface consistency but not currently used in filtering.
def
get_lcms_obj( self, load_raw=True, load_light=False, use_original_parser=True, raw_file_path=None, time_range: Union[Tuple[float, float], List[Tuple[float, float]], NoneType] = None) -> corems.mass_spectra.factory.lc_class.LCMSBase:
773 def get_lcms_obj( 774 self, load_raw=True, load_light=False, use_original_parser=True, raw_file_path=None, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None 775 ) -> LCMSBase: 776 """ 777 Return LCMSBase object, populating attributes on the LCMSBase object from the HDF5 file. 778 779 Parameters 780 ---------- 781 load_raw : bool 782 If True, load raw data (unprocessed) from HDF5 files for overall lcms object and individual mass spectra. Default is True. 783 load_light : bool 784 If True, only load the parameters, mass features, and scan info. Default is False. 785 use_original_parser : bool 786 If True, use the original parser to populate the LCMS object. Default is True. 787 raw_file_path : str 788 The location of the raw file to parse if attempting to use original parser. 789 Default is None, which attempts to get the raw file path from the HDF5 file. 790 If the original file path has moved, this parameter can be used to specify the new location. 791 time_range : tuple or list of tuples, optional 792 Retention time range(s) to load. Can be: 793 - Single range: (start_time, end_time) in minutes 794 - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes 795 If None, loads all scans. Note: For HDF5 files, this parameter is accepted for 796 interface consistency. If use_original_parser=True, time_range can be passed to the 797 original parser for filtering. 798 """ 799 # Instantiate the LCMS object 800 lcms_obj = LCMSBase( 801 file_location=self.file_location, 802 analyzer=self.analyzer, 803 instrument_label=self.instrument_label, 804 sample_name=self.sample_name, 805 ) 806 807 # This will populate the majority of the attributes on the LCMS object 808 self.run(lcms_obj, load_raw=load_raw, load_light=load_light) 809 810 # Set final attributes of the LCMS object 811 lcms_obj.polarity = self.h5pydata.attrs["polarity"] 812 lcms_obj._scans_number_list = list(lcms_obj.scan_df.scan) 813 lcms_obj._retention_time_list = list(lcms_obj.scan_df.scan_time) 814 lcms_obj._tic_list = list(lcms_obj.scan_df.tic) 815 816 # If use_original_parser is True, instantiate the original parser and populate the LCMS object 817 if use_original_parser: 818 lcms_obj = self.add_original_parser(lcms_obj, raw_file_path=raw_file_path) 819 else: 820 lcms_obj.spectra_parser_class = self.__class__ 821 822 return lcms_obj
Return LCMSBase object, populating attributes on the LCMSBase object from the HDF5 file.
Parameters
- load_raw (bool): If True, load raw data (unprocessed) from HDF5 files for overall lcms object and individual mass spectra. Default is True.
- load_light (bool): If True, only load the parameters, mass features, and scan info. Default is False.
- use_original_parser (bool): If True, use the original parser to populate the LCMS object. Default is True.
- raw_file_path (str): The location of the raw file to parse if attempting to use original parser. Default is None, which attempts to get the raw file path from the HDF5 file. If the original file path has moved, this parameter can be used to specify the new location.
- 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. Note: For HDF5 files, this parameter is accepted for interface consistency. If use_original_parser=True, time_range can be passed to the original parser for filtering.
def
get_raw_file_location(self):
824 def get_raw_file_location(self): 825 """ 826 Get the raw file location from the HDF5 file attributes. 827 828 Returns 829 ------- 830 str 831 The raw file location. 832 """ 833 if "original_file_location" in self.h5pydata.attrs: 834 return self.h5pydata.attrs["original_file_location"] 835 else: 836 return None
Get the raw file location from the HDF5 file attributes.
Returns
- str: The raw file location.
def
add_original_parser(self, mass_spectra, raw_file_path=None):
838 def add_original_parser(self, mass_spectra, raw_file_path=None): 839 """ 840 Add the original parser to the mass spectra object. 841 842 Parameters 843 ---------- 844 mass_spectra : MassSpectraBase | LCMSBase 845 The MassSpectraBase or LCMSBase object to add the original parser to. 846 raw_file_path : str 847 The location of the raw file to parse. Default is None, which attempts to get the raw file path from the HDF5 file. 848 """ 849 # Get the original parser type 850 og_parser_type = self.h5pydata.attrs["parser_type"] 851 852 # If raw_file_path is None, get it from the HDF5 file attributes 853 if raw_file_path is None: 854 raw_file_path = self.get_raw_file_location() 855 if raw_file_path is None: 856 raise ValueError( 857 "Raw file path not found in HDF5 file attributes, cannot instantiate original parser." 858 ) 859 860 # Set the raw file path on the mass_spectra object so the parser knows where to find the raw file 861 mass_spectra.raw_file_location = raw_file_path 862 863 if og_parser_type == "ImportMassSpectraThermoMSFileReader": 864 # Check that the parser can be instantiated with the raw file path 865 parser_class = ImportMassSpectraThermoMSFileReader 866 elif og_parser_type == "MZMLSpectraParser": 867 # Check that the parser can be instantiated with the raw file path 868 parser_class = MZMLSpectraParser 869 870 # Set the spectra parser class on the mass_spectra object so the spectra_parser property can be used with the original parser 871 mass_spectra.spectra_parser_class = parser_class 872 873 return mass_spectra
Add the original parser to the mass spectra object.
Parameters
- mass_spectra (MassSpectraBase | LCMSBase): The MassSpectraBase or LCMSBase object to add the original parser to.
- raw_file_path (str): The location of the raw file to parse. Default is None, which attempts to get the raw file path from the HDF5 file.
def
get_original_creation_time(self):
875 def get_original_creation_time(self): 876 """ 877 Get the creation time of the original raw data file. 878 879 First checks if creation_time is saved in the HDF5 file attributes. 880 If not found, attempts to instantiate the original parser and get the creation time. 881 882 Returns 883 ------- 884 datetime 885 The creation time of the original raw data file, or None if not available. 886 """ 887 # Check if creation_time is saved in HDF5 attributes 888 if "creation_time" in self.h5pydata.attrs: 889 from datetime import datetime 890 return datetime.fromisoformat(self.h5pydata.attrs["creation_time"]) 891 892 # Fall back to using original parser to get creation time 893 try: 894 # Get the original parser type and raw file path 895 og_parser_type = self.h5pydata.attrs.get("parser_type") 896 raw_file_path = self.get_raw_file_location() 897 898 if og_parser_type is None or raw_file_path is None: 899 warnings.warn( 900 "Cannot retrieve creation time: parser_type or original_file_location not found in HDF5 attributes." 901 ) 902 return None 903 904 # Check if raw file exists 905 from pathlib import Path 906 if not Path(raw_file_path).exists(): 907 warnings.warn( 908 f"Cannot retrieve creation time: original raw file not found at {raw_file_path}" 909 ) 910 return None 911 912 # Instantiate the original parser 913 if og_parser_type == "ImportMassSpectraThermoMSFileReader": 914 parser = ImportMassSpectraThermoMSFileReader(raw_file_path) 915 elif og_parser_type == "MZMLSpectraParser": 916 parser = MZMLSpectraParser(raw_file_path) 917 else: 918 warnings.warn( 919 f"Unknown parser type: {og_parser_type}, cannot retrieve creation time." 920 ) 921 return None 922 923 # Get creation time from parser 924 return parser.get_creation_time() 925 926 except Exception as e: 927 warnings.warn( 928 f"Failed to retrieve creation time from original parser: {e}" 929 ) 930 return None
Get the creation time of the original raw data file.
First checks if creation_time is saved in the HDF5 file attributes. If not found, attempts to instantiate the original parser and get the creation time.
Returns
- datetime: The creation time of the original raw data file, or None if not available.
def
get_creation_time(self):
932 def get_creation_time(self): 933 """ 934 Get the creation time of the original raw data file. 935 936 This is an alias for get_original_creation_time() for backward compatibility. 937 938 Returns 939 ------- 940 datetime 941 The creation time of the original raw data file, or None if not available. 942 """ 943 return self.get_original_creation_time()
Get the creation time of the original raw data file.
This is an alias for get_original_creation_time() for backward compatibility.
Returns
- datetime: The creation time of the original raw data file, or None if not available.
def
get_instrument_info(self):
945 def get_instrument_info(self): 946 """ 947 Raise a NotImplemented Warning, as instrument info is not available in CoreMS HDF5 files and returning None. 948 """ 949 warnings.warn( 950 "Instrument info is not available in CoreMS HDF5 files, returning None." 951 "This should be accessed through the original parser.", 952 ) 953 return None
Raise a NotImplemented Warning, as instrument info is not available in CoreMS HDF5 files and returning None.
def
get_scans_in_time_range( self, time_range: Union[Tuple[float, float], List[Tuple[float, float]]], ms_level: Optional[int] = None) -> List[int]:
955 def get_scans_in_time_range( 956 self, 957 time_range: Union[Tuple[float, float], List[Tuple[float, float]]], 958 ms_level: Optional[int] = None 959 ) -> List[int]: 960 """Return scan numbers within specified retention time range(s). 961 962 Parameters 963 ---------- 964 time_range : tuple or list of tuples 965 Retention time range(s) in minutes. Can be: 966 - Single range: (start_time, end_time) 967 - Multiple ranges: [(start1, end1), (start2, end2), ...] 968 ms_level : int, optional 969 If specified, only return scans of this MS level (e.g., 1 for MS1, 2 for MS2). 970 If None, returns scans of all MS levels. 971 972 Returns 973 ------- 974 list of int 975 List of scan numbers within the specified time range(s) and MS level. 976 """ 977 # Normalize time range to list of tuples 978 time_ranges = self._normalize_time_range(time_range) 979 980 # Get all scan data 981 scan_df = self.get_scan_df() 982 983 # Filter by time range 984 mask = pd.Series([False] * len(scan_df), index=scan_df.index) 985 for start_time, end_time in time_ranges: 986 mask |= (scan_df.scan_time >= start_time) & (scan_df.scan_time <= end_time) 987 988 filtered_df = scan_df[mask] 989 990 # Filter by MS level if specified 991 if ms_level is not None: 992 filtered_df = filtered_df[filtered_df.ms_level == ms_level] 993 994 return filtered_df.scan.tolist()
Return scan numbers within specified retention time range(s).
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.
Inherited Members
- corems.mass_spectrum.input.coremsHDF5.ReadCoreMSHDF_MassSpectrum
- h5pydata
- load_raw_data
- get_mass_spectrum
- load_settings
- get_dataframe
- get_time_index_to_pull
- get_high_level_attr_data
- get_scan_group_attr_data
- get_raw_data_attr_data
- get_output_parameters
- corems.mass_spectrum.input.baseClass.MassListBaseClass
- file_location
- header_lines
- isCentroid
- isThermoProfile
- headerless
- analyzer
- instrument_label
- sample_name
- parameters
- set_parameter_from_toml
- set_parameter_from_json
- data_type
- delimiter
- encoding_detector
- set_data_type
- clean_data_frame
- check_columns
- read_xml_peaks
- get_xml_polarity
class
ReadCoreMSHDFMassSpectraCollection:
997class ReadCoreMSHDFMassSpectraCollection: 998 """Read a collection of CoreMS HDF5 files and populate an LCMSCollection object. 999 1000 Parameters 1001 ---------- 1002 folder_location : Path 1003 Folder containing .corems subdirectories with HDF5 files. 1004 manifest_file : Path, optional 1005 Manifest CSV with columns: sample_name, order, batch, center, time. 1006 One sample must have center='TRUE' for RT alignment. 1007 If None, checks if auto-generated manifest_auto.csv exists in the folder. If not, 1008 auto-generates from folder contents. Default: None. 1009 cores : int, optional 1010 Number of cores for multiprocessing. Default: 1. 1011 auto_manifest_batch_threshold_hours : float, optional 1012 Time gap (hours) for auto-generated batch separation. Default: 12.0. 1013 auto_manifest_center_name : str, optional 1014 Sample name for RT alignment center when auto-generating. 1015 Must match a discovered sample. If None, uses middle sample. Default: None. 1016 1017 Attributes 1018 ---------- 1019 folder_location : Path 1020 Folder containing CoreMS HDF5 files. 1021 manifest_filepath : Path 1022 Path to manifest file. 1023 manifest : dict 1024 Manifest data indexed by sample_name. 1025 """ 1026 def __init__( 1027 self, 1028 folder_location: Path, 1029 manifest_file: Path = None, 1030 cores: int = 1, 1031 auto_manifest_batch_threshold_hours: float = 12.0, 1032 auto_manifest_center_name: str = None 1033 ): 1034 # Check for folder location 1035 folder_location = Path(folder_location) 1036 if not folder_location.exists(): 1037 raise FileNotFoundError(f"Folder location {folder_location} not found.") 1038 1039 # Auto-generate manifest if not provided 1040 if manifest_file is None: 1041 # Check if manifest_auto.csv already exists 1042 auto_manifest_path = folder_location / "manifest_auto.csv" 1043 if auto_manifest_path.exists(): 1044 print(f"No manifest file provided. Using existing manifest_auto.csv from {folder_location}") 1045 manifest_file = auto_manifest_path 1046 else: 1047 print(f"No manifest file provided. Auto-generating manifest from {folder_location}") 1048 manifest_file = create_manifest_from_folder( 1049 folder_path=folder_location, 1050 output_path=auto_manifest_path, 1051 batch_time_threshold_hours=auto_manifest_batch_threshold_hours, 1052 center_name=auto_manifest_center_name, 1053 overwrite=True 1054 ) 1055 else: 1056 manifest_file = Path(manifest_file) 1057 if not manifest_file.exists(): 1058 raise FileNotFoundError(f"Manifest file {manifest_file} not found.") 1059 1060 # Check if the manifest file is a CSV 1061 if manifest_file.suffix != ".csv": 1062 raise ValueError("Manifest file must be a CSV.") 1063 1064 self.folder_location = folder_location 1065 self._manifest_dict = None 1066 self._parse_manifest(manifest_file) 1067 self._validate_manifest() 1068 self._validate_parameters() 1069 self._validate_cores(cores) 1070 1071 def _validate_cores(self, cores): 1072 # Check if the cores parameter is an integer greater than 0 and less than the number of cores available 1073 if not isinstance(cores, int) or cores < 1: 1074 raise ValueError("Cores must be an integer greater than 0.") 1075 if cores > multiprocessing.cpu_count(): 1076 raise ValueError( 1077 f"Cores must be less than or equal to the number of cores available ({multiprocessing.cpu_count()})." 1078 ) 1079 self._cores = cores 1080 1081 def _parse_manifest(self, manifest_file): 1082 """Parse the manifest file and set the manifest dictionary.""" 1083 self.manifest_filepath = manifest_file 1084 manifest = pd.read_csv(manifest_file) 1085 # Check if the following columns exisit in the manifest file 1086 if not all( 1087 col in manifest.columns for col in ["sample_name", "order", "batch"] 1088 ): 1089 raise ValueError( 1090 "Manifest file must contain the following columns: 'sample_name', 'order', 'batch'." 1091 ) 1092 # Set index to the 'sample_name' column 1093 manifest.set_index("sample_name", inplace=True) 1094 self._manifest_dict = manifest.to_dict(orient="index") 1095 1096 def _validate_manifest(self): 1097 """Validate the manifest dictionary against the CoreMS folder location.""" 1098 # Check if the folder location contains HDF5 files for each sample 1099 for sample_name in self._manifest_dict.keys(): 1100 corems_dir = self.folder_location / f"{sample_name}.corems" 1101 if not corems_dir.exists(): 1102 raise FileNotFoundError(f"CoreMS folder for {sample_name} not found.") 1103 hdf5_file = corems_dir / f"{sample_name}.hdf5" 1104 if not hdf5_file.exists(): 1105 raise FileNotFoundError(f"HDF5 file for {sample_name} not found.") 1106 1107 # Check that at least one sample has center='TRUE' for retention time alignment 1108 center_values = [sample_data.get('center') for sample_data in self._manifest_dict.values()] 1109 if not any(center_val == 'TRUE' or center_val == True for center_val in center_values): 1110 raise ValueError( 1111 "Manifest must contain at least one sample with center='TRUE' for retention time alignment. " 1112 "None of the samples in the manifest have center='TRUE'." 1113 ) 1114 1115 def _validate_parameters(self): 1116 """Validate that the parameters used for all samples within a batch are the same.""" 1117 # Check if parameters files are saved as JSON or TOML 1118 if self.parameters_files[0].suffix == ".json": 1119 importer = json 1120 suffix = ".json" 1121 1122 elif self.parameters_files[0].suffix == ".toml": 1123 importer = toml 1124 suffix = ".toml" 1125 1126 manfiest_df = self.manifest_dataframe 1127 1128 # Split up samples by batch 1129 batches = manfiest_df["batch"].unique() 1130 1131 for batch in batches: 1132 samples = manfiest_df[manfiest_df["batch"] == batch].index 1133 # check if self.parameters_files end with .json or .toml 1134 batch_param_files = [ 1135 self.folder_location / f"{sample_name}.corems/{sample_name}{suffix}" 1136 for sample_name in self._manifest_dict.keys() 1137 if sample_name in samples 1138 ] 1139 with open( 1140 batch_param_files[0], 1141 "r", 1142 encoding="utf8", 1143 ) as stream: 1144 first_parameters = importer.load(stream) 1145 for parameters_file in batch_param_files[1:]: 1146 with open( 1147 parameters_file, 1148 "r", 1149 encoding="utf8", 1150 ) as stream: 1151 parameters = importer.load(stream) 1152 if parameters != first_parameters: 1153 raise ValueError( 1154 f"Parameters files for samples in batch {batch} are not equal." 1155 ) 1156 1157 def get_lcms_obj(self, sample_name: str, load_raw=False, load_light=True, use_original_parser=True, raw_file_path=None) -> LCMSBase: 1158 """Return a LCMSBase object for a given sample name within the collection. 1159 1160 Parameters 1161 ---------- 1162 sample_name : str 1163 The sample name to retrieve the LCMS object for. 1164 load_raw : bool 1165 If True, load raw data from HDF5 files. Default is False. 1166 load_light : bool 1167 If True, only load the parameters, mass features, and scan info are initially loaded for each lcms object. Default is True. 1168 """ 1169 hdf5_file = self.folder_location / f"{sample_name}.corems/{sample_name}.hdf5" 1170 with ReadCoreMSHDFMassSpectra(hdf5_file) as parser: 1171 lcms_obj = parser.get_lcms_obj(load_raw=load_raw, load_light=load_light, use_original_parser=use_original_parser, raw_file_path=raw_file_path) 1172 if load_light: 1173 mf_df = lcms_obj.mass_features_to_df() 1174 # Add ._eic_mz to mf_df for each mass_feature 1175 eic_mz_list = [] 1176 for mf_id, mf in lcms_obj.mass_features.items(): 1177 if hasattr(mf, "_eic_mz"): 1178 eic_mz_list.append(mf._eic_mz) 1179 else: 1180 eic_mz_list.append(None) 1181 mf_df["_eic_mz"] = eic_mz_list 1182 lcms_obj.mass_features = {} 1183 lcms_obj.light_mf_df = mf_df 1184 return lcms_obj 1185 1186 def get_lcms_collection(self, load_raw = False, load_light = True, use_original_parser = True) -> LCMSCollection: 1187 """Return a LCMSCollection object 1188 1189 Parameters 1190 ---------- 1191 load_raw : bool 1192 If True, load raw data from HDF5 files. Default is False. 1193 load_light : bool 1194 If True, only load the parameters, mass features, and scan info are initially loaded for each lcms object. 1195 After concatenating the mass_features, remove the mass_features attribute from the individual LCMS objects for memory efficiency. Default is True. 1196 Default is True. 1197 """ 1198 # Instantiate the LCMSCollection object 1199 lcms_coll = LCMSCollection( 1200 collection_location=self.folder_location, 1201 manifest=self.manifest, 1202 collection_parser=self 1203 ) 1204 1205 # Set the number of cores on the LCMSCollection object from the ReadCoreMSHDFMassSpectraCollection object 1206 lcms_coll.parameters.lcms_collection.cores = self._cores 1207 1208 # Add LCMS objects to the collection 1209 samples = self._manifest_dict.keys() 1210 1211 # Initialize the LCMS object dictionary 1212 if self._cores > 1: 1213 if self._cores > len(samples): 1214 ncores = len(samples) 1215 else: 1216 ncores = self._cores 1217 # Create a pool of workers (one for each core or sample, whichever is smaller) 1218 pool = multiprocessing.Pool(ncores) 1219 args = [(sample, load_raw, load_light, use_original_parser) for sample in samples] 1220 lcms_objs = pool.starmap(self.get_lcms_obj, args) 1221 for sample_name, lcms_obj in zip(samples, lcms_objs): 1222 lcms_coll._lcms[sample_name] = lcms_obj 1223 1224 elif self._cores == 1: 1225 # Load the LCMS objects sequentially 1226 for sample_name in samples: 1227 lcms_coll._lcms[sample_name] = self.get_lcms_obj(sample_name, load_raw=load_raw, load_light=load_light, use_original_parser=use_original_parser) 1228 1229 else: 1230 raise ValueError("Number of cores must be greater than 0 and set on the ReadCoreMSHDFMassSpectraCollection object.") 1231 1232 # Check that all LCMS objects have the same polarity 1233 if len(set([x.polarity for k, x in lcms_coll._lcms.items()])) != 1: 1234 raise ValueError("All samples must have the same polarity.") 1235 1236 # Set ids on the LCMS objects in the manifest 1237 i = 0 1238 for sample in lcms_coll.samples: 1239 lcms_coll._manifest_dict[sample]["collection_id"] = i 1240 i += 1 1241 1242 # Reorder the LCMS objects 1243 lcms_coll._reorder_lcms_objects() 1244 1245 # Collect the mass features from the LCMS objects and combine them into a single dataframe for the collection 1246 lcms_coll._combine_mass_features() 1247 1248 # If load_light, remove the mass_feature attribute from the individual LCMS objects 1249 if load_light: 1250 for sample_name in lcms_coll.samples: 1251 lcms_coll._lcms[sample_name].mass_features = {} 1252 # Remove the light_mf_df attribute from the individual LCMS objects 1253 del lcms_coll._lcms[sample_name].light_mf_df 1254 1255 1256 return lcms_coll 1257 1258 @property 1259 def manifest(self): 1260 return self._manifest_dict 1261 1262 @property 1263 def manifest_dataframe(self): 1264 return pd.DataFrame(self._manifest_dict).T 1265 1266 @property 1267 def hdf5_files(self): 1268 return [ 1269 self.folder_location / f"{sample_name}.corems/{sample_name}.hdf5" 1270 for sample_name in self._manifest_dict.keys() 1271 ] 1272 1273 @property 1274 def parameters_files(self): 1275 # Check if parameters files are saved as JSON or TOML 1276 json_files = [ 1277 self.folder_location / f"{sample_name}.corems/{sample_name}.json" 1278 for sample_name in self._manifest_dict.keys() 1279 ] 1280 toml_files = [ 1281 self.folder_location / f"{sample_name}.corems/{sample_name}.toml" 1282 for sample_name in self._manifest_dict.keys() 1283 ] 1284 if all([x.exists() for x in json_files]): 1285 return json_files 1286 elif all([x.exists() for x in toml_files]): 1287 return toml_files 1288 else: 1289 raise ValueError("Parameters files are not saved for all samples.")
Read a collection of CoreMS HDF5 files and populate an LCMSCollection object.
Parameters
- folder_location (Path): Folder containing .corems subdirectories with HDF5 files.
- manifest_file (Path, optional): Manifest CSV with columns: sample_name, order, batch, center, time. One sample must have center='TRUE' for RT alignment. If None, checks if auto-generated manifest_auto.csv exists in the folder. If not, auto-generates from folder contents. Default: None.
- cores (int, optional): Number of cores for multiprocessing. Default: 1.
- auto_manifest_batch_threshold_hours (float, optional): Time gap (hours) for auto-generated batch separation. Default: 12.0.
- auto_manifest_center_name (str, optional): Sample name for RT alignment center when auto-generating. Must match a discovered sample. If None, uses middle sample. Default: None.
Attributes
- folder_location (Path): Folder containing CoreMS HDF5 files.
- manifest_filepath (Path): Path to manifest file.
- manifest (dict): Manifest data indexed by sample_name.
ReadCoreMSHDFMassSpectraCollection( folder_location: pathlib._local.Path, manifest_file: pathlib._local.Path = None, cores: int = 1, auto_manifest_batch_threshold_hours: float = 12.0, auto_manifest_center_name: str = None)
1026 def __init__( 1027 self, 1028 folder_location: Path, 1029 manifest_file: Path = None, 1030 cores: int = 1, 1031 auto_manifest_batch_threshold_hours: float = 12.0, 1032 auto_manifest_center_name: str = None 1033 ): 1034 # Check for folder location 1035 folder_location = Path(folder_location) 1036 if not folder_location.exists(): 1037 raise FileNotFoundError(f"Folder location {folder_location} not found.") 1038 1039 # Auto-generate manifest if not provided 1040 if manifest_file is None: 1041 # Check if manifest_auto.csv already exists 1042 auto_manifest_path = folder_location / "manifest_auto.csv" 1043 if auto_manifest_path.exists(): 1044 print(f"No manifest file provided. Using existing manifest_auto.csv from {folder_location}") 1045 manifest_file = auto_manifest_path 1046 else: 1047 print(f"No manifest file provided. Auto-generating manifest from {folder_location}") 1048 manifest_file = create_manifest_from_folder( 1049 folder_path=folder_location, 1050 output_path=auto_manifest_path, 1051 batch_time_threshold_hours=auto_manifest_batch_threshold_hours, 1052 center_name=auto_manifest_center_name, 1053 overwrite=True 1054 ) 1055 else: 1056 manifest_file = Path(manifest_file) 1057 if not manifest_file.exists(): 1058 raise FileNotFoundError(f"Manifest file {manifest_file} not found.") 1059 1060 # Check if the manifest file is a CSV 1061 if manifest_file.suffix != ".csv": 1062 raise ValueError("Manifest file must be a CSV.") 1063 1064 self.folder_location = folder_location 1065 self._manifest_dict = None 1066 self._parse_manifest(manifest_file) 1067 self._validate_manifest() 1068 self._validate_parameters() 1069 self._validate_cores(cores)
def
get_lcms_obj( self, sample_name: str, load_raw=False, load_light=True, use_original_parser=True, raw_file_path=None) -> corems.mass_spectra.factory.lc_class.LCMSBase:
1157 def get_lcms_obj(self, sample_name: str, load_raw=False, load_light=True, use_original_parser=True, raw_file_path=None) -> LCMSBase: 1158 """Return a LCMSBase object for a given sample name within the collection. 1159 1160 Parameters 1161 ---------- 1162 sample_name : str 1163 The sample name to retrieve the LCMS object for. 1164 load_raw : bool 1165 If True, load raw data from HDF5 files. Default is False. 1166 load_light : bool 1167 If True, only load the parameters, mass features, and scan info are initially loaded for each lcms object. Default is True. 1168 """ 1169 hdf5_file = self.folder_location / f"{sample_name}.corems/{sample_name}.hdf5" 1170 with ReadCoreMSHDFMassSpectra(hdf5_file) as parser: 1171 lcms_obj = parser.get_lcms_obj(load_raw=load_raw, load_light=load_light, use_original_parser=use_original_parser, raw_file_path=raw_file_path) 1172 if load_light: 1173 mf_df = lcms_obj.mass_features_to_df() 1174 # Add ._eic_mz to mf_df for each mass_feature 1175 eic_mz_list = [] 1176 for mf_id, mf in lcms_obj.mass_features.items(): 1177 if hasattr(mf, "_eic_mz"): 1178 eic_mz_list.append(mf._eic_mz) 1179 else: 1180 eic_mz_list.append(None) 1181 mf_df["_eic_mz"] = eic_mz_list 1182 lcms_obj.mass_features = {} 1183 lcms_obj.light_mf_df = mf_df 1184 return lcms_obj
Return a LCMSBase object for a given sample name within the collection.
Parameters
- sample_name (str): The sample name to retrieve the LCMS object for.
- load_raw (bool): If True, load raw data from HDF5 files. Default is False.
- load_light (bool): If True, only load the parameters, mass features, and scan info are initially loaded for each lcms object. Default is True.
def
get_lcms_collection( self, load_raw=False, load_light=True, use_original_parser=True) -> corems.mass_spectra.factory.lc_class.LCMSCollection:
1186 def get_lcms_collection(self, load_raw = False, load_light = True, use_original_parser = True) -> LCMSCollection: 1187 """Return a LCMSCollection object 1188 1189 Parameters 1190 ---------- 1191 load_raw : bool 1192 If True, load raw data from HDF5 files. Default is False. 1193 load_light : bool 1194 If True, only load the parameters, mass features, and scan info are initially loaded for each lcms object. 1195 After concatenating the mass_features, remove the mass_features attribute from the individual LCMS objects for memory efficiency. Default is True. 1196 Default is True. 1197 """ 1198 # Instantiate the LCMSCollection object 1199 lcms_coll = LCMSCollection( 1200 collection_location=self.folder_location, 1201 manifest=self.manifest, 1202 collection_parser=self 1203 ) 1204 1205 # Set the number of cores on the LCMSCollection object from the ReadCoreMSHDFMassSpectraCollection object 1206 lcms_coll.parameters.lcms_collection.cores = self._cores 1207 1208 # Add LCMS objects to the collection 1209 samples = self._manifest_dict.keys() 1210 1211 # Initialize the LCMS object dictionary 1212 if self._cores > 1: 1213 if self._cores > len(samples): 1214 ncores = len(samples) 1215 else: 1216 ncores = self._cores 1217 # Create a pool of workers (one for each core or sample, whichever is smaller) 1218 pool = multiprocessing.Pool(ncores) 1219 args = [(sample, load_raw, load_light, use_original_parser) for sample in samples] 1220 lcms_objs = pool.starmap(self.get_lcms_obj, args) 1221 for sample_name, lcms_obj in zip(samples, lcms_objs): 1222 lcms_coll._lcms[sample_name] = lcms_obj 1223 1224 elif self._cores == 1: 1225 # Load the LCMS objects sequentially 1226 for sample_name in samples: 1227 lcms_coll._lcms[sample_name] = self.get_lcms_obj(sample_name, load_raw=load_raw, load_light=load_light, use_original_parser=use_original_parser) 1228 1229 else: 1230 raise ValueError("Number of cores must be greater than 0 and set on the ReadCoreMSHDFMassSpectraCollection object.") 1231 1232 # Check that all LCMS objects have the same polarity 1233 if len(set([x.polarity for k, x in lcms_coll._lcms.items()])) != 1: 1234 raise ValueError("All samples must have the same polarity.") 1235 1236 # Set ids on the LCMS objects in the manifest 1237 i = 0 1238 for sample in lcms_coll.samples: 1239 lcms_coll._manifest_dict[sample]["collection_id"] = i 1240 i += 1 1241 1242 # Reorder the LCMS objects 1243 lcms_coll._reorder_lcms_objects() 1244 1245 # Collect the mass features from the LCMS objects and combine them into a single dataframe for the collection 1246 lcms_coll._combine_mass_features() 1247 1248 # If load_light, remove the mass_feature attribute from the individual LCMS objects 1249 if load_light: 1250 for sample_name in lcms_coll.samples: 1251 lcms_coll._lcms[sample_name].mass_features = {} 1252 # Remove the light_mf_df attribute from the individual LCMS objects 1253 del lcms_coll._lcms[sample_name].light_mf_df 1254 1255 1256 return lcms_coll
Return a LCMSCollection object
Parameters
- load_raw (bool): If True, load raw data from HDF5 files. Default is False.
- load_light (bool): If True, only load the parameters, mass features, and scan info are initially loaded for each lcms object. After concatenating the mass_features, remove the mass_features attribute from the individual LCMS objects for memory efficiency. Default is True. Default is True.
parameters_files
1273 @property 1274 def parameters_files(self): 1275 # Check if parameters files are saved as JSON or TOML 1276 json_files = [ 1277 self.folder_location / f"{sample_name}.corems/{sample_name}.json" 1278 for sample_name in self._manifest_dict.keys() 1279 ] 1280 toml_files = [ 1281 self.folder_location / f"{sample_name}.corems/{sample_name}.toml" 1282 for sample_name in self._manifest_dict.keys() 1283 ] 1284 if all([x.exists() for x in json_files]): 1285 return json_files 1286 elif all([x.exists() for x in toml_files]): 1287 return toml_files 1288 else: 1289 raise ValueError("Parameters files are not saved for all samples.")
1291class ReadSavedLCMSCollection(ReadCoreMSHDFMassSpectraCollection): 1292 """ 1293 Subclass to read and re-instantiate a LCMSCollection from a saved HDF5 file. 1294 1295 1296 Parameters 1297 ---------- 1298 collection_hdf5_path : str or Path 1299 Path to the saved LCMSCollection HDF5 file. 1300 cores : int, optional 1301 Number of cores for processing. Default is 1. 1302 """ 1303 1304 def __init__( 1305 self, 1306 collection_hdf5_path: str, 1307 cores: int = 1 1308 ): 1309 # Convert to Path objects 1310 self.collection_hdf5_path = Path(collection_hdf5_path) 1311 1312 # Validate the collection file exists 1313 if not self.collection_hdf5_path.exists(): 1314 raise FileNotFoundError(f"Collection HDF5 file {self.collection_hdf5_path} not found.") 1315 1316 # Validate cores 1317 self._validate_cores(cores) 1318 1319 # Load metadata from saved collection 1320 self._load_collection_metadata() 1321 1322 if not self.folder_location.exists(): 1323 raise FileNotFoundError(f"Folder location {self.folder_location} not found.") 1324 1325 # Load the mass spectra data 1326 self._validate_manifest() 1327 1328 # Set the parameters file location 1329 self.parameters_location = self._get_parameters_location() 1330 1331 def _get_parameters_location(self): 1332 """Find the parameters file (JSON or TOML) associated with the collection HDF5 file.""" 1333 # Check for TOML file first (preferred) 1334 toml_path = self.collection_hdf5_path.with_suffix('.toml') 1335 if toml_path.exists(): 1336 return toml_path 1337 1338 # Check for JSON file 1339 json_path = self.collection_hdf5_path.with_suffix('.json') 1340 if json_path.exists(): 1341 return json_path 1342 1343 # No parameters file found 1344 return None 1345 1346 def _load_collection_metadata(self): 1347 """Load metadata and manifest from the saved collection HDF5 file.""" 1348 with h5py.File(self.collection_hdf5_path, 'r') as f: 1349 self.folder_location = Path(f.attrs.get('lcms_objects_folder', '')) 1350 self.missing_mass_features_searched = f.attrs.get('missing_mass_features_searched', False) 1351 1352 # Call the _load_manifest function to process the manifest 1353 self._manifest_dict = self._load_manifest(f) 1354 1355 def _load_manifest(self, hdf_handle): 1356 """Load and clean the manifest from the HDF5 file.""" 1357 manifest_json = hdf_handle.attrs.get('manifest', '{}') 1358 if isinstance(manifest_json, bytes): 1359 manifest_json = manifest_json.decode('utf-8') 1360 loaded_manifest = json.loads(manifest_json) 1361 1362 # Convert integer values for 'use_rt_alignment' back to booleans 1363 def convert_back_to_bool(data): 1364 if isinstance(data, dict): 1365 # Process each key-value pair recursively 1366 return {k: (bool(v) if k == 'use_rt_alignment' and isinstance(v, int) else convert_back_to_bool(v)) for k, v in data.items()} 1367 elif isinstance(data, list): 1368 # Recursively process lists 1369 return [convert_back_to_bool(item) for item in data] 1370 else: 1371 # Return non-dict/list types unchanged 1372 return data 1373 1374 # Clean the loaded manifest 1375 return convert_back_to_bool(loaded_manifest) 1376 1377 def _load_rt_alignments(self, lcms_collection): 1378 """Load retention time alignments from the saved collection HDF5 file.""" 1379 # First Set the rt_aligned flag from the collection-level attribute saved directly 1380 with h5py.File(self.collection_hdf5_path, 'r') as f: 1381 lcms_collection.rt_aligned = f.attrs.get('rt_aligned', False) 1382 lcms_collection.rt_alignment_attempted = f.attrs.get('rt_alignment_attempted', False) 1383 1384 if lcms_collection.rt_aligned: 1385 with h5py.File(self.collection_hdf5_path, 'r') as f: 1386 if "rt_alignments" in f: 1387 # Iterate over the group `rt_alignments` containing datasets and add to the corresponding lcms object 1388 rt_alignments_group = f["rt_alignments"] 1389 for sample_idx, lcms_obj in zip(rt_alignments_group.keys(), lcms_collection): 1390 alignment_data = rt_alignments_group[sample_idx][:] 1391 scan_df = lcms_obj.scan_df 1392 scan_df["scan_time_aligned"] = alignment_data 1393 lcms_obj.scan_df = scan_df 1394 elif lcms_collection.rt_alignment_attempted: 1395 # This means it was attempted and not used, so we populate the "scan_time_aligned" 1396 for lcms_obj in lcms_collection: 1397 scan_df = lcms_obj.scan_df 1398 scan_df["scan_time_aligned"] = scan_df["scan_time"] 1399 lcms_obj.scan_df = scan_df 1400 1401 def _load_cluster_assignments(self, lcms_collection): 1402 """Load cluster assignments from the saved collection HDF5 file.""" 1403 with h5py.File(self.collection_hdf5_path, 'r') as f: 1404 if "cluster_assignments" in f: 1405 # Access the group containing cluster assignments 1406 cluster_grp = f["cluster_assignments"] 1407 1408 # Reload index and cluster data 1409 index = cluster_grp["index"][:] # Extract index 1410 index = [idx.decode('utf-8') for idx in index] # Convert byte strings back to regular strings 1411 cluster_data = cluster_grp["cluster"][:] # Extract cluster column 1412 1413 # Reassemble the DataFrame 1414 cluster_df = pd.DataFrame({"cluster": cluster_data}, index=index) 1415 1416 # Assign cluster data back to lcms_collection.mass_features_dataframe 1417 lcms_collection.mass_features_dataframe = lcms_collection.mass_features_dataframe.join(cluster_df, how='left') 1418 1419 # Drop rows with NaN cluster values 1420 lcms_collection.mass_features_dataframe.dropna(subset=['cluster'], inplace=True) 1421 1422 def get_lcms_collection(self, load_raw=False, load_light=False, load_representatives=False, load_eics=False, load_ms1=False, load_ms2=False): 1423 """Get the LCMS collection from the saved HDF5 file. 1424 1425 Parameters 1426 ---------- 1427 load_raw : bool, optional 1428 If True, load raw data. Default is False. 1429 load_light : bool, optional 1430 If True, load light data (minimal). Default is False. 1431 load_representatives : bool, optional 1432 If True, load representative mass features from clusters. Default is False. 1433 load_eics : bool, optional 1434 If True, load EIC data for clustered mass features. Default is False. 1435 load_ms1 : bool, optional 1436 If True, load MS1 spectra for loaded mass features. Default is False. 1437 load_ms2 : bool, optional 1438 If True, load MS2 spectra for loaded mass features. Default is False. 1439 1440 Returns 1441 ------- 1442 LCMSCollection 1443 The loaded LCMS collection object. 1444 """ 1445 # First load the LCMSCollection object exactly as in the parent class 1446 lcms_collection = super().get_lcms_collection(load_raw=load_raw, load_light=load_light) 1447 1448 # Set the missing_mass_features_searched flag from saved metadata 1449 lcms_collection.missing_mass_features_searched = self.missing_mass_features_searched 1450 1451 # Load parameters if a parameters file exists 1452 if self.parameters_location: 1453 self._load_parameters(lcms_collection) 1454 1455 # Add retention time alignments if they exist 1456 self._load_rt_alignments(lcms_collection) 1457 1458 # Add cluster assignments if they exist 1459 self._load_cluster_assignments(lcms_collection) 1460 1461 # Load induced mass features if they exist 1462 self._load_induced_mass_features(lcms_collection) 1463 1464 # Load EICs for induced mass features from collection HDF5 1465 if lcms_collection.missing_mass_features_searched and load_eics: 1466 self._load_induced_eics_from_collection(lcms_collection) 1467 1468 # Combine induced mass features into the collection-level dataframe if any were loaded 1469 if lcms_collection.missing_mass_features_searched: 1470 lcms_collection._combine_mass_features(induced_features=True) 1471 1472 # Load representative mass features if requested 1473 if load_representatives: 1474 self._load_representative_mass_features(lcms_collection) 1475 1476 # Load MS1 and/or MS2 spectra for loaded mass features if requested 1477 if load_ms1 or load_ms2: 1478 # Reuse the existing ReloadFeaturesOperation from the pipeline system 1479 from corems.mass_spectra.calc.lc_calc_operations import ReloadFeaturesOperation 1480 1481 operations = [ReloadFeaturesOperation('reload_spectra', add_ms1=load_ms1, add_ms2=load_ms2)] 1482 lcms_collection.process_samples_pipeline(operations, keep_raw_data=False, show_progress=False) 1483 1484 # Load EICs for clustered features if requested 1485 if load_eics: 1486 # Reuse the existing LoadEICsOperation from the pipeline system 1487 from corems.mass_spectra.calc.lc_calc_operations import LoadEICsOperation 1488 1489 operations = [LoadEICsOperation('load_eics')] 1490 lcms_collection.process_samples_pipeline(operations, keep_raw_data=False, show_progress=False) 1491 1492 # Associate EICs with mass features (same as in process_consensus_features) 1493 for sample_id in range(len(lcms_collection.samples)): 1494 sample = lcms_collection[sample_id] 1495 if sample.eics: # Only if EICs were loaded 1496 # Associate EICs with regular mass features 1497 sample.associate_eics_with_mass_features(induced=False) 1498 # Associate EICs with induced mass features 1499 sample.associate_eics_with_mass_features(induced=True) 1500 1501 return lcms_collection 1502 1503 def _load_parameters(self, lcms_collection): 1504 """Load collection-level parameters from the saved parameters file.""" 1505 from corems.encapsulation.input.parameter_from_json import ( 1506 load_and_set_json_parameters_lcms_collection, 1507 load_and_set_toml_parameters_lcms_collection, 1508 ) 1509 1510 if self.parameters_location.suffix == ".json": 1511 load_and_set_json_parameters_lcms_collection(lcms_collection, self.parameters_location) 1512 elif self.parameters_location.suffix == ".toml": 1513 load_and_set_toml_parameters_lcms_collection(lcms_collection, self.parameters_location) 1514 else: 1515 warnings.warn(f"Unknown parameter file format: {self.parameters_location.suffix}. Skipping parameter loading.") 1516 1517 def _load_induced_mass_features(self, lcms_collection): 1518 """Load induced mass features from the saved collection HDF5 file. 1519 1520 Induced mass features are gap-filled features that exist at the collection level. 1521 This method loads them from the collection HDF5 file with all their attributes 1522 and datasets, and distributes them to individual LCMS objects. 1523 1524 Parameters 1525 ---------- 1526 lcms_collection : LCMSCollection 1527 The LCMS collection object to populate with induced mass features. 1528 """ 1529 with h5py.File(self.collection_hdf5_path, 'r') as f: 1530 if "induced_mass_features" not in f: 1531 return 1532 1533 # Access the top-level induced mass features group 1534 imf_group = f["induced_mass_features"] 1535 1536 # Iterate through each sample's induced mass features 1537 for sample_idx in imf_group.keys(): 1538 lcms_obj = lcms_collection[int(sample_idx)] 1539 sample_group = imf_group[sample_idx] 1540 1541 # Load each mass feature for this sample 1542 for mf_id_str in sample_group.keys(): 1543 mf_group = sample_group[mf_id_str] 1544 1545 # Note: Induced mass feature IDs are strings like 'c2923_0_i', not integers 1546 # Keep them as strings since that's how they're stored 1547 mf_id = mf_id_str 1548 1549 # Instantiate the LCMSMassFeature object with required attributes 1550 mass_feature = LCMSMassFeature( 1551 lcms_obj, 1552 mz=mf_group.attrs["_mz_exp"], 1553 retention_time=mf_group.attrs["_retention_time"], 1554 intensity=mf_group.attrs["_intensity"], 1555 apex_scan=mf_group.attrs["_apex_scan"], 1556 persistence=mf_group.attrs.get("_persistence", 0), 1557 id=mf_id, 1558 ) 1559 1560 # Populate additional attributes from HDF5 attributes 1561 for key in mf_group.attrs.keys() - { 1562 "_mz_exp", 1563 "_mz_cal", 1564 "_retention_time", 1565 "_intensity", 1566 "_apex_scan", 1567 "_persistence", 1568 }: 1569 setattr(mass_feature, key, mf_group.attrs[key]) 1570 1571 # Populate attributes from HDF5 datasets (arrays) 1572 for key in mf_group.keys(): 1573 setattr(mass_feature, key, mf_group[key][:]) 1574 # Convert _noise_score from array to tuple 1575 if key == "_noise_score": 1576 mass_feature._noise_score = tuple(mass_feature._noise_score) 1577 1578 # Add to the LCMS object's induced_mass_features dictionary 1579 lcms_obj.induced_mass_features[mf_id] = mass_feature 1580 1581 def _load_induced_eics_from_collection(self, lcms_collection): 1582 """Load EICs for induced mass features from the collection HDF5 file. 1583 1584 Induced mass features are gap-filled features. Their EICs are saved at the 1585 collection level and need to be loaded and associated with the induced mass features. 1586 1587 Parameters 1588 ---------- 1589 lcms_collection : LCMSCollection 1590 The LCMS collection object with induced mass features to associate EICs with. 1591 """ 1592 with h5py.File(self.collection_hdf5_path, 'r') as f: 1593 if "induced_eics" not in f: 1594 return 1595 1596 # Access the top-level induced EICs group 1597 induced_eics_group = f["induced_eics"] 1598 1599 # Iterate through each sample's induced EICs 1600 for sample_idx in induced_eics_group.keys(): 1601 lcms_obj = lcms_collection[int(sample_idx)] 1602 sample_group = induced_eics_group[sample_idx] 1603 1604 # Use the static helper to load EICs 1605 loaded_eics = ReadCoreMSHDFMassSpectra._load_eics_from_hdf5_group(sample_group, lcms_obj) 1606 1607 # Ensure eics dictionary exists (should already be initialized in __init__) 1608 if not hasattr(lcms_obj, 'eics') or lcms_obj.eics is None: 1609 lcms_obj.eics = {} 1610 1611 # Add to lcms_obj.eics dictionary 1612 for eic_mz, eic in loaded_eics.items(): 1613 lcms_obj.eics[eic_mz] = eic 1614 1615 # Associate EICs with induced mass features after all samples processed 1616 # This is done outside the loop to handle all samples at once 1617 for lcms_obj in lcms_collection: 1618 if len(lcms_obj.induced_mass_features) > 0: 1619 lcms_obj.associate_eics_with_mass_features(induced=True) 1620 1621 def _load_representative_mass_features(self, lcms_collection): 1622 """Load representative mass features for all clusters from HDF5 files. 1623 1624 This method uses the same logic as process_consensus_features() when loading 1625 representatives, calling get_sample_mf_map_for_representatives() (DRY helper) 1626 to determine which features to load. 1627 1628 Parameters 1629 ---------- 1630 lcms_collection : LCMSCollection 1631 The LCMS collection object to populate with representative mass features. 1632 """ 1633 # Get cluster assignments from the mass_features_dataframe 1634 if "cluster" not in lcms_collection.mass_features_dataframe.columns: 1635 return 1636 1637 # Use DRY helper method to build sample_mf_map with cluster IDs 1638 sample_mf_map = lcms_collection.get_sample_mf_map_for_representatives(include_cluster_id=True) 1639 1640 # Load mass features for each sample 1641 for sample_id, mf_list in sample_mf_map.items(): 1642 lcms_obj = lcms_collection[sample_id] 1643 1644 # Load each mass feature 1645 for mf_id, cluster_id in mf_list: 1646 self._load_single_mass_feature(lcms_obj, mf_id, cluster_id) 1647 1648 def _load_single_mass_feature(self, lcms_obj, feature_id, cluster_index=None): 1649 """Load a single mass feature from an LCMS object's HDF5 file. 1650 1651 Parameters 1652 ---------- 1653 lcms_obj : LCMSBase 1654 The LCMS object to add the mass feature to. 1655 feature_id : int 1656 The ID of the mass feature to load. 1657 cluster_index : int, optional 1658 The cluster index to assign to the loaded mass feature. 1659 """ 1660 hdf5_path = lcms_obj.file_location.with_suffix('.hdf5') 1661 1662 if not hdf5_path.exists(): 1663 return 1664 1665 with h5py.File(hdf5_path, 'r') as f: 1666 if 'mass_features' not in f: 1667 return 1668 1669 mf_group = f['mass_features'] 1670 feature_id_str = str(feature_id) 1671 1672 if feature_id_str not in mf_group: 1673 return 1674 1675 mf_data = mf_group[feature_id_str] 1676 1677 # Create LCMSMassFeature object 1678 mass_feature = LCMSMassFeature( 1679 lcms_obj, 1680 mz=mf_data.attrs["_mz_exp"], 1681 retention_time=mf_data.attrs["_retention_time"], 1682 intensity=mf_data.attrs["_intensity"], 1683 apex_scan=mf_data.attrs["_apex_scan"], 1684 persistence=mf_data.attrs.get("_persistence", 0), 1685 id=feature_id, 1686 ) 1687 1688 # Set cluster_index if provided 1689 if cluster_index is not None: 1690 mass_feature.cluster_index = cluster_index 1691 1692 # Populate additional attributes from HDF5 attributes 1693 for key in mf_data.attrs.keys() - { 1694 "_mz_exp", 1695 "_mz_cal", 1696 "_retention_time", 1697 "_intensity", 1698 "_apex_scan", 1699 "_persistence", 1700 }: 1701 setattr(mass_feature, key, mf_data.attrs[key]) 1702 1703 # Populate attributes from HDF5 datasets (arrays) 1704 for key in mf_data.keys(): 1705 setattr(mass_feature, key, mf_data[key][:]) 1706 # Convert _noise_score from array to tuple 1707 if key == "_noise_score": 1708 mass_feature._noise_score = tuple(mass_feature._noise_score) 1709 1710 # Add to the LCMS object's mass_features dictionary 1711 lcms_obj.mass_features[feature_id] = mass_feature
Subclass to read and re-instantiate a LCMSCollection from a saved HDF5 file.
Parameters
- collection_hdf5_path (str or Path): Path to the saved LCMSCollection HDF5 file.
- cores (int, optional): Number of cores for processing. Default is 1.
ReadSavedLCMSCollection(collection_hdf5_path: str, cores: int = 1)
1304 def __init__( 1305 self, 1306 collection_hdf5_path: str, 1307 cores: int = 1 1308 ): 1309 # Convert to Path objects 1310 self.collection_hdf5_path = Path(collection_hdf5_path) 1311 1312 # Validate the collection file exists 1313 if not self.collection_hdf5_path.exists(): 1314 raise FileNotFoundError(f"Collection HDF5 file {self.collection_hdf5_path} not found.") 1315 1316 # Validate cores 1317 self._validate_cores(cores) 1318 1319 # Load metadata from saved collection 1320 self._load_collection_metadata() 1321 1322 if not self.folder_location.exists(): 1323 raise FileNotFoundError(f"Folder location {self.folder_location} not found.") 1324 1325 # Load the mass spectra data 1326 self._validate_manifest() 1327 1328 # Set the parameters file location 1329 self.parameters_location = self._get_parameters_location()
def
get_lcms_collection( self, load_raw=False, load_light=False, load_representatives=False, load_eics=False, load_ms1=False, load_ms2=False):
1422 def get_lcms_collection(self, load_raw=False, load_light=False, load_representatives=False, load_eics=False, load_ms1=False, load_ms2=False): 1423 """Get the LCMS collection from the saved HDF5 file. 1424 1425 Parameters 1426 ---------- 1427 load_raw : bool, optional 1428 If True, load raw data. Default is False. 1429 load_light : bool, optional 1430 If True, load light data (minimal). Default is False. 1431 load_representatives : bool, optional 1432 If True, load representative mass features from clusters. Default is False. 1433 load_eics : bool, optional 1434 If True, load EIC data for clustered mass features. Default is False. 1435 load_ms1 : bool, optional 1436 If True, load MS1 spectra for loaded mass features. Default is False. 1437 load_ms2 : bool, optional 1438 If True, load MS2 spectra for loaded mass features. Default is False. 1439 1440 Returns 1441 ------- 1442 LCMSCollection 1443 The loaded LCMS collection object. 1444 """ 1445 # First load the LCMSCollection object exactly as in the parent class 1446 lcms_collection = super().get_lcms_collection(load_raw=load_raw, load_light=load_light) 1447 1448 # Set the missing_mass_features_searched flag from saved metadata 1449 lcms_collection.missing_mass_features_searched = self.missing_mass_features_searched 1450 1451 # Load parameters if a parameters file exists 1452 if self.parameters_location: 1453 self._load_parameters(lcms_collection) 1454 1455 # Add retention time alignments if they exist 1456 self._load_rt_alignments(lcms_collection) 1457 1458 # Add cluster assignments if they exist 1459 self._load_cluster_assignments(lcms_collection) 1460 1461 # Load induced mass features if they exist 1462 self._load_induced_mass_features(lcms_collection) 1463 1464 # Load EICs for induced mass features from collection HDF5 1465 if lcms_collection.missing_mass_features_searched and load_eics: 1466 self._load_induced_eics_from_collection(lcms_collection) 1467 1468 # Combine induced mass features into the collection-level dataframe if any were loaded 1469 if lcms_collection.missing_mass_features_searched: 1470 lcms_collection._combine_mass_features(induced_features=True) 1471 1472 # Load representative mass features if requested 1473 if load_representatives: 1474 self._load_representative_mass_features(lcms_collection) 1475 1476 # Load MS1 and/or MS2 spectra for loaded mass features if requested 1477 if load_ms1 or load_ms2: 1478 # Reuse the existing ReloadFeaturesOperation from the pipeline system 1479 from corems.mass_spectra.calc.lc_calc_operations import ReloadFeaturesOperation 1480 1481 operations = [ReloadFeaturesOperation('reload_spectra', add_ms1=load_ms1, add_ms2=load_ms2)] 1482 lcms_collection.process_samples_pipeline(operations, keep_raw_data=False, show_progress=False) 1483 1484 # Load EICs for clustered features if requested 1485 if load_eics: 1486 # Reuse the existing LoadEICsOperation from the pipeline system 1487 from corems.mass_spectra.calc.lc_calc_operations import LoadEICsOperation 1488 1489 operations = [LoadEICsOperation('load_eics')] 1490 lcms_collection.process_samples_pipeline(operations, keep_raw_data=False, show_progress=False) 1491 1492 # Associate EICs with mass features (same as in process_consensus_features) 1493 for sample_id in range(len(lcms_collection.samples)): 1494 sample = lcms_collection[sample_id] 1495 if sample.eics: # Only if EICs were loaded 1496 # Associate EICs with regular mass features 1497 sample.associate_eics_with_mass_features(induced=False) 1498 # Associate EICs with induced mass features 1499 sample.associate_eics_with_mass_features(induced=True) 1500 1501 return lcms_collection
Get the LCMS collection from the saved HDF5 file.
Parameters
- load_raw (bool, optional): If True, load raw data. Default is False.
- load_light (bool, optional): If True, load light data (minimal). Default is False.
- load_representatives (bool, optional): If True, load representative mass features from clusters. Default is False.
- load_eics (bool, optional): If True, load EIC data for clustered mass features. Default is False.
- load_ms1 (bool, optional): If True, load MS1 spectra for loaded mass features. Default is False.
- load_ms2 (bool, optional): If True, load MS2 spectra for loaded mass features. Default is False.
Returns
- LCMSCollection: The loaded LCMS collection object.