drugapp / app.py
mgbam's picture
Update app.py
62a6d51 verified
raw
history blame
24.9 kB
import streamlit as st
import requests
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from rdkit import Chem
from rdkit.Chem import Draw
from fpdf import FPDF
import tempfile
import logging
import os
import plotly.graph_objects as go
import networkx as nx
from typing import Optional, Dict, List, Any
from datetime import datetime
# -------------------------------
# STREAMLIT CONFIGURATION & LOGGING
# -------------------------------
st.set_page_config(page_title="Pharma Research Expert Platform", layout="wide")
logging.basicConfig(level=logging.ERROR)
# -------------------------------
# API ENDPOINTS (Stable Sources Only)
# -------------------------------
API_ENDPOINTS = {
"clinical_trials": "https://clinicaltrials.gov/api/v2/studies",
"pubchem": "https://pubchem.ncbi.nlm.nih.gov/rest/pug/compound/name/{}/JSON",
"pubmed": "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi",
"fda_drug_approval": "https://api.fda.gov/drug/label.json",
"faers_adverse_events": "https://api.fda.gov/drug/event.json",
# PharmGKB endpoints for gene variants (if available)
"pharmgkb_gene_variants": "https://api.pharmgkb.org/v1/data/gene/{}/variants",
# BioPortal for ontology searches
"bioportal_search": "https://data.bioontology.org/search",
# RxNorm & RxClass endpoints
"rxnorm_rxcui": "https://rxnav.nlm.nih.gov/REST/rxcui.json",
"rxnorm_properties": "https://rxnav.nlm.nih.gov/REST/rxcui/{}/properties.json",
"rxclass_by_drug": "https://rxnav.nlm.nih.gov/REST/class/byDrugName.json"
}
# -------------------------------
# TRADE-TO-GENERIC MAPPING (FALLBACK)
# -------------------------------
TRADE_TO_GENERIC = {
"tylenol": "acetaminophen",
"advil": "ibuprofen",
# Extend with additional mappings as desired
}
# -------------------------------
# SECRETS RETRIEVAL
# -------------------------------
OPENAI_API_KEY = st.secrets.get("OPENAI_API_KEY")
BIOPORTAL_API_KEY = st.secrets.get("BIOPORTAL_API_KEY")
PUB_EMAIL = st.secrets.get("PUB_EMAIL")
OPENFDA_KEY = st.secrets.get("OPENFDA_KEY")
if not PUB_EMAIL:
st.error("PUB_EMAIL is not configured in secrets.")
if not BIOPORTAL_API_KEY:
st.error("BIOPORTAL_API_KEY is not configured in secrets.")
if not OPENFDA_KEY:
st.error("OPENFDA_KEY is not configured in secrets.")
if not OPENAI_API_KEY:
st.error("OPENAI_API_KEY is not configured in secrets.")
# -------------------------------
# INITIALIZE OPENAI (GPT-4) CLIENT
# -------------------------------
from openai import OpenAI
openai_client = OpenAI(api_key=OPENAI_API_KEY)
def generate_ai_content(prompt: str) -> str:
"""Call GPT-4 to generate innovative insights."""
try:
response = openai_client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
max_tokens=300
)
return response.choices[0].message.content.strip()
except Exception as e:
st.error(f"GPT-4 generation error: {e}")
logging.error(e)
return "AI content generation failed."
# -------------------------------
# UTILITY FUNCTIONS WITH CACHING
# -------------------------------
@st.cache_data(show_spinner=False)
def query_api(endpoint: str, params: Optional[Dict] = None, headers: Optional[Dict] = None) -> Optional[Dict]:
"""Wrapper for HTTP GET requests with error handling."""
try:
response = requests.get(endpoint, params=params, headers=headers, timeout=15)
response.raise_for_status()
return response.json()
except Exception as e:
st.error(f"API error for {endpoint}: {e}")
logging.error(f"API error for {endpoint}: {e}")
return None
@st.cache_data(show_spinner=False)
def get_pubchem_smiles(drug_name: str) -> Optional[str]:
"""Retrieve canonical SMILES using PubChem."""
url = API_ENDPOINTS["pubchem"].format(drug_name)
data = query_api(url)
if data and data.get("PC_Compounds"):
for prop in data["PC_Compounds"][0].get("props", []):
if prop.get("name") == "Canonical SMILES":
return prop["value"]["sval"]
return None
def draw_molecule(smiles: str) -> Optional[Any]:
"""Generate a 2D molecule image using RDKit."""
try:
mol = Chem.MolFromSmiles(smiles)
if mol:
return Draw.MolToImage(mol)
else:
st.error("Invalid SMILES string provided.")
except Exception as e:
st.error(f"Error drawing molecule: {e}")
logging.error(e)
return None
@st.cache_data(show_spinner=False)
def get_pubchem_drug_details(drug_name: str) -> Optional[Dict[str, str]]:
"""Retrieve drug details from PubChem."""
url = API_ENDPOINTS["pubchem"].format(drug_name)
data = query_api(url)
details = {}
if data and data.get("PC_Compounds"):
compound = data["PC_Compounds"][0]
for prop in compound.get("props", []):
urn = prop.get("urn", {})
if urn.get("label") == "Molecular Formula":
details["Molecular Formula"] = prop["value"]["sval"]
if urn.get("name") == "Preferred":
details["IUPAC Name"] = prop["value"]["sval"]
if prop.get("name") == "Canonical SMILES":
details["Canonical SMILES"] = prop["value"]["sval"]
return details
return None
@st.cache_data(show_spinner=False)
def get_clinical_trials(query: str) -> Optional[Dict]:
"""Query ClinicalTrials.gov."""
if query.upper().startswith("NCT") and query[3:].isdigit():
params = {"id": query, "fmt": "json"}
else:
params = {"query.term": query, "retmax": 10, "retmode": "json"}
return query_api(API_ENDPOINTS["clinical_trials"], params)
@st.cache_data(show_spinner=False)
def get_pubmed(query: str) -> Optional[Dict]:
"""Query PubMed."""
params = {"db": "pubmed", "term": query, "retmax": 10, "retmode": "json", "email": PUB_EMAIL}
return query_api(API_ENDPOINTS["pubmed"], params)
@st.cache_data(show_spinner=False)
def get_fda_approval(drug_name: str) -> Optional[Dict]:
"""Retrieve FDA approval info using openFDA."""
if not OPENFDA_KEY:
st.error("OpenFDA key not configured.")
return None
query = f'openfda.brand_name:"{drug_name}"'
params = {"api_key": OPENFDA_KEY, "search": query, "limit": 1}
data = query_api(API_ENDPOINTS["fda_drug_approval"], params)
if data and data.get("results"):
return data["results"][0]
return None
@st.cache_data(show_spinner=False)
def analyze_adverse_events(drug_name: str, limit: int = 5) -> Optional[Dict]:
"""Retrieve FAERS adverse events."""
if not OPENFDA_KEY:
st.error("OpenFDA key not configured.")
return None
query = f'patient.drug.medicinalproduct:"{drug_name}"'
params = {"api_key": OPENFDA_KEY, "search": query, "limit": limit}
return query_api(API_ENDPOINTS["faers_adverse_events"], params)
@st.cache_data(show_spinner=False)
def get_pharmgkb_variants_for_gene(pharmgkb_gene_id: str) -> Optional[List[str]]:
"""Return variant IDs for a PharmGKB gene accession."""
if not pharmgkb_gene_id.startswith("PA"):
st.warning("Enter a valid PharmGKB gene accession (e.g., PA1234).")
return None
endpoint = API_ENDPOINTS["pharmgkb_gene_variants"].format(pharmgkb_gene_id)
data = query_api(endpoint)
if data and data.get("data"):
return [variant["id"] for variant in data["data"]]
st.warning(f"No variants found for PharmGKB gene {pharmgkb_gene_id}.")
return None
@st.cache_data(show_spinner=False)
def get_rxnorm_rxcui(drug_name: str) -> Optional[str]:
"""Return RxCUI for a drug."""
url = f"{API_ENDPOINTS['rxnorm_rxcui']}?name={drug_name}"
data = query_api(url)
if data and "idGroup" in data and data["idGroup"].get("rxnormId"):
return data["idGroup"]["rxnormId"][0]
st.warning(f"No RxCUI found for {drug_name}.")
return None
@st.cache_data(show_spinner=False)
def get_rxnorm_properties(rxcui: str) -> Optional[Dict]:
"""Return RxNorm properties for a given RxCUI."""
url = API_ENDPOINTS["rxnorm_properties"].format(rxcui)
return query_api(url)
@st.cache_data(show_spinner=False)
def get_rxclass_by_drug_name(drug_name: str) -> Optional[Dict]:
"""Return RxClass information for a drug."""
url = f"{API_ENDPOINTS['rxclass_by_drug']}?drugName={drug_name}"
data = query_api(url)
if data and "classMember" in data:
return data
return None
# -------------------------------
# AI-DRIVEN DRUG INSIGHTS
# -------------------------------
def generate_drug_insights(drug_name: str) -> str:
"""
Gather FDA, PubChem, RxNorm, and RxClass data (using generic fallback) and build a GPT‑4 prompt
for an innovative, bullet‑point drug analysis.
"""
query_name = TRADE_TO_GENERIC.get(drug_name.lower(), drug_name)
# Get FDA data
fda_info = get_fda_approval(query_name)
fda_status = "Not Approved"
if fda_info and fda_info.get("openfda", {}).get("brand_name"):
fda_status = ", ".join(fda_info["openfda"]["brand_name"])
# Get PubChem details
pubchem_details = get_pubchem_drug_details(query_name)
if pubchem_details:
formula = pubchem_details.get("Molecular Formula", "N/A")
iupac = pubchem_details.get("IUPAC Name", "N/A")
canon_smiles = pubchem_details.get("Canonical SMILES", "N/A")
else:
formula = iupac = canon_smiles = "Not Available"
# RxNorm and RxClass
rxnorm_id = get_rxnorm_rxcui(query_name)
if rxnorm_id:
rx_props = get_rxnorm_properties(rxnorm_id)
rxnorm_info = f"RxCUI: {rxnorm_id}\nProperties: {rx_props}"
else:
rxnorm_info = "No RxNorm data available."
rxclass_data = get_rxclass_by_drug_name(query_name)
rxclass_info = rxclass_data if rxclass_data else "No RxClass data available."
# Construct prompt for GPT-4
prompt = (
f"Please provide an advanced, innovative drug analysis report for '{drug_name}' "
f"(generic: {query_name}).\n\n"
f"**FDA Approval Status:** {fda_status}\n"
f"**PubChem Details:** Molecular Formula: {formula}, IUPAC Name: {iupac}, Canonical SMILES: {canon_smiles}\n"
f"**RxNorm Info:** {rxnorm_info}\n"
f"**RxClass Info:** {rxclass_info}\n\n"
f"Include the following in bullet points:\n"
f"- Pharmacogenomic considerations (including genetic variants that might affect metabolism and toxicity).\n"
f"- Potential repurposing opportunities based on drug mechanism.\n"
f"- Regulatory insights and challenges, particularly for expanding indications or personalized medicine.\n"
f"- Innovative suggestions for future research and data integration approaches.\n"
)
return generate_ai_content(prompt)
# -------------------------------
# STREAMLIT APP LAYOUT
# -------------------------------
tabs = st.tabs([
"💊 Drug Development",
"📊 Trial Analytics",
"🧬 Molecular Profiling",
"📜 Regulatory Intelligence",
"📚 Literature Search",
"📈 Dashboard",
"🧪 Drug Data Integration",
"🤖 AI Insights"
])
# ----- Tab 1: Drug Development -----
with tabs[0]:
st.header("AI-Driven Drug Development Strategy")
target = st.text_input("Target Disease/Pathway:", placeholder="Enter disease mechanism or target")
target_gene = st.text_input("Target Gene (PharmGKB Accession):", placeholder="e.g., PA1234")
strategy = st.selectbox("Development Strategy:", ["First-in-class", "Me-too", "Repurposing", "Biologic"])
if st.button("Generate Development Plan"):
with st.spinner("Generating comprehensive development plan..."):
plan_prompt = (
f"Develop a detailed drug development plan for treating {target} using a {strategy} strategy. "
"Include sections on target validation, lead optimization, preclinical testing, clinical trial design, "
"regulatory strategy, market analysis, competitive landscape, and relevant pharmacogenomic considerations."
)
plan = generate_ai_content(plan_prompt)
st.subheader("Comprehensive Development Plan")
st.markdown(plan)
st.subheader("FDA Regulatory Insights")
if target:
fda_data = get_fda_approval(target.split()[0])
if fda_data:
st.json(fda_data)
else:
st.write("No FDA data found for the given target.")
st.subheader("Pharmacogenomic Considerations")
if target_gene:
if not target_gene.startswith("PA"):
st.warning("Enter a valid PharmGKB accession (e.g., PA1234).")
else:
variants = get_pharmgkb_variants_for_gene(target_gene)
if variants:
st.write("PharmGKB Variants:")
st.write(variants)
for vid in variants[:3]:
annotations = _get_pharmgkb_clinical_annotations(vid)
st.write(f"Annotations for Variant {vid}:")
st.json(annotations if annotations else {"message": "No annotations found."})
else:
st.write("No variants found for the specified PharmGKB gene accession.")
else:
st.write("Provide a PharmGKB gene accession to retrieve pharmacogenomic data.")
# ----- Tab 2: Clinical Trial Analytics -----
with tabs[1]:
st.header("Clinical Trial Landscape Analytics")
trial_query = st.text_input("Search Clinical Trials:", placeholder="Enter condition, intervention, or NCT number")
if st.button("Analyze Trial Landscape"):
with st.spinner("Fetching trial data..."):
trials = get_clinical_trials(trial_query)
if trials and trials.get("studies"):
trial_data = []
for study in trials["studies"][:5]:
trial_data.append({
"Title": study.get("protocolSection", {}).get("identificationModule", {}).get("briefTitle", "N/A"),
"Status": study.get("protocolSection", {}).get("statusModule", {}).get("overallStatus", "N/A"),
"Phase": study.get("protocolSection", {}).get("designModule", {}).get("phases", ["Not Available"])[0],
"Enrollment": study.get("protocolSection", {}).get("designModule", {}).get("enrollmentInfo", {}).get("count", "N/A")
})
_display_dataframe(trial_data, list(trial_data[0].keys()))
else:
st.warning("No clinical trials found for the query.")
ae_data = analyze_adverse_events(trial_query)
if ae_data and ae_data.get("results"):
st.subheader("Adverse Event Profile (Top 5)")
ae_results = ae_data["results"][:5]
ae_df = pd.json_normalize(ae_results)
st.dataframe(ae_df)
if "patient.reaction.reactionmeddrapt" in ae_df.columns:
try:
reactions = ae_df["patient.reaction.reactionmeddrapt"].explode().dropna()
top_reactions = reactions.value_counts().nlargest(10)
fig, ax = plt.subplots(figsize=(10, 6))
sns.barplot(x=top_reactions.values, y=top_reactions.index, ax=ax)
ax.set_title("Top Adverse Reactions")
ax.set_xlabel("Frequency")
ax.set_ylabel("Reaction")
st.pyplot(fig)
except Exception as e:
st.error(f"Error visualizing adverse events: {e}")
else:
st.write("No adverse event data available.")
# ----- Tab 3: Molecular Profiling -----
with tabs[2]:
st.header("Advanced Molecular Profiling")
compound_input = st.text_input("Compound Identifier:", placeholder="Enter drug name, SMILES, or INN")
if st.button("Analyze Compound"):
with st.spinner("Querying PubChem for molecular structure..."):
query_compound = TRADE_TO_GENERIC.get(compound_input.lower(), compound_input)
smiles = _get_pubchem_smiles(query_compound)
if smiles:
mol_image = draw_molecule(smiles)
if mol_image:
st.image(mol_image, caption="2D Molecular Structure")
else:
st.error("Molecular structure not found. Try a more specific compound name.")
pubchem_data = query_api(API_ENDPOINTS["pubchem"].format(query_compound))
if pubchem_data and pubchem_data.get("PC_Compounds"):
st.subheader("Physicochemical Properties")
props = pubchem_data["PC_Compounds"][0].get("props", [])
mw = next((prop["value"]["sval"] for prop in props if prop.get("name") == "Molecular Weight"), "N/A")
logp = next((prop["value"]["sval"] for prop in props if prop.get("name") == "LogP"), "N/A")
st.write(f"**Molecular Weight:** {mw}")
st.write(f"**LogP:** {logp}")
else:
st.error("Physicochemical properties not available.")
# ----- Tab 4: Regulatory Intelligence -----
with tabs[3]:
st.header("Global Regulatory Monitoring")
st.markdown("**Note:** Due to persistent issues with EMA/WHO/DailyMed APIs, this section focuses on FDA data and PubChem drug details.")
drug_prod = st.text_input("Drug Product:", placeholder="Enter generic or brand name")
if st.button("Generate Regulatory Report"):
with st.spinner("Compiling regulatory data..."):
fda_data = get_fda_approval(drug_prod)
fda_status = "Not Approved"
if fda_data and fda_data.get("openfda", {}).get("brand_name"):
fda_status = ", ".join(fda_data["openfda"]["brand_name"])
pubchem_details = get_pubchem_drug_details(drug_prod)
if pubchem_details:
formula = pubchem_details.get("Molecular Formula", "N/A")
iupac = pubchem_details.get("IUPAC Name", "N/A")
canon_smiles = pubchem_details.get("Canonical SMILES", "N/A")
else:
formula = iupac = canon_smiles = "Not Available"
col1, col2 = st.columns(2)
with col1:
st.markdown("**FDA Status**")
st.write(fda_status)
with col2:
st.markdown("**Drug Details (PubChem)**")
st.write(f"**Molecular Formula:** {formula}")
st.write(f"**IUPAC Name:** {iupac}")
st.write(f"**Canonical SMILES:** {canon_smiles}")
report_text = (
f"### Regulatory Report for {drug_prod}\n\n"
f"**FDA Status:** {fda_status}\n\n"
f"**Molecular Formula:** {formula}\n\n"
f"**IUPAC Name:** {iupac}\n\n"
f"**Canonical SMILES:** {canon_smiles}\n"
)
with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as tmp:
pdf_file = _save_pdf_report(report_text, tmp.name)
if pdf_file:
with open(pdf_file, "rb") as f:
st.download_button("Download Regulatory Report (PDF)", data=f, file_name=f"{drug_prod}_report.pdf", mime="application/pdf")
os.remove(pdf_file)
# ----- Tab 5: Literature Search -----
with tabs[4]:
st.header("Literature Search")
lit_query = st.text_input("Enter search query for PubMed:", placeholder="e.g., Alzheimer's disease genetics")
if st.button("Search PubMed"):
with st.spinner("Searching PubMed..."):
pubmed_results = get_pubmed(lit_query)
if pubmed_results and pubmed_results.get("esearchresult", {}).get("idlist"):
id_list = pubmed_results["esearchresult"]["idlist"]
st.subheader(f"Found {len(id_list)} PubMed Results")
for pmid in id_list:
st.markdown(f"- [PMID: {pmid}](https://pubmed.ncbi.nlm.nih.gov/{pmid}/)")
else:
st.write("No PubMed results found.")
st.header("Ontology Search")
ont_query = st.text_input("Enter search query for Ontology:", placeholder="e.g., Alzheimer's disease")
ont_select = st.selectbox("Select Ontology", ["MESH", "NCIT", "GO", "SNOMEDCT"])
if st.button("Search BioPortal"):
with st.spinner("Searching BioPortal..."):
bioportal_results = _get_bioportal_data(ont_select, ont_query)
if bioportal_results and bioportal_results.get("collection"):
st.subheader(f"BioPortal Results for {ont_select}")
for item in bioportal_results["collection"]:
label = item.get("prefLabel", "N/A")
ont_id = item.get("@id", "N/A")
st.markdown(f"- **{label}** ({ont_id})")
else:
st.write("No ontology results found.")
# ----- Tab 6: Comprehensive Dashboard -----
with tabs[5]:
st.header("Comprehensive Dashboard")
# Example KPIs (these could later be replaced by dynamic queries)
kpi_fda = 5000
kpi_trials = 12000
kpi_pubs = 250000
col1, col2, col3 = st.columns(3)
col1.metric("FDA Approved Drugs", kpi_fda)
col2.metric("Ongoing Trials", kpi_trials)
col3.metric("Publications", kpi_pubs)
st.subheader("Trend Analysis")
years = list(range(2000, 2026))
approvals = [kpi_fda // len(years)] * len(years)
fig_trend, ax_trend = plt.subplots(figsize=(10, 6))
sns.lineplot(x=years, y=approvals, marker="o", ax=ax_trend)
ax_trend.set_title("FDA Approvals Over Time")
ax_trend.set_xlabel("Year")
ax_trend.set_ylabel("Number of Approvals")
st.pyplot(fig_trend)
st.subheader("Gene-Variant-Drug Network (Sample)")
sample_gene = "CYP2C19"
sample_variants = ["rs4244285", "rs12248560"]
sample_annots = {"rs4244285": ["Clopidogrel", "Omeprazole"], "rs12248560": ["Sertraline"]}
try:
net_fig = _create_variant_network(sample_gene, sample_variants, sample_annots)
st.plotly_chart(net_fig, use_container_width=True)
except Exception as e:
st.error(f"Network graph error: {e}")
# ----- Tab 7: Drug Data Integration -----
with tabs[6]:
st.header("🧪 Drug Data Integration")
drug_integration = st.text_input("Enter Drug Name for API Integration:", placeholder="e.g., aspirin")
if st.button("Retrieve Drug Data"):
with st.spinner("Fetching drug data from multiple sources..."):
query_drug = TRADE_TO_GENERIC.get(drug_integration.lower(), drug_integration)
rxnorm_id = get_rxnorm_rxcui(query_drug)
rx_props = get_rxnorm_properties(rxnorm_id) if rxnorm_id else None
rxclass_info = get_rxclass_by_drug_name(query_drug)
st.subheader("RxNorm Data")
if rxnorm_id:
st.write(f"RxCUI for {drug_integration}: {rxnorm_id}")
st.json(rx_props if rx_props else {"message": "No RxNorm properties found."})
else:
st.write("No RxCUI found for the given drug name.")
st.subheader("RxClass Information")
if rxclass_info:
st.json(rxclass_info)
else:
st.write("No RxClass data found for the given drug.")
pubchem_info = get_pubchem_drug_details(query_drug)
st.subheader("PubChem Drug Details")
if pubchem_info:
st.write(f"**Molecular Formula:** {pubchem_info.get('Molecular Formula', 'N/A')}")
st.write(f"**IUPAC Name:** {pubchem_info.get('IUPAC Name', 'N/A')}")
st.write(f"**Canonical SMILES:** {pubchem_info.get('Canonical SMILES', 'N/A')}")
else:
st.write("No PubChem details found.")
# ----- Tab 8: AI Insights -----
with tabs[7]:
st.header("🤖 AI Insights")
ai_drug = st.text_input("Enter Drug Name for AI-Driven Analysis:", placeholder="e.g., tylenol")
if st.button("Generate AI Insights"):
with st.spinner("Generating AI-driven insights..."):
query_ai_drug = TRADE_TO_GENERIC.get(ai_drug.lower(), ai_drug)
insights_text = generate_drug_insights(query_ai_drug)
st.subheader("AI-Driven Drug Analysis")
st.markdown(insights_text)
# -------------------------------
# SIDEBAR INFORMATION
# -------------------------------
st.sidebar.header("About")
st.sidebar.info("""
**Pharma Research Expert Platform**
An innovative, AI-driven tool for advanced drug discovery, clinical research, and regulatory analysis.
**Developed by:** Your Name
**Contact:** [[email protected]](mailto:[email protected])
Last updated: {}
""".format(datetime.now().strftime("%Y-%m-%d")))