corems.molecular_id.factory.MolecularLookupTable

  1__author__ = "Yuri E. Corilo"
  2__date__ = "Jul 02, 2019"
  3
  4import contextlib
  5import cProfile
  6import io
  7import itertools
  8import json
  9import pstats
 10from copy import deepcopy
 11from typing import Dict
 12
 13from sqlalchemy import create_engine, func
 14from sqlalchemy.orm import sessionmaker
 15from tqdm import tqdm
 16
 17from corems import chunks, timeit
 18from corems.encapsulation.constant import Atoms
 19from corems.encapsulation.spawn_pool import spawn_pool
 20from corems.encapsulation.factory.parameters import MSParameters
 21from corems.encapsulation.factory.processingSetting import (
 22    MolecularLookupDictSettings,
 23    validate_used_atoms_keys,
 24)
 25from corems.molecular_id.factory.molecularSQL import (
 26    CarbonHydrogen,
 27    HeteroAtoms,
 28    MolecularFormulaLink,
 29    MolForm_SQL,
 30    psycopg3_url,
 31)
 32
 33
 34@contextlib.contextmanager
 35def profiled():
 36    """A context manager for profiling."""
 37    pr = cProfile.Profile()
 38    pr.enable()
 39    yield
 40    pr.disable()
 41    s = io.StringIO()
 42    ps = pstats.Stats(pr, stream=s).sort_stats("cumulative")
 43    ps.print_stats()
 44    # uncomment this to see who's calling what
 45    # ps.print_callers()
 46    print(s.getvalue())
 47
 48
 49def insert_database_worker(args):
 50    """Inserts data into the database."""
 51    results, url = args
 52
 53    if not url:
 54        url = "sqlite:///db/molformulas.sqlite"
 55
 56    url = psycopg3_url(url)
 57
 58    if url[0:6] == "sqlite":
 59        engine = create_engine(url, echo=False)
 60    else:
 61        engine = create_engine(url, echo=False, isolation_level="AUTOCOMMIT")
 62
 63    session_factory = sessionmaker(bind=engine)
 64    session = session_factory()
 65    insert_query = MolecularFormulaLink.__table__.insert().values(results)
 66    session.execute(insert_query)
 67    session.commit()
 68    session.close()
 69    engine.dispose()
 70
 71
 72class MolecularCombinations:
 73    """A class for generating molecular formula combinations.
 74
 75    Parameters
 76    ----------
 77    molecular_search_settings : object
 78        An object containing user-defined settings.
 79
 80    Attributes
 81    ----------
 82    sql_db : MolForm_SQL
 83        The SQLite database object.
 84    len_existing_classes : int
 85        The number of existing classes in the SQLite database.
 86    odd_ch_id : list
 87        A list of odd carbon and hydrogen atom IDs.
 88    odd_ch_dict : list
 89        A list of odd carbon and hydrogen atom dictionaries.
 90    odd_ch_mass : list
 91        A list of odd carbon and hydrogen atom masses.
 92    odd_ch_dbe : list
 93        A list of odd carbon and hydrogen atom double bond equivalents.
 94    even_ch_id : list
 95        A list of even carbon and hydrogen atom IDs.
 96    even_ch_dict : list
 97        A list of even carbon and hydrogen atom dictionaries.
 98    even_ch_mass : list
 99        A list of even carbon and hydrogen atom masses.
100    even_ch_dbe : list
101        A list of even carbon and hydrogen atom double bond equivalents.
102
103    Methods
104    -------
105    * cProfile_worker(args)
106        A cProfile worker for the get_mol_formulas function.
107    * check_database_get_class_list(molecular_search_settings)
108        Checks if the database has all the classes, if not create the missing classes.
109    * get_carbonsHydrogens(settings, odd_even)
110        Retrieves carbon and hydrogen atoms from the molecular lookup table based on user-defined settings.
111    * add_carbonsHydrogens(settings, existing_classes_objs)
112        Adds carbon and hydrogen atoms to the molecular lookup table based on user-defined settings.
113    * runworker(molecular_search_settings)
114        Runs the molecular formula lookup table worker.
115    * get_classes_in_order(molecular_search_settings)
116        Gets the classes in order.
117    * sort_classes(atoms_in_order, combination_dict)
118        Sorts the classes in order.
119    * get_fixed_initial_number_of_hydrogen(min_h, odd_even)
120        Gets the fixed initial number of hydrogen atoms.
121    * calc_mz(datadict, class_mass=0)
122        Calculates the mass-to-charge ratio (m/z) of a molecular formula.
123    * calc_dbe_class(datadict)
124        Calculates the double bond equivalent (DBE) of a molecular formula.
125    * populate_combinations(classe_tuple, settings)
126        Populates the combinations.
127    * get_or_add(SomeClass, kw)
128        Gets or adds a class.
129    * get_mol_formulas(odd_even_tag, classe_tuple, settings)
130        Gets the molecular formulas.
131    * get_h_odd_or_even(class_dict)
132        Gets the hydrogen odd or even.
133    * get_total_halogen_atoms(class_dict)
134        Gets the total number of halogen atoms.
135    * get_total_hetero_valence(class_dict)
136        Gets the total valence of heteroatoms other than N, F, Cl, and Br
137    """
138
139    def __init__(self, sql_db=None):
140        if not sql_db:
141            self.sql_db = MolForm_SQL()
142        else:
143            self.sql_db = sql_db
144
145    def cProfile_worker(self, args):
146        """cProfile worker for the get_mol_formulas function"""
147        cProfile.runctx(
148            "self.get_mol_formulas(*args)",
149            globals(),
150            locals(),
151            "mf_database_cprofile.prof",
152        )
153
154    def check_database_get_class_list(self, molecular_search_settings):
155        """check if the database has all the classes, if not create the missing classes
156
157        Parameters
158        ----------
159        molecular_search_settings : object
160            An object containing user-defined settings.
161
162        Returns
163        -------
164        list
165            list of tuples with the class name and the class dictionary
166        """
167        all_class_to_create = []
168
169        classes_dict = self.get_classes_in_order(molecular_search_settings)
170
171        class_str_set = set(classes_dict.keys())
172
173        existing_classes_objs = self.sql_db.session.query(HeteroAtoms).distinct().all()
174
175        existing_classes_str = set([classe.name for classe in existing_classes_objs])
176
177        self.len_existing_classes = len(existing_classes_str)
178
179        class_to_create = class_str_set - existing_classes_str
180
181        class_count = len(existing_classes_objs)
182
183        data_classes = list()
184        for index, class_str in enumerate(class_to_create):
185            class_dict = classes_dict.get(class_str)
186            halogen_count = self.get_total_halogen_atoms(class_dict)
187            data_classes.append(
188                {
189                    "name": class_str,
190                    "id": class_count + index + 1,
191                    "halogensCount": halogen_count,
192                }
193            )
194
195        # data_classes = [{"name":class_str, "id":class_count+ index + 1} for index, class_str in enumerate(class_to_create)]
196
197        if data_classes:
198            list_insert_chunks = chunks(data_classes, self.sql_db.chunks_count)
199            for insert_chunk in list_insert_chunks:
200                insert_query = HeteroAtoms.__table__.insert().values(insert_chunk)
201                self.sql_db.session.execute(insert_query)
202
203        for index, class_str in enumerate(class_to_create):
204            class_tuple = (
205                class_str,
206                classes_dict.get(class_str),
207                class_count + index + 1,
208            )
209
210            all_class_to_create.append(class_tuple)
211
212        return (
213            [(c_s, c_d) for c_s, c_d in classes_dict.items()],
214            all_class_to_create,
215            existing_classes_objs,
216        )
217
218    def get_carbonsHydrogens(self, settings, odd_even):
219        """Retrieve carbon and hydrogen atoms from the molecular lookup table based on user-defined settings.
220
221        Parameters
222        ----------
223        settings : object
224             An object containing user-defined settings.
225        odd_even : str
226            A string indicating whether to retrieve even or odd hydrogen atoms.
227
228        Returns
229        -------
230        list
231            A list of CarbonHydrogen objects that satisfy the specified conditions.
232        """
233        operator = "==" if odd_even == "even" else "!="
234        usedAtoms = settings.usedAtoms
235        user_min_c, user_max_c = usedAtoms.get("C")
236        user_min_h, user_max_h = usedAtoms.get("H")
237
238        return eval(
239            "self.sql_db.session.query(CarbonHydrogen).filter("
240            "CarbonHydrogen.C >= user_min_c,"
241            "CarbonHydrogen.H >= user_min_h,"
242            "CarbonHydrogen.C <= user_max_c,"
243            "CarbonHydrogen.H <= user_max_h,"
244            "CarbonHydrogen.H % 2" + operator + "0).all()"
245        )
246
247    def add_carbonsHydrogens(self, settings, existing_classes_objs):
248        """Add carbon and hydrogen atoms to the molecular lookup table based on user-defined settings.
249
250        Parameters
251        ----------
252        settings : object
253            An object containing user-defined settings.
254        existing_classes_objs : list
255            A list of HeteroAtoms objects.
256        """
257        usedAtoms = settings.usedAtoms
258
259        user_min_c, user_max_c = usedAtoms.get("C")
260        user_min_h, user_max_h = usedAtoms.get("H")
261
262        query_obj = self.sql_db.session.query(
263            func.max(CarbonHydrogen.C).label("max_c"),
264            func.min(CarbonHydrogen.C).label("min_c"),
265            func.max(CarbonHydrogen.H).label("max_h"),
266            func.min(CarbonHydrogen.H).label("min_h"),
267        )
268
269        database = query_obj.first()
270        if (
271            database.max_c == user_max_c
272            and database.min_c == user_min_c
273            and database.max_h == user_max_h
274            and database.min_h == user_min_h
275        ):
276            # all data is already available at the database
277            pass
278
279        else:
280            current_count = self.sql_db.session.query(CarbonHydrogen.C).count()
281
282            databaseCarbonHydrogen = self.sql_db.session.query(CarbonHydrogen).all()
283
284            userCarbon = set(range(user_min_c, user_max_c + 1))
285            userHydrogen = set(range(user_min_h, user_max_h + 1))
286
287            carbon_hydrogen_objs_database = {}
288            for obj in databaseCarbonHydrogen:
289                str_data = "C:{},H:{}".format(obj.C, obj.H)
290                carbon_hydrogen_objs_database[str_data] = str_data
291
292            carbon_hydrogen_objs_to_create = {"even": {}, "odd": {}}
293
294            list_ch_obj_to_add = list()
295            i = 0
296            for comb in itertools.product(userCarbon, userHydrogen):
297                C = comb[0]
298                H = comb[1]
299                data = {
300                    "C": C,
301                    "H": H,
302                }
303
304                data_insert = {
305                    "C": C,
306                    "H": H,
307                }
308
309                str_data = "C:{},H:{}".format(C, H)
310
311                if str_data not in carbon_hydrogen_objs_database.keys():
312                    label = "even" if comb[1] % 2 == 0 else "odd"
313                    data["mass"] = (C * Atoms.atomic_masses.get("C")) + (
314                        H * Atoms.atomic_masses.get("H")
315                    )
316                    data["dbe"] = C - (H / 2) + 1
317                    data["id"] = i + current_count + 1
318                    data_insert["id"] = i + current_count + 1
319                    i = i + 1
320                    carbon_hydrogen_objs_to_create[label][str_data] = data
321
322                    list_ch_obj_to_add.append(data_insert)
323
324            if list_ch_obj_to_add:
325                # insert carbon hydrogen objs
326                list_insert_chunks = chunks(
327                    list_ch_obj_to_add, self.sql_db.chunks_count
328                )
329                for insert_chunk in list_insert_chunks:
330                    insert_query = CarbonHydrogen.__table__.insert().values(
331                        insert_chunk
332                    )
333                    self.sql_db.session.execute(insert_query)
334                self.sql_db.session.commit()
335
336                list_molecular_form = list()
337                for classe_obj in existing_classes_objs:
338                    classe_dict = classe_obj.to_dict()
339                    classe_mass = self.calc_mz(classe_dict)
340                    classe_dbe = self.calc_dbe_class(classe_dict)
341
342                    odd_even_label = self.get_h_odd_or_even(classe_dict)
343
344                    ch_datalist = carbon_hydrogen_objs_to_create.get(
345                        odd_even_label
346                    ).values()
347
348                    for ch_dict in ch_datalist:
349                        mass = ch_dict.get("mass") + classe_mass
350                        dbe = ch_dict.get("dbe") + classe_dbe
351
352                        if settings.min_mz <= mass <= settings.max_mz:
353                            if settings.min_dbe <= dbe <= settings.max_dbe:
354                                list_molecular_form.append(
355                                    {
356                                        "heteroAtoms_id": classe_obj.id,
357                                        "carbonHydrogen_id": ch_dict.get("id"),
358                                        "mass": mass,
359                                        "DBE": dbe,
360                                    }
361                                )
362
363                list_insert_chunks = chunks(
364                    list_molecular_form, self.sql_db.chunks_count
365                )
366                for insert_chunk in list_insert_chunks:
367                    insert_query = MolecularFormulaLink.__table__.insert().values(
368                        insert_chunk
369                    )
370                    self.sql_db.session.execute(insert_query)
371                self.sql_db.session.commit()
372
373    @timeit(print_time=True)
374    def runworker(self, molecular_search_settings, **kwargs):
375        """Run the molecular formula lookup table worker.
376
377        Parameters
378        ----------
379        molecular_search_settings : object
380            An object containing user-defined settings.
381        kwargs : dict
382            A dictionary of keyword arguments.
383            Most notably, the print_time argument which is passed to the timeit decorator.
384
385        Returns
386        -------
387        list
388            A list of tuples with the class name and the class dictionary.
389
390
391        """
392        verbose = molecular_search_settings.verbose_processing
393
394        classes_list, class_to_create, existing_classes_objs = (
395            self.check_database_get_class_list(molecular_search_settings)
396        )
397
398        settings = MolecularLookupDictSettings()
399        settings.usedAtoms = deepcopy(molecular_search_settings.usedAtoms)
400        settings.url_database = molecular_search_settings.url_database
401        settings.db_jobs = molecular_search_settings.db_jobs
402
403        self.add_carbonsHydrogens(settings, existing_classes_objs)
404
405        if class_to_create:
406            settings = MolecularLookupDictSettings()
407            settings.usedAtoms = deepcopy(molecular_search_settings.usedAtoms)
408            settings.url_database = molecular_search_settings.url_database
409            settings.db_jobs = molecular_search_settings.db_jobs
410
411            self.sql_db.session.commit()
412            odd_ch_obj = self.get_carbonsHydrogens(settings, "odd")
413            self.odd_ch_id = [obj.id for obj in odd_ch_obj]
414            self.odd_ch_dict = [{"C": obj.C, "H": obj.H} for obj in odd_ch_obj]
415            self.odd_ch_mass = [obj.mass for obj in odd_ch_obj]
416            self.odd_ch_dbe = [obj.dbe for obj in odd_ch_obj]
417
418            even_ch_obj = self.get_carbonsHydrogens(settings, "even")
419            self.even_ch_id = [obj.id for obj in even_ch_obj]
420            self.even_ch_dict = [{"C": obj.C, "H": obj.H} for obj in even_ch_obj]
421            self.even_ch_mass = [obj.mass for obj in even_ch_obj]
422            self.even_ch_dbe = [obj.dbe for obj in even_ch_obj]
423
424            all_results = list()
425            for class_tuple in tqdm(class_to_create, disable=not verbose):
426                results = self.populate_combinations(class_tuple, settings)
427                all_results.extend(results)
428                if settings.db_jobs == 1:
429                    # if len(all_results) >= self.sql_db.chunks_count:
430                    list_insert_chunks = list(chunks(results, self.sql_db.chunks_count))
431                    for chunk in list_insert_chunks:
432                        insert_query = MolecularFormulaLink.__table__.insert().values(
433                            chunk
434                        )
435                        self.sql_db.session.execute(insert_query)
436                    # all_results = list()
437            self.sql_db.session.commit()
438            # each chunk takes ~600Mb of memory, so if using 8 processes the total free memory needs to be 5GB
439            if settings.db_jobs > 1:
440                list_insert_chunks = list(chunks(all_results, self.sql_db.chunks_count))
441                if verbose:
442                    print(
443                        "Started database insert using {} iterations for a total of {} rows".format(
444                            len(list_insert_chunks), len(all_results)
445                        )
446                    )
447                worker_args = [
448                    (chunk, settings.url_database) for chunk in list_insert_chunks
449                ]
450                with spawn_pool(settings.db_jobs) as p:
451                    for class_list in tqdm(
452                        p.imap_unordered(insert_database_worker, worker_args),
453                        disable=not verbose,
454                    ):
455                        pass
456
457        return classes_list
458
459    def get_classes_in_order(self, molecular_search_settings):
460        """Get the classes in order
461
462        Parameters
463        ----------
464        molecular_search_settings : object
465            An object containing user-defined settings.
466
467        Returns
468        -------
469        dict
470            A dictionary of classes in order.
471            structure is  ('HC', {'HC': 1})
472        """
473        validate_used_atoms_keys(molecular_search_settings.usedAtoms)
474
475        usedAtoms = deepcopy(molecular_search_settings.usedAtoms)
476
477        usedAtoms.pop("C")
478        usedAtoms.pop("H")
479
480        min_n, max_n = usedAtoms.get("N") if usedAtoms.get("N") else (0, 0)
481        min_o, max_o = usedAtoms.get("O") if usedAtoms.get("O") else (0, 0)
482        min_s, max_s = usedAtoms.get("S") if usedAtoms.get("S") else (0, 0)
483        min_p, max_p = usedAtoms.get("P") if usedAtoms.get("P") else (0, 0)
484
485        possible_n = [n for n in range(min_n, max_n + 1)]
486        possible_o = [o for o in range(min_o, max_o + 1)]
487        possible_s = [s for s in range(min_s, max_s + 1)]
488        possible_p = [p for p in range(min_p, max_p + 1)]
489
490        atoms_in_order = ["N", "O", "S", "P"]
491
492        classe_in_order = {}
493
494        all_atoms_tuples = itertools.product(
495            possible_n, possible_o, possible_s, possible_p
496        )
497
498        for atom in atoms_in_order:
499            usedAtoms.pop(atom, None)
500
501        for selected_atom, min_max_tuple in usedAtoms.items():
502            min_x = min_max_tuple[0]
503            max_x = min_max_tuple[1]
504
505            possible_x = [x for x in range(min_x, max_x + 1)]
506
507            all_atoms_tuples = itertools.product(all_atoms_tuples, possible_x)
508            all_atoms_tuples = [
509                all_atoms_combined[0] + (all_atoms_combined[1],)
510                for all_atoms_combined in all_atoms_tuples
511            ]
512            atoms_in_order.append(selected_atom)
513
514        for all_atoms_tuple in all_atoms_tuples:
515            classe_str = ""
516            classe_dict = {}
517
518            for each_atoms_index, atom_number in enumerate(all_atoms_tuple):
519                if atom_number != 0:
520                    classe_dict[atoms_in_order[each_atoms_index]] = atom_number
521
522            if not classe_dict:
523                classe_in_order["HC"] = {"HC": ""}
524                continue
525
526            classe_str = json.dumps(classe_dict)
527
528            if len(classe_str) > 0:
529                classe_in_order[classe_str] = classe_dict
530
531        classe_in_order_dict = self.sort_classes(atoms_in_order, classe_in_order)
532
533        return classe_in_order_dict
534
535    @staticmethod
536    def sort_classes(atoms_in_order, combination_dict) -> Dict[str, Dict[str, int]]:
537        """Sort the classes in order
538
539        Parameters
540        ----------
541        atoms_in_order : list
542            A list of atoms in order.
543        combination_dict : dict
544            A dictionary of classes.
545
546        Returns
547        -------
548        dict
549            A dictionary of classes in order.
550        """
551        # ensures atoms are always in the order defined at atoms_in_order list
552        join_dict_classes = dict()
553        atoms_in_order = ["N", "S", "P", "O"] + atoms_in_order[4:] + ["HC"]
554
555        sort_method = lambda atoms_keys: [atoms_in_order.index(atoms_keys)]
556        for class_str, class_dict in combination_dict.items():
557            sorted_dict_keys = sorted(class_dict, key=sort_method)
558            class_dict = {atom: class_dict[atom] for atom in sorted_dict_keys}
559            class_str = json.dumps(class_dict)
560            # using json for the new database, class
561            # class_str = ' '.join([atom + str(class_dict[atom]) for atom in sorted_dict_keys])
562            join_dict_classes[class_str] = class_dict
563
564        return join_dict_classes
565
566    @staticmethod
567    def get_fixed_initial_number_of_hydrogen(min_h, odd_even):
568        """Get the fixed initial number of hydrogen atoms
569
570        Parameters
571        ----------
572        min_h : int
573            The minimum number of hydrogen atoms.
574        odd_even : str
575            A string indicating whether to retrieve even or odd hydrogen atoms.
576        """
577        remaining_h = min_h % 2
578
579        if odd_even == "even":
580            if remaining_h == 0:
581                return remaining_h
582
583            else:
584                return remaining_h + 1
585
586        else:
587            if remaining_h == 0:
588                return remaining_h + 1
589
590            else:
591                return remaining_h
592
593    def calc_mz(self, datadict, class_mass=0):
594        """Calculate the mass-to-charge ratio (m/z) of a molecular formula.
595
596        Parameters
597        ----------
598        datadict : dict
599            A dictionary of classes.
600        class_mass : int
601            The mass of the class.
602
603        Returns
604        -------
605        float
606            The mass-to-charge ratio (m/z) of a molecular formula.
607        """
608        mass = class_mass
609
610        for atom in datadict.keys():
611            if atom != "HC":
612                mass = mass + Atoms.atomic_masses[atom] * datadict.get(atom)
613
614        return mass
615
616    def calc_dbe_class(self, datadict):
617        """Calculate the double bond equivalent (DBE) of a molecular formula.
618
619        Parameters
620        ----------
621        datadict : dict
622            A dictionary of classes.
623
624        Returns
625        -------
626        float
627            The double bond equivalent (DBE) of a molecular formula.
628        """
629        init_dbe = 0
630        for atom in datadict.keys():
631            if atom == "HC":
632                continue
633
634            n_atom = int(datadict.get(atom))
635
636            clean_atom = "".join([i for i in atom if not i.isdigit()])
637
638            valencia = MSParameters.molecular_search.used_atom_valences.get(clean_atom)
639
640            if type(valencia) is tuple:
641                valencia = valencia[0]
642            if valencia > 0:
643                # print atom, valencia, n_atom, init_dbe
644                init_dbe = init_dbe + (n_atom * (valencia - 2))
645            else:
646                continue
647
648        return 0.5 * init_dbe
649
650    def populate_combinations(self, classe_tuple, settings):
651        """Populate the combinations
652
653        Parameters
654        ----------
655        classe_tuple : tuple
656            A tuple containing the class name, the class dictionary, and the class ID.
657        settings : object
658            An object containing user-defined settings.
659
660        Returns
661        -------
662        list
663            A list of molecular formula data dictionaries.
664        """
665        class_dict = classe_tuple[1]
666        odd_or_even = self.get_h_odd_or_even(class_dict)
667
668        return self.get_mol_formulas(odd_or_even, classe_tuple, settings)
669
670    def get_or_add(self, SomeClass, kw):
671        """Get or add a class
672
673        Parameters
674        ----------
675        SomeClass : object
676            A class object.
677        kw : dict
678            A dictionary of classes.
679
680        Returns
681        -------
682        object
683            A class object.
684        """
685        obj = self.sql_db.session.query(SomeClass).filter_by(**kw).first()
686        if not obj:
687            obj = SomeClass(**kw)
688        return obj
689
690    def get_mol_formulas(self, odd_even_tag, classe_tuple, settings):
691        """Get the molecular formulas
692
693        Parameters
694        ----------
695        odd_even_tag : str
696            A string indicating whether to retrieve even or odd hydrogen atoms.
697        classe_tuple : tuple
698
699        settings : object
700            An object containing user-defined settings.
701
702        Returns
703        -------
704        list
705            A list of molecular formula data dictionaries.
706
707        """
708        class_str = classe_tuple[0]
709        class_dict = classe_tuple[1]
710        classe_id = classe_tuple[2]
711
712        results = list()
713
714        if "HC" in class_dict:
715            del class_dict["HC"]
716
717        class_dbe = self.calc_dbe_class(class_dict)
718        class_mass = self.calc_mz(class_dict)
719
720        carbonHydrogen_mass = (
721            self.odd_ch_mass if odd_even_tag == "odd" else self.even_ch_mass
722        )
723        carbonHydrogen_dbe = (
724            self.odd_ch_dbe if odd_even_tag == "odd" else self.even_ch_dbe
725        )
726        carbonHydrogen_id = self.odd_ch_id if odd_even_tag == "odd" else self.even_ch_id
727
728        for index, carbonHydrogen_obj in enumerate(carbonHydrogen_id):
729            mass = carbonHydrogen_mass[index] + class_mass
730            dbe = carbonHydrogen_dbe[index] + class_dbe
731
732            if settings.min_mz <= mass <= settings.max_mz:
733                if settings.min_dbe <= dbe <= settings.max_dbe:
734                    molecularFormula = {
735                        "heteroAtoms_id": classe_id,
736                        "carbonHydrogen_id": carbonHydrogen_id[index],
737                        "mass": mass,
738                        "DBE": dbe,
739                    }
740
741                    results.append(molecularFormula)
742
743        return results
744
745    def get_h_odd_or_even(self, class_dict):
746        """Get the hydrogen odd or even
747
748        Parameters
749        ----------
750        class_dict : dict
751            A dictionary of classes.
752
753        Returns
754        -------
755        str
756            A string indicating whether to retrieve even or odd hydrogen atoms.
757        """
758
759        HAS_NITROGEN = "N" in class_dict.keys()
760
761        number_of_halogen = self.get_total_halogen_atoms(class_dict)
762        number_of_hetero = self.get_total_heteroatoms(class_dict)
763
764        if number_of_halogen > 0:
765            HAS_HALOGEN = True
766
767        else:
768            HAS_HALOGEN = False
769
770        if HAS_HALOGEN:
771            remaining_halogen = number_of_halogen % 2
772
773        else:
774            remaining_halogen = 0
775
776        if number_of_hetero > 0:
777            HAS_OTHER_HETERO = True
778
779            total_hetero_valence = self.get_total_hetero_valence(class_dict)
780
781        else:
782            HAS_OTHER_HETERO = False
783
784            total_hetero_valence = 0
785
786        if HAS_OTHER_HETERO:
787            remaining_hetero_valence = total_hetero_valence % 2
788
789        else:
790            remaining_hetero_valence = 0
791
792        if HAS_NITROGEN and not HAS_OTHER_HETERO:
793            number_of_n = class_dict.get("N")
794            remaining_n = number_of_n % 2
795
796        elif HAS_NITROGEN and HAS_OTHER_HETERO:
797            number_of_n = class_dict.get("N")
798            remaining_n = (number_of_n + remaining_hetero_valence) % 2
799
800        elif HAS_OTHER_HETERO and not HAS_NITROGEN:
801            remaining_n = remaining_hetero_valence
802
803        else:
804            remaining_n = -1
805
806        if remaining_n > 0.0:
807            if HAS_NITROGEN or HAS_OTHER_HETERO:
808                if HAS_HALOGEN:
809                    if remaining_halogen == 0:
810                        return "odd"
811                    else:
812                        return "even"
813
814                else:
815                    return "odd"
816
817        elif remaining_n == 0.0:
818            if HAS_NITROGEN or HAS_OTHER_HETERO:
819                if HAS_HALOGEN:
820                    if remaining_halogen == 0:
821                        return "even"
822                    else:
823                        return "odd"
824
825                else:
826                    return "even"
827
828        else:
829            if HAS_HALOGEN:
830                if remaining_halogen == 0:
831                    return "even"
832                else:
833                    return "odd"
834
835            else:
836                return "even"
837
838    @staticmethod
839    def get_total_heteroatoms(class_dict):
840        """Get the total number of heteroatoms other than N, F, Cl, Br
841
842        Parameters
843        ----------
844        class_dict : dict
845            A dictionary of classes.
846
847        Returns
848        -------
849        int
850            The total number of heteroatoms.
851        """
852
853        total_number = 0
854
855        for atom in class_dict.keys():
856            if atom not in ["HC", "C", "H", "O", "N", "F", "Cl", "Br"]:
857                total_number = total_number + class_dict.get(atom)
858
859        return total_number
860
861    @staticmethod
862    def get_total_hetero_valence(class_dict):
863        """Get the total valence of heteroatoms other than N, F, Cl, Br
864
865        Parameters
866        ----------
867        class_dict : dict
868            A dictionary of classes.
869
870        Returns
871        -------
872        int
873            The total heteroatom valence.
874        """
875        total_valence = 0
876
877        for atom in class_dict.keys():
878            if atom not in ["HC", "C", "H", "O", "N", "F", "Cl", "Br"]:
879                clean_atom = "".join([i for i in atom if not i.isdigit()])
880
881                atom_valence = MSParameters.molecular_search.used_atom_valences.get(
882                    clean_atom
883                )
884
885                if type(atom_valence) is tuple:
886                    atom_valence = atom_valence[0]
887
888                n_atom = int(class_dict.get(atom))
889
890                n_atom_valence = atom_valence * n_atom
891
892                total_valence = total_valence + n_atom_valence
893
894        return total_valence
895
896    @staticmethod
897    def get_total_halogen_atoms(class_dict):
898        """Get the total number of halogen atoms
899
900        Parameters
901        ----------
902        class_dict : dict
903            A dictionary of classes.
904
905        Returns
906        -------
907        int
908            The total number of halogen atoms.
909        """
910        atoms = ["F", "Cl", "Br"]
911
912        total_number = 0
913
914        for atom in atoms:
915            if atom in class_dict.keys():
916                total_number = total_number + class_dict.get(atom)
917
918        return total_number
@contextlib.contextmanager
def profiled():
35@contextlib.contextmanager
36def profiled():
37    """A context manager for profiling."""
38    pr = cProfile.Profile()
39    pr.enable()
40    yield
41    pr.disable()
42    s = io.StringIO()
43    ps = pstats.Stats(pr, stream=s).sort_stats("cumulative")
44    ps.print_stats()
45    # uncomment this to see who's calling what
46    # ps.print_callers()
47    print(s.getvalue())

