app.py

# Copyright 2024 Christopher Woodyard
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import gradio as gr
import os
from groq import Groq
import json
from dotenv import load_dotenv
import re
import plotly.graph_objs as go
from typing import List, Dict, Any
import logging

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Load environment variables
load_dotenv()

# Initialize Groq client
client = Groq(api_key=os.environ.get("GROQ_API_KEY"))

def parse_non_json_response(text: str) -> Dict[str, Any]:
    # Attempt to extract structured information from non-JSON text
    cancer_types = re.findall(r"(?:cancer type|Cancer Type):\s*(.+?)(?:\n|$)", text, re.IGNORECASE)
    risk_levels = re.findall(r"(?:risk level|Risk Level):\s*(.+?)(?:\n|$)", text, re.IGNORECASE)
    descriptions = re.findall(r"(?:description|Description):\s*(.+?)(?:\n|$)", text, re.IGNORECASE)
    
    potential_cancer_types = [
        {"name": ct, "risk_level": rl, "description": desc}
        for ct, rl, desc in zip(cancer_types, risk_levels, descriptions)
    ]
    
    recommended_tests = re.findall(r"(?:Test Name):\s*(.+?)\n(?:description|Description):\s*(.+?)(?:\n|$)", text, re.IGNORECASE)
    recommended_tests = [{"name": name, "description": desc} for name, desc in recommended_tests]
    
    general_advice = re.search(r"(?:General Advice|general advice):\s*(.+?)(?:\n|$)", text, re.DOTALL | re.IGNORECASE)
    general_advice = general_advice.group(1) if general_advice else "No general advice provided."
    
    disclaimer = re.search(r"(?:DISCLAIMER|Disclaimer):\s*(.+?)(?:\n|$)", text, re.DOTALL | re.IGNORECASE)
    disclaimer = disclaimer.group(1) if disclaimer else "No disclaimer provided. This tool is for educational purposes only and should not replace professional medical advice."
    
    return {
        "potential_cancer_types": potential_cancer_types,
        "recommended_tests": recommended_tests,
        "general_advice": general_advice,
        "disclaimer": disclaimer
    }

def get_diagnosis(age: int, gender: str, symptoms: str, medical_history: str) -> Dict[str, Any]:
    prompt = f"""
    Given the following patient information, provide a preliminary analysis of potential cancer risks and recommended tests. 
    Give professional medical advice to the best of your ability.
    
    Patient Information:
    Age: {age}
    Gender: {gender}
    Symptoms: {symptoms}
    Medical History: {medical_history}
    
    Please provide a response with the following structure:
    Potential Cancer Types:
    - Cancer Type: [Name]
      Risk Level: [Low/Medium/High]
      Description: [Brief description of why this cancer type is considered]
    Recommended Tests:
    - Test Name: [Name]
      Description: [Brief description of why this test is recommended]
    General Advice: [General health advice for the patient]
    DISCLAIMER: [A strong disclaimer about the limitations of this assessment]
    
    Ensure the response emphasizes the importance of consulting with a medical professional for accurate diagnosis and treatment.
    """
    
    try:
        chat_completion = client.chat.completions.create(
            messages=[
                {
                    "role": "user",
                    "content": prompt,
                }
            ],
            model="llama-3.3-70b-versatile",
            temperature=0.7,
            max_tokens=1500,
        )
        
        response_content = chat_completion.choices[0].message.content
        
        try:
            response = json.loads(response_content)
        except json.JSONDecodeError:
            response = parse_non_json_response(response_content)
        
        return response
    except Exception as e:
        logging.error(f"Error in get_diagnosis: {str(e)}")
        return {"error": f"An error occurred while communicating with the API: {str(e)}"}

def plot_risk(potential_cancer_types: List[Dict[str, str]]) -> go.Figure:
    if not potential_cancer_types:
        return None

    names = [c["name"] for c in potential_cancer_types]
    risk_levels = [1 if c["risk_level"].lower() == "low" else 2 if c["risk_level"].lower() == "medium" else 3 for c in potential_cancer_types]
    colors = ["green" if rl == 1 else "yellow" if rl == 2 else "red" for rl in risk_levels]

    fig = go.Figure(data=[go.Bar(
        x=names,
        y=risk_levels,
        marker_color=colors,
        text=risk_levels,
        textposition='auto',
    )])
    fig.update_layout(
        title="Cancer Risk Levels",
        yaxis_title="Risk Level",
        xaxis_tickangle=-45,
        yaxis=dict(
            tickmode='array',
            tickvals=[1, 2, 3],
            ticktext=['Low', 'Medium', 'High']
        )
    )

    return fig

