Improvements to support building models from other annotations

Author: Christopher Henry

modelseedpy/core/annotationontology.py

@@ -452,4 +452,39 @@ def get_msgenome(self,prioritized_event_list=None,ontologies=None,merge_all=Fals
         )
         newgenome.annoont = self
         return newgenome
-        
+    
+    def get_events_from_priority_list(self,priority_list):
+        event_list = []
+        for item in priority_list:
+            selected_merge = None
+            for event in self.events:
+                if item == "all":
+                    if event.id not in event_list:
+                        event_list.append(event.id)
+                elif item == "RAST":
+                    if len(event.method) > 4 and event.method[0:4] == "RAST" and event.id not in event_list:
+                        event_list.append(event.id)
+                elif item == "Prokka":
+                    if len(event.method) > 6 and event.method[0:6] == "Prokka" and event.id not in event_list:
+                        event_list.append(event.id)
+                elif item == "DRAM":
+                    if len(event.method) > 4 and event.method[0:4] == "DRAM" and event.id not in event_list:
+                        event_list.append(event.id)
+                elif item == "GLM4EC":
+                    if len(event.method) > 6 and event.method[0:6] == "GLM4EC" and event.id not in event_list:
+                        event_list.append(event.id)
+                elif item == "PDB":
+                    if event.method == "KBAnnotationApps.PDBAnnotation" and event.ontology.id == "EC" and event.id not in event_list:
+                        event_list.append(event.id)
+                elif item == "SNEKMER":
+                    if len(event.method) > 7 and event.method[0:7] == "Snekmer" and event.id not in event_list:
+                        event_list.append(event.id)
+                elif item == "Import":
+                    if len(event.method) > 6 and event.method[0:6] == "Import" and event.id not in event_list:
+                        event_list.append(event.id)
+                elif item == "Merge":
+                    if len(event.method) > 5 and event.method[0:5] == "Merge" and event.id not in event_list:
+                        selected_merge = event.id
+            if selected_merge:
+                event_list.append(selected_merge)
+        return event_list

modelseedpy/core/msmodelutl.py

