app.py

from groq import Groq
import gradio as gr
from gtts import gTTS
import uuid
import base64
from io import BytesIO
import os
import logging
import spacy
from transformers import pipeline
import torch
from PIL import Image
from torchvision import transforms
import pathlib
import cv2  # Import OpenCV
import numpy as np

# Pathlib adjustment for Windows compatibility
# temp = pathlib.PosixPath
# pathlib.PosixPath = pathlib.WindowsPath

# Set up logger
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
console_handler = logging.StreamHandler()
file_handler = logging.FileHandler('chatbot_log.log')
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
console_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)
logger.addHandler(console_handler)
logger.addHandler(file_handler)

#Initialize Groq Client
client = Groq(api_key=os.getenv("GROQ_API_KEY_1"))
# logger.info(f"API Key: {client}")  # Just for debugging

# # Initialize Groq Client
#client = Groq(api_key="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")

# Initialize spaCy NLP model for named entity recognition (NER)
spacy.cli.download("en_core_web_sm")
nlp = spacy.load("en_core_web_sm")

# Initialize sentiment analysis model using Hugging Face
sentiment_analyzer = pipeline("sentiment-analysis")

# Load pre-trained YOLOv5 model
def load_yolov5_model():
    model = torch.hub.load(r"ultralytics/yolov5", 'custom', path=r'./models/best.pt')
    model.eval()
    return model

model = load_yolov5_model()

# Function to preprocess user input for better NLP understanding
def preprocess_input(user_input):
    user_input = user_input.strip().lower()
    return user_input

# Function for sentiment analysis (optional)
def analyze_sentiment(user_input):
    result = sentiment_analyzer(user_input)
    return result[0]['label']

# Function to extract medical entities from input using NER
symptoms = [
    "fever", "cough", "headache", "nausea", "pain", "fatigue", "dizziness",
    "shortness of breath", "sore throat", "runny nose", "congestion", "diarrhea",
    "vomiting", "chills", "sweating", "loss of appetite", "insomnia", 
    "itching", "rash", "swelling", "bleeding", "burning sensation",
    "weakness", "tingling", "numbness", "muscle cramps", "joint pain",
    "blurred vision", "double vision", "dry eyes", "sensitivity to light",
    "difficulty breathing", "palpitations", "chest pain", "back pain",
    "stomach ache", "abdominal pain", "weight loss", "weight gain",
    "frequent urination", "difficulty urinating", "anxiety", "depression",
    "irritability", "confusion", "memory loss", "bruising"
]
diseases = [
    "diabetes", "cancer", "asthma", "flu", "pneumonia", "hypertension",
    "arthritis", "bronchitis", "migraine", "stroke", "heart attack",
    "coronary artery disease", "tuberculosis", "malaria", "dengue",
    "hepatitis", "anemia", "thyroid disease", "eczema", "psoriasis",
    "osteoporosis", "parkinson's", "alzheimer's", "depression",
    "anxiety disorder", "schizophrenia", "epilepsy", "bipolar disorder",
    "chronic kidney disease", "liver cirrhosis", "HIV", "AIDS",
    "covid-19", "cholera", "smallpox", "measles", "mumps", 
    "rubella", "whooping cough", "obesity", "GERD", "IBS", 
    "celiac disease", "ulcerative colitis", "Crohn's disease", 
    "sleep apnea", "hypothyroidism", "hyperthyroidism"
]

def extract_medical_entities(user_input):
    user_input = preprocess_input(user_input)
    medical_entities = []
    for word in user_input.split():
        if word in symptoms or word in diseases:
            medical_entities.append(word)
    return medical_entities

# Function to encode the image
def encode_image(uploaded_image):
    try:
        logger.debug("Encoding image...")
        buffered = BytesIO()
        uploaded_image.save(buffered, format="PNG")
        logger.debug("Image encoding complete.")
        return base64.b64encode(buffered.getvalue()).decode("utf-8")
    except Exception as e:
        logger.error(f"Error encoding image: {e}")
        raise