A context manager for profiling.

def insert_database_worker(args):
50def insert_database_worker(args):
51    """Inserts data into the database."""
52    results, url = args
53
54    if not url:
55        url = "sqlite:///db/molformulas.sqlite"
56
57    url = psycopg3_url(url)
58
59    if url[0:6] == "sqlite":
60        engine = create_engine(url, echo=False)
61    else:
62        engine = create_engine(url, echo=False, isolation_level="AUTOCOMMIT")
63
64    session_factory = sessionmaker(bind=engine)
65    session = session_factory()
66    insert_query = MolecularFormulaLink.__table__.insert().values(results)
67    session.execute(insert_query)
68    session.commit()
69    session.close()
70    engine.dispose()

Inserts data into the database.

class MolecularCombinations:
 73class MolecularCombinations:
 74    """A class for generating molecular formula combinations.
 75
 76    Parameters
 77    ----------
 78    molecular_search_settings : object
 79        An object containing user-defined settings.
 80
 81    Attributes
 82    ----------
 83    sql_db : MolForm_SQL
 84        The SQLite database object.
 85    len_existing_classes : int
 86        The number of existing classes in the SQLite database.
 87    odd_ch_id : list
 88        A list of odd carbon and hydrogen atom IDs.
 89    odd_ch_dict : list
 90        A list of odd carbon and hydrogen atom dictionaries.
 91    odd_ch_mass : list
 92        A list of odd carbon and hydrogen atom masses.
 93    odd_ch_dbe : list
 94        A list of odd carbon and hydrogen atom double bond equivalents.
 95    even_ch_id : list
 96        A list of even carbon and hydrogen atom IDs.
 97    even_ch_dict : list
 98        A list of even carbon and hydrogen atom dictionaries.
 99    even_ch_mass : list
100        A list of even carbon and hydrogen atom masses.
101    even_ch_dbe : list
102        A list of even carbon and hydrogen atom double bond equivalents.
103
104    Methods
105    -------
106    * cProfile_worker(args)
107        A cProfile worker for the get_mol_formulas function.
108    * check_database_get_class_list(molecular_search_settings)
109        Checks if the database has all the classes, if not create the missing classes.
110    * get_carbonsHydrogens(settings, odd_even)
111        Retrieves carbon and hydrogen atoms from the molecular lookup table based on user-defined settings.
112    * add_carbonsHydrogens(settings, existing_classes_objs)
113        Adds carbon and hydrogen atoms to the molecular lookup table based on user-defined settings.
114    * runworker(molecular_search_settings)
115        Runs the molecular formula lookup table worker.
116    * get_classes_in_order(molecular_search_settings)
117        Gets the classes in order.
118    * sort_classes(atoms_in_order, combination_dict)
119        Sorts the classes in order.
120    * get_fixed_initial_number_of_hydrogen(min_h, odd_even)
121        Gets the fixed initial number of hydrogen atoms.
122    * calc_mz(datadict, class_mass=0)
123        Calculates the mass-to-charge ratio (m/z) of a molecular formula.
124    * calc_dbe_class(datadict)
125        Calculates the double bond equivalent (DBE) of a molecular formula.
126    * populate_combinations(classe_tuple, settings)
127        Populates the combinations.
128    * get_or_add(SomeClass, kw)
129        Gets or adds a class.
130    * get_mol_formulas(odd_even_tag, classe_tuple, settings)
131        Gets the molecular formulas.
132    * get_h_odd_or_even(class_dict)
133        Gets the hydrogen odd or even.
134    * get_total_halogen_atoms(class_dict)
135        Gets the total number of halogen atoms.
136    * get_total_hetero_valence(class_dict)
137        Gets the total valence of heteroatoms other than N, F, Cl, and Br
138    """
139
140    def __init__(self, sql_db=None):
141        if not sql_db:
142            self.sql_db = MolForm_SQL()
143        else:
144            self.sql_db = sql_db
145
146    def cProfile_worker(self, args):
147        """cProfile worker for the get_mol_formulas function"""
148        cProfile.runctx(
149            "self.get_mol_formulas(*args)",
150            globals(),
151            locals(),
152            "mf_database_cprofile.prof",
153        )
154
155    def check_database_get_class_list(self, molecular_search_settings):
156        """check if the database has all the classes, if not create the missing classes
157
158        Parameters
159        ----------
160        molecular_search_settings : object
161            An object containing user-defined settings.
162
163        Returns
164        -------
165        list
166            list of tuples with the class name and the class dictionary
167        """
168        all_class_to_create = []
169
170        classes_dict = self.get_classes_in_order(molecular_search_settings)
171
172        class_str_set = set(classes_dict.keys())
173
174        existing_classes_objs = self.sql_db.session.query(HeteroAtoms).distinct().all()
175
176        existing_classes_str = set([classe.name for classe in existing_classes_objs])
177
178        self.len_existing_classes = len(existing_classes_str)
179
180        class_to_create = class_str_set - existing_classes_str
181
182        class_count = len(existing_classes_objs)
183
184        data_classes = list()
185        for index, class_str in enumerate(class_to_create):
186            class_dict = classes_dict.get(class_str)
187            halogen_count = self.get_total_halogen_atoms(class_dict)
188            data_classes.append(
189                {
190                    "name": class_str,
191                    "id": class_count + index + 1,
192                    "halogensCount": halogen_count,
193                }
194            )
195
196        # data_classes = [{"name":class_str, "id":class_count+ index + 1} for index, class_str in enumerate(class_to_create)]
197
198        if data_classes:
199            list_insert_chunks = chunks(data_classes, self.sql_db.chunks_count)
200            for insert_chunk in list_insert_chunks:
201                insert_query = HeteroAtoms.__table__.insert().values(insert_chunk)
202                self.sql_db.session.execute(insert_query)
203
204        for index, class_str in enumerate(class_to_create):
205            class_tuple = (
206                class_str,
207                classes_dict.get(class_str),
208                class_count + index + 1,
209            )
210
211            all_class_to_create.append(class_tuple)
212
213        return (
214            [(c_s, c_d) for c_s, c_d in classes_dict.items()],
215            all_class_to_create,
216            existing_classes_objs,
217        )
218
219    def get_carbonsHydrogens(self, settings, odd_even):
220        """Retrieve carbon and hydrogen atoms from the molecular lookup table based on user-defined settings.
221
222        Parameters
223        ----------
224        settings : object
225             An object containing user-defined settings.
226        odd_even : str
227            A string indicating whether to retrieve even or odd hydrogen atoms.
228
229        Returns
230        -------
231        list
232            A list of CarbonHydrogen objects that satisfy the specified conditions.
233        """
234        operator = "==" if odd_even == "even" else "!="
235        usedAtoms = settings.usedAtoms
236        user_min_c, user_max_c = usedAtoms.get("C")
237        user_min_h, user_max_h = usedAtoms.get("H")
238
239        return eval(
240            "self.sql_db.session.query(CarbonHydrogen).filter("
241            "CarbonHydrogen.C >= user_min_c,"
242            "CarbonHydrogen.H >= user_min_h,"
243            "CarbonHydrogen.C <= user_max_c,"
244            "CarbonHydrogen.H <= user_max_h,"
245            "CarbonHydrogen.H % 2" + operator + "0).all()"
246        )
247
248    def add_carbonsHydrogens(self, settings, existing_classes_objs):
249        """Add carbon and hydrogen atoms to the molecular lookup table based on user-defined settings.
250
251        Parameters
252        ----------
253        settings : object
254            An object containing user-defined settings.
255        existing_classes_objs : list
256            A list of HeteroAtoms objects.
257        """
258        usedAtoms = settings.usedAtoms
259
260        user_min_c, user_max_c = usedAtoms.get("C")
261        user_min_h, user_max_h = usedAtoms.get("H")
262
263        query_obj = self.sql_db.session.query(
264            func.max(CarbonHydrogen.C).label("max_c"),
265            func.min(CarbonHydrogen.C).label("min_c"),
266            func.max(CarbonHydrogen.H).label("max_h"),
267            func.min(CarbonHydrogen.H).label("min_h"),
268        )
269
270        database = query_obj.first()
271        if (
272            database.max_c == user_max_c
273            and database.min_c == user_min_c
274            and database.max_h == user_max_h
275            and database.min_h == user_min_h
276        ):
277            # all data is already available at the database
278            pass
279
280        else:
281            current_count = self.sql_db.session.query(CarbonHydrogen.C).count()
282
283            databaseCarbonHydrogen = self.sql_db.session.query(CarbonHydrogen).all()
284
285            userCarbon = set(range(user_min_c, user_max_c + 1))
286            userHydrogen = set(range(user_min_h, user_max_h + 1))
287
288            carbon_hydrogen_objs_database = {}
289            for obj in databaseCarbonHydrogen:
290                str_data = "C:{},H:{}".format(obj.C, obj.H)
291                carbon_hydrogen_objs_database[str_data] = str_data
292
293            carbon_hydrogen_objs_to_create = {"even": {}, "odd": {}}
294
295            list_ch_obj_to_add = list()
296            i = 0
297            for comb in itertools.product(userCarbon, userHydrogen):
298                C = comb[0]
299                H = comb[1]
300                data = {
301                    "C": C,
302                    "H": H,
303                }
304
305                data_insert = {
306                    "C": C,
307                    "H": H,
308                }
309
310                str_data = "C:{},H:{}".format(C, H)
311
312                if str_data not in carbon_hydrogen_objs_database.keys():
313                    label = "even" if comb[1] % 2 == 0 else "odd"
314                    data["mass"] = (C * Atoms.atomic_masses.get("C")) + (
315                        H * Atoms.atomic_masses.get("H")
316                    )
317                    data["dbe"] = C - (H / 2) + 1
318                    data["id"] = i + current_count + 1
319                    data_insert["id"] = i + current_count + 1
320                    i = i + 1
321                    carbon_hydrogen_objs_to_create[label][str_data] = data
322
323                    list_ch_obj_to_add.append(data_insert)
324
325            if list_ch_obj_to_add:
326                # insert carbon hydrogen objs
327                list_insert_chunks = chunks(
328                    list_ch_obj_to_add, self.sql_db.chunks_count
329                )
330                for insert_chunk in list_insert_chunks:
331                    insert_query = CarbonHydrogen.__table__.insert().values(
332                        insert_chunk
333                    )
334                    self.sql_db.session.execute(insert_query)
335                self.sql_db.session.commit()
336
337                list_molecular_form = list()
338                for classe_obj in existing_classes_objs:
339                    classe_dict = classe_obj.to_dict()
340                    classe_mass = self.calc_mz(classe_dict)
341                    classe_dbe = self.calc_dbe_class(classe_dict)
342
343                    odd_even_label = self.get_h_odd_or_even(classe_dict)
344
345                    ch_datalist = carbon_hydrogen_objs_to_create.get(
346                        odd_even_label
347                    ).values()
348
349                    for ch_dict in ch_datalist:
350                        mass = ch_dict.get("mass") + classe_mass
351                        dbe = ch_dict.get("dbe") + classe_dbe
352
353                        if settings.min_mz <= mass <= settings.max_mz:
354                            if settings.min_dbe <= dbe <= settings.max_dbe:
355                                list_molecular_form.append(
356                                    {
357                                        "heteroAtoms_id": classe_obj.id,
358                                        "carbonHydrogen_id": ch_dict.get("id"),
359                                        "mass": mass,
360                                        "DBE": dbe,
361                                    }
362                                )
363
364                list_insert_chunks = chunks(
365                    list_molecular_form, self.sql_db.chunks_count
366                )
367                for insert_chunk in list_insert_chunks:
368                    insert_query = MolecularFormulaLink.__table__.insert().values(
369                        insert_chunk
370                    )
371                    self.sql_db.session.execute(insert_query)
372                self.sql_db.session.commit()
373
374    @timeit(print_time=True)
375    def runworker(self, molecular_search_settings, **kwargs):
376        """Run the molecular formula lookup table worker.
377
378        Parameters
379        ----------
380        molecular_search_settings : object
381            An object containing user-defined settings.
382        kwargs : dict
383            A dictionary of keyword arguments.
384            Most notably, the print_time argument which is passed to the timeit decorator.
385
386        Returns
387        -------
388        list
389            A list of tuples with the class name and the class dictionary.
390
391
392        """
393        verbose = molecular_search_settings.verbose_processing
394
395        classes_list, class_to_create, existing_classes_objs = (
396            self.check_database_get_class_list(molecular_search_settings)
397        )
398
399        settings = MolecularLookupDictSettings()
400        settings.usedAtoms = deepcopy(molecular_search_settings.usedAtoms)
401        settings.url_database = molecular_search_settings.url_database
402        settings.db_jobs = molecular_search_settings.db_jobs
403
404        self.add_carbonsHydrogens(settings, existing_classes_objs)
405
406        if class_to_create:
407            settings = MolecularLookupDictSettings()
408            settings.usedAtoms = deepcopy(molecular_search_settings.usedAtoms)
409            settings.url_database = molecular_search_settings.url_database
410            settings.db_jobs = molecular_search_settings.db_jobs
411
412            self.sql_db.session.commit()
413            odd_ch_obj = self.get_carbonsHydrogens(settings, "odd")
414            self.odd_ch_id = [obj.id for obj in odd_ch_obj]
415            self.odd_ch_dict = [{"C": obj.C, "H": obj.H} for obj in odd_ch_obj]
416            self.odd_ch_mass = [obj.mass for obj in odd_ch_obj]
417            self.odd_ch_dbe = [obj.dbe for obj in odd_ch_obj]
418
419            even_ch_obj = self.get_carbonsHydrogens(settings, "even")
420            self.even_ch_id = [obj.id for obj in even_ch_obj]
421            self.even_ch_dict = [{"C": obj.C, "H": obj.H} for obj in even_ch_obj]
422            self.even_ch_mass = [obj.mass for obj in even_ch_obj]
423            self.even_ch_dbe = [obj.dbe for obj in even_ch_obj]
424
425            all_results = list()
426            for class_tuple in tqdm(class_to_create, disable=not verbose):
427                results = self.populate_combinations(class_tuple, settings)
428                all_results.extend(results)
429                if settings.db_jobs == 1:
430                    # if len(all_results) >= self.sql_db.chunks_count:
431                    list_insert_chunks = list(chunks(results, self.sql_db.chunks_count))
432                    for chunk in list_insert_chunks:
433                        insert_query = MolecularFormulaLink.__table__.insert().values(
434                            chunk
435                        )
436                        self.sql_db.session.execute(insert_query)
437                    # all_results = list()
438            self.sql_db.session.commit()
439            # each chunk takes ~600Mb of memory, so if using 8 processes the total free memory needs to be 5GB
440            if settings.db_jobs > 1:
441                list_insert_chunks = list(chunks(all_results, self.sql_db.chunks_count))
442                if verbose:
443                    print(
444                        "Started database insert using {} iterations for a total of {} rows".format(
445                            len(list_insert_chunks), len(all_results)
446                        )
447                    )
448                worker_args = [
449                    (chunk, settings.url_database) for chunk in list_insert_chunks
450                ]
451                with spawn_pool(settings.db_jobs) as p:
452                    for class_list in tqdm(
453                        p.imap_unordered(insert_database_worker, worker_args),
454                        disable=not verbose,
455                    ):
456                        pass
457
458        return classes_list
459
460    def get_classes_in_order(self, molecular_search_settings):
461        """Get the classes in order
462
463        Parameters
464        ----------
465        molecular_search_settings : object
466            An object containing user-defined settings.
467
468        Returns
469        -------
470        dict
471            A dictionary of classes in order.
472            structure is  ('HC', {'HC': 1})
473        """
474        validate_used_atoms_keys(molecular_search_settings.usedAtoms)
475
476        usedAtoms = deepcopy(molecular_search_settings.usedAtoms)
477
478        usedAtoms.pop("C")
479        usedAtoms.pop("H")
480
481        min_n, max_n = usedAtoms.get("N") if usedAtoms.get("N") else (0, 0)
482        min_o, max_o = usedAtoms.get("O") if usedAtoms.get("O") else (0, 0)
483        min_s, max_s = usedAtoms.get("S") if usedAtoms.get("S") else (0, 0)
484        min_p, max_p = usedAtoms.get("P") if usedAtoms.get("P") else (0, 0)
485
486        possible_n = [n for n in range(min_n, max_n + 1)]
487        possible_o = [o for o in range(min_o, max_o + 1)]
488        possible_s = [s for s in range(min_s, max_s + 1)]
489        possible_p = [p for p in range(min_p, max_p + 1)]
490
491        atoms_in_order = ["N", "O", "S", "P"]
492
493        classe_in_order = {}
494
495        all_atoms_tuples = itertools.product(
496            possible_n, possible_o, possible_s, possible_p
497        )
498
499        for atom in atoms_in_order:
500            usedAtoms.pop(atom, None)
501
502        for selected_atom, min_max_tuple in usedAtoms.items():
503            min_x = min_max_tuple[0]
504            max_x = min_max_tuple[1]
505
506            possible_x = [x for x in range(min_x, max_x + 1)]
507
508            all_atoms_tuples = itertools.product(all_atoms_tuples, possible_x)
509            all_atoms_tuples = [
510                all_atoms_combined[0] + (all_atoms_combined[1],)
511                for all_atoms_combined in all_atoms_tuples
512            ]
513            atoms_in_order.append(selected_atom)
514
515        for all_atoms_tuple in all_atoms_tuples:
516            classe_str = ""
517            classe_dict = {}
518
519            for each_atoms_index, atom_number in enumerate(all_atoms_tuple):
520                if atom_number != 0:
521                    classe_dict[atoms_in_order[each_atoms_index]] = atom_number
522
523            if not classe_dict:
524                classe_in_order["HC"] = {"HC": ""}
525                continue
526
527            classe_str = json.dumps(classe_dict)
528
529            if len(classe_str) > 0:
530                classe_in_order[classe_str] = classe_dict
531
532        classe_in_order_dict = self.sort_classes(atoms_in_order, classe_in_order)
533
534        return classe_in_order_dict
535
536    @staticmethod
537    def sort_classes(atoms_in_order, combination_dict) -> Dict[str, Dict[str, int]]:
538        """Sort the classes in order
539
540        Parameters
541        ----------
542        atoms_in_order : list
543            A list of atoms in order.
544        combination_dict : dict
545            A dictionary of classes.
546
547        Returns
548        -------
549        dict
550            A dictionary of classes in order.
551        """
552        # ensures atoms are always in the order defined at atoms_in_order list
553        join_dict_classes = dict()
554        atoms_in_order = ["N", "S", "P", "O"] + atoms_in_order[4:] + ["HC"]
555
556        sort_method = lambda atoms_keys: [atoms_in_order.index(atoms_keys)]
557        for class_str, class_dict in combination_dict.items():
558            sorted_dict_keys = sorted(class_dict, key=sort_method)
559            class_dict = {atom: class_dict[atom] for atom in sorted_dict_keys}
560            class_str = json.dumps(class_dict)
561            # using json for the new database, class
562            # class_str = ' '.join([atom + str(class_dict[atom]) for atom in sorted_dict_keys])
563            join_dict_classes[class_str] = class_dict
564
565        return join_dict_classes
566
567    @staticmethod
568    def get_fixed_initial_number_of_hydrogen(min_h, odd_even):
569        """Get the fixed initial number of hydrogen atoms
570
571        Parameters
572        ----------
573        min_h : int
574            The minimum number of hydrogen atoms.
575        odd_even : str
576            A string indicating whether to retrieve even or odd hydrogen atoms.
577        """
578        remaining_h = min_h % 2
579
580        if odd_even == "even":
581            if remaining_h == 0:
582                return remaining_h
583
584            else:
585                return remaining_h + 1
586
587        else:
588            if remaining_h == 0:
589                return remaining_h + 1
590
591            else:
592                return remaining_h
593
594    def calc_mz(self, datadict, class_mass=0):
595        """Calculate the mass-to-charge ratio (m/z) of a molecular formula.
596
597        Parameters
598        ----------
599        datadict : dict
600            A dictionary of classes.
601        class_mass : int
602            The mass of the class.
603
604        Returns
605        -------
606        float
607            The mass-to-charge ratio (m/z) of a molecular formula.
608        """
609        mass = class_mass
610
611        for atom in datadict.keys():
612            if atom != "HC":
613                mass = mass + Atoms.atomic_masses[atom] * datadict.get(atom)
614
615        return mass
616
617    def calc_dbe_class(self, datadict):
618        """Calculate the double bond equivalent (DBE) of a molecular formula.
619
620        Parameters
621        ----------
622        datadict : dict
623            A dictionary of classes.
624
625        Returns
626        -------
627        float
628            The double bond equivalent (DBE) of a molecular formula.
629        """
630        init_dbe = 0
631        for atom in datadict.keys():
632            if atom == "HC":
633                continue
634
635            n_atom = int(datadict.get(atom))
636
637            clean_atom = "".join([i for i in atom if not i.isdigit()])
638
639            valencia = MSParameters.molecular_search.used_atom_valences.get(clean_atom)
640
641            if type(valencia) is tuple:
642                valencia = valencia[0]
643            if valencia > 0:
644                # print atom, valencia, n_atom, init_dbe
645                init_dbe = init_dbe + (n_atom * (valencia - 2))
646            else:
647                continue
648
649        return 0.5 * init_dbe
650
651    def populate_combinations(self, classe_tuple, settings):
652        """Populate the combinations
653
654        Parameters
655        ----------
656        classe_tuple : tuple
657            A tuple containing the class name, the class dictionary, and the class ID.
658        settings : object
659            An object containing user-defined settings.
660
661        Returns
662        -------
663        list
664            A list of molecular formula data dictionaries.
665        """
666        class_dict = classe_tuple[1]
667        odd_or_even = self.get_h_odd_or_even(class_dict)
668
669        return self.get_mol_formulas(odd_or_even, classe_tuple, settings)
670
671    def get_or_add(self, SomeClass, kw):
672        """Get or add a class
673
674        Parameters
675        ----------
676        SomeClass : object
677            A class object.
678        kw : dict
679            A dictionary of classes.
680
681        Returns
682        -------
683        object
684            A class object.
685        """
686        obj = self.sql_db.session.query(SomeClass).filter_by(**kw).first()
687        if not obj:
688            obj = SomeClass(**kw)
689        return obj
690
691    def get_mol_formulas(self, odd_even_tag, classe_tuple, settings):
692        """Get the molecular formulas
693
694        Parameters
695        ----------
696        odd_even_tag : str
697            A string indicating whether to retrieve even or odd hydrogen atoms.
698        classe_tuple : tuple
699
700        settings : object
701            An object containing user-defined settings.
702
703        Returns
704        -------
705        list
706            A list of molecular formula data dictionaries.
707
708        """
709        class_str = classe_tuple[0]
710        class_dict = classe_tuple[1]
711        classe_id = classe_tuple[2]
712
713        results = list()
714
715        if "HC" in class_dict:
716            del class_dict["HC"]
717
718        class_dbe = self.calc_dbe_class(class_dict)
719        class_mass = self.calc_mz(class_dict)
720
721        carbonHydrogen_mass = (
722            self.odd_ch_mass if odd_even_tag == "odd" else self.even_ch_mass
723        )
724        carbonHydrogen_dbe = (
725            self.odd_ch_dbe if odd_even_tag == "odd" else self.even_ch_dbe
726        )
727        carbonHydrogen_id = self.odd_ch_id if odd_even_tag == "odd" else self.even_ch_id
728
729        for index, carbonHydrogen_obj in enumerate(carbonHydrogen_id):
730            mass = carbonHydrogen_mass[index] + class_mass
731            dbe = carbonHydrogen_dbe[index] + class_dbe
732
733            if settings.min_mz <= mass <= settings.max_mz:
734                if settings.min_dbe <= dbe <= settings.max_dbe:
735                    molecularFormula = {
736                        "heteroAtoms_id": classe_id,
737                        "carbonHydrogen_id": carbonHydrogen_id[index],
738                        "mass": mass,
739                        "DBE": dbe,
740                    }
741
742                    results.append(molecularFormula)
743
744        return results
745
746    def get_h_odd_or_even(self, class_dict):
747        """Get the hydrogen odd or even
748
749        Parameters
750        ----------
751        class_dict : dict
752            A dictionary of classes.
753
754        Returns
755        -------
756        str
757            A string indicating whether to retrieve even or odd hydrogen atoms.
758        """
759
760        HAS_NITROGEN = "N" in class_dict.keys()
761
762        number_of_halogen = self.get_total_halogen_atoms(class_dict)
763        number_of_hetero = self.get_total_heteroatoms(class_dict)
764
765        if number_of_halogen > 0:
766            HAS_HALOGEN = True
767
768        else:
769            HAS_HALOGEN = False
770
771        if HAS_HALOGEN:
772            remaining_halogen = number_of_halogen % 2
773
774        else:
775            remaining_halogen = 0
776
777        if number_of_hetero > 0:
778            HAS_OTHER_HETERO = True
779
780            total_hetero_valence = self.get_total_hetero_valence(class_dict)
781
782        else:
783            HAS_OTHER_HETERO = False
784
785            total_hetero_valence = 0
786
787        if HAS_OTHER_HETERO:
788            remaining_hetero_valence = total_hetero_valence % 2
789
790        else:
791            remaining_hetero_valence = 0
792
793        if HAS_NITROGEN and not HAS_OTHER_HETERO:
794            number_of_n = class_dict.get("N")
795            remaining_n = number_of_n % 2
796
797        elif HAS_NITROGEN and HAS_OTHER_HETERO:
798            number_of_n = class_dict.get("N")
799            remaining_n = (number_of_n + remaining_hetero_valence) % 2
800
801        elif HAS_OTHER_HETERO and not HAS_NITROGEN:
802            remaining_n = remaining_hetero_valence
803
804        else:
805            remaining_n = -1
806
807        if remaining_n > 0.0:
808            if HAS_NITROGEN or HAS_OTHER_HETERO:
809                if HAS_HALOGEN:
810                    if remaining_halogen == 0:
811                        return "odd"
812                    else:
813                        return "even"
814
815                else:
816                    return "odd"
817
818        elif remaining_n == 0.0:
819            if HAS_NITROGEN or HAS_OTHER_HETERO:
820                if HAS_HALOGEN:
821                    if remaining_halogen == 0:
822                        return "even"
823                    else:
824                        return "odd"
825
826                else:
827                    return "even"
828
829        else:
830            if HAS_HALOGEN:
831                if remaining_halogen == 0:
832                    return "even"
833                else:
834                    return "odd"
835
836            else:
837                return "even"
838
839    @staticmethod
840    def get_total_heteroatoms(class_dict):
841        """Get the total number of heteroatoms other than N, F, Cl, Br
842
843        Parameters
844        ----------
845        class_dict : dict
846            A dictionary of classes.
847
848        Returns
849        -------
850        int
851            The total number of heteroatoms.
852        """
853
854        total_number = 0
855
856        for atom in class_dict.keys():
857            if atom not in ["HC", "C", "H", "O", "N", "F", "Cl", "Br"]:
858                total_number = total_number + class_dict.get(atom)
859
860        return total_number
861
862    @staticmethod
863    def get_total_hetero_valence(class_dict):
864        """Get the total valence of heteroatoms other than N, F, Cl, Br
865
866        Parameters
867        ----------
868        class_dict : dict
869            A dictionary of classes.
870
871        Returns
872        -------
873        int
874            The total heteroatom valence.
875        """
876        total_valence = 0
877
878        for atom in class_dict.keys():
879            if atom not in ["HC", "C", "H", "O", "N", "F", "Cl", "Br"]:
880                clean_atom = "".join([i for i in atom if not i.isdigit()])
881
882                atom_valence = MSParameters.molecular_search.used_atom_valences.get(
883                    clean_atom
884                )
885
886                if type(atom_valence) is tuple:
887                    atom_valence = atom_valence[0]
888
889                n_atom = int(class_dict.get(atom))
890
891                n_atom_valence = atom_valence * n_atom
892
893                total_valence = total_valence + n_atom_valence
894
895        return total_valence
896
897    @staticmethod
898    def get_total_halogen_atoms(class_dict):
899        """Get the total number of halogen atoms
900
901        Parameters
902        ----------
903        class_dict : dict
904            A dictionary of classes.
905
906        Returns
907        -------
908        int
909            The total number of halogen atoms.
910        """
911        atoms = ["F", "Cl", "Br"]
912
913        total_number = 0
914
915        for atom in atoms:
916            if atom in class_dict.keys():
917                total_number = total_number + class_dict.get(atom)
918
919        return total_number