@@ -498,7 +498,7 @@ def get_atp_tests(self,core_template=None,atp_media_filename=None,recompute=Fals
             return self.atp_tests
         #Attempting to pull ATP tests from attributes
         if not recompute:
-            print("Getting tests from attributes")
+            logger.debug("Getting tests from attributes")
             atp_analysis = self.get_attributes("ATP_analysis",None)
             if atp_analysis:
                 if "tests" in atp_analysis:
@@ -948,7 +948,7 @@ def is_core(self,rxn):
         if not isinstance(rxn, str):
             rxn = rxn.id
         if "core_reactions" in self.get_attributes():
-            print("Using core reactions attribute!")
+            logger.debug("Using core reactions attribute!")
             if rxn in self.get_attributes("core_reactions"):
                 return True
             return False
@@ -1291,7 +1291,7 @@ def test_solution(self,solution,targets,medias,thresholds=[0.1],remove_unneeded_
         return unneeded
 
     def add_gapfilling(self, solution):
-        print("Adding gapfilling",str(solution))
+        logger.debug("Adding gapfilling:"+str(solution))
         self.integrated_gapfillings.append(solution)
 
     def create_kb_gapfilling_data(self, kbmodel, atpmedia_ws="94026"):
@@ -1439,7 +1439,7 @@ def test_single_condition(self, condition, apply_condition=True, model=None,repo
                 self.model.reactions.get_by_id(condition["objective"]).lower_bound = 0
             return False
         elif value <= condition["threshold"] and not condition["is_max_threshold"]:
-            print("Failed low:"+condition["media"].id+":"+str(new_objective)+";"+str(condition["threshold"]))
+            logger.debug("Failed low:"+condition["media"].id+":"+str(new_objective)+";"+str(condition["threshold"]))
             return False
         self.test_objective = new_objective
         logger.debug("Passed:"+condition["media"].id+":"+str(new_objective)+";"+str(condition["threshold"]))
@@ -1544,11 +1544,11 @@ def binary_expansion_test(self, reaction_list, condition, currmodel, depth=0,pos
         filtered_list = []
         # First run the full test
         if self.test_single_condition(condition,apply_condition=False,model=currmodel,rxn_list=reaction_list):
-            #print("Reaction set passed"," ".join(map(str, reaction_list)))
+            #logger.debug("Reaction set passed"," ".join(map(str, reaction_list)))
             return []
         # Check if input list contains only one reaction:
         if len(reaction_list) == 1:
-            #print("Failed:"+reaction_list[0][1]+reaction_list[0][0].id)
+            #logger.debug("Failed:"+reaction_list[0][1]+reaction_list[0][0].id)
             if reaction_list[0][1] == ">":
                 reaction_list[0].append(reaction_list[0][0].upper_bound)
                 reaction_list[0][0].upper_bound = 0
@@ -1561,7 +1561,7 @@ def binary_expansion_test(self, reaction_list, condition, currmodel, depth=0,pos
                 #Testing positive growth conditions
                 for pos_condition in positive_growth:
                     if not self.test_single_condition(pos_condition,apply_condition=True,model=currmodel):
-                        print("Does not pass positive growth tests:"+reaction_list[0][1]+reaction_list[0][0].id)
+                        logger.debug("Does not pass positive growth tests:"+reaction_list[0][1]+reaction_list[0][0].id)
                         success = False
                         break
                 #Restoring current test condition
@@ -1601,7 +1601,7 @@ def binary_expansion_test(self, reaction_list, condition, currmodel, depth=0,pos
         for item in new_filter:
             filtered_list.append(item)
         if self.breaking_reaction != None:
-            print("Ending early due to breaking reaction:"+self.breaking_reaction.id)
+            logger.debug("Ending early due to breaking reaction:"+self.breaking_reaction.id)
             return filtered_list
         # Submitting second half of reactions for testing - now only breaking reactions are removed from the first list
         for i, item in enumerate(reaction_list):
@@ -1667,14 +1667,14 @@ def reaction_expansion_test(
             scores = self.assign_reliability_scores_to_reactions()
             reaction_list = sorted(reaction_list, key=lambda x: scores[x[0].id][x[1]])
             for item in reaction_list:
-                print(item[0].id+":"+item[1]+":"+str(scores[item[0].id][item[1]]))
+                logger.debug(item[0].id+":"+item[1]+":"+str(scores[item[0].id][item[1]]))
         for condition in condition_list:
             logger.debug(f"testing condition {condition}")
             currmodel = self.model
             tic = time.perf_counter()
             new_filtered = []
             if not self.check_if_solution_exists(reaction_list, condition, currmodel):
-                print("No solution exists that passes tests for condition "+condition["media"].id)
+                logger.debug("No solution exists that passes tests for condition "+condition["media"].id)
                 return None
             with currmodel:
                 self.apply_test_condition(condition)
@@ -1691,13 +1691,13 @@ def reaction_expansion_test(
                             done = True
                         else:
                             #Remove breaking reaction from reaction_list
-                            print("Keeping breaking reaction:"+self.breaking_reaction.id)
+                            logger.debug("Keeping breaking reaction:"+self.breaking_reaction.id)
                             for i in range(len(reaction_list)):
                                 if reaction_list[i][0] == self.breaking_reaction:
                                     del reaction_list[i]
                                     break
                             if not self.check_if_solution_exists(reaction_list, condition, currmodel):
-                                print("No solution exists after retaining breaking reaction:"+self.breaking_reaction.id)
+                                logger.debug("No solution exists after retaining breaking reaction:"+self.breaking_reaction.id)
                                 return None
                             self.breaking_reaction = None
                 else:

modelseedpy/core/mstemplate.py

@@ -1515,7 +1515,16 @@ def remove_reactions(
                 associated_groups = self.get_associated_groups(reaction)
                 for group in associated_groups:
                     group.remove_members(reaction) """
-
+    
+    #*************************Curation Functions*************************
+    def auto_fix_protons(self):
+        for rxn in self.reactions:
+            mb = rxn.check_mass_balance()
+            if 'charge' in mb and mb.get('H') == mb.get('charge'):
+                print(f'auto fix charge for {rxn.id}')
+                rxn.add_metabolites({
+                    self.compcompounds.cpd00067_c: -1 * mb['charge']
+                })
 
 class MSTemplateBuilder:
     def __init__(

modelseedpy/fbapkg/gapfillingpkg.py

@@ -24,7 +24,7 @@
     logging.INFO
 )  # When debugging - set this to INFO then change needed messages below from DEBUG to INFO
 
-base_blacklist = {"rxn04656":"=","rxn07589":"=","rxn07588":"="}
+base_blacklist = {"rxn04656":"=","rxn07589":"<","rxn07588":"<"}
 zero_threshold = 1e-8