# Initialize messages
def initialize_messages():
    return [{"role": "system", "content": '''You are Dr. HealthBuddy, a professional, empathetic, and knowledgeable virtual doctor chatbot.'''}]

messages = initialize_messages()

# Function for image prediction using YOLOv5
def predict_image(image):
    try:
        # Debug: Check if the image is None
        if image is None:
            return "Error: No image uploaded.", "No description available."

        # Convert PIL image to NumPy array (OpenCV format)
        image_np = np.array(image)  # Convert PIL image to NumPy array

        # Convert RGB to BGR (OpenCV uses BGR by default)
        image_np = cv2.cvtColor(image_np, cv2.COLOR_RGB2BGR)

        # Resize the image to match the model's expected input size
        image_resized = cv2.resize(image_np, (224, 224))

        # Transform the image for the model
        transform = transforms.Compose([
            transforms.ToTensor(),  # Convert image to tensor
            transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),  # Normalize
        ])
        im = transform(image_resized).unsqueeze(0)  # Add batch dimension (BCHW)

        # Get predictions
        with torch.no_grad():
            output = model(im)  # Raw model output (logits)

        # Apply softmax to get confidence scores
        softmax = torch.nn.Softmax(dim=1)
        probs = softmax(output)

        # Get the predicted class and its confidence score
        predicted_class_id = torch.argmax(probs, dim=1).item()
        confidence_score = probs[0, predicted_class_id].item()

        # Get predicted class name if available
        if hasattr(model, 'names'):
            class_name = model.names[predicted_class_id]
            prediction_result = f"Predicted Class: {class_name}\nConfidence: {confidence_score:.4f}"
            description = get_description(class_name)  # Function to get description
        else:
            prediction_result = f"Predicted Class ID: {predicted_class_id}\nConfidence: {confidence_score:.4f}"
            description = "No description available."

        # Display the image with OpenCV (optional)
        cv2.imshow("Processed Image", image_resized)
        cv2.waitKey(1)  # Wait for 1 ms to display the image

        return prediction_result, description

    except Exception as e:
        logger.error(f"Error in image prediction: {e}")
        return f"An error occurred during image prediction: {e}", "No description available."

# Function to get description based on predicted class
def get_description(class_name):
    descriptions = {
        "bcc": "Basal cell carcinoma (BCC) is a type of skin cancer that begins in the basal cells. It often appears as a slightly transparent bump on the skin, though it can take other forms. BCC grows slowly and is unlikely to spread to other parts of the body, but early treatment is important to prevent damage to surrounding tissues.",
        "atopic": "Atopic dermatitis is a chronic skin condition characterized by itchy, inflamed skin. It is common in individuals with a family history of allergies or asthma.",
        "acne": "Acne is a skin condition that occurs when hair follicles become clogged with oil and dead skin cells. It often causes pimples, blackheads, and whiteheads, and is most common among teenagers.",
        # Add more descriptions as needed
    }
    return descriptions.get(class_name.lower(), "No description available.")

# Custom LLM Bot Function
def customLLMBot(user_input, uploaded_image, chat_history):
    try:
        global messages
        logger.info("Processing input...")

        # Preprocess the user input
        user_input = preprocess_input(user_input)
        
        # Analyze sentiment (Optional)
        sentiment = analyze_sentiment(user_input)
        logger.info(f"Sentiment detected: {sentiment}")

        # Extract medical entities (Optional)
        medical_entities = extract_medical_entities(user_input)
        logger.info(f"Extracted medical entities: {medical_entities}")

        # Append user input to the chat history
        chat_history.append(("user", user_input))

        if uploaded_image is not None:
            # Encode the image to base64
            base64_image = encode_image(uploaded_image)

            logger.debug(f"Image received, size: {len(base64_image)} bytes")

            # Create a message for the image prompt
            messages_image = [
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": "What's in this image?"},
                        {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{base64_image}"}}
                    ]
                }
            ]

            logger.info("Sending image to Groq API for processing...")
            response = client.chat.completions.create(
                model="llama-3.2-11b-vision-preview",
                messages=messages_image,
            )
            logger.info("Image processed successfully.")
        else:
            # Process text input
            logger.info("Processing text input...")
            messages.append({
                "role": "user",
                "content": user_input
            })
            response = client.chat.completions.create(
                model="llama-3.2-11b-vision-preview",
                messages=messages,
            )
            logger.info("Text processed successfully.")

        # Extract the reply
        LLM_reply = response.choices[0].message.content
        logger.debug(f"LLM reply: {LLM_reply}")

        # Append the bot's response to the chat history
        chat_history.append(("bot", LLM_reply))
        messages.append({"role": "assistant", "content": LLM_reply})

        # Generate audio for response
        audio_file = f"response_{uuid.uuid4().hex}.mp3"
        tts = gTTS(LLM_reply, lang='en')
        tts.save(audio_file)
        logger.info(f"Audio response saved as {audio_file}")

        # Return chat history and audio file
        return chat_history, audio_file

    except Exception as e:
        logger.error(f"Error in customLLMBot function: {e}")
        return [("user", user_input or "Image uploaded"), ("bot", f"An error occurred: {e}")], None

