Upload 3 files

app.py

@@ -0,0 +1,170 @@
+from openai import OpenAI
+import gradio as gr
+import os
+from dotenv import load_dotenv
+from fetch_all_proteins import *
+import json
+import requests
+
+
+load_dotenv()
+
+email = os.getenv("EMAIL")
+
+
+
+client = OpenAI(
+     api_key=os.environ.get("OPENAI_API_KEY"),
+ )
+
+def chatApiCall(messages):
+    payload = {
+            "messages": messages,
+            "web_access": False
+        }
+    url = "https://open-ai21.p.rapidapi.com/claude3"
+    headers = {
+            "x-rapidapi-key": os.environ.get("OPENAI_API_KEY"),
+            "x-rapidapi-host": "open-ai21.p.rapidapi.com",
+            "Content-Type": "application/json"
+        }
+    # response = requests.post(url, json=payload, headers=headers)
+    response = client.chat.completions.create(
+        messages=messages,
+        model="gpt-4o-mini",
+    )
+    res_json=response.choices[0].message.content
+    print("response",res_json)
+    return res_json
+
+## Create a function that determines whether the request relates to a protein or not
+def IsProteinRequest(history, message): 
+    prompt = """Respond only with true when the conditions below are met otherwise respond only with false. The conditions
+     are as follows: 
+      1: Within the context of the chat history, the message refers to a specific protein.
+       2: A specific protein name is mentioned in the message. GCPR proteins alone does not count. 
+        3: If there are no proteins mentioned in the chat history, there should be a specific protein name mentioned in the message.
+         4: If there are no chat histories, look at the following message. 
+         5: If you detect any generalized requests like "Tell me about proteins" "Tell me about receptors" or any request that has no 
+         specific protein mentioned like Rhodopsin or OR51E2, respond with false. 
+           The message is as follows: """ + message
+    history.append({"role":"user", "content": f"{prompt}"})
+    # function_client = OpenAI(
+    #     api_key=os.environ.get("OPENAI_API_KEY")
+    # )
+    # function_client = OpenAI()
+    # response = function_client.chat.completions.create(
+    #     model="gpt-4o-mini",
+    #     messages=history,
+
+    # )
+    response=chatApiCall(history)
+    # print("prompt",history,"res",response)
+    return response
+
+## Create a function that returns the name of the protein 
+def ProteinName(history, message):
+    prompt = """Respond only with the name of the protein the message is referring to with respect to both
+    the chat history above and the message itself. The message is as follows: """ + message
+    
+    history.append({"role":"user", "content": f"{prompt}"})
+    # function_client = OpenAI(
+    #     api_key=os.environ.get("OPENAI_API_KEY")
+    # )
+    # function_client = OpenAI()
+    # response = function_client.chat.completions.create(
+    #     model="gpt-4o-mini",
+    #     messages=history,
+
+    # )
+    response=chatApiCall(history)
+    # print("prompt protein name",history,"res",response)
+    return response
+
+
+## Create a function that takes in a protein name and returns protein info 
+def ProteinInfo(protein): 
+    print("caall hua hai",protein)
+    accession, full_name = fetch_protein_info(protein)
+    all_data = {
+        "uniprot": fetch_uniprot_info(accession, email),
+        "interpro": fetch_comprehensive_interpro_info(accession, email),
+        # "string": fetch_string_info(accession, 9606, email),  # Assuming human (9606)
+        "quickgo": {}
+    }
+    
+    go_terms = fetch_protein_go_terms(accession, email)
+    all_data["quickgo"]["go_terms"] = go_terms
+    # for go_term in go_terms:
+    #     all_data["quickgo"][go_term] = fetch_go_info(go_term, email)
+    # print(all_data)
+    # with open(f"{protein}.json",'w') as f:
+    #     json.dump(all_data,f)
+    return json.dumps(all_data)
+
+## Create a function that takes in a message and a protein information and returns an informed response
+def InformedResponse(proteinInfo, message):
+    prompt = f"{proteinInfo} From the following information given, answer this question: " + message
+    history=[]
+    Agent = {"role": "system", "content": "You are a helpful assistant with extensive background in protein analysis."}
+    history.append(Agent)
+    history.append({"role":"user", "content": f"{prompt}"})
+    # function_client = OpenAI(
+    #     api_key=os.environ.get("OPENAI_API_KEY")
+    # )
+    # function_client = OpenAI()
+    # response = function_client.chat.completions.create(
+    #     model="gpt-4o-mini",
+    #     messages=history,
+    # )
+    response=chatApiCall(history)
+    # print("prompt",history,"res",response)
+    return response
+
+def HistoryConverter(history): 
+    Agent = {"role": "system", "content": "You are a helpful assistant with extensive background in protein analysis."}
+
+    formatted_history = []
+    formatted_history.append(Agent)
+    for each in history:
+        formatted_history.append({"role": "user", "content": f"{each[0]}"})
+        formatted_history.append({"role": "assistant", "content": f"{each[1]}"})
+    return formatted_history
+
+
+
+def openai_chatbot(message, history):
+    formatted_history = HistoryConverter(history=history)
+
+    isProteinRequest = IsProteinRequest(history=formatted_history, message=message)
+
+    if isProteinRequest == "true": 
+        proteinName = ProteinName(history=formatted_history, message=message)
+        proteinInfo = ProteinInfo(protein=proteinName)
+        print(proteinName,proteinInfo)
+        return InformedResponse(proteinInfo=proteinInfo, message=message)
+    else:
+        # client = OpenAI()
+        messages = HistoryConverter(history=history)
+        messages.append({"role":"user","content": f"{message}"})
+        # payload = {
+        #     "messages": messages,
+        #     "web_access": False
+        # }
+        # response=chatApiCall({"messages":history})
+
+
+        # response = client.chat.completions.create(
+        # model="gpt-4o-mini",
+        # # messages=[{"role":"system", "content": "You are a helpful assistant"}, {"role":"user", "content":"Tell me about peter pan"}]
+        # messages = messages
+
+        # )
+        
+        response=chatApiCall(messages)
+        # print("prompt",messages,"res",response)
+        return response
+
+if __name__=="__main__":
+    demo_chatbot = gr.ChatInterface(openai_chatbot, title="OpenAI Chatbot", description="Start Chatting")
+    demo_chatbot.launch()