A class for generating molecular formula combinations.

Parameters
  • molecular_search_settings (object): An object containing user-defined settings.
Attributes
  • sql_db (MolForm_SQL): The SQLite database object.
  • len_existing_classes (int): The number of existing classes in the SQLite database.
  • odd_ch_id (list): A list of odd carbon and hydrogen atom IDs.
  • odd_ch_dict (list): A list of odd carbon and hydrogen atom dictionaries.
  • odd_ch_mass (list): A list of odd carbon and hydrogen atom masses.
  • odd_ch_dbe (list): A list of odd carbon and hydrogen atom double bond equivalents.
  • even_ch_id (list): A list of even carbon and hydrogen atom IDs.
  • even_ch_dict (list): A list of even carbon and hydrogen atom dictionaries.
  • even_ch_mass (list): A list of even carbon and hydrogen atom masses.
  • even_ch_dbe (list): A list of even carbon and hydrogen atom double bond equivalents.
Methods
  • cProfile_worker(args) A cProfile worker for the get_mol_formulas function.
  • check_database_get_class_list(molecular_search_settings) Checks if the database has all the classes, if not create the missing classes.
  • get_carbonsHydrogens(settings, odd_even) Retrieves carbon and hydrogen atoms from the molecular lookup table based on user-defined settings.
  • add_carbonsHydrogens(settings, existing_classes_objs) Adds carbon and hydrogen atoms to the molecular lookup table based on user-defined settings.
  • runworker(molecular_search_settings) Runs the molecular formula lookup table worker.
  • get_classes_in_order(molecular_search_settings) Gets the classes in order.
  • sort_classes(atoms_in_order, combination_dict) Sorts the classes in order.
  • get_fixed_initial_number_of_hydrogen(min_h, odd_even) Gets the fixed initial number of hydrogen atoms.
  • calc_mz(datadict, class_mass=0) Calculates the mass-to-charge ratio (m/z) of a molecular formula.
  • calc_dbe_class(datadict) Calculates the double bond equivalent (DBE) of a molecular formula.
  • populate_combinations(classe_tuple, settings) Populates the combinations.
  • get_or_add(SomeClass, kw) Gets or adds a class.
  • get_mol_formulas(odd_even_tag, classe_tuple, settings) Gets the molecular formulas.
  • get_h_odd_or_even(class_dict) Gets the hydrogen odd or even.
  • get_total_halogen_atoms(class_dict) Gets the total number of halogen atoms.
  • get_total_hetero_valence(class_dict) Gets the total valence of heteroatoms other than N, F, Cl, and Br
