corems.mass_spectra.output.export

   1__author__ = "Yuri E. Corilo"
   2__date__ = "Dec 14, 2010"
   3
   4
   5import csv
   6import json
   7import re
   8import uuid
   9import warnings
  10from datetime import datetime, timezone
  11from pathlib import Path
  12
  13import h5py
  14import numpy as np
  15import pandas as pd
  16from openpyxl import load_workbook
  17from pandas import DataFrame, ExcelWriter, read_excel
  18
  19from corems import __version__, corems_md5
  20from corems.encapsulation.output import parameter_to_dict
  21from corems.encapsulation.output.parameter_to_json import (
  22    dump_lcms_settings_json,
  23    dump_lcms_settings_toml,
  24    dump_lcms_collection_settings_json,
  25    dump_lcms_collection_settings_toml,
  26)
  27from corems.mass_spectrum.output.export import HighResMassSpecExport
  28from corems.molecular_formula.factory.MolecularFormulaFactory import MolecularFormula
  29from corems.molecular_id.calc.SpectralSimilarity import methods_name
  30
  31ion_type_dict = {
  32    # adduct : [atoms to add, atoms to subtract when calculating formula of ion
  33    "M+": [{}, {}],
  34    "[M]+": [{}, {}],
  35    "protonated": [{"H": 1}, {}],
  36    "[M+H]+": [{"H": 1}, {}],
  37    "[M+NH4]+": [{"N": 1, "H": 4}, {}],  # ammonium
  38    "[M+Na]+": [{"Na": 1}, {}],
  39    "[M+K]+": [{"K": 1}, {}],
  40    "[M+2Na+Cl]+": [{"Na": 2, "Cl": 1}, {}],
  41    "[M+2Na-H]+": [{"Na": 2}, {"H": 1}],
  42    "[M+C2H3Na2O2]+": [{"C": 2, "H": 3, "Na": 2, "O": 2}, {}],
  43    "[M+C4H10N3]+": [{"C": 4, "H": 10, "N": 3}, {}],
  44    "[M+NH4+ACN]+": [{"C": 2, "H": 7, "N": 2}, {}],
  45    "[M+H-H2O]+": [{}, {"H": 1, "O": 1}],
  46    "de-protonated": [{}, {"H": 1}],
  47    "[M-H]-": [{}, {"H": 1}],
  48    "[M+Cl]-": [{"Cl": 1}, {}],
  49    "[M+HCOO]-": [{"C": 1, "H": 1, "O": 2}, {}],  # formate
  50    "[M+CH3COO]-": [{"C": 2, "H": 3, "O": 2}, {}],  # acetate
  51    "[M+2NaAc+Cl]-": [{"Na": 2, "C": 2, "H": 3, "O": 2, "Cl": 1}, {}],
  52    "[M+K-2H]-": [{"K": 1}, {"H": 2}],
  53    "[M+Na-2H]-": [{"Na": 1}, {"H": 2}],
  54}
  55
  56
  57class LowResGCMSExport:
  58    """A class to export low resolution GC-MS data.
  59
  60    This class provides methods to export low resolution GC-MS data to various formats such as Excel, CSV, HDF5, and Pandas DataFrame.
  61
  62    Parameters:
  63    ----------
  64    out_file_path : str
  65        The output file path.
  66    gcms : object
  67        The low resolution GCMS object.
  68
  69    Attributes:
  70    ----------
  71    output_file : Path
  72        The output file path as a Path object.
  73    gcms : object
  74        The low resolution GCMS object.
  75
  76    Methods:
  77    -------
  78    * get_pandas_df(id_label="corems:"). Get the exported data as a Pandas DataFrame.
  79    * get_json(nan=False, id_label="corems:"). Get the exported data as a JSON string.
  80    * to_pandas(write_metadata=True, id_label="corems:"). Export the data to a Pandas DataFrame and save it as a pickle file.
  81    * to_excel(write_mode='a', write_metadata=True, id_label="corems:"),
  82        Export the data to an Excel file.
  83    * to_csv(separate_output=False, write_mode="w", write_metadata=True, id_label="corems:").
  84        Export the data to a CSV file.
  85    * to_hdf(id_label="corems:").
  86        Export the data to an HDF5 file.
  87    * get_data_stats(gcms).
  88        Get statistics about the GCMS data.
  89
  90    """
  91
  92    def __init__(self, out_file_path, gcms):
  93        self.output_file = Path(out_file_path)
  94
  95        self.gcms = gcms
  96
  97        self._init_columns()
  98
  99    def _init_columns(self):
 100        """Initialize the column names for the exported data.
 101
 102        Returns:
 103        -------
 104        list
 105            The list of column names.
 106        """
 107
 108        columns = [
 109            "Sample name",
 110            "Peak Index",
 111            "Retention Time",
 112            "Retention Time Ref",
 113            "Peak Height",
 114            "Peak Area",
 115            "Retention index",
 116            "Retention index Ref",
 117            "Retention Index Score",
 118            "Similarity Score",
 119            "Spectral Similarity Score",
 120            "Compound Name",
 121            "Chebi ID",
 122            "Kegg Compound ID",
 123            "Inchi",
 124            "Inchi Key",
 125            "Smiles",
 126            "Molecular Formula",
 127            "IUPAC Name",
 128            "Traditional Name",
 129            "Common Name",
 130            "Derivatization",
 131        ]
 132
 133        if self.gcms.molecular_search_settings.exploratory_mode:
 134            columns.extend(
 135                [
 136                    "Weighted Cosine Correlation",
 137                    "Cosine Correlation",
 138                    "Stein Scott Similarity",
 139                    "Pearson Correlation",
 140                    "Spearman Correlation",
 141                    "Kendall Tau Correlation",
 142                    "Euclidean Distance",
 143                    "Manhattan Distance",
 144                    "Jaccard Distance",
 145                    "DWT Correlation",
 146                    "DFT Correlation",
 147                ]
 148            )
 149
 150            columns.extend(list(methods_name.values()))
 151
 152        return columns
 153
 154    def get_pandas_df(self, id_label="corems:"):
 155        """Get the exported data as a Pandas DataFrame.
 156
 157        Parameters:
 158        ----------
 159        id_label : str, optional
 160            The ID label for the data. Default is "corems:".
 161
 162        Returns:
 163        -------
 164        DataFrame
 165            The exported data as a Pandas DataFrame.
 166        """
 167
 168        columns = self._init_columns()
 169
 170        dict_data_list = self.get_list_dict_data(self.gcms)
 171
 172        df = DataFrame(dict_data_list, columns=columns)
 173
 174        df.name = self.gcms.sample_name
 175
 176        return df
 177
 178    def get_json(self, nan=False, id_label="corems:"):
 179        """Get the exported data as a JSON string.
 180
 181        Parameters:
 182        ----------
 183        nan : bool, optional
 184            Whether to include NaN values in the JSON string. Default is False.
 185        id_label : str, optional
 186            The ID label for the data. Default is "corems:".
 187
 188        """
 189
 190        import json
 191
 192        dict_data_list = self.get_list_dict_data(self.gcms)
 193
 194        return json.dumps(
 195            dict_data_list, sort_keys=False, indent=4, separators=(",", ": ")
 196        )
 197
 198    def to_pandas(self, write_metadata=True, id_label="corems:"):
 199        """Export the data to a Pandas DataFrame and save it as a pickle file.
 200
 201        Parameters:
 202        ----------
 203        write_metadata : bool, optional
 204            Whether to write metadata to the output file.
 205        id_label : str, optional
 206            The ID label for the data.
 207        """
 208
 209        columns = self._init_columns()
 210
 211        dict_data_list = self.get_list_dict_data(self.gcms)
 212
 213        df = DataFrame(dict_data_list, columns=columns)
 214
 215        df.to_pickle(self.output_file.with_suffix(".pkl"))
 216
 217        if write_metadata:
 218            self.write_settings(
 219                self.output_file.with_suffix(".pkl"), self.gcms, id_label="corems:"
 220            )
 221
 222    def to_excel(self, write_mode="a", write_metadata=True, id_label="corems:"):
 223        """Export the data to an Excel file.
 224
 225        Parameters:
 226        ----------
 227        write_mode : str, optional
 228            The write mode for the Excel file. Default is 'a' (append).
 229        write_metadata : bool, optional
 230            Whether to write metadata to the output file. Default is True.
 231        id_label : str, optional
 232            The ID label for the data. Default is "corems:".
 233        """
 234
 235        out_put_path = self.output_file.with_suffix(".xlsx")
 236
 237        columns = self._init_columns()
 238
 239        dict_data_list = self.get_list_dict_data(self.gcms)
 240
 241        df = DataFrame(dict_data_list, columns=columns)
 242
 243        if write_mode == "a" and out_put_path.exists():
 244            writer = ExcelWriter(out_put_path, engine="openpyxl")
 245            # try to open an existing workbook
 246            writer.book = load_workbook(out_put_path)
 247            # copy existing sheets
 248            writer.sheets = dict((ws.title, ws) for ws in writer.book.worksheets)
 249            # read existing file
 250            reader = read_excel(out_put_path)
 251            # write out the new sheet
 252            df.to_excel(writer, index=False, header=False, startrow=len(reader) + 1)
 253
 254            writer.close()
 255        else:
 256            df.to_excel(
 257                self.output_file.with_suffix(".xlsx"), index=False, engine="openpyxl"
 258            )
 259
 260        if write_metadata:
 261            self.write_settings(out_put_path, self.gcms, id_label=id_label)
 262
 263    def to_csv(
 264        self,
 265        separate_output=False,
 266        write_mode="w",
 267        write_metadata=True,
 268        id_label="corems:",
 269    ):
 270        """Export the data to a CSV file.
 271
 272        Parameters:
 273        ----------
 274        separate_output : bool, optional
 275            Whether to separate the output into multiple files. Default is False.
 276        write_mode : str, optional
 277            The write mode for the CSV file. Default is 'w' (write).
 278        write_metadata : bool, optional
 279            Whether to write metadata to the output file. Default is True.
 280        id_label : str, optional
 281            The ID label for the data. Default is "corems:".
 282        """
 283
 284        if separate_output:
 285            # set write mode to write
 286            # this mode will overwrite the file without warning
 287            write_mode = "w"
 288        else:
 289            # set write mode to append
 290            write_mode = "a"
 291
 292        columns = self._init_columns()
 293
 294        dict_data_list = self.get_list_dict_data(self.gcms)
 295
 296        out_put_path = self.output_file.with_suffix(".csv")
 297
 298        write_header = not out_put_path.exists()
 299
 300        try:
 301            with open(out_put_path, write_mode, newline="") as csvfile:
 302                writer = csv.DictWriter(csvfile, fieldnames=columns)
 303                if write_header:
 304                    writer.writeheader()
 305                for data in dict_data_list:
 306                    writer.writerow(data)
 307
 308            if write_metadata:
 309                self.write_settings(out_put_path, self.gcms, id_label=id_label)
 310
 311        except IOError as ioerror:
 312            print(ioerror)
 313
 314    def to_hdf(self, id_label="corems:"):
 315        """Export the data to an HDF5 file.
 316
 317        Parameters:
 318        ----------
 319        id_label : str, optional
 320            The ID label for the data. Default is "corems:".
 321        """
 322
 323        # save sample at a time
 324        def add_compound(gc_peak, compound_obj):
 325            modifier = compound_obj.classify if compound_obj.classify else ""
 326            compound_group = compound_obj.name.replace("/", "") + " " + modifier
 327
 328            if compound_group not in peak_group:
 329                compound_group = peak_group.create_group(compound_group)
 330
 331                # compound_group.attrs["retention_time"] = compound_obj.retention_time
 332                compound_group.attrs["retention_index"] = compound_obj.ri
 333                compound_group.attrs["retention_index_score"] = compound_obj.ri_score
 334                compound_group.attrs["spectral_similarity_score"] = (
 335                    compound_obj.spectral_similarity_score
 336                )
 337                compound_group.attrs["similarity_score"] = compound_obj.similarity_score
 338
 339                compond_mz = compound_group.create_dataset(
 340                    "mz", data=np.array(compound_obj.mz), dtype="f8"
 341                )
 342                compond_abundance = compound_group.create_dataset(
 343                    "abundance", data=np.array(compound_obj.abundance), dtype="f8"
 344                )
 345
 346                if self.gcms.molecular_search_settings.exploratory_mode:
 347                    compound_group.attrs["Spectral Similarities"] = json.dumps(
 348                        compound_obj.spectral_similarity_scores,
 349                        sort_keys=False,
 350                        indent=4,
 351                        separators=(",", ":"),
 352                    )
 353            else:
 354                warnings.warn("Skipping duplicate reference compound.")
 355
 356        import json
 357        from datetime import datetime, timezone
 358
 359        import h5py
 360        import numpy as np
 361
 362        output_path = self.output_file.with_suffix(".hdf5")
 363
 364        with h5py.File(output_path, "w") as hdf_handle:
 365            timenow = str(datetime.now(timezone.utc).strftime("%d/%m/%Y %H:%M:%S %Z"))
 366            hdf_handle.attrs["time_stamp"] = timenow
 367            hdf_handle.attrs["data_structure"] = "gcms"
 368            hdf_handle.attrs["analyzer"] = self.gcms.analyzer
 369            hdf_handle.attrs["instrument_label"] = self.gcms.instrument_label
 370
 371            hdf_handle.attrs["sample_id"] = "self.gcms.id"
 372            hdf_handle.attrs["sample_name"] = self.gcms.sample_name
 373            hdf_handle.attrs["input_data"] = str(self.gcms.file_location)
 374            hdf_handle.attrs["output_data"] = str(output_path)
 375            hdf_handle.attrs["output_data_id"] = id_label + uuid.uuid4().hex
 376            hdf_handle.attrs["corems_version"] = __version__
 377
 378            hdf_handle.attrs["Stats"] = json.dumps(
 379                self.get_data_stats(self.gcms),
 380                sort_keys=False,
 381                indent=4,
 382                separators=(",", ": "),
 383            )
 384            hdf_handle.attrs["Calibration"] = json.dumps(
 385                self.get_calibration_stats(self.gcms, id_label),
 386                sort_keys=False,
 387                indent=4,
 388                separators=(",", ": "),
 389            )
 390            hdf_handle.attrs["Blank"] = json.dumps(
 391                self.get_blank_stats(self.gcms),
 392                sort_keys=False,
 393                indent=4,
 394                separators=(",", ": "),
 395            )
 396
 397            corems_dict_setting = parameter_to_dict.get_dict_data_gcms(self.gcms)
 398            hdf_handle.attrs["CoreMSParameters"] = json.dumps(
 399                corems_dict_setting, sort_keys=False, indent=4, separators=(",", ": ")
 400            )
 401
 402            scans_dataset = hdf_handle.create_dataset(
 403                "scans", data=np.array(self.gcms.scans_number), dtype="f8"
 404            )
 405            rt_dataset = hdf_handle.create_dataset(
 406                "rt", data=np.array(self.gcms.retention_time), dtype="f8"
 407            )
 408            tic_dataset = hdf_handle.create_dataset(
 409                "tic", data=np.array(self.gcms.tic), dtype="f8"
 410            )
 411            processed_tic_dataset = hdf_handle.create_dataset(
 412                "processed_tic", data=np.array(self.gcms.processed_tic), dtype="f8"
 413            )
 414
 415            output_score_method = (
 416                self.gcms.molecular_search_settings.output_score_method
 417            )
 418
 419            for gc_peak in self.gcms:
 420                # print(gc_peak.retention_time)
 421                # print(gc_peak.tic)
 422
 423                # check if there is a compound candidate
 424                peak_group = hdf_handle.create_group(str(gc_peak.retention_time))
 425                peak_group.attrs["deconvolution"] = int(
 426                    self.gcms.chromatogram_settings.use_deconvolution
 427                )
 428
 429                peak_group.attrs["start_scan"] = gc_peak.start_scan
 430                peak_group.attrs["apex_scan"] = gc_peak.apex_scan
 431                peak_group.attrs["final_scan"] = gc_peak.final_scan
 432
 433                peak_group.attrs["retention_index"] = gc_peak.ri
 434                peak_group.attrs["retention_time"] = gc_peak.retention_time
 435                peak_group.attrs["area"] = gc_peak.area
 436
 437                mz = peak_group.create_dataset(
 438                    "mz", data=np.array(gc_peak.mass_spectrum.mz_exp), dtype="f8"
 439                )
 440                abundance = peak_group.create_dataset(
 441                    "abundance",
 442                    data=np.array(gc_peak.mass_spectrum.abundance),
 443                    dtype="f8",
 444                )
 445
 446                if gc_peak:
 447                    if output_score_method == "highest_sim_score":
 448                        compound_obj = gc_peak.highest_score_compound
 449                        add_compound(gc_peak, compound_obj)
 450
 451                    elif output_score_method == "highest_ss":
 452                        compound_obj = gc_peak.highest_ss_compound
 453                        add_compound(gc_peak, compound_obj)
 454
 455                    else:
 456                        for compound_obj in gc_peak:
 457                            add_compound(gc_peak, compound_obj)
 458
 459    def get_data_stats(self, gcms):
 460        """Get statistics about the GCMS data.
 461
 462        Parameters:
 463        ----------
 464        gcms : object
 465            The low resolution GCMS object.
 466
 467        Returns:
 468        -------
 469        dict
 470            A dictionary containing the data statistics.
 471        """
 472
 473        matched_peaks = gcms.matched_peaks
 474        no_matched_peaks = gcms.no_matched_peaks
 475        unique_metabolites = gcms.unique_metabolites
 476
 477        peak_matchs_above_0p85 = 0
 478        unique_peak_match_above_0p85 = 0
 479        for match_peak in matched_peaks:
 480            gc_peak_above_85 = 0
 481            matches_above_85 = list(
 482                filter(lambda m: m.similarity_score >= 0.85, match_peak)
 483            )
 484            if matches_above_85:
 485                peak_matchs_above_0p85 += 1
 486            if len(matches_above_85) == 1:
 487                unique_peak_match_above_0p85 += 1
 488
 489        data_stats = {}
 490        data_stats["average_signal_noise"] = "ni"
 491        data_stats["chromatogram_dynamic_range"] = gcms.dynamic_range
 492        data_stats["total_number_peaks"] = len(gcms)
 493        data_stats["total_peaks_matched"] = len(matched_peaks)
 494        data_stats["total_peaks_without_matches"] = len(no_matched_peaks)
 495        data_stats["total_matches_above_similarity_score_0.85"] = peak_matchs_above_0p85
 496        data_stats["single_matches_above_similarity_score_0.85"] = (
 497            unique_peak_match_above_0p85
 498        )
 499        data_stats["unique_metabolites"] = len(unique_metabolites)
 500
 501        return data_stats
 502
 503    def get_calibration_stats(self, gcms, id_label):
 504        """Get statistics about the GC-MS calibration.
 505
 506        Parameters:
 507        ----------
 508        """
 509        calibration_parameters = {}
 510
 511        calibration_parameters["calibration_rt_ri_pairs_ref"] = gcms.ri_pairs_ref
 512        calibration_parameters["data_url"] = str(gcms.cal_file_path)
 513        calibration_parameters["has_input"] = id_label + corems_md5(gcms.cal_file_path)
 514        calibration_parameters["data_name"] = str(gcms.cal_file_path.stem)
 515        calibration_parameters["calibration_method"] = ""
 516
 517        return calibration_parameters
 518
 519    def get_blank_stats(self, gcms):
 520        """Get statistics about the GC-MS blank."""
 521        blank_parameters = {}
 522
 523        blank_parameters["data_name"] = "ni"
 524        blank_parameters["blank_id"] = "ni"
 525        blank_parameters["data_url"] = "ni"
 526        blank_parameters["has_input"] = "ni"
 527        blank_parameters["common_features_to_blank"] = "ni"
 528
 529        return blank_parameters
 530
 531    def get_instrument_metadata(self, gcms):
 532        """Get metadata about the GC-MS instrument."""
 533        instrument_metadata = {}
 534
 535        instrument_metadata["analyzer"] = gcms.analyzer
 536        instrument_metadata["instrument_label"] = gcms.instrument_label
 537        instrument_metadata["instrument_id"] = uuid.uuid4().hex
 538
 539        return instrument_metadata
 540
 541    def get_data_metadata(self, gcms, id_label, output_path):
 542        """Get metadata about the GC-MS data.
 543
 544        Parameters:
 545        ----------
 546        gcms : object
 547            The low resolution GCMS object.
 548        id_label : str
 549            The ID label for the data.
 550        output_path : str
 551            The output file path.
 552
 553        Returns:
 554        -------
 555        dict
 556            A dictionary containing the data metadata.
 557        """
 558        if isinstance(output_path, str):
 559            output_path = Path(output_path)
 560
 561        paramaters_path = output_path.with_suffix(".json")
 562
 563        if paramaters_path.exists():
 564            with paramaters_path.open() as current_param:
 565                metadata = json.load(current_param)
 566                data_metadata = metadata.get("Data")
 567        else:
 568            data_metadata = {}
 569            data_metadata["data_name"] = []
 570            data_metadata["input_data_url"] = []
 571            data_metadata["has_input"] = []
 572
 573        data_metadata["data_name"].append(gcms.sample_name)
 574        data_metadata["input_data_url"].append(str(gcms.file_location))
 575        data_metadata["has_input"].append(id_label + corems_md5(gcms.file_location))
 576
 577        data_metadata["output_data_name"] = str(output_path.stem)
 578        data_metadata["output_data_url"] = str(output_path)
 579        data_metadata["has_output"] = id_label + corems_md5(output_path)
 580
 581        return data_metadata
 582
 583    def get_parameters_json(self, gcms, id_label, output_path):
 584        """Get the parameters as a JSON string.
 585
 586        Parameters:
 587        ----------
 588        gcms : GCMS object
 589            The low resolution GCMS object.
 590        id_label : str
 591            The ID label for the data.
 592        output_path : str
 593            The output file path.
 594
 595        Returns:
 596        -------
 597        str
 598            The parameters as a JSON string.
 599        """
 600
 601        output_parameters_dict = {}
 602        output_parameters_dict["Data"] = self.get_data_metadata(
 603            gcms, id_label, output_path
 604        )
 605        output_parameters_dict["Stats"] = self.get_data_stats(gcms)
 606        output_parameters_dict["Calibration"] = self.get_calibration_stats(
 607            gcms, id_label
 608        )
 609        output_parameters_dict["Blank"] = self.get_blank_stats(gcms)
 610        output_parameters_dict["Instrument"] = self.get_instrument_metadata(gcms)
 611        corems_dict_setting = parameter_to_dict.get_dict_data_gcms(gcms)
 612        corems_dict_setting["corems_version"] = __version__
 613        output_parameters_dict["CoreMSParameters"] = corems_dict_setting
 614        output_parameters_dict["has_metabolite"] = gcms.metabolites_data
 615        output = json.dumps(
 616            output_parameters_dict, sort_keys=False, indent=4, separators=(",", ": ")
 617        )
 618
 619        return output
 620
 621    def write_settings(self, output_path, gcms, id_label="emsl:"):
 622        """Write the settings to a JSON file.
 623
 624        Parameters:
 625        ----------
 626        output_path : str
 627            The output file path.
 628        gcms : GCMS object
 629            The low resolution GCMS object.
 630        id_label : str
 631            The ID label for the data. Default is "emsl:".
 632
 633        """
 634
 635        output = self.get_parameters_json(gcms, id_label, output_path)
 636
 637        with open(
 638            output_path.with_suffix(".json"),
 639            "w",
 640            encoding="utf8",
 641        ) as outfile:
 642            outfile.write(output)
 643
 644    def get_list_dict_data(self, gcms, include_no_match=True, no_match_inline=False):
 645        """Get the exported data as a list of dictionaries.
 646
 647        Parameters:
 648        ----------
 649        gcms : object
 650            The low resolution GCMS object.
 651        include_no_match : bool, optional
 652            Whether to include no match data. Default is True.
 653        no_match_inline : bool, optional
 654            Whether to include no match data inline. Default is False.
 655
 656        Returns:
 657        -------
 658        list
 659            The exported data as a list of dictionaries.
 660        """
 661
 662        output_score_method = gcms.molecular_search_settings.output_score_method
 663
 664        dict_data_list = []
 665
 666        def add_match_dict_data():
 667            derivatization = "{}:{}:{}".format(
 668                compound_obj.classify,
 669                compound_obj.derivativenum,
 670                compound_obj.derivatization,
 671            )
 672            out_dict = {
 673                "Sample name": gcms.sample_name,
 674                "Peak Index": gcpeak_index,
 675                "Retention Time": gc_peak.retention_time,
 676                "Retention Time Ref": compound_obj.retention_time,
 677                "Peak Height": gc_peak.tic,
 678                "Peak Area": gc_peak.area,
 679                "Retention index": gc_peak.ri,
 680                "Retention index Ref": compound_obj.ri,
 681                "Retention Index Score": compound_obj.ri_score,
 682                "Spectral Similarity Score": compound_obj.spectral_similarity_score,
 683                "Similarity Score": compound_obj.similarity_score,
 684                "Compound Name": compound_obj.name,
 685                "Chebi ID": compound_obj.metadata.chebi,
 686                "Kegg Compound ID": compound_obj.metadata.kegg,
 687                "Inchi": compound_obj.metadata.inchi,
 688                "Inchi Key": compound_obj.metadata.inchikey,
 689                "Smiles": compound_obj.metadata.smiles,
 690                "Molecular Formula": compound_obj.formula,
 691                "IUPAC Name": compound_obj.metadata.iupac_name,
 692                "Traditional Name": compound_obj.metadata.traditional_name,
 693                "Common Name": compound_obj.metadata.common_name,
 694                "Derivatization": derivatization,
 695            }
 696
 697            if self.gcms.molecular_search_settings.exploratory_mode:
 698                out_dict.update(
 699                    {
 700                        "Weighted Cosine Correlation": compound_obj.spectral_similarity_scores.get(
 701                            "weighted_cosine_correlation"
 702                        ),
 703                        "Cosine Correlation": compound_obj.spectral_similarity_scores.get(
 704                            "cosine_correlation"
 705                        ),
 706                        "Stein Scott Similarity": compound_obj.spectral_similarity_scores.get(
 707                            "stein_scott_similarity"
 708                        ),
 709                        "Pearson Correlation": compound_obj.spectral_similarity_scores.get(
 710                            "pearson_correlation"
 711                        ),
 712                        "Spearman Correlation": compound_obj.spectral_similarity_scores.get(
 713                            "spearman_correlation"
 714                        ),
 715                        "Kendall Tau Correlation": compound_obj.spectral_similarity_scores.get(
 716                            "kendall_tau_correlation"
 717                        ),
 718                        "DFT Correlation": compound_obj.spectral_similarity_scores.get(
 719                            "dft_correlation"
 720                        ),
 721                        "DWT Correlation": compound_obj.spectral_similarity_scores.get(
 722                            "dwt_correlation"
 723                        ),
 724                        "Euclidean Distance": compound_obj.spectral_similarity_scores.get(
 725                            "euclidean_distance"
 726                        ),
 727                        "Manhattan Distance": compound_obj.spectral_similarity_scores.get(
 728                            "manhattan_distance"
 729                        ),
 730                        "Jaccard Distance": compound_obj.spectral_similarity_scores.get(
 731                            "jaccard_distance"
 732                        ),
 733                    }
 734                )
 735                for method in methods_name:
 736                    out_dict[methods_name.get(method)] = (
 737                        compound_obj.spectral_similarity_scores.get(method)
 738                    )
 739
 740            dict_data_list.append(out_dict)
 741
 742        def add_no_match_dict_data():
 743            dict_data_list.append(
 744                {
 745                    "Sample name": gcms.sample_name,
 746                    "Peak Index": gcpeak_index,
 747                    "Retention Time": gc_peak.retention_time,
 748                    "Peak Height": gc_peak.tic,
 749                    "Peak Area": gc_peak.area,
 750                    "Retention index": gc_peak.ri,
 751                }
 752            )
 753
 754        for gcpeak_index, gc_peak in enumerate(gcms.sorted_gcpeaks):
 755            # check if there is a compound candidate
 756            if gc_peak:
 757                if output_score_method == "highest_sim_score":
 758                    compound_obj = gc_peak.highest_score_compound
 759                    add_match_dict_data()
 760
 761                elif output_score_method == "highest_ss":
 762                    compound_obj = gc_peak.highest_ss_compound
 763                    add_match_dict_data()
 764
 765                else:
 766                    for compound_obj in gc_peak:
 767                        add_match_dict_data()  # add monoisotopic peak
 768
 769            else:
 770                # include not_match
 771                if include_no_match and no_match_inline:
 772                    add_no_match_dict_data()
 773
 774        if include_no_match and not no_match_inline:
 775            for gcpeak_index, gc_peak in enumerate(gcms.sorted_gcpeaks):
 776                if not gc_peak:
 777                    add_no_match_dict_data()
 778
 779        return dict_data_list
 780
 781
 782class HighResMassSpectraExport(HighResMassSpecExport):
 783    """A class to export high resolution mass spectra data.
 784
 785    This class provides methods to export high resolution mass spectra data to various formats
 786    such as Excel, CSV, HDF5, and Pandas DataFrame.
 787
 788    Parameters
 789    ----------
 790    out_file_path : str | Path
 791        The output file path.
 792    mass_spectra : object
 793        The high resolution mass spectra object.
 794    output_type : str, optional
 795        The output type. Default is 'excel'.
 796
 797    Attributes
 798    ----------
 799    output_file : Path
 800        The output file path without suffix
 801    dir_loc : Path
 802        The directory location for the output file,
 803        by default this will be the output_file + ".corems" and all output files will be
 804        written into this location
 805    mass_spectra : MassSpectraBase
 806        The high resolution mass spectra object.
 807    """
 808
 809    def __init__(self, out_file_path, mass_spectra, output_type="excel"):
 810        super().__init__(
 811            out_file_path=out_file_path, mass_spectrum=None, output_type=output_type
 812        )
 813
 814        self.dir_loc = Path(out_file_path + ".corems")
 815        self.dir_loc.mkdir(exist_ok=True)
 816        # Place the output file in the directory
 817        self.output_file = self.dir_loc / Path(out_file_path).name
 818        self._output_type = output_type  # 'excel', 'csv', 'pandas' or 'hdf5'
 819        self.mass_spectra = mass_spectra
 820        self.atoms_order_list = None
 821        self._init_columns()
 822
 823    def get_pandas_df(self):
 824        """Get the mass spectra as a list of Pandas DataFrames."""
 825
 826        list_df = []
 827
 828        for mass_spectrum in self.mass_spectra:
 829            columns = self.columns_label + self.get_all_used_atoms_in_order(
 830                mass_spectrum
 831            )
 832
 833            dict_data_list = self.get_list_dict_data(mass_spectrum)
 834
 835            df = DataFrame(dict_data_list, columns=columns)
 836
 837            scan_number = mass_spectrum.scan_number
 838
 839            df.name = str(self.output_file) + "_" + str(scan_number)
 840
 841            list_df.append(df)
 842
 843        return list_df
 844
 845    def to_pandas(self, write_metadata=True):
 846        """Export the data to a Pandas DataFrame and save it as a pickle file.
 847
 848        Parameters:
 849        ----------
 850        write_metadata : bool, optional
 851            Whether to write metadata to the output file. Default is True.
 852        """
 853
 854        for mass_spectrum in self.mass_spectra:
 855            columns = self.columns_label + self.get_all_used_atoms_in_order(
 856                mass_spectrum
 857            )
 858
 859            dict_data_list = self.get_list_dict_data(mass_spectrum)
 860
 861            df = DataFrame(dict_data_list, columns=columns)
 862
 863            scan_number = mass_spectrum.scan_number
 864
 865            out_filename = Path(
 866                "%s_scan%s%s" % (self.output_file, str(scan_number), ".pkl")
 867            )
 868
 869            df.to_pickle(self.dir_loc / out_filename)
 870
 871            if write_metadata:
 872                self.write_settings(
 873                    self.dir_loc / out_filename.with_suffix(""), mass_spectrum
 874                )
 875
 876    def to_excel(self, write_metadata=True):
 877        """Export the data to an Excel file.
 878
 879        Parameters:
 880        ----------
 881        write_metadata : bool, optional
 882            Whether to write metadata to the output file. Default is True.
 883        """
 884        for mass_spectrum in self.mass_spectra:
 885            columns = self.columns_label + self.get_all_used_atoms_in_order(
 886                mass_spectrum
 887            )
 888
 889            dict_data_list = self.get_list_dict_data(mass_spectrum)
 890
 891            df = DataFrame(dict_data_list, columns=columns)
 892
 893            scan_number = mass_spectrum.scan_number
 894
 895            out_filename = Path(
 896                "%s_scan%s%s" % (self.output_file, str(scan_number), ".xlsx")
 897            )
 898
 899            df.to_excel(self.dir_loc / out_filename)
 900
 901            if write_metadata:
 902                self.write_settings(
 903                    self.dir_loc / out_filename.with_suffix(""), mass_spectrum
 904                )
 905
 906    def to_csv(self, write_metadata=True):
 907        """Export the data to a CSV file.
 908
 909        Parameters:
 910        ----------
 911        write_metadata : bool, optional
 912            Whether to write metadata to the output file. Default is True.
 913        """
 914        import csv
 915
 916        for mass_spectrum in self.mass_spectra:
 917            columns = self.columns_label + self.get_all_used_atoms_in_order(
 918                mass_spectrum
 919            )
 920
 921            scan_number = mass_spectrum.scan_number
 922
 923            dict_data_list = self.get_list_dict_data(mass_spectrum)
 924
 925            out_filename = Path(
 926                "%s_scan%s%s" % (self.output_file, str(scan_number), ".csv")
 927            )
 928
 929            with open(self.dir_loc / out_filename, "w", newline="") as csvfile:
 930                writer = csv.DictWriter(csvfile, fieldnames=columns)
 931                writer.writeheader()
 932                for data in dict_data_list:
 933                    writer.writerow(data)
 934
 935            if write_metadata:
 936                self.write_settings(
 937                    self.dir_loc / out_filename.with_suffix(""), mass_spectrum
 938                )
 939
 940    def get_mass_spectra_attrs(self):
 941        """Get the mass spectra attributes as a JSON string.
 942
 943        Parameters:
 944        ----------
 945        mass_spectra : object
 946            The high resolution mass spectra object.
 947
 948        Returns:
 949        -------
 950        str
 951            The mass spectra attributes as a JSON string.
 952        """
 953        dict_ms_attrs = {}
 954        dict_ms_attrs["analyzer"] = self.mass_spectra.analyzer
 955        dict_ms_attrs["instrument_label"] = self.mass_spectra.instrument_label
 956        dict_ms_attrs["sample_name"] = self.mass_spectra.sample_name
 957
 958        return json.dumps(
 959            dict_ms_attrs, sort_keys=False, indent=4, separators=(",", ": ")
 960        )
 961
 962    def to_hdf(self, overwrite=False, export_raw=True):
 963        """Export the data to an HDF5 file.
 964
 965        Parameters
 966        ----------
 967        overwrite : bool, optional
 968            Whether to overwrite the output file. Default is False.
 969        export_raw : bool, optional
 970            Whether to export the raw mass spectra data. Default is True.
 971        """
 972        if overwrite:
 973            if self.output_file.with_suffix(".hdf5").exists():
 974                self.output_file.with_suffix(".hdf5").unlink()
 975
 976        with h5py.File(self.output_file.with_suffix(".hdf5"), "a") as hdf_handle:
 977            if not hdf_handle.attrs.get("date_utc"):
 978                # Set metadata for all mass spectra
 979                timenow = str(
 980                    datetime.now(timezone.utc).strftime("%d/%m/%Y %H:%M:%S %Z")
 981                )
 982                hdf_handle.attrs["date_utc"] = timenow
 983                hdf_handle.attrs["filename"] = self.mass_spectra.file_location.name
 984                hdf_handle.attrs["data_structure"] = "mass_spectra"
 985                hdf_handle.attrs["analyzer"] = self.mass_spectra.analyzer
 986                hdf_handle.attrs["instrument_label"] = (
 987                    self.mass_spectra.instrument_label
 988                )
 989                hdf_handle.attrs["sample_name"] = self.mass_spectra.sample_name
 990                hdf_handle.attrs["polarity"] = self.mass_spectra.polarity
 991                hdf_handle.attrs["parser_type"] = (
 992                    self.mass_spectra.spectra_parser_class.__name__
 993                )
 994                hdf_handle.attrs["original_file_location"] = (
 995                    self.mass_spectra.file_location._str
 996                )
 997                
 998                # Save creation time from original parser if available
 999                try:
1000                    if hasattr(self.mass_spectra, 'spectra_parser') and self.mass_spectra.spectra_parser is not None:
1001                        creation_time = self.mass_spectra.spectra_parser.get_creation_time()
1002                        if creation_time is not None:
1003                            hdf_handle.attrs["creation_time"] = creation_time.isoformat()
1004                except Exception:
1005                    pass  # If creation time cannot be retrieved, skip it
1006
1007            if "mass_spectra" not in hdf_handle:
1008                mass_spectra_group = hdf_handle.create_group("mass_spectra")
1009            else:
1010                mass_spectra_group = hdf_handle.get("mass_spectra")
1011
1012            for mass_spectrum in self.mass_spectra:
1013                group_key = str(int(mass_spectrum.scan_number))
1014
1015                self.add_mass_spectrum_to_hdf5(
1016                    hdf_handle, mass_spectrum, group_key, mass_spectra_group, export_raw
1017                )
1018
1019
1020class LCMSExport(HighResMassSpectraExport):
1021    """A class to export high resolution LC-MS data.
1022
1023    This class provides methods to export high resolution LC-MS data to HDF5.
1024
1025    Parameters
1026    ----------
1027    out_file_path : str | Path
1028        The output file path, do not include the file extension.
1029    lcms_object : LCMSBase
1030        The high resolution lc-ms object.
1031    """
1032
1033    def __init__(self, out_file_path, mass_spectra):
1034        super().__init__(out_file_path, mass_spectra, output_type="hdf5")
1035
1036    @staticmethod
1037    def _save_mass_features_dict_to_hdf5(mass_features_dict, mass_features_group, overwrite=False):
1038        """Save a dictionary of mass features to an HDF5 group.
1039        
1040        This is a helper method that can be reused by different export classes.
1041        
1042        Parameters
1043        ----------
1044        mass_features_dict : dict
1045            Dictionary of mass features to save, keyed by mass feature ID.
1046        mass_features_group : h5py.Group
1047            The HDF5 group to save the mass features to.
1048        overwrite : bool, optional
1049            Whether to overwrite existing mass features. Default is False.
1050        """
1051
1052        # Create group for each mass feature, with key as the mass feature id
1053        for k, v in mass_features_dict.items():
1054                if str(k) not in mass_features_group or overwrite:
1055                    if str(k) in mass_features_group and overwrite:
1056                        del mass_features_group[str(k)]
1057                    mass_features_group.create_group(str(k))
1058                    # Loop through each of the mass feature attributes and add them as attributes (if single value) or datasets (if array)
1059                    for k2, v2 in v.__dict__.items():
1060                        if v2 is not None:
1061                            # Check if the attribute is an integer or float and set as an attribute in the mass feature group
1062                            if k2 not in [
1063                                "chromatogram_parent",
1064                                "ms2_mass_spectra",
1065                                "mass_spectrum",
1066                                "_eic_data",
1067                                "ms2_similarity_results",
1068                            ]:
1069                                if k2 == "ms2_scan_numbers":
1070                                    array = np.array(v2)
1071                                    # Convert int64 to int32
1072                                    if array.dtype == np.int64:
1073                                        array = array.astype(np.int32)
1074                                    mass_features_group[str(k)].create_dataset(
1075                                        str(k2), data=array, compression="gzip", compression_opts=9, chunks=True
1076                                    )
1077                                elif k2 == "_half_height_width":
1078                                    array = np.array(v2)
1079                                    # Convert float64 to float32
1080                                    if array.dtype == np.float64:
1081                                        array = array.astype(np.float32)
1082                                    mass_features_group[str(k)].create_dataset(
1083                                        str(k2), data=array, compression="gzip", compression_opts=9, chunks=True
1084                                    )
1085                                elif k2 == "_ms_deconvoluted_idx":
1086                                    array = np.array(v2)
1087                                    # Convert int64 to int32
1088                                    if array.dtype == np.int64:
1089                                        array = array.astype(np.int32)
1090                                    mass_features_group[str(k)].create_dataset(
1091                                        str(k2), data=array, compression="gzip", compression_opts=9, chunks=True
1092                                    )
1093                                elif k2 == "associated_mass_features_deconvoluted":
1094                                    array = np.array(v2)
1095                                    # Convert int64 to int32
1096                                    if array.dtype == np.int64:
1097                                        array = array.astype(np.int32)
1098                                    mass_features_group[str(k)].create_dataset(
1099                                        str(k2), data=array, compression="gzip", compression_opts=9, chunks=True
1100                                    )
1101                                elif k2 == "_noise_score":
1102                                    array = np.array(v2)
1103                                    # Convert float64 to float32
1104                                    if array.dtype == np.float64:
1105                                        array = array.astype(np.float32)
1106                                    mass_features_group[str(k)].create_dataset(
1107                                        str(k2), data=array, compression="gzip", compression_opts=9, chunks=True
1108                                    )
1109                                elif (
1110                                    isinstance(v2, int)
1111                                    or isinstance(v2, float)
1112                                    or isinstance(v2, str)
1113                                    or isinstance(v2, np.integer)
1114                                    or isinstance(v2, np.float32)
1115                                    or isinstance(v2, np.float64)
1116                                    or isinstance(v2, np.bool_)
1117                                ):
1118                                    # Convert numpy types to smaller precision for storage
1119                                    if isinstance(v2, np.int64):
1120                                        v2 = np.int32(v2)
1121                                    elif isinstance(v2, np.float64):
1122                                        v2 = np.float32(v2)
1123                                    mass_features_group[str(k)].attrs[str(k2)] = v2
1124    
1125    @staticmethod
1126    def _save_eics_dict_to_hdf5(eics_dict, eics_group, overwrite=False):
1127        """Save a dictionary of EICs to an HDF5 group.
1128        
1129        This is a static helper method that can be reused by different export classes
1130        to save EIC data in a consistent format.
1131        
1132        Parameters
1133        ----------
1134        eics_dict : dict
1135            Dictionary of EIC_Data objects, keyed by m/z value.
1136        eics_group : h5py.Group
1137            The HDF5 group to save the EICs to.
1138        overwrite : bool, optional
1139            Whether to overwrite existing EICs. Default is False.
1140        """
1141        for mz, eic_data in eics_dict.items():
1142            mz_str = str(mz)
1143            if mz_str not in eics_group or overwrite:
1144                if mz_str in eics_group and overwrite:
1145                    del eics_group[mz_str]
1146                eic_grp = eics_group.create_group(mz_str)
1147                eic_grp.attrs["mz"] = mz
1148                
1149                # Save all EIC_Data attributes as datasets
1150                for attr_name, attr_value in eic_data.__dict__.items():
1151                    if attr_value is not None:
1152                        array = np.array(attr_value)
1153                        # Apply data type optimization and compression
1154                        if array.dtype == np.int64:
1155                            array = array.astype(np.int32)
1156                        elif array.dtype == np.float64:
1157                            array = array.astype(np.float32)
1158                        elif array.dtype.str[0:2] == "<U":
1159                            # Convert Unicode strings to UTF-8 encoded strings
1160                            string_data = [str(item) for item in array]
1161                            string_dtype = h5py.string_dtype(encoding='utf-8')
1162                            eic_grp.create_dataset(str(attr_name), data=string_data, dtype=string_dtype, compression="gzip", compression_opts=9, chunks=True)
1163                            continue
1164                        eic_grp.create_dataset(str(attr_name), data=array, compression="gzip", compression_opts=9, chunks=True)
1165    
1166    def _save_mass_features_to_hdf5(self, hdf_handle, group_name = "mass_features", overwrite=False):
1167        """Save the mass features to the HDF5 file.
1168
1169        Parameters
1170        ----------
1171        hdf_handle : h5py.File
1172            The HDF5 file handle.
1173        group_name : str, optional
1174            The name of the group to save the mass features to. Default is 'mass_features'.
1175        overwrite : bool, optional
1176            Whether to overwrite the group if it exists. Default is False.
1177        """
1178        # Determine which mass features to save based on group_name
1179        if group_name == "induced_mass_features":
1180            if len(self.mass_spectra.induced_mass_features) == 0:
1181                return  # No induced mass features to save
1182            mass_features_dict = self.mass_spectra.induced_mass_features
1183        else:
1184            if len(self.mass_spectra.mass_features) == 0:
1185                return  # No mass features to save
1186            mass_features_dict = self.mass_spectra.mass_features
1187
1188        # Add LCMS mass features to hdf5 file
1189        if group_name not in hdf_handle:
1190            mass_features_group = hdf_handle.create_group(group_name)
1191        else:
1192            mass_features_group = hdf_handle.get(group_name)
1193        
1194        # Use the static helper method to save the mass features
1195        self._save_mass_features_dict_to_hdf5(mass_features_dict, mass_features_group, overwrite)
1196
1197    def to_hdf(self, overwrite=False, save_parameters=True, parameter_format="toml"):
1198        """Export the data to an HDF5.
1199
1200        Parameters
1201        ----------
1202        overwrite : bool, optional
1203            Whether to overwrite the output file. Default is False.
1204        save_parameters : bool, optional
1205            Whether to save the parameters as a separate json or toml file. Default is True.
1206        parameter_format : str, optional
1207            The format to save the parameters in. Default is 'toml'.
1208
1209        Raises
1210        ------
1211        ValueError
1212            If parameter_format is not 'json' or 'toml'.
1213        """
1214        export_profile_spectra = (
1215            self.mass_spectra.parameters.lc_ms.export_profile_spectra
1216        )
1217
1218        # Write the mass spectra data to the hdf5 file
1219        super().to_hdf(overwrite=overwrite, export_raw=export_profile_spectra)
1220
1221        # Write scan info, ms_unprocessed, mass features, eics, and ms2_search results to the hdf5 file
1222        with h5py.File(self.output_file.with_suffix(".hdf5"), "a") as hdf_handle:
1223            # Add scan_info to hdf5 file
1224            if "scan_info" not in hdf_handle or overwrite:
1225                if "scan_info" in hdf_handle and overwrite:
1226                    del hdf_handle["scan_info"]
1227                scan_info_group = hdf_handle.create_group("scan_info")
1228                for k, v in self.mass_spectra._scan_info.items():
1229                    array = np.array(list(v.values()))
1230                    if array.dtype.str[0:2] == "<U":
1231                        array = array.astype("S")
1232                    scan_info_group.create_dataset(k, data=array)
1233
1234            # Add ms_unprocessed to hdf5 file
1235            export_unprocessed_ms1 = (
1236                self.mass_spectra.parameters.lc_ms.export_unprocessed_ms1
1237            )
1238            if self.mass_spectra._ms_unprocessed and export_unprocessed_ms1:
1239                if "ms_unprocessed" not in hdf_handle or overwrite:
1240                    if "ms_unprocessed" in hdf_handle and overwrite:
1241                        del hdf_handle["ms_unprocessed"]
1242                    ms_unprocessed_group = hdf_handle.create_group("ms_unprocessed")
1243                else:
1244                    ms_unprocessed_group = hdf_handle.get("ms_unprocessed")
1245                for k, v in self.mass_spectra._ms_unprocessed.items():
1246                    if str(k) not in ms_unprocessed_group or overwrite:
1247                        if str(k) in ms_unprocessed_group and overwrite:
1248                            del ms_unprocessed_group[str(k)]
1249                        array = np.array(v)
1250                        ms_unprocessed_group.create_dataset(str(k), data=array)
1251
1252            # Add LCMS mass features to hdf5 file
1253            self._save_mass_features_to_hdf5(hdf_handle, group_name="mass_features", overwrite=overwrite)
1254            self._save_mass_features_to_hdf5(hdf_handle, group_name="induced_mass_features", overwrite=overwrite)
1255
1256            # Add EIC data to hdf5 file
1257            export_eics = self.mass_spectra.parameters.lc_ms.export_eics
1258            if len(self.mass_spectra.eics) > 0 and export_eics:
1259                if "eics" not in hdf_handle or overwrite:
1260                    if "eics" in hdf_handle and overwrite:
1261                        del hdf_handle["eics"]
1262                    eic_group = hdf_handle.create_group("eics")
1263                else:
1264                    eic_group = hdf_handle.get("eics")
1265
1266                # Use the static helper method to save the EICs
1267                self._save_eics_dict_to_hdf5(self.mass_spectra.eics, eic_group, overwrite)
1268
1269            # Add ms2_search results to hdf5 file (parameterized)
1270            if len(self.mass_spectra.spectral_search_results) > 0:
1271                if "spectral_search_results" not in hdf_handle or overwrite:
1272                    if "spectral_search_results" in hdf_handle and overwrite:
1273                        del hdf_handle["spectral_search_results"]
1274                    spectral_search_results = hdf_handle.create_group(
1275                        "spectral_search_results"
1276                    )
1277                else:
1278                    spectral_search_results = hdf_handle.get("spectral_search_results")
1279                # Create group for each search result by ms2_scan / precursor_mz
1280                for k, v in self.mass_spectra.spectral_search_results.items():
1281                    #TODO KRH: Fix to handle if export_only_relevant and k not in relevant_scan_numbers: continue!
1282                    if str(k) not in spectral_search_results or overwrite:
1283                        if str(k) in spectral_search_results and overwrite:
1284                            del spectral_search_results[str(k)]
1285                        spectral_search_results.create_group(str(k))
1286                        for k2, v2 in v.items():
1287                            spectral_search_results[str(k)].create_group(str(k2))
1288                            spectral_search_results[str(k)][str(k2)].attrs[
1289                                "precursor_mz"
1290                            ] = v2.precursor_mz
1291                            spectral_search_results[str(k)][str(k2)].attrs[
1292                                "query_spectrum_id"
1293                            ] = v2.query_spectrum_id
1294                            # Loop through each of the attributes and add them as datasets (if array)
1295                            for k3, v3 in v2.__dict__.items():
1296                                if v3 is not None and k3 not in [
1297                                    "query_spectrum",
1298                                    "precursor_mz",
1299                                    "query_spectrum_id",
1300                                ]:
1301                                    if k3 == "query_frag_types" or k3 == "ref_frag_types":
1302                                        v3 = [", ".join(x) for x in v3]
1303                                    if all(v3 is not None for v3 in v3):
1304                                        array = np.array(v3)
1305                                    if array.dtype.str[0:2] == "<U":
1306                                        array = array.astype("S")
1307                                    spectral_search_results[str(k)][str(k2)].create_dataset(
1308                                        str(k3), data=array
1309                                    )
1310
1311            # Save parameters as separate json
1312            if save_parameters:
1313                # Check if parameter_format is valid
1314                if parameter_format not in ["json", "toml"]:
1315                    raise ValueError("parameter_format must be 'json' or 'toml'")
1316
1317                if parameter_format == "json":
1318                    dump_lcms_settings_json(
1319                        filename=self.output_file.with_suffix(".json"),
1320                        lcms_obj=self.mass_spectra,
1321                    )
1322                elif parameter_format == "toml":
1323                    dump_lcms_settings_toml(
1324                        filename=self.output_file.with_suffix(".toml"),
1325                        lcms_obj=self.mass_spectra,
1326                    )
1327
1328class LCMSMetabolomicsExport(LCMSExport):
1329    """A class to export LCMS metabolite data.
1330
1331    This class provides methods to export LCMS metabolite data to various formats and summarize the metabolite report.
1332
1333    Parameters
1334    ----------
1335    out_file_path : str | Path
1336        The output file path, do not include the file extension.
1337    mass_spectra : object
1338        The high resolution mass spectra object.
1339    """
1340
1341    def __init__(self, out_file_path, mass_spectra):
1342        super().__init__(out_file_path, mass_spectra)
1343        self.ion_type_dict = ion_type_dict
1344    
1345    @staticmethod
1346    def get_ion_formula(neutral_formula, ion_type):
1347        """From a neutral formula and an ion type, return the formula of the ion.
1348
1349        Notes
1350        -----
1351        This is a static method.
1352        If the neutral_formula is not a string, this method will return None.
1353
1354        Parameters
1355        ----------
1356        neutral_formula : str
1357            The neutral formula, this should be a string form from the MolecularFormula class
1358            (e.g. 'C2 H4 O2', isotopes OK), or simple string (e.g. 'C2H4O2', no isotope handling in this case).
1359            In the case of a simple string, the atoms are parsed based on the presence of capital letters,
1360            e.g. MgCl2 is parsed as 'Mg Cl2.
1361        ion_type : str
1362            The ion type, e.g. 'protonated', '[M+H]+', '[M+Na]+', etc.
1363            See the self.ion_type_dict for the available ion types.
1364
1365        Returns
1366        -------
1367        str
1368            The formula of the ion as a string (like 'C2 H4 O2'); or None if the neutral_formula is not a string.
1369        """
1370        # If neutral_formula is not a string, return None
1371        if not isinstance(neutral_formula, str):
1372            return None
1373
1374        # Check if there are spaces in the formula (these are outputs of the MolecularFormula class and do not need to be processed before being passed to the class)
1375        if re.search(r"\s", neutral_formula):
1376            neutral_formula = MolecularFormula(neutral_formula, ion_charge=0)
1377        else:
1378            form_pre = re.sub(r"([A-Z])", r" \1", neutral_formula)[1:]
1379            elements = [re.findall(r"[A-Z][a-z]*", x) for x in form_pre.split()]
1380            counts = [re.findall(r"\d+", x) for x in form_pre.split()]
1381            neutral_formula = MolecularFormula(
1382                dict(
1383                    zip(
1384                        [x[0] for x in elements],
1385                        [int(x[0]) if x else 1 for x in counts],
1386                    )
1387                ),
1388                ion_charge=0,
1389            )
1390        neutral_formula_dict = neutral_formula.to_dict().copy()
1391
1392        adduct_add_dict = ion_type_dict[ion_type][0]
1393        for key in adduct_add_dict:
1394            if key in neutral_formula_dict.keys():
1395                neutral_formula_dict[key] += adduct_add_dict[key]
1396            else:
1397                neutral_formula_dict[key] = adduct_add_dict[key]
1398
1399        adduct_subtract = ion_type_dict[ion_type][1]
1400        for key in adduct_subtract:
1401            neutral_formula_dict[key] -= adduct_subtract[key]
1402
1403        return MolecularFormula(neutral_formula_dict, ion_charge=0).string
1404
1405    @staticmethod
1406    def get_isotope_type(ion_formula):
1407        """From an ion formula, return the 13C isotope type of the ion.
1408
1409        Notes
1410        -----
1411        This is a static method.
1412        If the ion_formula is not a string, this method will return None.
1413        This is currently only functional for 13C isotopes.
1414
1415        Parameters
1416        ----------
1417        ion_formula : str
1418            The formula of the ion, expected to be a string like 'C2 H4 O2'.
1419
1420        Returns
1421        -------
1422        str
1423            The isotope type of the ion, e.g. '13C1', '13C2', etc; or None if the ion_formula does not contain a 13C isotope.
1424
1425        Raises
1426        ------
1427        ValueError
1428            If the ion_formula is not a string.
1429        """
1430        if not isinstance(ion_formula, str):
1431            return None
1432
1433        if re.search(r"\s", ion_formula):
1434            ion_formula = MolecularFormula(ion_formula, ion_charge=0)
1435        else:
1436            raise ValueError('ion_formula should be a string like "C2 H4 O2"')
1437        ion_formula_dict = ion_formula.to_dict().copy()
1438
1439        try:
1440            iso_class = "13C" + str(ion_formula_dict.pop("13C"))
1441        except KeyError:
1442            iso_class = None
1443
1444        return iso_class
1445    
1446    def report_to_csv(self, molecular_metadata=None):
1447        """Create a report of the mass features and their annotations and save it as a CSV file.
1448
1449        Parameters
1450        ----------
1451        molecular_metadata : dict, optional
1452            The molecular metadata. Default is None.
1453        """
1454        report = self.to_report(molecular_metadata=molecular_metadata)
1455        out_file = self.output_file.with_suffix(".csv")
1456        report.to_csv(out_file, index=False)
1457    
1458    def clean_ms1_report(self, ms1_summary_full):
1459        """Clean the MS1 report.
1460
1461        Parameters
1462        ----------
1463        ms1_summary_full : DataFrame
1464            The full MS1 summary DataFrame.
1465
1466        Returns
1467        -------
1468        DataFrame
1469            The cleaned MS1 summary DataFrame.
1470        """
1471        ms1_summary_full = ms1_summary_full.reset_index()
1472        cols_to_keep = [
1473            "mf_id",
1474            "Molecular Formula",
1475            "Ion Type",
1476            "Calculated m/z",
1477            "m/z Error (ppm)",
1478            "m/z Error Score",
1479            "Is Isotopologue",
1480            "Isotopologue Similarity",
1481            "Confidence Score",
1482        ]
1483        ms1_summary = ms1_summary_full[cols_to_keep].copy()
1484        ms1_summary["ion_formula"] = [
1485            self.get_ion_formula(f, a)
1486            for f, a in zip(ms1_summary["Molecular Formula"], ms1_summary["Ion Type"])
1487        ]
1488        ms1_summary["isotopologue_type"] = [
1489            self.get_isotope_type(f) for f in ms1_summary["ion_formula"].tolist()
1490        ]
1491
1492        # Reorder columns
1493        ms1_summary = ms1_summary[
1494            [
1495                "mf_id",
1496                "ion_formula",
1497                "isotopologue_type",
1498                "Calculated m/z",
1499                "m/z Error (ppm)",
1500                "m/z Error Score",
1501                "Isotopologue Similarity",
1502                "Confidence Score",
1503            ]
1504        ]
1505
1506        # Set the index to mf_id
1507        ms1_summary = ms1_summary.set_index("mf_id")
1508
1509        return ms1_summary
1510    
1511    def summarize_ms2_report(self, ms2_annot_report):
1512        """
1513        Summarize the MS2 report.
1514
1515        Parameters
1516        ----------
1517        ms2_annot_report : DataFrame
1518            The MS2 annotation DataFrame with all annotations, output of mass_features_ms2_annot_to_df.
1519        
1520        Returns
1521        -------
1522        """
1523    
1524    def summarize_metabolomics_report(self, ms2_annot_report):
1525        """Summarize the MS2 hits for a metabolomics report
1526        
1527        Parameters
1528        ----------
1529        ms2_annot : DataFrame
1530            The MS2 annotation DataFrame with all annotations.
1531
1532        Returns
1533        -------
1534        DataFrame
1535            The summarized metabolomics report.
1536        """
1537        columns_to_drop = [
1538            "precursor_mz",
1539            "precursor_mz_error_ppm",
1540            "cas",
1541            "data_id",
1542            "iupac_name",
1543            "traditional_name",
1544            "common_name",
1545            "casno",
1546        ]
1547        ms2_annot = ms2_annot_report.drop(
1548            columns=[col for col in columns_to_drop if col in ms2_annot_report.columns]
1549        )
1550        
1551        # Prepare information about the search results, pulling out the best hit for the single report
1552        # Group by mf_id,ref_mol_id grab row with highest entropy similarity
1553        ms2_annot = ms2_annot.reset_index()
1554        # Add column called "n_spectra_contributing" that is the number of unique values in query_spectrum_id per mf_id,ref_mol_id
1555        ms2_annot["n_spectra_contributing"] = (
1556            ms2_annot.groupby(["mf_id", "ref_mol_id"])["query_spectrum_id"]
1557            .transform("nunique")
1558        )
1559        # Sort by entropy similarity
1560        ms2_annot = ms2_annot.sort_values(
1561            by=["mf_id", "ref_mol_id", "entropy_similarity"], ascending=[True, True, False]
1562        )
1563        best_entropy = ms2_annot.drop_duplicates(
1564            subset=["mf_id", "ref_mol_id"], keep="first"
1565        )
1566
1567        return best_entropy
1568
1569    def clean_ms2_report(self, metabolite_summary):
1570        """Clean the MS2 report.
1571
1572        Parameters
1573        ----------
1574        metabolite_summary : DataFrame
1575            The full metabolomics summary DataFrame.
1576
1577        Returns
1578        -------
1579        DataFrame
1580            The cleaned metabolomics summary DataFrame.
1581        """
1582        metabolite_summary = metabolite_summary.reset_index()
1583        metabolite_summary["ion_formula"] = [
1584            self.get_ion_formula(f, a)
1585            for f, a in zip(metabolite_summary["formula"], metabolite_summary["ref_ion_type"])
1586        ]
1587
1588        col_order = [
1589            "mf_id",
1590            "ion_formula",
1591            "ref_ion_type",
1592            "formula",
1593            "inchikey",
1594            "name",
1595            "inchi",
1596            "chebi",
1597            "smiles",
1598            "kegg",
1599            "cas",
1600            "database_name",
1601            "ref_ms_id",
1602            "entropy_similarity",
1603            "ref_mz_in_query_fract",
1604            "n_spectra_contributing",
1605        ]
1606
1607        # Reorder columns
1608        metabolite_summary = metabolite_summary[
1609            [col for col in col_order if col in metabolite_summary.columns]
1610        ]
1611
1612        # Convert chebi (if present) to int:
1613        if "chebi" in metabolite_summary.columns:
1614            metabolite_summary["chebi"] = metabolite_summary["chebi"].astype(
1615                "Int64", errors="ignore"
1616            )
1617
1618        # Set the index to mf_id
1619        metabolite_summary = metabolite_summary.set_index("mf_id")
1620
1621        return metabolite_summary
1622    
1623    def combine_reports(self, mf_report, ms1_annot_report, ms2_annot_report):
1624        """Combine the mass feature report with the MS1 and MS2 reports.
1625
1626        Parameters
1627        ----------
1628        mf_report : DataFrame
1629            The mass feature report DataFrame.
1630        ms1_annot_report : DataFrame
1631            The MS1 annotation report DataFrame.
1632        ms2_annot_report : DataFrame
1633            The MS2 annotation report DataFrame.
1634        """
1635        # If there is an ms1_annot_report, merge it with the mf_report
1636        if ms1_annot_report is not None and not ms1_annot_report.empty:
1637            # MS1 has been run and has molecular formula information
1638            mf_report = pd.merge(
1639                mf_report,
1640                ms1_annot_report,
1641                how="left",
1642                on=["mf_id", "isotopologue_type"],
1643            )
1644        if ms2_annot_report is not None:
1645            # If both reports contain 'ion_formula', prefer a merge that respects it.
1646            # Otherwise fall back to merging on 'mf_id' only to remain robust when
1647            # MS1 formula assignment wasn't performed or MS2 summary lacks the field.
1648            if "ion_formula" in mf_report.columns and "ion_formula" in ms2_annot_report.columns:
1649                # pull out the records without ion_formula and merge on mf_id only
1650                mf_no_ion_formula = mf_report[mf_report["ion_formula"].isna()]
1651                mf_no_ion_formula = mf_no_ion_formula.drop(columns=["ion_formula"]) if "ion_formula" in mf_no_ion_formula.columns else mf_no_ion_formula
1652                mf_no_ion_formula = pd.merge(
1653                    mf_no_ion_formula, ms2_annot_report, how="left", on=["mf_id"]
1654                )
1655
1656                # pull out the records with ion_formula and merge on mf_id + ion_formula
1657                mf_with_ion_formula = mf_report[~mf_report["ion_formula"].isna()]
1658                mf_with_ion_formula = pd.merge(
1659                    mf_with_ion_formula,
1660                    ms2_annot_report,
1661                    how="left",
1662                    on=["mf_id", "ion_formula"],
1663                )
1664
1665                # put back together
1666                mf_report = pd.concat([mf_no_ion_formula, mf_with_ion_formula])
1667            else:
1668                # Fall back to merging on mf_id only (robust when ion_formula missing)
1669                mf_report = pd.merge(
1670                    mf_report, ms2_annot_report, how="left", on=["mf_id"]
1671                )
1672
1673        # Rename colums
1674        rename_dict = {
1675            "mf_id": "Mass Feature ID",
1676            "scan_time": "Retention Time (min)",
1677            "mz": "m/z",
1678            "apex_scan": "Apex Scan Number",
1679            "intensity": "Intensity",
1680            "persistence": "Persistence",
1681            "area": "Area",
1682            "half_height_width": "Half Height Width (min)",
1683            "tailing_factor": "Tailing Factor",
1684            "dispersity_index": "Dispersity Index",
1685            "ms2_spectrum": "MS2 Spectrum",
1686            "monoisotopic_mf_id": "Monoisotopic Mass Feature ID",
1687            "isotopologue_type": "Isotopologue Type",
1688            "mass_spectrum_deconvoluted_parent": "Is Largest Ion after Deconvolution",
1689            "associated_mass_features": "Associated Mass Features after Deconvolution",
1690            "ion_formula": "Ion Formula",
1691            "formula": "Molecular Formula",
1692            "ref_ion_type": "Ion Type",
1693            "annot_level": "Lipid Annotation Level",
1694            "lipid_molecular_species_id": "Lipid Molecular Species",
1695            "lipid_summed_name": "Lipid Species",
1696            "lipid_subclass": "Lipid Subclass",
1697            "lipid_class": "Lipid Class",
1698            "lipid_category": "Lipid Category",
1699            "entropy_similarity": "Entropy Similarity",
1700            "ref_mz_in_query_fract": "Library mzs in Query (fraction)",
1701            "n_spectra_contributing": "Spectra with Annotation (n)",
1702        }
1703        mf_report = mf_report.rename(columns=rename_dict)
1704        mf_report["Sample Name"] = self.mass_spectra.sample_name
1705        mf_report["Polarity"] = self.mass_spectra.polarity
1706        mf_report = mf_report[
1707            ["Mass Feature ID", "Sample Name", "Polarity"]
1708            + [
1709                col
1710                for col in mf_report.columns
1711                if col not in ["Mass Feature ID", "Sample Name", "Polarity"]
1712            ]
1713        ]
1714
1715        # Reorder rows by "Mass Feature ID", then "Entropy Similarity" (descending), then "Confidence Score" (descending)
1716        if "Entropy Similarity" in mf_report.columns and "Confidence Score" in mf_report.columns:
1717            mf_report = mf_report.sort_values(
1718                by=["Mass Feature ID", "Entropy Similarity", "Confidence Score"],
1719                ascending=[True, False, False],
1720            )
1721        elif "Entropy Similarity" in mf_report.columns:
1722             mf_report = mf_report.sort_values(
1723                by=["Mass Feature ID", "Entropy Similarity"],
1724                ascending=[True, False],
1725            )
1726        elif "Confidence Score" in mf_report.columns:
1727             mf_report = mf_report.sort_values(
1728                by=["Mass Feature ID", "Confidence Score"],
1729                ascending=[True, False],
1730            )
1731        # If neither "Entropy Similarity" nor "Confidence Score" are in the columns, just sort by "Mass Feature ID"
1732        else:
1733            mf_report = mf_report.sort_values("Mass Feature ID")
1734
1735        # Reset index
1736        mf_report = mf_report.reset_index(drop=True)
1737
1738        return mf_report
1739    
1740    def to_report(self, molecular_metadata=None, suppress_warnings=False):
1741        """Create a report of the mass features and their annotations.
1742
1743        Parameters
1744        ----------
1745        molecular_metadata : dict, optional
1746            The molecular metadata. Default is None.
1747        suppress_warnings : bool, optional
1748            If True, suppresses warnings from mass_features_ms2_annot_to_df.
1749            Default is False.
1750
1751        Returns
1752        -------
1753        DataFrame
1754            The report as a Pandas DataFrame.
1755        """
1756        # Get mass feature dataframe
1757        mf_report = self.mass_spectra.mass_features_to_df()
1758        mf_report = mf_report.reset_index(drop=False)
1759
1760        # Get and clean ms1 annotation dataframe
1761        ms1_annot_report = self.mass_spectra.mass_features_ms1_annot_to_df(suppress_warnings=suppress_warnings)
1762        if ms1_annot_report is not None:
1763            ms1_annot_report = ms1_annot_report.copy()
1764            ms1_annot_report = self.clean_ms1_report(ms1_annot_report)
1765            ms1_annot_report = ms1_annot_report.reset_index(drop=False)
1766        else:
1767            ms1_annot_report = None
1768
1769        # Get, summarize, and clean ms2 annotation dataframe
1770        ms2_annot_report = self.mass_spectra.mass_features_ms2_annot_to_df(
1771            molecular_metadata=molecular_metadata,
1772            suppress_warnings=suppress_warnings
1773        )
1774        if ms2_annot_report is not None and molecular_metadata is not None:
1775            ms2_annot_report = self.summarize_metabolomics_report(ms2_annot_report)
1776            ms2_annot_report = self.clean_ms2_report(ms2_annot_report)
1777            ms2_annot_report = ms2_annot_report.dropna(axis=1, how="all")
1778            ms2_annot_report = ms2_annot_report.reset_index(drop=False)
1779        else:
1780            ms2_annot_report = None
1781
1782        report = self.combine_reports(
1783            mf_report=mf_report,
1784            ms1_annot_report=ms1_annot_report,
1785            ms2_annot_report=ms2_annot_report
1786        )
1787
1788        return report
1789class LipidomicsExport(LCMSMetabolomicsExport):
1790    """A class to export lipidomics data.
1791
1792    This class provides methods to export lipidomics data to various formats and summarize the lipid report.
1793
1794    Parameters
1795    ----------
1796    out_file_path : str | Path
1797        The output file path, do not include the file extension.
1798    mass_spectra : object
1799        The high resolution mass spectra object.
1800    """
1801
1802    def __init__(self, out_file_path, mass_spectra):
1803        super().__init__(out_file_path, mass_spectra)
1804
1805    def summarize_lipid_report(self, ms2_annot):
1806        """Summarize the lipid report.
1807
1808        Parameters
1809        ----------
1810        ms2_annot : DataFrame
1811            The MS2 annotation DataFrame with all annotations.
1812
1813        Returns
1814        -------
1815        DataFrame
1816            The summarized lipid report.
1817        """
1818        # Drop unnecessary columns for easier viewing
1819        columns_to_drop = [
1820            "precursor_mz",
1821            "precursor_mz_error_ppm",
1822            "ref_mol_id",
1823            "ref_precursor_mz",
1824            "cas",
1825            "inchikey",
1826            "inchi",
1827            "chebi",
1828            "smiles",
1829            "kegg",
1830            "data_id",
1831            "iupac_name",
1832            "traditional_name",
1833            "common_name",
1834            "casno",
1835        ]
1836        ms2_annot = ms2_annot.drop(
1837            columns=[col for col in columns_to_drop if col in ms2_annot.columns]
1838        )
1839
1840        # If ion_types_excluded is not empty, remove those ion types
1841        ion_types_excluded = self.mass_spectra.parameters.mass_spectrum[
1842            "ms2"
1843        ].molecular_search.ion_types_excluded
1844        if len(ion_types_excluded) > 0:
1845            ms2_annot = ms2_annot[~ms2_annot["ref_ion_type"].isin(ion_types_excluded)]
1846
1847        # If mf_id is not present, check that the index name is mf_id and reset the index
1848        if "mf_id" not in ms2_annot.columns:
1849            if ms2_annot.index.name == "mf_id":
1850                ms2_annot = ms2_annot.reset_index()
1851            else:
1852                raise ValueError("mf_id is not present in the dataframe")
1853
1854        # Attempt to get consensus annotations to the MLF level
1855        mlf_results_all = []
1856        for mf_id in ms2_annot["mf_id"].unique():
1857            mlf_results_perid = []
1858            ms2_annot_mf = ms2_annot[ms2_annot["mf_id"] == mf_id].copy()
1859            ms2_annot_mf["n_spectra_contributing"] = ms2_annot_mf.query_spectrum_id.nunique()
1860
1861            for query_scan in ms2_annot["query_spectrum_id"].unique():
1862                ms2_annot_sub = ms2_annot_mf[
1863                    ms2_annot_mf["query_spectrum_id"] == query_scan
1864                ].copy()
1865
1866                if ms2_annot_sub["lipid_summed_name"].nunique() == 1:
1867                    # If there is only one lipid_summed_name, let's try to get consensus molecular species annotation
1868                    if ms2_annot_sub["lipid_summed_name"].nunique() == 1:
1869                        ms2_annot_sub["entropy_max"] = (
1870                            ms2_annot_sub["entropy_similarity"]
1871                            == ms2_annot_sub["entropy_similarity"].max()
1872                        )
1873                        ms2_annot_sub["ref_match_fract_max"] = (
1874                            ms2_annot_sub["ref_mz_in_query_fract"]
1875                            == ms2_annot_sub["ref_mz_in_query_fract"].max()
1876                        )
1877                        ms2_annot_sub["frag_max"] = ms2_annot_sub[
1878                            "query_frag_types"
1879                        ].apply(lambda x: True if "MLF" in x else False)
1880
1881                        # New column that looks if there is a consensus between the ranks (one row that is highest in all ranks)
1882                        ms2_annot_sub["consensus"] = ms2_annot_sub[
1883                            ["entropy_max", "ref_match_fract_max", "frag_max"]
1884                        ].all(axis=1)
1885
1886                        # If there is a consensus, take the row with the highest entropy_similarity
1887                        if ms2_annot_sub["consensus"].any():
1888                            ms2_annot_sub = ms2_annot_sub[
1889                                ms2_annot_sub["entropy_similarity"]
1890                                == ms2_annot_sub["entropy_similarity"].max()
1891                            ].head(1)
1892                            mlf_results_perid.append(ms2_annot_sub)
1893            if len(mlf_results_perid) == 0:
1894                mlf_results_perid = pd.DataFrame()
1895            else:
1896                mlf_results_perid = pd.concat(mlf_results_perid)
1897                if mlf_results_perid["name"].nunique() == 1:
1898                    mlf_results_perid = mlf_results_perid[
1899                        mlf_results_perid["entropy_similarity"]
1900                        == mlf_results_perid["entropy_similarity"].max()
1901                    ].head(1)
1902                else:
1903                    mlf_results_perid = pd.DataFrame()
1904                mlf_results_all.append(mlf_results_perid)
1905
1906        # These are the consensus annotations to the MLF level
1907        if len(mlf_results_all) > 0:
1908            mlf_results_all = pd.concat(mlf_results_all)
1909            mlf_results_all["annot_level"] = mlf_results_all["structure_level"]
1910        else:
1911            # Make an empty dataframe
1912            mlf_results_all = ms2_annot.head(0)
1913
1914        # For remaining mf_ids, try to get a consensus annotation to the species level
1915        species_results_all = []
1916        # Remove mf_ids that have consensus annotations to the MLF level
1917        ms2_annot_spec = ms2_annot[
1918            ~ms2_annot["mf_id"].isin(mlf_results_all["mf_id"].unique())
1919        ]
1920        for mf_id in ms2_annot_spec["mf_id"].unique():
1921            # Do all the hits have the same lipid_summed_name?
1922            ms2_annot_sub = ms2_annot_spec[ms2_annot_spec["mf_id"] == mf_id].copy()
1923            ms2_annot_sub["n_spectra_contributing"] = len(ms2_annot_sub)
1924
1925            if ms2_annot_sub["lipid_summed_name"].nunique() == 1:
1926                # Grab the highest entropy_similarity result
1927                ms2_annot_sub = ms2_annot_sub[
1928                    ms2_annot_sub["entropy_similarity"]
1929                    == ms2_annot_sub["entropy_similarity"].max()
1930                ].head(1)
1931                species_results_all.append(ms2_annot_sub)
1932
1933        # These are the consensus annotations to the species level
1934        if len(species_results_all) > 0:
1935            species_results_all = pd.concat(species_results_all)
1936            species_results_all["annot_level"] = "species"
1937        else:
1938            # Make an empty dataframe
1939            species_results_all = ms2_annot.head(0)
1940
1941        # Deal with the remaining mf_ids that do not have consensus annotations to the species level or MLF level
1942        # Remove mf_ids that have consensus annotations to the species level
1943        ms2_annot_remaining = ms2_annot_spec[
1944            ~ms2_annot_spec["mf_id"].isin(species_results_all["mf_id"].unique())
1945        ]
1946        no_consensus = []
1947        for mf_id in ms2_annot_remaining["mf_id"].unique():
1948            id_sub = []
1949            id_no_con = []
1950            ms2_annot_sub_mf = ms2_annot_remaining[
1951                ms2_annot_remaining["mf_id"] == mf_id
1952            ].copy()
1953            for query_scan in ms2_annot_sub_mf["query_spectrum_id"].unique():
1954                ms2_annot_sub = ms2_annot_sub_mf[
1955                    ms2_annot_sub_mf["query_spectrum_id"] == query_scan
1956                ].copy()
1957
1958                # New columns for ranking [HIGHER RANK = BETTER]
1959                ms2_annot_sub["entropy_max"] = (
1960                    ms2_annot_sub["entropy_similarity"]
1961                    == ms2_annot_sub["entropy_similarity"].max()
1962                )
1963                ms2_annot_sub["ref_match_fract_max"] = (
1964                    ms2_annot_sub["ref_mz_in_query_fract"]
1965                    == ms2_annot_sub["ref_mz_in_query_fract"].max()
1966                )
1967                ms2_annot_sub["frag_max"] = ms2_annot_sub["query_frag_types"].apply(
1968                    lambda x: True if "MLF" in x else False
1969                )
1970
1971                # New column that looks if there is a consensus between the ranks (one row that is highest in all ranks)
1972                ms2_annot_sub["consensus"] = ms2_annot_sub[
1973                    ["entropy_max", "ref_match_fract_max", "frag_max"]
1974                ].all(axis=1)
1975                ms2_annot_sub_con = ms2_annot_sub[ms2_annot_sub["consensus"]]
1976                id_sub.append(ms2_annot_sub_con)
1977                id_no_con.append(ms2_annot_sub)
1978            id_sub = pd.concat(id_sub)
1979            id_no_con = pd.concat(id_no_con)
1980
1981            # Scenario 1: Multiple scans are being resolved to different MLFs [could be coelutions and should both be kept and annotated to MS level]
1982            if (
1983                id_sub["query_frag_types"]
1984                .apply(lambda x: True if "MLF" in x else False)
1985                .all()
1986                and len(id_sub) > 0
1987            ):
1988                idx = id_sub.groupby("name")["entropy_similarity"].idxmax()
1989                id_sub = id_sub.loc[idx]
1990                # Reorder so highest entropy_similarity is first
1991                id_sub = id_sub.sort_values("entropy_similarity", ascending=False)
1992                id_sub["annot_level"] = id_sub["structure_level"]
1993                no_consensus.append(id_sub)
1994
1995            # Scenario 2: Multiple scans are being resolved to different species, keep both and annotate to appropriate level
1996            elif len(id_sub) == 0:
1997                for lipid_summed_name in id_no_con["lipid_summed_name"].unique():
1998                    summed_sub = id_no_con[
1999                        id_no_con["lipid_summed_name"] == lipid_summed_name
2000                    ]
2001                    # Any consensus to MLF?
2002                    if summed_sub["consensus"].any():
2003                        summed_sub = summed_sub[summed_sub["consensus"]]
2004                        summed_sub["annot_level"] = summed_sub["structure_level"]
2005                        no_consensus.append(summed_sub)
2006                    else:
2007                        # Grab the highest entropy_similarity, if there are multiple, grab the first one
2008                        summed_sub = summed_sub[
2009                            summed_sub["entropy_similarity"]
2010                            == summed_sub["entropy_similarity"].max()
2011                        ].head(1)
2012                        # get first row
2013                        summed_sub["annot_level"] = "species"
2014                        summed_sub["name"] = ""
2015                        no_consensus.append(summed_sub)
2016            else:
2017                raise ValueError("Unexpected scenario for summarizing mf_id: ", mf_id)
2018
2019        if len(no_consensus) > 0:
2020            no_consensus = pd.concat(no_consensus)
2021        else:
2022            no_consensus = ms2_annot.head(0)
2023
2024        # Combine all the consensus annotations and reformat the dataframe for output
2025        species_results_all = species_results_all.drop(columns=["name"])
2026        species_results_all["lipid_molecular_species_id"] = ""
2027        mlf_results_all["lipid_molecular_species_id"] = mlf_results_all["name"]
2028        no_consensus["lipid_molecular_species_id"] = no_consensus["name"]
2029        consensus_annotations = pd.concat(
2030            [mlf_results_all, species_results_all, no_consensus]
2031        )
2032        consensus_annotations = consensus_annotations.sort_values(
2033            "mf_id", ascending=True
2034        )
2035        cols_to_keep = [
2036            "mf_id",
2037            "ref_ion_type",
2038            "entropy_similarity",
2039            "ref_mz_in_query_fract",
2040            "lipid_molecular_species_id",
2041            "lipid_summed_name",
2042            "lipid_subclass",
2043            "lipid_class",
2044            "lipid_category",
2045            "formula",
2046            "annot_level",
2047            "n_spectra_contributing",
2048        ]
2049        consensus_annotations = consensus_annotations[cols_to_keep]
2050        consensus_annotations = consensus_annotations.set_index("mf_id")
2051
2052        return consensus_annotations
2053
2054    def clean_ms2_report(self, lipid_summary):
2055        """Clean the MS2 report.
2056
2057        Parameters
2058        ----------
2059        lipid_summary : DataFrame
2060            The full lipid summary DataFrame.
2061
2062        Returns
2063        -------
2064        DataFrame
2065            The cleaned lipid summary DataFrame.
2066        """
2067        lipid_summary = lipid_summary.reset_index()
2068        lipid_summary["ion_formula"] = [
2069            self.get_ion_formula(f, a)
2070            for f, a in zip(lipid_summary["formula"], lipid_summary["ref_ion_type"])
2071        ]
2072
2073        # Reorder columns
2074        lipid_summary = lipid_summary[
2075            [
2076                "mf_id",
2077                "ion_formula",
2078                "ref_ion_type",
2079                "formula",
2080                "annot_level",
2081                "lipid_molecular_species_id",
2082                "lipid_summed_name",
2083                "lipid_subclass",
2084                "lipid_class",
2085                "lipid_category",
2086                "entropy_similarity",
2087                "ref_mz_in_query_fract",
2088                "n_spectra_contributing",
2089            ]
2090        ]
2091
2092        # Set the index to mf_id
2093        lipid_summary = lipid_summary.set_index("mf_id")
2094
2095        return lipid_summary
2096
2097    def to_report(self, molecular_metadata=None):
2098        """Create a report of the mass features and their annotations.
2099
2100        Parameters
2101        ----------
2102        molecular_metadata : dict, optional
2103            The molecular metadata. Default is None.
2104
2105        Returns
2106        -------
2107        DataFrame
2108            The report of the mass features and their annotations.
2109
2110        Notes
2111        -----
2112        The report will contain the mass features and their annotations from MS1 and MS2 (if available).
2113        """
2114        # Get mass feature dataframe
2115        mf_report = self.mass_spectra.mass_features_to_df()
2116        mf_report = mf_report.reset_index(drop=False)
2117
2118        # Get and clean ms1 annotation dataframe
2119        ms1_annot_report = self.mass_spectra.mass_features_ms1_annot_to_df().copy()
2120        ms1_annot_report = self.clean_ms1_report(ms1_annot_report)
2121        ms1_annot_report = ms1_annot_report.reset_index(drop=False)
2122
2123        # Get, summarize, and clean ms2 annotation dataframe
2124        ms2_annot_report = self.mass_spectra.mass_features_ms2_annot_to_df(
2125            molecular_metadata=molecular_metadata
2126        )
2127        if ms2_annot_report is not None and molecular_metadata is not None:
2128            ms2_annot_report = self.summarize_lipid_report(ms2_annot_report)
2129            ms2_annot_report = self.clean_ms2_report(ms2_annot_report)
2130            ms2_annot_report = ms2_annot_report.dropna(axis=1, how="all")
2131            ms2_annot_report = ms2_annot_report.reset_index(drop=False)
2132        report = self.combine_reports(
2133            mf_report=mf_report,
2134            ms1_annot_report=ms1_annot_report,
2135            ms2_annot_report=ms2_annot_report
2136        )
2137        return report
2138
2139        
2140class LCMSCollectionExport():
2141    """A class to export an LCMS collection to HDF5 format.
2142
2143    This class provides methods to export collection-level data from multi-sample LC-MS
2144    experiments to HDF5 files. It handles the export of metadata, retention time alignments,
2145    cluster assignments, and induced mass features (gap-filled features) across the collection.
2146
2147    The exporter is designed to work with LCMSCollection objects and complements the individual
2148    LCMSExport class by focusing on collection-wide data rather than individual sample data.
2149
2150    Parameters
2151    ----------
2152    out_file_path : str | Path
2153        The output file path, do not include the file extension. The .hdf5 extension
2154        will be added automatically.
2155    mass_spectra_collection : LCMSCollection
2156        The LCMS collection object containing multiple LCMS samples with processed mass features,
2157        alignments, and clustering information.
2158
2159    Attributes
2160    ----------
2161    out_file_path : Path
2162        The output file path as a Path object.
2163    mass_spectra_collection : LCMSCollection
2164        The LCMS collection object to be exported.
2165
2166    Methods
2167    -------
2168    export_to_hdf5(overwrite=False)
2169        Export the LCMS collection to an HDF5 file with collection-level data.
2170
2171    Notes
2172    -----
2173    This class exports collection-level data including:
2174    - Sample manifest (metadata about all samples in the collection)
2175    - Retention time alignment data (if RT alignment has been performed)
2176    - Cluster assignments (consensus mass feature groupings across samples)
2177    - Induced mass features (gap-filled features saved to individual LCMS object HDF5 files)
2178
2179    Individual sample data (mass spectra, mass features, EICs, etc.) should be exported
2180    separately using the LCMSExport class for each LCMS object in the collection.
2181
2182    Examples
2183    --------
2184    Export a collection after clustering and gap-filling:
2185
2186    >>> from corems.mass_spectra.output.export import LCMSCollectionExporter
2187    >>> exporter = LCMSCollectionExporter("my_collection", lcms_collection)
2188    >>> exporter.export_to_hdf5(overwrite=True)
2189
2190    The resulting HDF5 file will contain collection-level metadata and can be used
2191    to reconstruct the collection state for further analysis.
2192
2193    See Also
2194    --------
2195    LCMSExport : Export individual LCMS objects to HDF5
2196    LCMSCollection : The collection object being exported
2197    """
2198    def __init__(self, out_file_path, mass_spectra_collection):
2199        self.out_file_path = Path(out_file_path)
2200        self.mass_spectra_collection = mass_spectra_collection
2201
2202    def export_to_hdf5(
2203            self, 
2204            overwrite = False, 
2205            save_parameters=True, 
2206            parameter_format="toml",
2207            update_lcms_objects=True):
2208        """Export the LCMS collection to an HDF5 file.
2209        
2210        This method saves the collection-level data to an HDF5 file, including:
2211        - Basic metadata (date, folder location, gap-filling status)
2212        - Sample manifest
2213        - Retention time alignments (if available)
2214        - Cluster assignments (if available)
2215        - Induced mass features for each LCMS object (if gap-filling was performed)
2216        
2217        Individual LCMS objects in the collection are not exported by this method.
2218        Use LCMSExport for exporting individual LCMS objects.
2219        
2220        Parameters
2221        ----------
2222        overwrite : bool, optional
2223            If True, overwrites the output file if it already exists and replaces
2224            existing groups within the HDF5 file. If False, appends new data to
2225            existing file without overwriting existing groups. Default is False.
2226        save_parameters : bool, optional
2227            If True, saves the collection-level parameters to a separate file in the specified format.
2228            Default is True.
2229        parameter_format : str, optional
2230            The format for saving parameters, either "json" or "toml". Default is "toml".
2231        update_lcms_objects : bool, optional
2232            If True, updates the individual LCMS object HDF5 files with new raw file locations and any additional 
2233            information produced during the processing of the collection (e.g. cluster mass feature associations). Default is True.
2234            
2235        Notes
2236        -----
2237        The HDF5 file structure includes:
2238        - Attributes: date_utc, lcms_objects_folder, missing_mass_features_searched, manifest
2239        - Groups: rt_alignments, cluster_assignments (if available)
2240        
2241        Induced mass features are saved to the individual LCMS object HDF5 files
2242        within the .corems folder structure, not in the collection-level HDF5 file.
2243        
2244        Examples
2245        --------
2246        >>> exporter = LCMSCollectionExporter("my_collection", lcms_collection)
2247        >>> exporter.export_to_hdf5(overwrite=True)
2248        """
2249        if overwrite:
2250            if self.out_file_path.with_suffix(".hdf5").exists():
2251                self.out_file_path.with_suffix(".hdf5").unlink()
2252
2253        with h5py.File(self.out_file_path.with_suffix(".hdf5"), "a") as hdf_handle:
2254            # Add basic attributes to the HDF5 file, always overwrite these
2255            timenow = str(
2256                    datetime.now(timezone.utc).strftime("%d/%m/%Y %H:%M:%S %Z")
2257                )
2258            hdf_handle.attrs["date_utc"] = timenow
2259            hdf_handle.attrs["lcms_objects_folder"] = str(self.mass_spectra_collection.collection_parser.folder_location)
2260            hdf_handle.attrs["missing_mass_features_searched"] = self.mass_spectra_collection.missing_mass_features_searched
2261            hdf_handle.attrs["rt_aligned"] = self.mass_spectra_collection.rt_aligned
2262            hdf_handle.attrs["rt_alignment_attempted"] = self.mass_spectra_collection.rt_alignment_attempted
2263
2264            # Add the manifest to the HDF5 file, always overwrite this
2265            hdf_handle.attrs["manifest"] = self._convert_manifest_to_json()
2266
2267            # Save retention time alignments if they exist, only overwrite if specified
2268            self._save_rt_alignments_to_hdf5(hdf_handle, overwrite)
2269
2270            # Save cluster assignments if they exist, only overwrite if specified
2271            self._save_cluster_assignments_to_hdf5(hdf_handle, overwrite)
2272
2273        # Save new raw file locations to each LCMS object's HDF5 file if needed
2274        if hasattr(self.mass_spectra_collection, 'raw_files_relocated') and self.mass_spectra_collection.raw_files_relocated:
2275            self._update_raw_file_locations_in_hdf5()
2276
2277        # Save induced mass features to the collection with associations to each individual, only if lcms_collection.missing_mass_features_searched is True
2278        if self.mass_spectra_collection.missing_mass_features_searched:
2279            self._save_induced_mass_features_to_hdf5(overwrite)
2280            # Save EICs for induced mass features at collection level
2281            self._save_induced_eics_to_hdf5(overwrite)
2282        
2283        # Build cluster mass feature map to know which features to update
2284        # This uses the same logic as process_consensus_features to determine loaded features
2285        cluster_mf_map = self._build_cluster_mf_map()
2286        
2287        # Save updated mass features for each LCMS object
2288        # This implements selective update: only loaded features are updated, non-cluster features are preserved
2289        if update_lcms_objects:
2290            self._save_lcms_objects_to_hdf5(cluster_mf_map, overwrite)
2291
2292        # Save collection-level parameters as separate file
2293        if save_parameters:
2294            # Check if parameter_format is valid
2295            if parameter_format not in ["json", "toml"]:
2296                raise ValueError("parameter_format must be 'json' or 'toml'")
2297
2298            if parameter_format == "json":
2299                dump_lcms_collection_settings_json(
2300                    filename=self.out_file_path.with_suffix(".json"),
2301                    lcms_collection=self.mass_spectra_collection,
2302                )
2303            elif parameter_format == "toml":
2304                dump_lcms_collection_settings_toml(
2305                    filename=self.out_file_path.with_suffix(".toml"),
2306                    lcms_collection=self.mass_spectra_collection,
2307                )
2308
2309    def _save_rt_alignments_to_hdf5(self, hdf_handle, overwrite):
2310        """Save retention time alignments to HDF5 file."""
2311        # If no rt_alignments, return early
2312        if not self.mass_spectra_collection.rt_aligned:
2313            return
2314        
2315        # If rt_alignments exist, save them
2316        if self.mass_spectra_collection.rt_aligned:
2317            group_name = "rt_alignments"
2318            # grab dictionary of rt_alignments
2319            rt_alignments = self.mass_spectra_collection.rt_alignments
2320
2321        if rt_alignments:
2322            # Check if group exists and handle overwrite logic
2323            if group_name in hdf_handle:
2324                if not overwrite:
2325                    return
2326                del hdf_handle[group_name]
2327            
2328            grp = hdf_handle.create_group(group_name)
2329            
2330            # Save each alignment as a dataset
2331            for sample_idx, alignment_data in rt_alignments.items():
2332                grp.create_dataset(str(sample_idx), data=alignment_data)
2333    
2334    def _convert_manifest_to_json(self):
2335        """Clean the manifest for export to HDF5."""
2336        manifest = self.mass_spectra_collection.collection_parser.manifest
2337        
2338        # Process the manifest to convert numpy.bool_ or bool values for the 'use_rt_alignment' key
2339        def convert_bool_values(data):
2340            if isinstance(data, dict):
2341                # Process each key-value pair recursively
2342                return {k: (int(v) if k == 'use_rt_alignment' and isinstance(v, (bool, np.bool_)) else convert_bool_values(v)) for k, v in data.items()}
2343            elif isinstance(data, list):
2344                # Recursively process lists
2345                return [convert_bool_values(item) for item in data]
2346            else:
2347                # Return non-dict/list types unchanged
2348                return data
2349        
2350        # Clean the whole manifest
2351        cleaned_manifest = convert_bool_values(manifest)
2352
2353        # Serialize the cleaned manifest into JSON format
2354        json_manifest = json.dumps(cleaned_manifest)
2355        return json_manifest
2356    
2357    def _save_cluster_assignments_to_hdf5(self, hdf_handle, overwrite):
2358        """Save cluster assignments to HDF5 file."""
2359        # Check if column "cluster" is present in self.mass_features_dataframe
2360        if "cluster" in self.mass_spectra_collection.mass_features_dataframe.columns:
2361            group_name = "cluster_assignments"
2362            cluster_assignments = self.mass_spectra_collection.mass_features_dataframe[["cluster"]].copy()
2363
2364            # Check if group exists and handle overwrite logic
2365            if group_name in hdf_handle:
2366                if not overwrite:
2367                    return
2368                del hdf_handle[group_name]
2369            
2370            grp = hdf_handle.create_group(group_name)
2371
2372            # Save the index, converting strings to bytes
2373            grp.create_dataset("index", data=cluster_assignments.index.astype(str).values.astype('S'))
2374            
2375            # Save the "cluster" column
2376            grp.create_dataset("cluster", data=cluster_assignments["cluster"].values)
2377    
2378    def _build_cluster_mf_map(self):
2379        """Build a mapping of which mass features should be saved for each sample.
2380        
2381        This uses the same logic as process_consensus_features to determine which
2382        mass features were loaded and should be updated in HDF5 files.
2383        
2384        Returns
2385        -------
2386        dict
2387            Dictionary mapping sample_id to list of tuples (mf_id, cluster_id).
2388            Only includes samples that have loaded representative features.
2389            Returns empty dict if no clusters exist.
2390        
2391        Notes
2392        -----
2393        This follows the DRY principle by using the same get_sample_mf_map_for_representatives
2394        method used by process_consensus_features and ReadSavedLCMSCollection.
2395        """
2396        # Check if clusters exist
2397        if "cluster" not in self.mass_spectra_collection.mass_features_dataframe.columns:
2398            return {}
2399        
2400        # Check if cluster_summary_dataframe exists (needed by get_sample_mf_map_for_representatives)
2401        if not hasattr(self.mass_spectra_collection, 'cluster_summary_dataframe') or \
2402           self.mass_spectra_collection.cluster_summary_dataframe is None:
2403            return {}
2404        
2405        # Use the same DRY helper method that process_consensus_features uses
2406        # This ensures consistency across the codebase
2407        cluster_mf_map = self.mass_spectra_collection.get_sample_mf_map_for_representatives(
2408            include_cluster_id=True
2409        )
2410        
2411        return cluster_mf_map
2412    
2413    def _update_raw_file_locations_in_hdf5(self):
2414        """Update raw file locations in each LCMS object's HDF5 file.
2415        
2416        This method updates the 'original_file_location' attribute in each LCMS object's
2417        HDF5 file to reflect the new raw file location after files have been relocated.
2418        """
2419        for lcms_obj in self.mass_spectra_collection:
2420            # Get the HDF5 file path for this LCMS object
2421            hdf5_path = lcms_obj.file_location.with_suffix('.hdf5')
2422            
2423            if hdf5_path.exists():
2424                with h5py.File(hdf5_path, 'a') as hdf_handle:
2425                    # Update the original_file_location attribute
2426                    if 'original_file_location' in hdf_handle.attrs:
2427                        hdf_handle.attrs['original_file_location'] = str(lcms_obj.raw_file_location)
2428                    # If the attribute does not exist, create it
2429                    else:
2430                        hdf_handle.attrs.create('original_file_location', str(lcms_obj.raw_file_location))
2431    
2432    def _save_induced_mass_features_to_hdf5(self, overwrite):
2433        """Save induced mass features to the collection HDF5 file.
2434        
2435        Induced mass features are gap-filled features that only exist at the collection level.
2436        They are saved with full detail (all attributes and datasets) in the collection HDF5 file
2437        and distributed to individual LCMS objects when the collection is loaded.
2438        
2439        The induced mass features are stored in the collection's induced_mass_features_dataframe
2440        and are regenerated as LCMSMassFeature objects for saving.
2441        
2442        Parameters
2443        ----------
2444        overwrite : bool
2445            If True, overwrites existing induced mass features group. If False, skips if group exists.
2446        """
2447        # Check if we have any induced mass features to save
2448        if (self.mass_spectra_collection.induced_mass_features_dataframe is None or 
2449            self.mass_spectra_collection.induced_mass_features_dataframe.empty):
2450            return
2451        
2452        # Open the collection HDF5 file to save induced mass features
2453        with h5py.File(self.out_file_path.with_suffix(".hdf5"), "a") as hdf_handle:
2454            group_name = "induced_mass_features"
2455            
2456            # Check if group exists and handle overwrite logic
2457            if group_name in hdf_handle:
2458                if not overwrite:
2459                    return
2460                del hdf_handle[group_name]
2461            
2462            # Create top-level group for induced mass features
2463            imf_group = hdf_handle.create_group(group_name)
2464            
2465            # Get the induced mass features dataframe
2466            induced_df = self.mass_spectra_collection.induced_mass_features_dataframe
2467            
2468            # Get unique sample IDs from the dataframe
2469            sample_ids = induced_df['sample_id'].unique()
2470            
2471            # Iterate through each sample and save its induced mass features
2472            for sample_id in sample_ids:
2473                # Filter dataframe to this sample
2474                sample_df = induced_df[induced_df['sample_id'] == sample_id].copy()
2475                
2476                if sample_df.empty:
2477                    continue
2478                
2479                # Regenerate mass features from the dataframe
2480                regenerated_features = self._regenerate_mass_features_from_sample_df(
2481                    sample_df, sample_id
2482                )
2483                
2484                if not regenerated_features:
2485                    continue
2486                
2487                # Create a subgroup for this sample's induced mass features
2488                sample_group = imf_group.create_group(str(sample_id))
2489                
2490                # Use the static helper method from LCMSExport to save the mass features
2491                LCMSExport._save_mass_features_dict_to_hdf5(
2492                    regenerated_features, 
2493                    sample_group, 
2494                    overwrite=overwrite
2495                )
2496    
2497    def _save_induced_eics_to_hdf5(self, overwrite):
2498        """Save EICs for induced mass features to the collection HDF5 file.
2499        
2500        Induced mass features are gap-filled features created during process_consensus_features.
2501        Their associated EICs need to be saved at the collection level so they can be reloaded.
2502        
2503        The induced mass features are identified from the collection's induced_mass_features_dataframe,
2504        and their EICs are retrieved from the individual LCMS objects.
2505        
2506        Parameters
2507        ----------
2508        overwrite : bool
2509            If True, overwrites existing induced EICs group. If False, skips if group exists.
2510        """
2511        # Check if we have any induced mass features to save
2512        if (self.mass_spectra_collection.induced_mass_features_dataframe is None or 
2513            self.mass_spectra_collection.induced_mass_features_dataframe.empty):
2514            return
2515        
2516        # Open the collection HDF5 file to save induced EICs
2517        with h5py.File(self.out_file_path.with_suffix(".hdf5"), "a") as hdf_handle:
2518            group_name = "induced_eics"
2519            
2520            # Check if group exists and handle overwrite logic
2521            if group_name in hdf_handle:
2522                if not overwrite:
2523                    return
2524                del hdf_handle[group_name]
2525            
2526            # Create top-level group for induced EICs
2527            induced_eics_group = hdf_handle.create_group(group_name)
2528            
2529            # Get the induced mass features dataframe
2530            induced_df = self.mass_spectra_collection.induced_mass_features_dataframe
2531            
2532            # Get unique sample IDs from the dataframe
2533            sample_ids = induced_df['sample_id'].unique()
2534            
2535            # Iterate through each sample and save EICs for its induced mass features
2536            for sample_id in sample_ids:
2537                lcms_obj = self.mass_spectra_collection[sample_id]
2538                
2539                # Filter dataframe to this sample
2540                sample_df = induced_df[induced_df['sample_id'] == sample_id].copy()
2541                
2542                if sample_df.empty:
2543                    continue
2544                
2545                # Collect EICs for induced mass features using _eic_mz from dataframe
2546                induced_eics = {}
2547                for _, row in sample_df.iterrows():
2548                    # Get the EIC m/z from the dataframe
2549                    eic_mz = row.get('_eic_mz')
2550                    
2551                    if eic_mz is not None and pd.notna(eic_mz):
2552                        # Try to get the EIC from the LCMS object
2553                        if hasattr(lcms_obj, 'eics') and lcms_obj.eics and eic_mz in lcms_obj.eics:
2554                            induced_eics[eic_mz] = lcms_obj.eics[eic_mz]
2555                
2556                if not induced_eics:
2557                    continue
2558                
2559                # Create a subgroup for this sample's induced EICs
2560                sample_group = induced_eics_group.create_group(str(sample_id))
2561                
2562                # Use the static helper method from LCMSExport to save the EICs
2563                LCMSExport._save_eics_dict_to_hdf5(induced_eics, sample_group, overwrite)
2564    
2565    def _regenerate_mass_features_from_sample_df(self, sample_df, sample_id):
2566        """Regenerate induced mass features from a sample-specific dataframe.
2567        
2568        This method creates LCMSMassFeature objects from rows in the induced_mass_features_dataframe
2569        for a specific sample. The regenerated features are used for saving to HDF5.
2570        
2571        Parameters
2572        ----------
2573        sample_df : pd.DataFrame
2574            DataFrame containing induced mass features for a specific sample.
2575        sample_id : int
2576            The sample ID (index in the collection).
2577            
2578        Returns
2579        -------
2580        dict
2581            Dictionary of regenerated LCMSMassFeature objects keyed by feature ID.
2582        """
2583        from corems.chroma_peak.factory.chroma_peak_classes import LCMSMassFeature
2584        
2585        if sample_df.empty:
2586            return {}
2587        
2588        # Get the corresponding LCMS object for proper parent reference
2589        lcms_obj = self.mass_spectra_collection[sample_id]
2590        
2591        # Regenerate mass features from the dataframe
2592        regenerated_features = {}
2593        
2594        for _, row in sample_df.iterrows():
2595            # Extract the original ID from mf_id (format: c{cluster}_{index}_i)
2596            # This is the ID used in lcms_obj.induced_mass_features dict
2597            original_id = row['mf_id']
2598            
2599            # Create a new LCMSMassFeature with proper parent reference
2600            # Note: dataframe uses 'scan_time' but __init__ parameter is 'retention_time'
2601            mass_feature = LCMSMassFeature(
2602                lcms_parent=lcms_obj,
2603                mz=row['mz'],
2604                retention_time=row['scan_time'],  # Column is 'scan_time' in dataframe
2605                intensity=row['intensity'],
2606                apex_scan=int(row['apex_scan']),
2607                persistence=row.get('persistence', None) if 'persistence' in row else None,
2608                id=original_id  # Use the original string ID from gap-filling
2609            )
2610            
2611            # Set additional attributes dynamically from dataframe columns
2612            # Skip columns already handled in __init__ or structural metadata
2613            skip_cols = {
2614                'sample_id', 'mf_id', 'mz', 'scan_time', 'scan_time_aligned',
2615                'intensity', 'apex_scan', 'persistence'}
2616            
2617            # Iterate through all columns and set via property setters
2618            for col_name in row.index:
2619                if col_name in skip_cols:
2620                    continue
2621                value = row[col_name]
2622                try:
2623                    if pd.isna(value):
2624                        continue
2625                except (TypeError, ValueError):
2626                    pass  # value is array-like; not NA, proceed
2627                
2628                # Set via property (public interface handles private attributes)
2629                # Don't save empty lists
2630                if isinstance(value, list) and len(value) == 0:
2631                    continue
2632                try:
2633                    setattr(mass_feature, col_name, value)
2634                except (AttributeError, TypeError):
2635                    pass  # Skip attributes that don't exist or can't be set
2636            
2637            # Set cluster_index if present
2638            if 'cluster' in row and pd.notna(row['cluster']):
2639                mass_feature.cluster_index = int(row['cluster'])
2640            
2641            regenerated_features[mass_feature.id] = mass_feature
2642        
2643        return regenerated_features
2644    
2645    def _save_lcms_objects_to_hdf5(self, cluster_mf_map, overwrite):
2646        """Save updated mass features for each LCMS object.
2647        
2648        This method implements a "selective update" strategy for mass features:
2649        - For mass features specified in cluster_mf_map (loaded representatives), we selectively
2650          update them by deleting their old entries and re-saving with new attributes.
2651        - Non-cluster features (not loaded) are never touched/overwritten.
2652        
2653        Note: EICs are NOT saved here. Induced feature EICs are saved at the collection level.
2654        
2655        Parameters
2656        ----------
2657        cluster_mf_map : dict
2658            Dictionary mapping sample_id to list of tuples (mf_id, cluster_id).
2659            This explicitly defines which mass features should be updated.
2660        overwrite : bool
2661            If True, allows overwriting of existing data. If False, skips if data exists.
2662        """
2663        for sample_id, lcms_obj in enumerate(self.mass_spectra_collection):
2664            hdf5_path = lcms_obj.file_location.with_suffix('.hdf5')
2665            
2666            if not hdf5_path.exists():
2667                # If HDF5 doesn't exist, we can't do selective update, raise error
2668                raise FileNotFoundError(
2669                    f"HDF5 file for LCMS object {lcms_obj.sample_name} not found at {hdf5_path}"
2670                )
2671            
2672            # Check if this sample has any loaded features in the map
2673            if sample_id not in cluster_mf_map or not cluster_mf_map[sample_id]:
2674                # Nothing loaded for this sample, nothing to update
2675                continue
2676            
2677            # Extract mf_ids from the map (cluster_mf_map contains tuples of (mf_id, cluster_id))
2678            mf_ids_to_update = [mf_id for mf_id, cluster_id in cluster_mf_map[sample_id]]
2679            
2680            # Perform selective update of mass features
2681            self._selective_update_mass_features(lcms_obj, hdf5_path, mf_ids_to_update, overwrite)
2682            
2683            # Save any new mass spectra that were added during processing
2684            self._save_new_mass_spectra(lcms_obj, hdf5_path, overwrite)
2685    
2686    def _save_new_mass_spectra(self, lcms_obj, hdf5_path, overwrite):
2687        """Save new mass spectra that were added during processing.
2688        
2689        This method checks what mass spectra are in lcms_obj._ms and saves any
2690        that aren't already in the HDF5 file's mass_spectra group. Uses the
2691        existing add_mass_spectrum_to_hdf5 method for consistency with original
2692        export logic.
2693        
2694        Parameters
2695        ----------
2696        lcms_obj : LCMSBase
2697            The LCMS object with potentially new mass spectra.
2698        hdf5_path : Path
2699            Path to the HDF5 file.
2700        overwrite : bool
2701            If True, allows overwriting existing spectra.
2702        """
2703        # Check if there are any mass spectra to save
2704        if not hasattr(lcms_obj, '_ms') or not lcms_obj._ms:
2705            return
2706        
2707        # Create an LCMS exporter instance for this LCMS object
2708        # This gives us access to add_mass_spectrum_to_hdf5 method inherited from HighResMassSpecExport
2709        # Turn hdf5_path into str without suffix for LCMSExport
2710        hdf5_path_str = str(hdf5_path.with_suffix(''))
2711        exporter = LCMSExport(
2712            out_file_path=hdf5_path_str,
2713            mass_spectra=lcms_obj
2714        )
2715        
2716        # Open HDF5 file and check existing mass spectra
2717        with h5py.File(hdf5_path, 'a') as hdf_handle:
2718            # Create mass_spectra group if it doesn't exist
2719            if 'mass_spectra' not in hdf_handle:
2720                ms_group = hdf_handle.create_group('mass_spectra')
2721                existing_scan_numbers = set()
2722            else:
2723                ms_group = hdf_handle['mass_spectra']
2724                existing_scan_numbers = set(int(k) for k in ms_group.keys())
2725            
2726            # Find new mass spectra (in _ms but not in HDF5)
2727            new_scan_numbers = set(lcms_obj._ms.keys()) - existing_scan_numbers
2728                        
2729            if not new_scan_numbers:
2730                return
2731            
2732            # Save new mass spectra using existing add_mass_spectrum_to_hdf5 method
2733            export_profile = lcms_obj.parameters.lc_ms.export_profile_spectra
2734            for scan_number in new_scan_numbers:
2735                mass_spec = lcms_obj._ms[scan_number]
2736                scan_group_name = str(scan_number)
2737                
2738                # Delete existing group if overwrite is True
2739                if scan_group_name in ms_group and overwrite:
2740                    del ms_group[scan_group_name]
2741                elif scan_group_name in ms_group:
2742                    continue
2743                
2744                # Use the existing method from HighResMassSpecExport
2745                exporter.add_mass_spectrum_to_hdf5(
2746                    hdf_handle=hdf_handle,
2747                    mass_spectrum=mass_spec,
2748                    group_key=scan_group_name,
2749                    mass_spectra_group=ms_group,
2750                    export_raw=export_profile
2751                )
2752    
2753    def _selective_update_mass_features(self, lcms_obj, hdf5_path, mf_ids_to_update, overwrite):
2754        """Selectively update mass features in HDF5 file.
2755        
2756        This method deletes only the mass features specified in mf_ids_to_update,
2757        then re-saves them with their potentially updated attributes. Non-cluster features
2758        in the HDF5 file are left untouched.
2759        
2760        Parameters
2761        ----------
2762        lcms_obj : LCMSBase
2763            The LCMS object with mass features to update.
2764        hdf5_path : Path
2765            Path to the HDF5 file.
2766        mf_ids_to_update : list of int
2767            List of mass feature IDs that should be updated. This explicitly defines
2768            which features were loaded and should be saved.
2769        overwrite : bool
2770            If True, allows overwriting. If False, skips if group exists.
2771        """
2772        if not mf_ids_to_update:
2773            return
2774        
2775        # Open HDF5 file and delete specified feature IDs, then re-save
2776        with h5py.File(hdf5_path, 'a') as hdf_handle:
2777            if 'mass_features' not in hdf_handle:
2778                return
2779            
2780            mf_group = hdf_handle['mass_features']
2781            
2782            # Delete features that are being updated
2783            for feature_id in mf_ids_to_update:
2784                feature_id_str = str(feature_id)
2785                if feature_id_str in mf_group:
2786                    del mf_group[feature_id_str]
2787            
2788            # Re-save updated features (only those that exist in mass_features dict)
2789            updated_features = {
2790                mf.id: mf for mf in lcms_obj.mass_features.values()
2791                if mf.id in mf_ids_to_update
2792            }
2793            
2794            if updated_features:
2795                LCMSExport._save_mass_features_dict_to_hdf5(
2796                    updated_features,
2797                    mf_group,
2798                    overwrite=overwrite
2799                )
2800