# Gradio Interface
def chatbot_ui():
    with gr.Blocks() as demo:
        gr.Markdown("# Healthcare Chatbot Doctor")

        # State for user chat history
        chat_history = gr.State([])

        # Layout for chatbot and input box alignment
        with gr.Row():
            with gr.Column(scale=3):  # Main column for chatbot
                chatbot = gr.Chatbot(label="Responses", elem_id="chatbot")
                user_input = gr.Textbox(
                    label="Ask a health-related question",
                    placeholder="Describe your symptoms...",
                    elem_id="user-input",
                    lines=1,
                )
            with gr.Column(scale=1):  # Side column for image and buttons
                uploaded_image = gr.Image(label="Upload an Image", type="pil")
                submit_btn = gr.Button("Submit")
                clear_btn = gr.Button("Clear")
                audio_output = gr.Audio(label="Audio Response")

        # New section for image prediction (left and right layout)
        with gr.Row():
            # Left side: Upload image
            with gr.Column():
                gr.Markdown("### Upload Image for Prediction")
                prediction_image = gr.Image(label="Upload Image", type="pil")
                predict_btn = gr.Button("Predict")

            # Right side: Prediction result and description
            with gr.Column():
                gr.Markdown("### Prediction Result")
                prediction_output = gr.Textbox(label="Result", interactive=False)

                # Description column
                gr.Markdown("### Description")
                description_output = gr.Textbox(label="Description", interactive=False)

                # Clear button for prediction result (below description box)
                clear_prediction_btn = gr.Button("Clear Prediction")

        # Define actions
        def handle_submit(user_query, image, history):
            logger.info("User submitted a query.")
            response, audio = customLLMBot(user_query, image, history)
            return response, audio, None, "", history

        # Clear prediction result and image
        def clear_prediction(prediction_image, prediction_output, description_output):
            return None, "", ""

        # Submit on pressing Enter key
        user_input.submit(
            handle_submit,
            inputs=[user_input, uploaded_image, chat_history],
            outputs=[chatbot, audio_output, uploaded_image, user_input, chat_history],
        )

        # Submit on button click
        submit_btn.click(
            handle_submit,
            inputs=[user_input, uploaded_image, chat_history],
            outputs=[chatbot, audio_output, uploaded_image, user_input, chat_history],
        )

        # Action for clearing all fields
        clear_btn.click(
            lambda: ([], "", None, []),
            inputs=[],
            outputs=[chatbot, user_input, uploaded_image, chat_history],
        )

        # Action for image prediction
        predict_btn.click(
            predict_image,
            inputs=[prediction_image],
            outputs=[prediction_output, description_output],  # Update both outputs
        )

        # Action for clearing prediction result and image
        clear_prediction_btn.click(
            clear_prediction,
            inputs=[prediction_image, prediction_output, description_output],
            outputs=[prediction_image, prediction_output, description_output],
        )

    return demo

# Launch the interface
#chatbot_ui().launch(server_name="localhost", server_port=7860)

# Launch the interface
chatbot_ui().launch(server_name="0.0.0.0", server_port=7860)