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_2")) # 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)