magma runner

magma/README.md

@@ -0,0 +1,21 @@
+# MAGMa
+
+This code directory is adapted from https://github.com/samgoldman97/mist/tree/nmi_paper_v1/src/mist/magma
+
+
+MAGMa is an algorithm which takes as input a molecule and provides as output a list of fragment molecules of the parent.
+
+In this project, MAGMa is used to label the fragment peaks of spectra datasets
+with chemical formulae and corresponding smiles, to be used as an extra
+training signal for models. The fragmentation code utilized is heavily inspired
+by the [original source code](https://github.com/NLeSC/MAGMa).
+
+`run_magma.py` can be run directly and requires the following arguments:
+
+- **--spectra-dir**: The directory path containing the SIRIUS program outputs.
+  To subset spectra, we use only peaks that have been preserved by SIRIUS as
+  an initial cleaning step. The program can be adapted to use other spectra
+  input sources.   
+- **--output-dir**: The chosen output directory path to save the magma output files    
+- **--lowest-penalty-filter**: If flag set, when selecting candidate chemical formulae and smiles to label spectra peaks, only candidates with the lowest penalty score (as assigned by the Magma fragmentation engine) will be selected    
+- **--spec-labels**: TSV file containing all the smiles for the spectra being used.   

magma/fragmentation.py

@@ -0,0 +1,376 @@
+"""fragmentation.py
+
+Code snippets taken from the MAGMa github project
+
+https://github.com/NLeSC/MAGMa
+
+"""
+
+
+import numpy
+from rdkit import Chem
+
+typew = {
+    Chem.rdchem.BondType.names["AROMATIC"]: 3.0,
+    Chem.rdchem.BondType.names["DOUBLE"]: 2.0,
+    Chem.rdchem.BondType.names["TRIPLE"]: 3.0,
+    Chem.rdchem.BondType.names["SINGLE"]: 1.0,
+}
+heterow = {False: 2, True: 1}
+missingfragmentpenalty = 10
+
+
+mims = {
+    "H": 1.0078250321,
+    "C": 12.0000000,
+    "N": 14.0030740052,
+    "O": 15.9949146221,
+    "F": 18.99840320,
+    "Na": 22.9897692809,
+    "P": 30.97376151,
+    "S": 31.97207069,
+    "Cl": 34.96885271,
+    "K": 38.96370668,
+    "Br": 78.9183376,
+    "I": 126.904468,
+    "Si": 28.0855,
+    "B": 10.811,
+    "Se": 78.97,
+    "Fe": 55.845,
+    "Co": 58.933,
+    "As": 74.9216
+}
+
+# Mass of hydrogen atom
+Hmass = mims["H"]
+elmass = 0.0005486
+
+ionmasses = {
+    1: {
+        "+H": mims["H"],
+        "+NH4": mims["N"] + 4 * mims["H"],
+        "+Na": mims["Na"],
+        "+K": mims["K"],
+    },
+    -1: {"-H": -mims["H"], "+Cl": mims["Cl"]},
+}
+
+
+class FragmentEngine(object):
+    def __init__(
+        self,
+        smiles,
+        max_broken_bonds,
+        max_water_losses,
+        ionisation_mode,
+        skip_fragmentation,
+        molcharge,
+    ):
+        try:
+            # self.mol = Chem.MolFromMolBlock(str(mol))
+            # self.mol = Chem.MolFromSmiles(smiles)
+            self.mol = Chem.MolFromSmiles(smiles)
+            self.accept = True
+            self.natoms = self.mol.GetNumAtoms()
+        except:
+            self.accept = False
+            return
+        self.max_broken_bonds = max_broken_bonds
+        self.max_water_losses = max_water_losses
+        self.ionisation_mode = ionisation_mode
+        self.skip_fragmentation = skip_fragmentation
+        self.molcharge = molcharge
+        self.atom_masses = []
+        self.atomHs = []
+        self.neutral_loss_atoms = []
+        self.bonded_atoms = []  # [[list of atom numbers]]
+        self.bonds = set([])
+        self.bondscore = {}
+        self.new_fragment = 0
+        self.template_fragment = 0
+        self.fragment_masses = ((max_broken_bonds + max_water_losses) * 2 + 1) * [0]
+        self.fragment_info = [[0, 0, 0]]
+        self.avg_score = None
+
+        for x in range(self.natoms):
+            self.bonded_atoms.append([])
+            atom = self.mol.GetAtomWithIdx(x)
+            self.atomHs.append(atom.GetNumImplicitHs() + atom.GetNumExplicitHs())
+            self.atom_masses.append(mims[atom.GetSymbol()] + Hmass * (self.atomHs[x]))
+            if (
+                atom.GetSymbol() == "O"
+                and self.atomHs[x] == 1
+                and len(atom.GetBonds()) == 1
+            ):
+                self.neutral_loss_atoms.append(x)
+            if (
+                atom.GetSymbol() == "N"
+                and self.atomHs[x] == 2
+                and len(atom.GetBonds()) == 1
+            ):
+                self.neutral_loss_atoms.append(x)
+        for bond in self.mol.GetBonds():
+            a1, a2 = bond.GetBeginAtomIdx(), bond.GetEndAtomIdx()
+            self.bonded_atoms[a1].append(a2)
+            self.bonded_atoms[a2].append(a1)
+            bondbits = 1 << a1 | 1 << a2
+            bondscore = (
+                typew[bond.GetBondType()]
+                * heterow[
+                    bond.GetBeginAtom().GetSymbol() != "C"
+                    or bond.GetEndAtom().GetSymbol() != "C"
+                ]
+            )
+            self.bonds.add(bondbits)
+            self.bondscore[bondbits] = bondscore
+
+    def extend(self, atom):
+        for a in self.bonded_atoms[atom]:
+            atombit = 1 << a
+            if atombit & self.template_fragment and not atombit & self.new_fragment:
+                self.new_fragment = self.new_fragment | atombit
+                self.extend(a)
+
+    def generate_fragments(self):
+        frag = (1 << self.natoms) - 1
+        all_fragments = set([frag])
+        total_fragments = set([frag])
+        current_fragments = set([frag])
+        new_fragments = set([frag])
+        self.add_fragment(frag, self.calc_fragment_mass(frag), 0, 0)
+
+        if self.skip_fragmentation:
+            self.convert_fragments_table()
+            return len(self.fragment_info)
+
+        # generate fragments for max_broken_bond steps
+        for step in range(self.max_broken_bonds):
+            # loop over all fragments to be fragmented
+            for fragment in current_fragments:
+                # loop over all atoms
+                for atom in range(self.natoms):
+                    # in the fragment
+                    if (1 << atom) & fragment:
+                        # remove the atom
+                        self.template_fragment = fragment ^ (1 << atom)
+                        list_ext_atoms = set([])
+                        extended_fragments = set([])
+                        # find all its neighbor atoms
+                        for a in self.bonded_atoms[atom]:
+                            # present in the fragment
+                            if (1 << a) & self.template_fragment:
+                                list_ext_atoms.add(a)
+                        # in case of one bonded atom, the new fragment is the remainder of the old fragment
+                        if len(list_ext_atoms) == 1:
+                            extended_fragments.add(self.template_fragment)
+                        else:
+                            # otherwise extend each neighbor atom to a complete fragment
+                            for a in list_ext_atoms:
+                                # except when deleted atom is in a ring and a previous extended
+                                # fragment already contains this neighbor atom, then
+                                # calculate fragment only once
+                                for frag in extended_fragments:
+                                    if (1 << a) & frag:
+                                        break
+                                else:
+                                    # extend atom to complete fragment
+                                    self.new_fragment = 1 << a
+                                    self.extend(a)
+                                    extended_fragments.add(self.new_fragment)
+                        for frag in extended_fragments:
+                            # add extended fragments, if not yet present, to the collection
+                            if frag not in all_fragments:
+                                all_fragments.add(frag)
+                                bondbreaks, score = self.score_fragment(frag)
+                                if bondbreaks <= self.max_broken_bonds and score < (
+                                    missingfragmentpenalty + 5
+                                ):
+                                    new_fragments.add(frag)
+                                    total_fragments.add(frag)
+                                    self.add_fragment(
+                                        frag,
+                                        self.calc_fragment_mass(frag),
+                                        score,
+                                        bondbreaks,
+                                    )
+            current_fragments = new_fragments
+            new_fragments = set([])
+        # number of OH losses
+        for step in range(self.max_water_losses):
+            # loop of all fragments
+            for fi in self.fragment_info:
+                # on which to apply neutral loss rules
+                if fi[2] == self.max_broken_bonds + step:
+                    fragment = fi[0]
+                    # loop over all atoms in the fragment
+                    for atom in self.neutral_loss_atoms:
+                        if (1 << atom) & fragment:
+                            frag = fragment ^ (1 << atom)
+                            # add extended fragments, if not yet present, to the collection
+                            if frag not in total_fragments:
+                                total_fragments.add(frag)
+                                bondbreaks, score = self.score_fragment(frag)
+                                if score < (missingfragmentpenalty + 5):
+                                    self.add_fragment(
+                                        frag,
+                                        self.calc_fragment_mass(frag),
+                                        score,
+                                        bondbreaks,
+                                    )
+        self.convert_fragments_table()
+        return len(self.fragment_info)
+
+    def score_fragment(self, fragment):
+        score = 0
+        bondbreaks = 0
+        for bond in self.bonds:
+            if 0 < (fragment & bond) < bond:
+                score += self.bondscore[bond]
+                bondbreaks += 1
+        if score == 0:
+            print("score=0: ", fragment, bondbreaks)
+        return bondbreaks, score
+
+    def score_fragment_rel2parent(self, fragment, parent):
+        score = 0
+        for bond in self.bonds:
+            if 0 < (fragment & bond) < (bond & parent):
+                score += self.bondscore[bond]
+        return score
+
+    def calc_fragment_mass(self, fragment):
+        fragment_mass = 0.0
+        for atom in range(self.natoms):
+            if fragment & (1 << atom):
+                fragment_mass += self.atom_masses[atom]
+        return fragment_mass
+
+    def add_fragment(self, fragment, fragmentmass, score, bondbreaks):
+        mass_range = (
+            (self.max_broken_bonds + self.max_water_losses - bondbreaks) * [0]
+            + list(
+                numpy.arange(
+                    -bondbreaks + self.ionisation_mode * (1 - self.molcharge),
+                    bondbreaks + self.ionisation_mode * (1 - self.molcharge) + 1,
+                )
+                * Hmass
+                + fragmentmass
+            )
+            + (self.max_broken_bonds + self.max_water_losses - bondbreaks) * [0]
+        )
+        if bondbreaks == 0:
+            # make sure that fragmentmass is included
+            mass_range[
+                self.max_broken_bonds + self.max_water_losses - self.ionisation_mode
+            ] = fragmentmass
+        self.fragment_masses += mass_range
+        self.fragment_info.append([fragment, score, bondbreaks])
+
+    def convert_fragments_table(self):
+        self.fragment_masses_np = numpy.array(self.fragment_masses).reshape(
+            len(self.fragment_info),
+            (self.max_broken_bonds + self.max_water_losses) * 2 + 1,
+        )
+
+    def calc_avg_score(self):
+        self.avg_score = numpy.average(self.scores)
+
+    def get_avg_score(self):
+        return self.avg_score
+
+    def find_fragments(self, mass, parent, precision, mz_precision_abs):
+        result = numpy.where(
+            numpy.where(
+                self.fragment_masses_np
+                < max(mass * precision, mass + mz_precision_abs),
+                self.fragment_masses_np,
+                0,
+            )
+            > min(mass / precision, mass - mz_precision_abs)
+        )
+        fragment_set = []
+        for i in range(len(result[0])):
+            fid = result[0][i]
+            fragment_set.append(
+                self.fragment_info[fid]
+                + [
+                    self.fragment_masses_np[fid][
+                        self.max_broken_bonds
+                        + self.max_water_losses
+                        - self.ionisation_mode * (1 - self.molcharge)
+                    ]
+                ]
+                + [
+                    self.ionisation_mode * (1 - self.molcharge)
+                    + result[1][i]
+                    - self.max_broken_bonds
+                    - self.max_water_losses
+                ]
+            )
+        return fragment_set
+
+    def get_fragment_info(self, fragment, deltaH):
+        atomlist = []
+        elements = {
+            "C": 0,
+            "H": 0,
+            "N": 0,
+            "O": 0,
+            "F": 0,
+            "P": 0,
+            "S": 0,
+            "Cl": 0,
+            "Br": 0,
+            "I": 0,
+            "Si": 0,
+            "B": 0,
+            "Se": 0,
+            "Fe": 0,
+            "Co": 0,
+            "As": 0
+        }
+        for atom in range(self.natoms):
+            if (1 << atom) & fragment:
+                atomlist.append(atom)
+                elements[self.mol.GetAtomWithIdx(atom).GetSymbol()] += 1
+                elements["H"] += self.atomHs[atom]
+        formula = ""
+        for el in (
+            "C",
+            "H",
+            "N",
+            "O",
+            "F",
+            "P",
+            "S",
+            "Cl",
+            "Br",
+            "I",
+            "Si",
+            "B",
+            "Se",
+            "Fe",
+            "Co",
+        ):
+            nel = elements[el]
+            if nel > 0:
+                formula += el
+            if nel > 1:
+                formula += str(nel)
+        atomstring = ",".join(str(a) for a in atomlist)
+        return atomstring, atomlist, formula, fragment2smiles(self.mol, atomlist)
+
+    def get_natoms(self):
+        return self.natoms
+
+    def accepted(self):
+        return self.accept
+
+
+def fragment2smiles(mol, atomlist):
+    emol = Chem.EditableMol(mol)
+    for atom in reversed(range(mol.GetNumAtoms())):
+        if atom not in atomlist:
+            emol.RemoveAtom(atom)
+    frag = emol.GetMol()
+    return Chem.MolToSmiles(frag)

magma/magma_utils.py

@@ -0,0 +1,87 @@
+""" magma_utils.py
+
+Additional utility file to assist with fingerprinting.
+
+"""
+
+import os
+from ast import literal_eval
+
+import numpy as np
+import pandas as pd
+
+from rdkit import Chem
+from rdkit.Chem import AllChem, DataStructs
+import re
+
+def extract_adduct_ion(adduct, default='+H'):
+    pattern = re.compile(r"\[M([+-][^\]]+)\]")
+    match = pattern.search(adduct)
+    if match:
+        return match.group(1)
+    return default
+
+
+def get_magma_fingerprint(smile):
+    """ get_magma_fingerprint. """
+    mol = Chem.MolFromSmiles(smile, sanitize=False)
+    Chem.SanitizeMol(
+        mol,
+        sanitizeOps=Chem.SanitizeFlags.SANITIZE_ALL
+        ^ Chem.SanitizeFlags.SANITIZE_KEKULIZE,
+    )
+    curr_fp = AllChem.GetMorganFingerprintAsBitVect(mol, 2, nBits=2048)
+
+    fingerprint = np.zeros((0,), dtype=np.uint8)
+    DataStructs.ConvertToNumpyArray(curr_fp, fingerprint)
+    return fingerprint
+
+
+def get_magma_fingerprint_bits(smile):
+    """ get magma fingerprint bits """
+    fingerprint = get_magma_fingerprint(smile)
+    hot_indices = list(np.where(np.array(list(fingerprint)) == 1)[0])
+    return hot_indices
+
+
+def read_magma_file(magma_frag_file):
+    """Read in magma file"""
+    if (
+        magma_frag_file is not None
+        and os.path.exists(magma_frag_file)
+        and os.path.getsize(magma_frag_file) > 0
+    ):
+
+        # correct for inconsistency by me in file parsing (sad)
+        sep = "\t"
+        spectra_df = pd.read_csv(magma_frag_file, index_col=0, sep=sep)
+        if (
+            "smiles" not in spectra_df.columns
+            or "chemical_formula" not in spectra_df.columns
+        ):
+            pass
+        else:
+            spectra_df = _convert_str_to_list(spectra_df, "smiles")
+            spectra_df = _convert_str_to_list(spectra_df, "chemical_formula")
+            if "mass_to_charge" not in spectra_df.columns:
+                spectra_df["mass_to_charge"] = spectra_df["mz"]
+
+            return spectra_df
+    spectra_df = pd.DataFrame(
+        columns=[
+            "mass_to_charge",
+            "intensity",
+            "chemical_formula",
+            "smiles",
+            "molecule_peak",
+        ]
+    )
+    return spectra_df
+
+
+def _convert_str_to_list(df, column):
+    """_convert_str_to_list"""
+    df.loc[:, column] = df.loc[:, column].apply(
+        lambda x: literal_eval(x) if x != "NAN" and not pd.isna(x) else []
+    )
+    return df

magma/run_magma.py

@@ -0,0 +1,275 @@
+""" run_magma.py
+
+Accept input processed spectra and make subformula peak assignments
+accordingly.
+
+"""
+import logging
+from pathlib import Path
+import numpy as np
+import pandas as pd
+import argparse
+import sys
+from multiprocessing import Pool
+from tqdm import tqdm
+from collections import defaultdict
+import json
+
+# add parent path
+import os
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+# Custom import
+from magma.fragmentation import FragmentEngine, ionmasses
+from magma import magma_utils
+from magma.fragmentation import ionmasses
+
+# Define basic logger
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s: %(message)s",
+    handlers=[
+        logging.StreamHandler(sys.stdout),
+    ],
+)
+
+
+FRAGMENT_ENGINE_PARAMS = {
+    "max_broken_bonds": 3,
+    "max_water_losses": 1,
+    "ionisation_mode": 1,
+    "skip_fragmentation": 0,
+    "molcharge": 0,
+}
+
+PEAK_ASSIGNMENT_PARAMS = {
+    'lowest_penalty_filter': True,
+    'tolerance': 1
+}
+
+def get_args():
+    """get args"""
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        '--data_pth',
+        required=True
+    )
+    parser.add_argument(
+        "--output_dir",
+        required=True,
+        help="Output directory to save MAGMA files",
+    )
+    parser.add_argument(
+        "--workers", default=30, action="store", type=int, help="Num workers"
+    )
+    return parser.parse_args()
+
+
+
+def get_matching_fragment(
+    fragment_df, mass_comparison_vector, lowest_penalty_filter: bool
+):
+    """get_matching_fragment.
+
+    Compare frag
+
+    Args:
+        fragment_df
+        mass_comparison_vec
+        lowest_penalty_filter
+    """
+    # Step 1 - Determine and filter for fragments whose mass range cover the peak mass
+    matched_fragments_df = fragment_df[mass_comparison_vector]
+
+    # If no candidate fragments exist, exit function
+    if matched_fragments_df.shape[0] == 0:
+        return []
+
+    # Step 2 - If multiple candidate substructures, filter for those with the lowest penalty scores
+    if lowest_penalty_filter:
+        if matched_fragments_df.shape[0] > 1:
+            min_score = matched_fragments_df["score"].min()
+            matched_fragments_df = matched_fragments_df[
+                matched_fragments_df["score"] == min_score
+            ]
+
+    # Step 3 - Save all remaining candidate fragments
+    matched_fragment_idxs = list(matched_fragments_df.index)
+
+    return matched_fragment_idxs
+
+
+def get_fragment_mass_range(fragment_engine, fragment_df, tolerance):
+    """get_fragment_mass_range.
+
+    Define min and max masses in the range that are available based upon
+    hydrogen diffs.
+
+    Args:
+        fragment_engine: Fragment engine
+        fragment_df: fragment_df
+        tolerance: Tolerance
+
+    """
+    fragment_masses_np = fragment_engine.fragment_masses_np
+
+    # Build a list of the min and max mass of each fragment
+    fragment_mass_min_max = []
+
+    for fragment_idx in range(fragment_masses_np.shape[0]):
+        fragment_masses = fragment_masses_np[fragment_idx, :]
+
+        if np.sum(fragment_masses) == 0:
+            min_frag_mass = 0
+            max_frag_mass = 0
+
+        else:
+            min_frag_mass = (
+                fragment_masses[np.nonzero(fragment_masses)[0][0]] - tolerance
+            )
+            max_frag_mass = max(fragment_masses) + tolerance
+
+        fragment_mass_min_max.append((min_frag_mass, max_frag_mass))
+
+    fragment_mass_min_max = np.array(fragment_mass_min_max)
+    fragment_df["min_mass"] = fragment_mass_min_max[:, 0]
+    fragment_df["max_mass"] = fragment_mass_min_max[:, 1]
+
+    return fragment_df
+
+def run_magma_wrapper(args):
+    if os.path.exists(args[-1]):     # skip over ones that have been processed
+        return 
+    return run_magma(*args)
+
+def run_magma(identifier, mzs, intensities, smiles, adduct, save_filename=''):
+    '''YZC
+        Run fragmentation, assignment, and save results 
+        
+    '''
+    # Step 1 - Load fragmentation engine and generate fragments
+    (
+        max_broken_bonds,
+        max_water_losses,
+        ionisation_mode,
+        skip_fragmentation,
+        molcharge,
+    ) = FRAGMENT_ENGINE_PARAMS.values()
+    try:
+        engine = FragmentEngine(
+            smiles=smiles,
+            max_broken_bonds=max_broken_bonds,
+            max_water_losses=max_water_losses,
+            ionisation_mode=ionisation_mode,
+            skip_fragmentation=skip_fragmentation,
+            molcharge=molcharge,
+        )
+        engine.generate_fragments()
+    except Exception as e:
+        logging.info(f"Error for spec {identifier}")
+        print(e)
+        return None
+    
+    # Step 2 - Assign fragments to peaks
+    assignment_dict = peak_fragment_assignment(
+        engine,
+        mzs,
+        intensities,
+        adduct,
+    )
+
+    # Step 3 - Save assignments
+    if save_filename:
+        with open(save_filename, 'w') as f:
+            json.dump(assignment_dict, f)
+    else:
+        return assignment_dict
+
+def peak_fragment_assignment(fragment_engine, mzs, intensities, adduct):
+    ''' returns a df with columns  
+
+    Args:
+        fragment_engine: FragmentEngine
+        mzs: np array of mz values
+        adduct: str eg. [M+H]+ [M+Na]+
+
+    Returns:
+        assignment_df
+    '''
+
+    fragments_info = fragment_engine.fragment_info
+
+    fragment_df = pd.DataFrame(
+        fragment_engine.fragment_info, columns=["id", "score", "bond_breaks"]
+    )
+    fragment_df = get_fragment_mass_range(fragment_engine, fragment_df, tolerance=PEAK_ASSIGNMENT_PARAMS['tolerance'])
+
+    # Need to build comparison values here
+    min_fragment_mass = fragment_df["min_mass"].values
+    max_fragment_mass = fragment_df["max_mass"].values
+    adduct = magma_utils.extract_adduct_ion(adduct)
+    charge = 1 if adduct.startswith('+') else -1
+    exact_masses = mzs + ionmasses[charge][adduct]
+
+    mass_comparison_matrix = np.logical_and(
+        exact_masses[None, :] >= min_fragment_mass[:, None],
+        exact_masses[None, :] <= max_fragment_mass[:, None],
+    )
+
+    # Iterate over each peak to find a match
+    assignments = defaultdict(list) # {mz, intensity, subformulas, candidates}
+    for k, (m, i) in enumerate(zip(mzs, intensities)):
+        mass_comparison_vector = mass_comparison_matrix[:, k]
+
+
+        matched_fragment_idxs = get_matching_fragment(
+            fragment_df,
+            mass_comparison_vector,
+            lowest_penalty_filter=PEAK_ASSIGNMENT_PARAMS['lowest_penalty_filter'],
+        )
+
+        # Save selected fragments info
+        subformulas = set([])
+        substructures = set([])
+        for idx in matched_fragment_idxs:
+            fragment_info = fragment_engine.get_fragment_info(fragments_info[idx][0], 0)
+
+            subformulas.add(fragment_info[2])
+            substructures.add(fragment_info[3])
+
+        subformulas = list(subformulas)
+        substructures = list(substructures)
+
+        assignments['mz'].append(m)
+        assignments['intensities'].append(i)
+        assignments['subformulas'].append(subformulas)
+        assignments['substructures'].append(substructures)
+    return assignments
+    
+
+if __name__ == "__main__":
+    import time
+
+    start_time = time.time()
+    args = get_args()
+    kwargs = args.__dict__
+
+    os.makedirs(args.output_dir, exist_ok=True)
+
+    df = pd.read_csv(args.data_pth, sep='\t')
+    df['save_filename'] = df['identifier'].apply(lambda x: os.path.join(args.output_dir, x + '.json'))
+
+    df['mzs'] = df['mzs'].apply(lambda x: np.array([float(m) for m in x.split(',')]))
+    df['intensities'] = df['intensities'].apply(lambda x: np.array([float(i) for i in x.split(',')]))
+
+    df = df[['identifier', 'mzs', 'intensities', 'smiles', 'adduct', 'save_filename']]
+
+
+    tasks = list(df.itertuples(index=False, name=None))
+
+    with Pool(processes=args.workers) as pool:
+        results = list(tqdm(pool.imap_unordered(run_magma_wrapper, tasks), total=len(tasks)))
+        # pool.starmap(run_magma, tasks)
+
+    end_time = time.time()
+    print(f"Program finished in: {end_time - start_time} seconds")