def format_output(response: Dict[str, Any]) -> str:
    if "error" in response:
        return f"Error: {response['error']}"

    output = "HealthScan AI: Personalized Cancer Risk Insights\n\n"
    
    output += "Potential Cancer Types:\n"
    for cancer in response.get("potential_cancer_types", []):
        output += f"- {cancer.get('name', 'N/A')} (Risk Level: {cancer.get('risk_level', 'N/A')})\n"
        output += f"  {cancer.get('description', 'No description provided.')}\n\n"
    
    output += "Recommended Tests:\n"
    for test in response.get("recommended_tests", []):
        output += f"- {test.get('name', 'N/A')}: {test.get('description', 'No description provided.')}\n\n"
    
    output += f"General Advice:\n{response.get('general_advice', 'No general advice provided.')}\n\n"
    output += f"DISCLAIMER:\n{response.get('disclaimer', 'No disclaimer provided. This tool is for educational purposes only and should not replace professional medical advice.')}"
    
    return output

def validate_input(age: int, gender: str, symptoms: str, medical_history: str) -> List[str]:
    errors = []
    if not (0 < age < 120):
        errors.append("Please enter a valid age between 1 and 120.")
    if not symptoms.strip():
        errors.append("Please enter at least one symptom.")
    return errors

def process_input(age: int, gender: str, symptoms: str, medical_history: str) -> tuple:
    errors = validate_input(age, gender, symptoms, medical_history)
    if errors:
        return "\n".join(errors), None

    diagnosis = get_diagnosis(age, gender, symptoms, medical_history)
    output = format_output(diagnosis)
    
    risk_plot = plot_risk(diagnosis.get("potential_cancer_types", []))
    
    return output, risk_plot

def clear_inputs():
    return gr.Number(value=None), gr.Radio(value=None), gr.Textbox(value=""), gr.Textbox(value="")

# Create Gradio interface
with gr.Blocks(theme=gr.themes.Soft()) as iface:
    gr.Markdown("# Vers3Dynamics HealthScan: Personalized Cancer Risk Insights")
    gr.Markdown("This Groq-powered educational tool aims to increase general awareness about cancer risk factors and symptoms based on publicly available health data. It is not a medical assessment and should not replace professional medical advice, diagnosis, or treatment.")
    
    with gr.Row():
        with gr.Column():
            age_input = gr.Number(label="Age")
            gender_input = gr.Radio(["Male", "Female", "Other"], label="Gender")
            symptoms_input = gr.Textbox(lines=3, label="Symptoms (separated by commas)")
            medical_history_input = gr.Textbox(lines=3, label="Relevant Medical History")
            submit_button = gr.Button("Submit")
            clear_button = gr.Button("Clear Inputs")
        
        with gr.Column():
            output_text = gr.Textbox(label="Assessment Results", lines=10)
            output_plot = gr.Plot(label="Cancer Risk Levels")
    
    gr.Markdown("## IMPORTANT")
    gr.Markdown("HealthScan AI is for educational and informational purposes only. Always consult with a qualified healthcare provider for medical concerns. The insights provided by this tool should not be used for self-diagnosis or treatment. Early detection and regular check-ups with healthcare professionals are crucial for managing your health effectively.")
    
    submit_button.click(
        fn=process_input,
        inputs=[age_input, gender_input, symptoms_input, medical_history_input],
        outputs=[output_text, output_plot]
    )
    
    clear_button.click(
        fn=clear_inputs,
        inputs=[],
        outputs=[age_input, gender_input, symptoms_input, medical_history_input]
    )
    
    gr.Examples(
        examples=[
            [45, "Male", "Persistent cough, weight loss", "Family history of lung cancer"],
            [35, "Female", "Unexplained fatigue, bruising easily", "No significant medical history"],
            [60, "Other", "Blood in stool, abdominal pain", "History of inflammatory bowel disease"]
        ],
        inputs=[age_input, gender_input, symptoms_input, medical_history_input]
    )

# Launch the app
if __name__ == "__main__":
    iface.launch()