MolecularCombinations(sql_db=None)
140    def __init__(self, sql_db=None):
141        if not sql_db:
142            self.sql_db = MolForm_SQL()
143        else:
144            self.sql_db = sql_db
def cProfile_worker(self, args):
146    def cProfile_worker(self, args):
147        """cProfile worker for the get_mol_formulas function"""
148        cProfile.runctx(
149            "self.get_mol_formulas(*args)",
150            globals(),
151            locals(),
152            "mf_database_cprofile.prof",
153        )

cProfile worker for the get_mol_formulas function

def check_database_get_class_list(self, molecular_search_settings):
155    def check_database_get_class_list(self, molecular_search_settings):
156        """check if the database has all the classes, if not create the missing classes
157
158        Parameters
159        ----------
160        molecular_search_settings : object
161            An object containing user-defined settings.
162
163        Returns
164        -------
165        list
166            list of tuples with the class name and the class dictionary
167        """
168        all_class_to_create = []
169
170        classes_dict = self.get_classes_in_order(molecular_search_settings)
171
172        class_str_set = set(classes_dict.keys())
173
174        existing_classes_objs = self.sql_db.session.query(HeteroAtoms).distinct().all()
175
176        existing_classes_str = set([classe.name for classe in existing_classes_objs])
177
178        self.len_existing_classes = len(existing_classes_str)
179
180        class_to_create = class_str_set - existing_classes_str
181
182        class_count = len(existing_classes_objs)
183
184        data_classes = list()
185        for index, class_str in enumerate(class_to_create):
186            class_dict = classes_dict.get(class_str)
187            halogen_count = self.get_total_halogen_atoms(class_dict)
188            data_classes.append(
189                {
190                    "name": class_str,
191                    "id": class_count + index + 1,
192                    "halogensCount": halogen_count,
193                }
194            )
195
196        # data_classes = [{"name":class_str, "id":class_count+ index + 1} for index, class_str in enumerate(class_to_create)]
197
198        if data_classes:
199            list_insert_chunks = chunks(data_classes, self.sql_db.chunks_count)
200            for insert_chunk in list_insert_chunks:
201                insert_query = HeteroAtoms.__table__.insert().values(insert_chunk)
202                self.sql_db.session.execute(insert_query)
203
204        for index, class_str in enumerate(class_to_create):
205            class_tuple = (
206                class_str,
207                classes_dict.get(class_str),
208                class_count + index + 1,
209            )
210
211            all_class_to_create.append(class_tuple)
212
213        return (
214            [(c_s, c_d) for c_s, c_d in classes_dict.items()],
215            all_class_to_create,
216            existing_classes_objs,
217        )