fetch_all_proteins.py

@@ -0,0 +1,238 @@
+import requests
+import json
+import time
+import urllib.parse
+
+
+def fetch_protein_info(protein_name):
+    url = "https://rest.uniprot.org/uniprotkb/search"
+    params = {
+        "query": protein_name,
+        "format": "json",
+        "fields": "accession,id,protein_name,gene_names,organism_name,reviewed",
+        "size": 10  # Increase size to get more results
+    }
+    try:
+        response = requests.get(url, params=params)
+        response.raise_for_status()
+        data = response.json()
+        if data.get('results'):
+            # Try to find an exact match for the gene name first
+            for result in data['results']:
+                gene_names = result.get('genes', [])
+                if gene_names and any(gene.get('geneName', {}).get('value') == protein_name for gene in gene_names):
+                    print("Exact gene match found:", result)
+                    return process_result(result)
+            
+            # If no exact match, return the first reviewed (Swiss-Prot) entry or the first result
+            reviewed_result = next((r for r in data['results'] if r.get('entryType') == 'UniProtKB reviewed (Swiss-Prot)'), None)
+            if reviewed_result:
+                print("Reviewed entry found:", reviewed_result)
+                return process_result(reviewed_result)
+            else:
+                print("Using first result:", data['results'][0])
+                return process_result(data['results'][0])
+        else:
+            print(f"No results found for '{protein_name}'")
+            return None, None
+    except requests.exceptions.RequestException as e:
+        print(f"Error occurred while fetching data: {e}")
+        return None, None
+    
+
+def process_result(result):
+    primary_accession = result.get('primaryAccession')
+    name = result.get('proteinName', [{}])[0].get('fullName', {}).get('value')
+    if not name:
+        name = result.get('proteinName', [{}])[0].get('shortName', [{}])[0].get('value')
+    if not name:
+        name = result.get('id')
+    return primary_accession, name
+
+def fetch_uniprot_info(accession, email):
+    uniprot_base_url = "https://rest.uniprot.org/uniprotkb/"
+    headers = {
+        "Accept": "application/json",
+        "User-Agent": f"Python script (mailto:{email})"
+    }
+    try:
+        response = requests.get(f"{uniprot_base_url}{accession}", headers=headers)
+        response.raise_for_status()
+        uniprot_data = response.json()
+        protein_info = {
+            "accession": accession,
+            "entry_type": uniprot_data.get('entryType'),
+            "entry_name": uniprot_data.get('uniProtkbId'),
+            "protein_name": uniprot_data.get('proteinDescription', {}).get('recommendedName', {}).get('fullName', {}).get('value'),
+            "gene_name": next((gene.get('geneName', {}).get('value') for gene in uniprot_data.get('genes', []) if gene.get('geneName')), None),
+            "organism": uniprot_data.get('organism', {}).get('scientificName'),
+            "sequence": uniprot_data.get('sequence', {}).get('value'),
+            "sequence_length": uniprot_data.get('sequence', {}).get('length'),
+            "function": next((comment.get('texts', [{}])[0].get('value') for comment in uniprot_data.get('comments', []) if comment.get('commentType') == 'FUNCTION'), None),
+            "subcellular_locations": [
+                loc.get('location', {}).get('value')
+                for comment in uniprot_data.get('comments', [])
+                if comment.get('commentType') == 'SUBCELLULAR LOCATION'
+                for loc in comment.get('subcellularLocations', [])
+            ],
+            "ec_numbers": [ec.get('value') for ec in uniprot_data.get('proteinDescription', {}).get('ecNumbers', [])],
+            "keywords": [kw.get('name') for kw in uniprot_data.get('keywords', [])],
+            "features": [{'type': f.get('type'), 'description': f.get('description')} for f in uniprot_data.get('features', [])]
+        }
+        return protein_info
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching UniProt data: {e}")
+        return None
+
+def fetch_comprehensive_interpro_info(accession, email):
+    base_url = "https://www.ebi.ac.uk/interpro/api/protein/uniprot/"
+    protein_url = f"{base_url}{accession}/entry_protein_locations/"
+    headers = {
+        "Accept": "application/json",
+        "User-Agent": f"Python script (mailto:{email})"
+    }
+    try:
+        response = requests.get(protein_url, headers=headers)
+        response.raise_for_status()
+        interpro_data = response.json()
+        return interpro_data
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching InterPro data: {e}")
+        return None
+
+def fetch_pdb_info(accession, email):
+    pdb_search_url = "https://search.rcsb.org/rcsbsearch/v2/query"
+    headers = {
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+        "User-Agent": f"Python script (mailto:{email})"
+    }
+
+    # Construct the search query
+    query = {
+        "query": {
+            "type": "terminal",
+            "service": "text",
+            "parameters": {
+                "attribute": "rcsb_polymer_entity_container_identifiers.reference_sequence_identifiers.database_accession",
+                "operator": "exact_match",
+                "value": accession
+            }
+        },
+        "return_type": "entry",
+        "request_options": {
+            "return_all_hits": True
+        }
+    }
+
+    try:
+        # Perform the search
+        response = requests.post(pdb_search_url, headers=headers, data=json.dumps(query))
+        response.raise_for_status()
+        search_results = response.json()
+        pdb_ids = [result['identifier'] for result in search_results.get('result_set', [])]
+
+        if not pdb_ids:
+            # No PDB entries found
+            return {
+                "message": "Protein not found in PDB.",
+                "alphafold_link": f"https://alphafold.ebi.ac.uk/entry/{accession}"
+            }
+
+        pdb_info_list = []
+        for pdb_id in pdb_ids:
+            time.sleep(0.1)  # Be polite to the API
+            pdb_entry_url = f"https://data.rcsb.org/rest/v1/core/entry/{pdb_id}"
+            response = requests.get(pdb_entry_url, headers=headers)
+            response.raise_for_status()
+            pdb_data = response.json()
+            pdb_info = {
+                "pdb_id": pdb_id,
+                "title": pdb_data.get('struct', {}).get('title'),
+                "deposition_date": pdb_data.get('rcsb_accession_info', {}).get('deposit_date'),
+                "release_date": pdb_data.get('rcsb_accession_info', {}).get('initial_release_date'),
+                "experimental_method": pdb_data.get('exptl', [{}])[0].get('method'),
+                "resolution": pdb_data.get('rcsb_entry_info', {}).get('resolution_combined', [None])[0],
+                "authors": [author.get("name") for author in pdb_data.get("audit_author", [])],
+                "ligands": [],
+                "pdb_structure_link": f"https://www.rcsb.org/3d-view/{pdb_id}"
+            }
+            # Fetch ligand information
+            ligand_entities = pdb_data.get('nonpolymer_entities', [])
+            for ligand in ligand_entities:
+                chem_comp = ligand.get('chem_comp', {})
+                ligand_info = {
+                    "chem_comp_id": chem_comp.get('id'),
+                    "name": chem_comp.get('name'),
+                    "formula": chem_comp.get('formula'),
+                    "weight": chem_comp.get('formula_weight')
+                }
+                pdb_info['ligands'].append(ligand_info)
+            pdb_info_list.append(pdb_info)
+        return {"pdb_entries": pdb_info_list}
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching PDB data: {e}")
+        return {
+            "message": "Error fetching PDB data.",
+            "alphafold_link": f"https://alphafold.ebi.ac.uk/entry/{accession}"
+        }
+
+def fetch_protein_go_terms(uniprot_id, email):
+    base_url = "https://www.ebi.ac.uk/QuickGO/services/annotation/search"
+    headers = {
+        "Accept": "application/json",
+        "User-Agent": f"Python script (mailto:{email})"
+    }
+    params = {
+        "geneProductId": uniprot_id,
+        "limit": 10  # Limit to top 10 GO terms
+    }
+    try:
+        response = requests.get(base_url, params=params, headers=headers)
+        response.raise_for_status()
+        data = response.json()
+        go_terms = []
+        for annotation in data.get('results', []):
+            go_terms.append({
+                "id": annotation.get('goId'),
+                "term": annotation.get('goName'),
+                "aspect": annotation.get('goAspect'),
+                "evidence": annotation.get('goEvidence'),
+                "reference": annotation.get('reference')
+            })
+        return go_terms
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching GO terms for {uniprot_id}: {str(e)}")
+        return []
+    
+
+def main():
+    email = "your_email@example.com"  # Replace with your actual email
+    protein_name = input("What protein would you like to know about? ")
+
+    print(f"\nFetching information for: {protein_name}")
+    accession, full_name = fetch_protein_info(protein_name)
+
+    if not accession:
+        print(f"No results found for '{protein_name}'")
+        return
+
+    print(f"Protein: {full_name}")
+    print(f"Accession: {accession}")
+
+    all_data = {
+        "uniprot": fetch_uniprot_info(accession, email),
+        "interpro": fetch_comprehensive_interpro_info(accession, email),
+        "pdb": fetch_pdb_info(accession, email),
+        "go_terms": fetch_protein_go_terms(accession, email)
+    }
+
+    # Save the data to a JSON file
+    filename = f"{accession}_comprehensive_info.json"
+    with open(filename, 'w') as f:
+        json.dump(all_data, f, indent=2)
+
+    print(f"\nComprehensive information has been saved to {filename}")
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,76 @@
+huggingface_hub==0.25.2
+aiofiles==23.2.1
+annotated-types==0.7.0
+anyio==4.4.0
+cachetools==5.4.0
+certifi==2024.7.4
+charset-normalizer==3.3.2
+click==8.1.7
+contourpy==1.2.1
+cycler==0.12.1
+fastapi==0.112.0
+ffmpy==0.4.0
+filelock==3.15.4
+fonttools==4.53.1
+fsspec==2024.6.1
+google-ai-generativelanguage==0.6.6
+google-api-core==2.19.1
+google-api-python-client==2.140.0
+google-auth==2.33.0
+google-auth-httplib2==0.2.0
+google-generativeai==0.7.2
+googleapis-common-protos==1.63.2
+gradio==4.41.0
+gradio_client==1.3.0
+grpcio==1.65.4
+grpcio-status==1.62.3
+h11==0.14.0
+httpcore==1.0.5
+httplib2==0.22.0
+httpx==0.27.0
+idna==3.7
+importlib_resources==6.4.0
+Jinja2==3.1.4
+kiwisolver==1.4.5
+markdown-it-py==3.0.0
+MarkupSafe==2.1.5
+matplotlib==3.9.2
+mdurl==0.1.2
+numpy==2.0.1
+orjson==3.10.7
+packaging==24.1
+pandas==2.2.2
+pillow==10.4.0
+proto-plus==1.24.0
+protobuf==4.25.4
+pyasn1==0.6.0
+pyasn1_modules==0.4.0
+pydantic==2.8.2
+pydantic_core==2.20.1
+pydub==0.25.1
+Pygments==2.18.0
+pyparsing==3.1.2
+python-dateutil==2.9.0.post0
+python-dotenv==1.0.1
+python-multipart==0.0.9
+pytz==2024.1
+PyYAML==6.0.2
+requests==2.32.3
+rich==13.7.1
+rsa==4.9
+ruff==0.5.7
+semantic-version==2.10.0
+shellingham==1.5.4
+six==1.16.0
+sniffio==1.3.1
+starlette==0.37.2
+tomlkit==0.12.0
+tqdm==4.66.5
+typer==0.12.3
+typing_extensions==4.12.2
+tzdata==2024.1
+uritemplate==4.1.1
+urllib3==2.2.2
+uvicorn==0.30.6
+websockets==12.0
+openai==1.52.0