Create biological_context_language_orchestrator.py

biological_context_language_orchestrator.py

@@ -0,0 +1,142 @@
+from databaseengine import DatabaseEngine
+from fetch_fasta import FetchFasta_Uniprot
+from mutant_constructor import ConstructMutant_using_MutationString
+from get_3d_structure import Get3D_Structure_of_Protein
+from domain_detector import Domain_Detector
+from epitope_prediction import Predict_Epitopes
+import requests
+
+de=DatabaseEngine()
+
+
+
+class Biological_Context_Orchestrator():
+    
+    def __init__(self):
+        self.version="1.0"
+       
+    
+    def BCL_Orchestrator(self,project_id,id,planned_experiments,target):
+
+        """
+        Property of xForce 
+        Proprietary Implementation ,Locked Down Access
+        Author : JUNAID BASHIR
+        """
+        experiments=planned_experiments.get("experiments")
+        
+
+        if not isinstance(experiments, list):
+                print("Error: 'planned_experiments' must be a list of dictionaries.")
+                return
+    
+        Experiment_to_Implementation_Mapper={
+
+             #"retrieve_uniprot_sequence":FetchFasta_Uniprot,
+            "introduce_point_mutation":ConstructMutant_using_MutationString,
+            "domain_determination":Domain_Detector,
+            "predict_structure":Get3D_Structure_of_Protein,
+            "analyze_epitopes":Predict_Epitopes
+
+        }
+        EXPERIMENT_OUTPUTS={}
+
+        if "retrieve_uniprot_sequence" not in EXPERIMENT_OUTPUTS:
+        
+            EXPERIMENT_OUTPUTS["retrieve_uniprot_sequence"]=FetchFasta_Uniprot(gene_symbol=target)
+            de.Insert_IE({
+                    
+                    "bcl_id":id,
+                    "operation":"retrieve_uniprot_sequence",
+                    "output":EXPERIMENT_OUTPUTS["retrieve_uniprot_sequence"]
+                    
+            })
+        
+        for experiment_object in experiments:
+            
+    
+            #if experiment_object["operation"] in Experiment_to_Implementation_Mapper and experiment_object["operation"] == "retrieve_uniprot_sequence":
+                  
+            #EXPERIMENT_TO_RUN="retrieve_uniprot_sequence"
+            #EXPERIMENT_OUTPUTS["retrieve_uniprot_sequence"]=FetchFasta_Uniprot(gene_symbol=target)
+
+            #''' Database Ops Begin '''
+                
+            #de.Update_Status(id)
+
+            '''
+            de.Insert_IE({
+                    
+                    "bcl_id":id,
+                    "operation":"retrieve_uniprot_sequence",
+                    "output":EXPERIMENT_OUTPUTS["retrieve_uniprot_sequence"]
+                    
+            })
+            '''
+            #'''Database Ops End'''
+                
+                  
+            if experiment_object["operation"] in Experiment_to_Implementation_Mapper:
+                    
+                    EXPERIMENT_TO_RUN=Experiment_to_Implementation_Mapper[experiment_object["operation"]]
+                
+                    '''checking for the presence of depends key'''
+
+                    if "depends" in experiment_object and experiment_object["depends"]=="None":
+                        pass
+                
+                    if "depends" in experiment_object and  experiment_object["depends"] in EXPERIMENT_OUTPUTS:
+
+            
+                        if "wildtype_sequence" in experiment_object["biological_inputs"]:
+                                    
+                                experiment_object["biological_inputs"]["wildtype_sequence"]=EXPERIMENT_OUTPUTS[experiment_object["depends"]]
+
+                        if "sequence_for_structure" in experiment_object["biological_inputs"]:
+                                experiment_object["biological_inputs"]["filename"]=id
+                                
+                        else:     
+                                keys=list( experiment_object["biological_inputs"].keys())
+                                if keys:
+                                    experiment_object["biological_inputs"][keys[0]]=EXPERIMENT_OUTPUTS[experiment_object["depends"]]
+
+                                    
+
+                        EXPERIMENT_OUTPUTS[experiment_object["operation"]]=EXPERIMENT_TO_RUN(**experiment_object["biological_inputs"])
+
+
+                        ''' checking if predict structure operation is executed then replacing its url output with raw content'''
+                        if "predict_structure" in EXPERIMENT_OUTPUTS:
+                            predict_structure_output=EXPERIMENT_OUTPUTS["predict_structure"]["sequence_domain_section"]
+                            
+                            if predict_structure_output.startswith("https"):
+                                predict_structure_request=requests.get(predict_structure_output)
+                                if predict_structure_request.status_code == 200:
+                                    pdb_content = predict_structure_request.text
+                                    EXPERIMENT_OUTPUTS["predict_structure"]=pdb_content
+                                
+
+                        ''' Database Ops Begin '''
+                        #de.Update_Status(id)
+                        
+                        de.Insert_IE({
+                    
+                            "bcl_id":id,
+                            "operation":experiment_object["operation"],
+                            "output":EXPERIMENT_OUTPUTS[experiment_object["operation"]]
+                    
+                        })
+                        ''' Database Ops End '''
+
+        
+        #return EXPERIMENT_OUTPUTS
+        #if not EXPERIMENT_OUTPUTS:
+        de.Update_Status(project_id,id)
+        
+        keys=list(EXPERIMENT_OUTPUTS.keys())
+        payload={
+            "bcl_id":id,
+            "executed_operations":keys,
+            "executed_operations_results":EXPERIMENT_OUTPUTS
+        }
+        de.InsertMemory(data=payload)