check if the database has all the classes, if not create the missing classes

Parameters
  • molecular_search_settings (object): An object containing user-defined settings.
Returns
  • list: list of tuples with the class name and the class dictionary
def get_carbonsHydrogens(self, settings, odd_even):
219    def get_carbonsHydrogens(self, settings, odd_even):
220        """Retrieve carbon and hydrogen atoms from the molecular lookup table based on user-defined settings.
221
222        Parameters
223        ----------
224        settings : object
225             An object containing user-defined settings.
226        odd_even : str
227            A string indicating whether to retrieve even or odd hydrogen atoms.
228
229        Returns
230        -------
231        list
232            A list of CarbonHydrogen objects that satisfy the specified conditions.
233        """
234        operator = "==" if odd_even == "even" else "!="
235        usedAtoms = settings.usedAtoms
236        user_min_c, user_max_c = usedAtoms.get("C")
237        user_min_h, user_max_h = usedAtoms.get("H")
238
239        return eval(
240            "self.sql_db.session.query(CarbonHydrogen).filter("
241            "CarbonHydrogen.C >= user_min_c,"
242            "CarbonHydrogen.H >= user_min_h,"
243            "CarbonHydrogen.C <= user_max_c,"
244            "CarbonHydrogen.H <= user_max_h,"
245            "CarbonHydrogen.H % 2" + operator + "0).all()"
246        )

