|
|
import gradio as gr |
|
|
from openfda_client import ( |
|
|
get_top_adverse_events, |
|
|
get_drug_event_pair_frequency, |
|
|
get_serious_outcomes, |
|
|
get_time_series_data, |
|
|
get_report_source_data |
|
|
) |
|
|
from plotting import ( |
|
|
create_bar_chart, |
|
|
create_outcome_chart, |
|
|
create_time_series_chart, |
|
|
create_pie_chart, |
|
|
create_placeholder_chart |
|
|
) |
|
|
import pandas as pd |
|
|
|
|
|
|
|
|
|
|
|
def format_pair_frequency_results(data: dict, drug_name: str, event_name: str) -> str: |
|
|
"""Formats the results for the drug-event pair frequency tool.""" |
|
|
if "error" in data: |
|
|
return f"An error occurred: {data['error']}" |
|
|
|
|
|
results = data.get("meta", {}).get("results", {}) |
|
|
total_reports = results.get("total", 0) |
|
|
total_for_drug = results.get("total_for_drug", 0) |
|
|
|
|
|
percentage_string = "" |
|
|
if total_for_drug > 0: |
|
|
percentage = (total_reports / total_for_drug) * 100 |
|
|
percentage_string = ( |
|
|
f"\n\nThis combination accounts for **{percentage:.2f}%** of the **{total_for_drug:,}** " |
|
|
f"total adverse event reports for '{drug_name.title()}' in the database." |
|
|
) |
|
|
|
|
|
result = ( |
|
|
f"Found **{total_reports:,}** reports for the combination of " |
|
|
f"'{drug_name.title()}' and '{event_name.title()}'.{percentage_string}\n\n" |
|
|
"**Source**: FDA FAERS via OpenFDA\n" |
|
|
"**Disclaimer**: Spontaneous reports do not prove causation. Consult a healthcare professional." |
|
|
) |
|
|
return result |
|
|
|
|
|
|
|
|
|
|
|
def top_adverse_events_tool(drug_name: str, top_n: int = 10, patient_sex: str = "all", min_age: int = 0, max_age: int = 120): |
|
|
""" |
|
|
MCP Tool: Finds the top reported adverse events for a given drug. |
|
|
|
|
|
Args: |
|
|
drug_name (str): The generic name of the drug is preferred! A small sample of brand names (e.g., 'Tylenol') are converted to generic names for demonstration purposes. |
|
|
top_n (int): The number of top adverse events to return. |
|
|
patient_sex (str): The patient's sex to filter by. |
|
|
min_age (int): The minimum age for the filter. |
|
|
max_age (int): The maximum age for the filter. |
|
|
|
|
|
Returns: |
|
|
tuple: A Plotly figure, a Pandas DataFrame, and a summary string. |
|
|
""" |
|
|
if top_n is None: |
|
|
top_n = 10 |
|
|
if patient_sex is None: |
|
|
patient_sex = "all" |
|
|
if min_age is None: |
|
|
min_age = 0 |
|
|
if max_age is None: |
|
|
max_age = 120 |
|
|
|
|
|
sex_code = None |
|
|
if patient_sex == "Male": |
|
|
sex_code = "1" |
|
|
elif patient_sex == "Female": |
|
|
sex_code = "2" |
|
|
|
|
|
age_range = None |
|
|
if min_age > 0 or max_age < 120: |
|
|
age_range = (min_age, max_age) |
|
|
|
|
|
data = get_top_adverse_events(drug_name, limit=top_n, patient_sex=sex_code, age_range=age_range) |
|
|
|
|
|
if "error" in data: |
|
|
error_message = f"An error occurred: {data['error']}" |
|
|
return create_placeholder_chart(error_message), pd.DataFrame(), error_message |
|
|
|
|
|
if "results" not in data or not data["results"]: |
|
|
message = f"No adverse event data found for '{drug_name}'. The drug may not be in the database or it might be misspelled." |
|
|
return create_placeholder_chart(message), pd.DataFrame(), message |
|
|
|
|
|
chart = create_bar_chart(data, drug_name) |
|
|
|
|
|
df = pd.DataFrame(data["results"]) |
|
|
df = df.rename(columns={"term": "Adverse Event", "count": "Report Count"}) |
|
|
|
|
|
total_reports = data.get("meta", {}).get("total_reports_for_query", 0) |
|
|
if total_reports > 0: |
|
|
df['Relative Frequency (%)'] = ((df['Report Count'] / total_reports) * 100).round(2) |
|
|
else: |
|
|
df['Relative Frequency (%)'] = 0.0 |
|
|
|
|
|
header = ( |
|
|
f"### Top {len(df)} Adverse Events for '{drug_name.title()}'\n" |
|
|
f"Based on **{total_reports:,}** total reports matching the given filters.\n" |
|
|
"**Source**: FDA FAERS via OpenFDA\n" |
|
|
"**Disclaimer**: Spontaneous reports do not prove causation. Consult a healthcare professional." |
|
|
) |
|
|
return chart, df, header |
|
|
|
|
|
def serious_outcomes_tool(drug_name: str, top_n: int = 6): |
|
|
""" |
|
|
MCP Tool: Finds the top reported serious outcomes for a given drug. |
|
|
|
|
|
Args: |
|
|
drug_name (str): The generic name of the drug is preferred. A small sample of brand names (e.g., 'Tylenol') are converted to generic names for demonstration purposes. |
|
|
top_n (int): The number of top serious outcomes to return. |
|
|
|
|
|
Returns: |
|
|
tuple: A Plotly figure, a Pandas DataFrame, and a summary string. |
|
|
""" |
|
|
if top_n is None: |
|
|
top_n = 6 |
|
|
data = get_serious_outcomes(drug_name, limit=top_n) |
|
|
|
|
|
if "error" in data: |
|
|
error_message = f"An error occurred: {data['error']}" |
|
|
return create_placeholder_chart(error_message), pd.DataFrame(), error_message |
|
|
|
|
|
if "results" not in data or not data["results"]: |
|
|
message = f"No serious outcome data found for '{drug_name}'. The drug may not be in the database or it might be misspelled." |
|
|
return create_placeholder_chart(message), pd.DataFrame(), message |
|
|
|
|
|
chart = create_outcome_chart(data, drug_name) |
|
|
|
|
|
df = pd.DataFrame(data["results"]) |
|
|
df = df.rename(columns={"term": "Serious Outcome", "count": "Report Count"}) |
|
|
|
|
|
total_serious_reports = data.get("meta", {}).get("total_reports_for_query", 0) |
|
|
if total_serious_reports > 0: |
|
|
df['% of Serious Reports'] = ((df['Report Count'] / total_serious_reports) * 100).round(2) |
|
|
else: |
|
|
df['% of Serious Reports'] = 0.0 |
|
|
|
|
|
header = ( |
|
|
f"### Top {len(df)} Serious Outcomes for '{drug_name.title()}'\n" |
|
|
f"Out of **{total_serious_reports:,}** total serious reports. " |
|
|
"Note: a single report may be associated with multiple outcomes.\n" |
|
|
"**Source**: FDA FAERS via OpenFDA\n" |
|
|
"**Disclaimer**: Spontaneous reports do not prove causation. Consult a healthcare professional." |
|
|
) |
|
|
return chart, df, header |
|
|
|
|
|
def drug_event_stats_tool(drug_name: str, event_name: str): |
|
|
""" |
|
|
MCP Tool: Gets the total number of reports for a specific drug and adverse event pair. |
|
|
|
|
|
Args: |
|
|
drug_name (str): The generic name of the drug is preferred. A small sample of brand names (e.g., 'Tylenol') are converted to generic names for demonstration purposes. |
|
|
event_name (str): The name of the adverse event to search for. |
|
|
|
|
|
Returns: |
|
|
str: A formatted string with the total count of reports. |
|
|
""" |
|
|
data = get_drug_event_pair_frequency(drug_name, event_name) |
|
|
return format_pair_frequency_results(data, drug_name, event_name) |
|
|
|
|
|
def time_series_tool(drug_name: str, event_name: str, aggregation: str): |
|
|
""" |
|
|
MCP Tool: Creates a time-series plot for a drug-event pair. |
|
|
|
|
|
Args: |
|
|
drug_name (str): The generic name of the drug is preferred. A small sample of brand names (e.g., 'Tylenol') are converted to generic names for demonstration purposes. |
|
|
event_name (str): The name of the adverse event. |
|
|
aggregation (str): Time aggregation ('Yearly' or 'Quarterly'). |
|
|
|
|
|
Returns: |
|
|
A Plotly figure. |
|
|
""" |
|
|
agg_code = 'Y' if aggregation == 'Yearly' else 'Q' |
|
|
data = get_time_series_data(drug_name, event_name) |
|
|
|
|
|
if "error" in data or not data.get("results"): |
|
|
return create_placeholder_chart(f"No time-series data found for '{drug_name}' and '{event_name}'.") |
|
|
|
|
|
chart = create_time_series_chart(data, drug_name, event_name, time_aggregation=agg_code) |
|
|
return chart |
|
|
|
|
|
def report_source_tool(drug_name: str, top_n: int = 5): |
|
|
""" |
|
|
MCP Tool: Creates a pie chart of report sources for a given drug. |
|
|
|
|
|
Args: |
|
|
drug_name (str): The generic name of the drug is preferred. A small sample of brand names (e.g., 'Tylenol') are converted to generic names for demonstration purposes. |
|
|
top_n (int): The number of top sources to return. |
|
|
|
|
|
Returns: |
|
|
tuple: A Plotly figure, a Pandas DataFrame, and a summary string. |
|
|
""" |
|
|
if top_n is None: |
|
|
top_n = 5 |
|
|
data = get_report_source_data(drug_name, limit=top_n) |
|
|
|
|
|
if "error" in data: |
|
|
error_message = f"An error occurred: {data['error']}" |
|
|
return create_placeholder_chart(error_message), pd.DataFrame(), error_message |
|
|
|
|
|
if not data or not data.get("results"): |
|
|
message = f"No report source data found for '{drug_name}'." |
|
|
return create_placeholder_chart(message), pd.DataFrame(), message |
|
|
|
|
|
chart = create_pie_chart(data, drug_name) |
|
|
|
|
|
df = pd.DataFrame(data['results']) |
|
|
df = df.rename(columns={"term": "Source", "count": "Report Count"}) |
|
|
|
|
|
total_reports = data.get("meta", {}).get("total_reports_for_query", 0) |
|
|
if total_reports > 0: |
|
|
df['Percentage'] = ((df['Report Count'] / total_reports) * 100).round(2) |
|
|
else: |
|
|
df['Percentage'] = 0.0 |
|
|
|
|
|
header = ( |
|
|
f"### Report Sources for '{drug_name.title()}'\n" |
|
|
f"Based on **{total_reports:,}** reports with source information." |
|
|
) |
|
|
return chart, df, header |
|
|
|
|
|
|
|
|
|
|
|
with open("gradio_readme.md", "r") as f: |
|
|
readme_content = f.read() |
|
|
|
|
|
with gr.Blocks(title="Medication Adverse-Event Explorer") as demo: |
|
|
gr.Markdown("# Medication Adverse-Event Explorer") |
|
|
|
|
|
with gr.Tabs(): |
|
|
with gr.TabItem("About"): |
|
|
gr.HTML(""" |
|
|
<div style="display: flex; justify-content: center;"> |
|
|
<iframe width="560" height="315" src="https://www.youtube.com/embed/noGG-lQwn2U" |
|
|
title="YouTube video player" frameborder="0" |
|
|
allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture" |
|
|
allowfullscreen> |
|
|
</iframe> |
|
|
</div> |
|
|
""") |
|
|
gr.Markdown(readme_content) |
|
|
|
|
|
with gr.TabItem("Top Events"): |
|
|
gr.Interface( |
|
|
fn=top_adverse_events_tool, |
|
|
inputs=[ |
|
|
gr.Textbox( |
|
|
label="Drug Name", |
|
|
info="Enter a brand or generic drug name (e.g., 'Aspirin', 'Lisinopril')." |
|
|
), |
|
|
gr.Slider( |
|
|
5, 50, |
|
|
value=10, |
|
|
label="Number of Events to Show", |
|
|
step=1 |
|
|
), |
|
|
gr.Radio( |
|
|
["All", "Male", "Female"], |
|
|
label="Patient Sex", |
|
|
value="All" |
|
|
), |
|
|
gr.Slider( |
|
|
0, 120, |
|
|
value=0, |
|
|
label="Minimum Age", |
|
|
step=1 |
|
|
), |
|
|
gr.Slider( |
|
|
0, 120, |
|
|
value=120, |
|
|
label="Maximum Age", |
|
|
step=1 |
|
|
), |
|
|
], |
|
|
outputs=[ |
|
|
gr.Plot(label="Top Adverse Events Chart"), |
|
|
gr.DataFrame(label="Top Adverse Events", interactive=False), |
|
|
gr.Markdown() |
|
|
], |
|
|
title="Top Adverse Events by Drug", |
|
|
description="Find the most frequently reported adverse events for a specific medication.", |
|
|
examples=[["Lisinopril"], ["Ozempic"], ["Metformin"]], |
|
|
allow_flagging="never", |
|
|
) |
|
|
|
|
|
with gr.TabItem("Serious Outcomes"): |
|
|
gr.Interface( |
|
|
fn=serious_outcomes_tool, |
|
|
inputs=[ |
|
|
gr.Textbox( |
|
|
label="Drug Name", |
|
|
info="Enter a brand or generic drug name (e.g., 'Aspirin', 'Lisinopril')." |
|
|
), |
|
|
gr.Slider(1, 6, value=6, label="Number of Outcomes to Show", step=1), |
|
|
], |
|
|
outputs=[ |
|
|
gr.Plot(label="Top Serious Outcomes Chart"), |
|
|
gr.DataFrame(label="Top Serious Outcomes", interactive=False), |
|
|
gr.Markdown() |
|
|
], |
|
|
title="Serious Outcome Analysis", |
|
|
description="Find the most frequently reported serious outcomes (e.g., hospitalization, death) for a specific medication.", |
|
|
examples=[["Lisinopril"], ["Ozempic"], ["Metformin"]], |
|
|
allow_flagging="never", |
|
|
) |
|
|
|
|
|
with gr.TabItem("Event Frequency"): |
|
|
gr.Interface( |
|
|
fn=drug_event_stats_tool, |
|
|
inputs=[ |
|
|
gr.Textbox(label="Drug Name", info="e.g., 'Ibuprofen'"), |
|
|
gr.Textbox(label="Adverse Event", info="e.g., 'Headache'") |
|
|
], |
|
|
outputs=[gr.Textbox(label="Report Count", lines=5)], |
|
|
title="Drug/Event Pair Frequency", |
|
|
description="Get the total number of reports for a specific drug and adverse event combination.", |
|
|
examples=[["Lisinopril", "Cough"], ["Ozempic", "Nausea"]], |
|
|
) |
|
|
|
|
|
with gr.TabItem("Time-Series Trends"): |
|
|
gr.Interface( |
|
|
fn=time_series_tool, |
|
|
inputs=[ |
|
|
gr.Textbox(label="Drug Name", info="e.g., 'Ibuprofen'"), |
|
|
gr.Textbox(label="Adverse Event", info="e.g., 'Headache'"), |
|
|
gr.Radio(["Yearly", "Quarterly"], label="Aggregation", value="Yearly") |
|
|
], |
|
|
outputs=[gr.Plot(label="Report Trends")], |
|
|
title="Time-Series Trend Plotting", |
|
|
description="Plot the number of adverse event reports over time for a specific drug-event pair.", |
|
|
examples=[["Lisinopril", "Cough", "Yearly"], ["Ozempic", "Nausea", "Quarterly"]], |
|
|
) |
|
|
|
|
|
with gr.TabItem("Report Sources"): |
|
|
gr.Interface( |
|
|
fn=report_source_tool, |
|
|
inputs=[ |
|
|
gr.Textbox(label="Drug Name", info="e.g., 'Aspirin', 'Lisinopril'"), |
|
|
gr.Slider(1, 5, value=5, label="Number of Sources to Show", step=1), |
|
|
], |
|
|
outputs=[ |
|
|
gr.Plot(label="Report Source Breakdown"), |
|
|
gr.DataFrame(label="Report Source Data", interactive=False), |
|
|
gr.Markdown() |
|
|
], |
|
|
title="Report Source Breakdown", |
|
|
description="Show a pie chart breaking down the source of the reports (e.g., Consumer, Physician).", |
|
|
examples=[["Lisinopril"], ["Ibuprofen"]], |
|
|
allow_flagging="never", |
|
|
) |
|
|
|
|
|
if __name__ == "__main__": |
|
|
demo.launch(mcp_server=True, server_name="0.0.0.0") |
