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
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.
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.
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
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
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
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.
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.
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.
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})
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.
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.
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.
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.
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.
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.
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.
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.
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.
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.
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.