Retrieve carbon and hydrogen atoms from the molecular lookup table based on user-defined settings.

Parameters
  • settings (object): An object containing user-defined settings.
  • odd_even (str): A string indicating whether to retrieve even or odd hydrogen atoms.
Returns
  • list: A list of CarbonHydrogen objects that satisfy the specified conditions.
def add_carbonsHydrogens(self, settings, existing_classes_objs):
248    def add_carbonsHydrogens(self, settings, existing_classes_objs):
249        """Add carbon and hydrogen atoms to the molecular lookup table based on user-defined settings.
250
251        Parameters
252        ----------
253        settings : object
254            An object containing user-defined settings.
255        existing_classes_objs : list
256            A list of HeteroAtoms objects.
257        """
258        usedAtoms = settings.usedAtoms
259
260        user_min_c, user_max_c = usedAtoms.get("C")
261        user_min_h, user_max_h = usedAtoms.get("H")
262
263        query_obj = self.sql_db.session.query(
264            func.max(CarbonHydrogen.C).label("max_c"),
265            func.min(CarbonHydrogen.C).label("min_c"),
266            func.max(CarbonHydrogen.H).label("max_h"),
267            func.min(CarbonHydrogen.H).label("min_h"),
268        )
269
270        database = query_obj.first()
271        if (
272            database.max_c == user_max_c
273            and database.min_c == user_min_c
274            and database.max_h == user_max_h
275            and database.min_h == user_min_h
276        ):
277            # all data is already available at the database
278            pass
279
280        else:
281            current_count = self.sql_db.session.query(CarbonHydrogen.C).count()
282
283            databaseCarbonHydrogen = self.sql_db.session.query(CarbonHydrogen).all()
284
285            userCarbon = set(range(user_min_c, user_max_c + 1))
286            userHydrogen = set(range(user_min_h, user_max_h + 1))
287
288            carbon_hydrogen_objs_database = {}
289            for obj in databaseCarbonHydrogen:
290                str_data = "C:{},H:{}".format(obj.C, obj.H)
291                carbon_hydrogen_objs_database[str_data] = str_data
292
293            carbon_hydrogen_objs_to_create = {"even": {}, "odd": {}}
294
295            list_ch_obj_to_add = list()
296            i = 0
297            for comb in itertools.product(userCarbon, userHydrogen):
298                C = comb[0]
299                H = comb[1]
300                data = {
301                    "C": C,
302                    "H": H,
303                }
304
305                data_insert = {
306                    "C": C,
307                    "H": H,
308                }
309
310                str_data = "C:{},H:{}".format(C, H)
311
312                if str_data not in carbon_hydrogen_objs_database.keys():
313                    label = "even" if comb[1] % 2 == 0 else "odd"
314                    data["mass"] = (C * Atoms.atomic_masses.get("C")) + (
315                        H * Atoms.atomic_masses.get("H")
316                    )
317                    data["dbe"] = C - (H / 2) + 1
318                    data["id"] = i + current_count + 1
319                    data_insert["id"] = i + current_count + 1
320                    i = i + 1
321                    carbon_hydrogen_objs_to_create[label][str_data] = data
322
323                    list_ch_obj_to_add.append(data_insert)
324
325            if list_ch_obj_to_add:
326                # insert carbon hydrogen objs
327                list_insert_chunks = chunks(
328                    list_ch_obj_to_add, self.sql_db.chunks_count
329                )
330                for insert_chunk in list_insert_chunks:
331                    insert_query = CarbonHydrogen.__table__.insert().values(
332                        insert_chunk
333                    )
334                    self.sql_db.session.execute(insert_query)
335                self.sql_db.session.commit()
336
337                list_molecular_form = list()
338                for classe_obj in existing_classes_objs:
339                    classe_dict = classe_obj.to_dict()
340                    classe_mass = self.calc_mz(classe_dict)
341                    classe_dbe = self.calc_dbe_class(classe_dict)
342
343                    odd_even_label = self.get_h_odd_or_even(classe_dict)
344
345                    ch_datalist = carbon_hydrogen_objs_to_create.get(
346                        odd_even_label
347                    ).values()
348
349                    for ch_dict in ch_datalist:
350                        mass = ch_dict.get("mass") + classe_mass
351                        dbe = ch_dict.get("dbe") + classe_dbe
352
353                        if settings.min_mz <= mass <= settings.max_mz:
354                            if settings.min_dbe <= dbe <= settings.max_dbe:
355                                list_molecular_form.append(
356                                    {
357                                        "heteroAtoms_id": classe_obj.id,
358                                        "carbonHydrogen_id": ch_dict.get("id"),
359                                        "mass": mass,
360                                        "DBE": dbe,
361                                    }
362                                )
363
364                list_insert_chunks = chunks(
365                    list_molecular_form, self.sql_db.chunks_count
366                )
367                for insert_chunk in list_insert_chunks:
368                    insert_query = MolecularFormulaLink.__table__.insert().values(
369                        insert_chunk
370                    )
371                    self.sql_db.session.execute(insert_query)
372                self.sql_db.session.commit()

Add carbon and hydrogen atoms to the molecular lookup table based on user-defined settings.

Parameters
  • settings (object): An object containing user-defined settings.
  • existing_classes_objs (list): A list of HeteroAtoms objects.
def runworker(*args, **kw):
39        def timed(*args, **kw):
40            # Extract print_time from kwargs if provided
41            local_print_time = kw.pop("print_time", print_time)
42            ts = time.time()
43            result = method(*args, **kw)
44            te = time.time()
45            if "log_time" in kw:
46                name = kw.get("log_name", method.__name__.upper())
47                kw["log_time"][name] = int((te - ts) * 1000)
48            elif local_print_time:
49                print("%r  %2.2f ms" % (method.__name__, (te - ts) * 1000))
50            return result

Run the molecular formula lookup table worker.

Parameters
  • molecular_search_settings (object): An object containing user-defined settings.
  • kwargs (dict): A dictionary of keyword arguments. Most notably, the print_time argument which is passed to the timeit decorator.
Returns
  • list: A list of tuples with the class name and the class dictionary.
def get_classes_in_order(self, molecular_search_settings):
460    def get_classes_in_order(self, molecular_search_settings):
461        """Get the classes in order
462
463        Parameters
464        ----------
465        molecular_search_settings : object
466            An object containing user-defined settings.
467
468        Returns
469        -------
470        dict
471            A dictionary of classes in order.
472            structure is  ('HC', {'HC': 1})
473        """
474        validate_used_atoms_keys(molecular_search_settings.usedAtoms)
475
476        usedAtoms = deepcopy(molecular_search_settings.usedAtoms)
477
478        usedAtoms.pop("C")
479        usedAtoms.pop("H")
480
481        min_n, max_n = usedAtoms.get("N") if usedAtoms.get("N") else (0, 0)
482        min_o, max_o = usedAtoms.get("O") if usedAtoms.get("O") else (0, 0)
483        min_s, max_s = usedAtoms.get("S") if usedAtoms.get("S") else (0, 0)
484        min_p, max_p = usedAtoms.get("P") if usedAtoms.get("P") else (0, 0)
485
486        possible_n = [n for n in range(min_n, max_n + 1)]
487        possible_o = [o for o in range(min_o, max_o + 1)]
488        possible_s = [s for s in range(min_s, max_s + 1)]
489        possible_p = [p for p in range(min_p, max_p + 1)]
490
491        atoms_in_order = ["N", "O", "S", "P"]
492
493        classe_in_order = {}
494
495        all_atoms_tuples = itertools.product(
496            possible_n, possible_o, possible_s, possible_p
497        )
498
499        for atom in atoms_in_order:
500            usedAtoms.pop(atom, None)
501
502        for selected_atom, min_max_tuple in usedAtoms.items():
503            min_x = min_max_tuple[0]
504            max_x = min_max_tuple[1]
505
506            possible_x = [x for x in range(min_x, max_x + 1)]
507
508            all_atoms_tuples = itertools.product(all_atoms_tuples, possible_x)
509            all_atoms_tuples = [
510                all_atoms_combined[0] + (all_atoms_combined[1],)
511                for all_atoms_combined in all_atoms_tuples
512            ]
513            atoms_in_order.append(selected_atom)
514
515        for all_atoms_tuple in all_atoms_tuples:
516            classe_str = ""
517            classe_dict = {}
518
519            for each_atoms_index, atom_number in enumerate(all_atoms_tuple):
520                if atom_number != 0:
521                    classe_dict[atoms_in_order[each_atoms_index]] = atom_number
522
523            if not classe_dict:
524                classe_in_order["HC"] = {"HC": ""}
525                continue
526
527            classe_str = json.dumps(classe_dict)
528
529            if len(classe_str) > 0:
530                classe_in_order[classe_str] = classe_dict
531
532        classe_in_order_dict = self.sort_classes(atoms_in_order, classe_in_order)
533
534        return classe_in_order_dict

Get the classes in order

Parameters
  • molecular_search_settings (object): An object containing user-defined settings.
Returns
  • dict: A dictionary of classes in order. structure is ('HC', {'HC': 1})
@staticmethod
def sort_classes(atoms_in_order, combination_dict) -> Dict[str, Dict[str, int]]:
536    @staticmethod
537    def sort_classes(atoms_in_order, combination_dict) -> Dict[str, Dict[str, int]]:
538        """Sort the classes in order
539
540        Parameters
541        ----------
542        atoms_in_order : list
543            A list of atoms in order.
544        combination_dict : dict
545            A dictionary of classes.
546
547        Returns
548        -------
549        dict
550            A dictionary of classes in order.
551        """
552        # ensures atoms are always in the order defined at atoms_in_order list
553        join_dict_classes = dict()
554        atoms_in_order = ["N", "S", "P", "O"] + atoms_in_order[4:] + ["HC"]
555
556        sort_method = lambda atoms_keys: [atoms_in_order.index(atoms_keys)]
557        for class_str, class_dict in combination_dict.items():
558            sorted_dict_keys = sorted(class_dict, key=sort_method)
559            class_dict = {atom: class_dict[atom] for atom in sorted_dict_keys}
560            class_str = json.dumps(class_dict)
561            # using json for the new database, class
562            # class_str = ' '.join([atom + str(class_dict[atom]) for atom in sorted_dict_keys])
563            join_dict_classes[class_str] = class_dict
564
565        return join_dict_classes

Sort the classes in order

Parameters
  • atoms_in_order (list): A list of atoms in order.
  • combination_dict (dict): A dictionary of classes.
Returns
  • dict: A dictionary of classes in order.
@staticmethod
def get_fixed_initial_number_of_hydrogen(min_h, odd_even):
567    @staticmethod
568    def get_fixed_initial_number_of_hydrogen(min_h, odd_even):
569        """Get the fixed initial number of hydrogen atoms
570
571        Parameters
572        ----------
573        min_h : int
574            The minimum number of hydrogen atoms.
575        odd_even : str
576            A string indicating whether to retrieve even or odd hydrogen atoms.
577        """
578        remaining_h = min_h % 2
579
580        if odd_even == "even":
581            if remaining_h == 0:
582                return remaining_h
583
584            else:
585                return remaining_h + 1
586
587        else:
588            if remaining_h == 0:
589                return remaining_h + 1
590
591            else:
592                return remaining_h

Get the fixed initial number of hydrogen atoms

Parameters
  • min_h (int): The minimum number of hydrogen atoms.
  • odd_even (str): A string indicating whether to retrieve even or odd hydrogen atoms.
def calc_mz(self, datadict, class_mass=0):
594    def calc_mz(self, datadict, class_mass=0):
595        """Calculate the mass-to-charge ratio (m/z) of a molecular formula.
596
597        Parameters
598        ----------
599        datadict : dict
600            A dictionary of classes.
601        class_mass : int
602            The mass of the class.
603
604        Returns
605        -------
606        float
607            The mass-to-charge ratio (m/z) of a molecular formula.
608        """
609        mass = class_mass
610
611        for atom in datadict.keys():
612            if atom != "HC":
613                mass = mass + Atoms.atomic_masses[atom] * datadict.get(atom)
614
615        return mass

Calculate the mass-to-charge ratio (m/z) of a molecular formula.

Parameters
  • datadict (dict): A dictionary of classes.
  • class_mass (int): The mass of the class.
Returns
  • float: The mass-to-charge ratio (m/z) of a molecular formula.
def calc_dbe_class(self, datadict):
617    def calc_dbe_class(self, datadict):
618        """Calculate the double bond equivalent (DBE) of a molecular formula.
619
620        Parameters
621        ----------
622        datadict : dict
623            A dictionary of classes.
624
625        Returns
626        -------
627        float
628            The double bond equivalent (DBE) of a molecular formula.
629        """
630        init_dbe = 0
631        for atom in datadict.keys():
632            if atom == "HC":
633                continue
634
635            n_atom = int(datadict.get(atom))
636
637            clean_atom = "".join([i for i in atom if not i.isdigit()])
638
639            valencia = MSParameters.molecular_search.used_atom_valences.get(clean_atom)
640
641            if type(valencia) is tuple:
642                valencia = valencia[0]
643            if valencia > 0:
644                # print atom, valencia, n_atom, init_dbe
645                init_dbe = init_dbe + (n_atom * (valencia - 2))
646            else:
647                continue
648
649        return 0.5 * init_dbe

Calculate the double bond equivalent (DBE) of a molecular formula.

Parameters
  • datadict (dict): A dictionary of classes.
Returns
  • float: The double bond equivalent (DBE) of a molecular formula.
def populate_combinations(self, classe_tuple, settings):
651    def populate_combinations(self, classe_tuple, settings):
652        """Populate the combinations
653
654        Parameters
655        ----------
656        classe_tuple : tuple
657            A tuple containing the class name, the class dictionary, and the class ID.
658        settings : object
659            An object containing user-defined settings.
660
661        Returns
662        -------
663        list
664            A list of molecular formula data dictionaries.
665        """
666        class_dict = classe_tuple[1]
667        odd_or_even = self.get_h_odd_or_even(class_dict)
668
669        return self.get_mol_formulas(odd_or_even, classe_tuple, settings)

Populate the combinations

Parameters
  • classe_tuple (tuple): A tuple containing the class name, the class dictionary, and the class ID.
  • settings (object): An object containing user-defined settings.
Returns
  • list: A list of molecular formula data dictionaries.
def get_or_add(self, SomeClass, kw):
671    def get_or_add(self, SomeClass, kw):
672        """Get or add a class
673
674        Parameters
675        ----------
676        SomeClass : object
677            A class object.
678        kw : dict
679            A dictionary of classes.
680
681        Returns
682        -------
683        object
684            A class object.
685        """
686        obj = self.sql_db.session.query(SomeClass).filter_by(**kw).first()
687        if not obj:
688            obj = SomeClass(**kw)
689        return obj

Get or add a class

Parameters
  • SomeClass (object): A class object.
  • kw (dict): A dictionary of classes.
Returns
  • object: A class object.
def get_mol_formulas(self, odd_even_tag, classe_tuple, settings):
691    def get_mol_formulas(self, odd_even_tag, classe_tuple, settings):
692        """Get the molecular formulas
693
694        Parameters
695        ----------
696        odd_even_tag : str
697            A string indicating whether to retrieve even or odd hydrogen atoms.
698        classe_tuple : tuple
699
700        settings : object
701            An object containing user-defined settings.
702
703        Returns
704        -------
705        list
706            A list of molecular formula data dictionaries.
707
708        """
709        class_str = classe_tuple[0]
710        class_dict = classe_tuple[1]
711        classe_id = classe_tuple[2]
712
713        results = list()
714
715        if "HC" in class_dict:
716            del class_dict["HC"]
717
718        class_dbe = self.calc_dbe_class(class_dict)
719        class_mass = self.calc_mz(class_dict)
720
721        carbonHydrogen_mass = (
722            self.odd_ch_mass if odd_even_tag == "odd" else self.even_ch_mass
723        )
724        carbonHydrogen_dbe = (
725            self.odd_ch_dbe if odd_even_tag == "odd" else self.even_ch_dbe
726        )
727        carbonHydrogen_id = self.odd_ch_id if odd_even_tag == "odd" else self.even_ch_id
728
729        for index, carbonHydrogen_obj in enumerate(carbonHydrogen_id):
730            mass = carbonHydrogen_mass[index] + class_mass
731            dbe = carbonHydrogen_dbe[index] + class_dbe
732
733            if settings.min_mz <= mass <= settings.max_mz:
734                if settings.min_dbe <= dbe <= settings.max_dbe:
735                    molecularFormula = {
736                        "heteroAtoms_id": classe_id,
737                        "carbonHydrogen_id": carbonHydrogen_id[index],
738                        "mass": mass,
739                        "DBE": dbe,
740                    }
741
742                    results.append(molecularFormula)
743
744        return results

Get the molecular formulas

Parameters
  • odd_even_tag (str): A string indicating whether to retrieve even or odd hydrogen atoms.
  • classe_tuple (tuple):

  • settings (object): An object containing user-defined settings.

Returns
  • list: A list of molecular formula data dictionaries.
def get_h_odd_or_even(self, class_dict):
746    def get_h_odd_or_even(self, class_dict):
747        """Get the hydrogen odd or even
748
749        Parameters
750        ----------
751        class_dict : dict
752            A dictionary of classes.
753
754        Returns
755        -------
756        str
757            A string indicating whether to retrieve even or odd hydrogen atoms.
758        """
759
760        HAS_NITROGEN = "N" in class_dict.keys()
761
762        number_of_halogen = self.get_total_halogen_atoms(class_dict)
763        number_of_hetero = self.get_total_heteroatoms(class_dict)
764
765        if number_of_halogen > 0:
766            HAS_HALOGEN = True
767
768        else:
769            HAS_HALOGEN = False
770
771        if HAS_HALOGEN:
772            remaining_halogen = number_of_halogen % 2
773
774        else:
775            remaining_halogen = 0
776
777        if number_of_hetero > 0:
778            HAS_OTHER_HETERO = True
779
780            total_hetero_valence = self.get_total_hetero_valence(class_dict)
781
782        else:
783            HAS_OTHER_HETERO = False
784
785            total_hetero_valence = 0
786
787        if HAS_OTHER_HETERO:
788            remaining_hetero_valence = total_hetero_valence % 2
789
790        else:
791            remaining_hetero_valence = 0
792
793        if HAS_NITROGEN and not HAS_OTHER_HETERO:
794            number_of_n = class_dict.get("N")
795            remaining_n = number_of_n % 2
796
797        elif HAS_NITROGEN and HAS_OTHER_HETERO:
798            number_of_n = class_dict.get("N")
799            remaining_n = (number_of_n + remaining_hetero_valence) % 2
800
801        elif HAS_OTHER_HETERO and not HAS_NITROGEN:
802            remaining_n = remaining_hetero_valence
803
804        else:
805            remaining_n = -1
806
807        if remaining_n > 0.0:
808            if HAS_NITROGEN or HAS_OTHER_HETERO:
809                if HAS_HALOGEN:
810                    if remaining_halogen == 0:
811                        return "odd"
812                    else:
813                        return "even"
814
815                else:
816                    return "odd"
817
818        elif remaining_n == 0.0:
819            if HAS_NITROGEN or HAS_OTHER_HETERO:
820                if HAS_HALOGEN:
821                    if remaining_halogen == 0:
822                        return "even"
823                    else:
824                        return "odd"
825
826                else:
827                    return "even"
828
829        else:
830            if HAS_HALOGEN:
831                if remaining_halogen == 0:
832                    return "even"
833                else:
834                    return "odd"
835
836            else:
837                return "even"

Get the hydrogen odd or even

Parameters
  • class_dict (dict): A dictionary of classes.
Returns
  • str: A string indicating whether to retrieve even or odd hydrogen atoms.
@staticmethod
def get_total_heteroatoms(class_dict):
839    @staticmethod
840    def get_total_heteroatoms(class_dict):
841        """Get the total number of heteroatoms other than N, F, Cl, Br
842
843        Parameters
844        ----------
845        class_dict : dict
846            A dictionary of classes.
847
848        Returns
849        -------
850        int
851            The total number of heteroatoms.
852        """
853
854        total_number = 0
855
856        for atom in class_dict.keys():
857            if atom not in ["HC", "C", "H", "O", "N", "F", "Cl", "Br"]:
858                total_number = total_number + class_dict.get(atom)
859
860        return total_number

Get the total number of heteroatoms other than N, F, Cl, Br

Parameters
  • class_dict (dict): A dictionary of classes.
Returns
  • int: The total number of heteroatoms.
@staticmethod
def get_total_hetero_valence(class_dict):
862    @staticmethod
863    def get_total_hetero_valence(class_dict):
864        """Get the total valence of heteroatoms other than N, F, Cl, Br
865
866        Parameters
867        ----------
868        class_dict : dict
869            A dictionary of classes.
870
871        Returns
872        -------
873        int
874            The total heteroatom valence.
875        """
876        total_valence = 0
877
878        for atom in class_dict.keys():
879            if atom not in ["HC", "C", "H", "O", "N", "F", "Cl", "Br"]:
880                clean_atom = "".join([i for i in atom if not i.isdigit()])
881
882                atom_valence = MSParameters.molecular_search.used_atom_valences.get(
883                    clean_atom
884                )
885
886                if type(atom_valence) is tuple:
887                    atom_valence = atom_valence[0]
888
889                n_atom = int(class_dict.get(atom))
890
891                n_atom_valence = atom_valence * n_atom
892
893                total_valence = total_valence + n_atom_valence
894
895        return total_valence

Get the total valence of heteroatoms other than N, F, Cl, Br

Parameters
  • class_dict (dict): A dictionary of classes.
Returns
  • int: The total heteroatom valence.
@staticmethod
def get_total_halogen_atoms(class_dict):
897    @staticmethod
898    def get_total_halogen_atoms(class_dict):
899        """Get the total number of halogen atoms
900
901        Parameters
902        ----------
903        class_dict : dict
904            A dictionary of classes.
905
906        Returns
907        -------
908        int
909            The total number of halogen atoms.
910        """
911        atoms = ["F", "Cl", "Br"]
912
913        total_number = 0
914
915        for atom in atoms:
916            if atom in class_dict.keys():
917                total_number = total_number + class_dict.get(atom)
918
919        return total_number

Get the total number of halogen atoms

Parameters
  • class_dict (dict): A dictionary of classes.
Returns
  • int: The total number of halogen atoms.