update space

app.py

@@ -1,17 +1,16 @@
+# app.py
+
 # Import necessary libraries
 import os
 import logging
 import dotenv
-import tensorflow as tf
-from keras.models import load_model
 import gradio as gr
-import cv2
 import numpy as np
-from huggingface_hub import login
 from pymongo import MongoClient
-from transformers import AutoModelForCausalLM, AutoTokenizer
 from datetime import datetime
 from werkzeug.security import generate_password_hash, check_password_hash
+from utils import PoultryFarmBot, llama3_response
+from transformers import AutoModelForCausalLM, AutoTokenizer
 
 # Load environment variables from .env file
 dotenv.load_dotenv()
@@ -29,7 +28,6 @@ enquiries_collection = db.enquiries  # Collection to store farmer enquiries
 users_collection = db.users  # Collection to store user credentials
 logs_collection = db.logs  # Collection to store application logs
 
-
 def log_to_db(level, message):
     log_entry = {
         "level": level,
@@ -38,14 +36,12 @@ def log_to_db(level, message):
     }
     logs_collection.insert_one(log_entry)
 
-
 # Override logger methods to also log to MongoDB
 class MongoHandler(logging.Handler):
     def emit(self, record):
         log_entry = self.format(record)
         log_to_db(record.levelname, log_entry)
 
-
 mongo_handler = MongoHandler()
 mongo_handler.setLevel(logging.INFO)
 formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
@@ -57,230 +53,9 @@ tok = os.getenv('HF_TOKEN')
 if tok:
     # Log in to Hugging Face using the token from environment variables
     logger.info("Logging in to Hugging Face.")
-    login(token=tok, add_to_git_credential=True)
 else:
     logger.warning("Hugging Face token not found in environment variables.")
 
-# GPU Setup for TensorFlow
-logger.info("TensorFlow version: %s", tf.__version__)
-logger.info("Eager execution: %s", tf.executing_eagerly())
-logger.info("TensorFlow GPU Available: %s", tf.config.list_physical_devices('GPU'))
-
-# Set TensorFlow to use mixed precision if GPU is available
-from tensorflow.keras import mixed_precision
-
-if len(tf.config.list_physical_devices('GPU')) > 0:
-    # Set mixed precision policy to improve GPU performance
-    policy = mixed_precision.Policy('mixed_float16')
-    mixed_precision.set_global_policy(policy)
-    logger.info("Using mixed precision with GPU")
-else:
-    logger.info("Using CPU without mixed precision")
-
-
-# Model loading utility
-def load_model_with_device(model_path, device_name):
-    """
-    Load a Keras model on the specified device.
-
-    Args:
-        model_path (str): Path to the model file.
-        device_name (str): Device to load the model on ('/GPU:0' or '/CPU:0').
-
-    Returns:
-        Model object if successful, None otherwise.
-    """
-    try:
-        logger.info(f"Loading model from '{model_path}' on {device_name}.")
-        with tf.device(device_name):
-            model = load_model(model_path, compile=True)
-            logger.info(f"Model '{model_path}' loaded successfully on {device_name}.")
-            return model
-    except Exception as e:
-        logger.error(f"Error loading model '{model_path}': {e}")
-        return None
-
-
-# Load disease detection model and authentication model
-device_name = '/GPU:0' if len(tf.config.list_physical_devices('GPU')) > 0 else '/CPU:0'
-logger.info("Loading disease detection model.")
-my_model = load_model_with_device('models/Final_Chicken_disease_model.h5', device_name)
-
-# Disease names and recommendations
-name_disease = {0: 'Coccidiosis', 1: 'Healthy', 2: 'New Castle Disease', 3: 'Salmonella'}
-result = {0: 'Critical', 1: 'No issue', 2: 'Critical', 3: 'Critical'}
-recommend = {
-    0: 'Panadol',
-    1: 'You have no need Medicine',
-    2: 'Paracetamol',
-    3: 'Ponstan'
-}
-
-
-# PoultryFarmBot class definition
-class PoultryFarmBot:
-    def __init__(self, db):
-        """
-        Initialize the PoultryFarmBot with a MongoDB database connection.
-
-        Args:
-            db (MongoClient): MongoDB database instance.
-        """
-        self.db = db  # MongoDB database for future use
-        logger.info("PoultryFarmBot initialized with MongoDB connection.")
-
-    def preprocess_image(self, image):
-        """
-        Preprocess the input image for disease detection.
-
-        Args:
-            image (numpy.ndarray): Input image to preprocess.
-
-        Returns:
-            numpy.ndarray: Preprocessed image ready for model input.
-        """
-        try:
-            logger.info("Preprocessing image for disease detection.")
-            # Resize the image to match the model input size (224x224)
-            image_check = cv2.resize(image, (224, 224))
-            # Add a batch dimension to the image
-            image_check = np.expand_dims(image_check, axis=0)
-            logger.info("Image preprocessing successful.")
-            return image_check
-        except Exception as e:
-            logger.error(f"Error in image preprocessing: {e}")
-            return None
-
-    def predict_disease(self, image):
-        """
-        Predict the disease from the given poultry fecal image.
-
-        Args:
-            image (numpy.ndarray): Input image to predict.
-
-        Returns:
-            tuple: Detailed response, disease name, status, and recommendation.
-        """
-        # Preprocess the image
-        logger.info("Starting disease prediction.")
-        image_check = self.preprocess_image(image)
-        if image_check is None:
-            logger.warning("Image preprocessing failed.")
-            return "Image preprocessing failed.", None, None, None
-
-        # Predict the disease using the loaded model
-        try:
-            logger.info("Running model prediction.")
-            indx = my_model.predict(image_check).argmax()
-            name = name_disease.get(indx, "Unknown disease")
-            status = result.get(indx, "unknown condition")
-            recom = recommend.get(indx, "no recommendation available")
-            logger.info(f"Disease prediction successful: {name}, Status: {status}, Recommendation: {recom}")
-            # Generate a detailed response about the disease
-            detailed_response = self.generate_disease_response(name, status, recom)
-            return detailed_response, name, status, recom
-        except Exception as e:
-            logger.error(f"Error during disease prediction: {e}")
-            return "Error during prediction.", None, None, None
-
-    def generate_disease_response(self, disease_name, status, recommendation):
-        """
-        Generate a detailed response about the detected disease.
-
-        Args:
-            disease_name (str): Name of the detected disease.
-            status (str): Status of the disease (e.g., critical or no issue).
-            recommendation (str): Recommended action.
-
-        Returns:
-            str: Detailed response generated by Llama 3.2 model.
-        """
-        logger.info(f"Generating detailed response for disease: {disease_name}")
-        prompt = (
-            f"The disease detected is {disease_name}, classified as {status}. "
-            f"Recommended action: {recommendation}. "
-            f"Here is some information about {disease_name}: causes, symptoms, and treatment methods "
-            "to effectively manage this condition on a poultry farm."
-        )
-        response = llama3_response(prompt)
-        return response.replace(prompt, "").strip()
-
-    def diagnose_disease(self, image):
-        """
-        Diagnose the disease from the given image.
-
-        Args:
-            image (numpy.ndarray): Input image to diagnose.
-
-        Returns:
-            tuple: Detailed response, disease name, status, and recommendation.
-        """
-        if image is not None and image.size > 0:
-            logger.info("Image provided, starting diagnosis.")
-            return self.predict_disease(image)
-        logger.warning("No image provided for diagnosis.")
-        return "Please provide an image of poultry fecal matter for disease detection.", None, None, None
-
-    def log_enquiry(self, enquiry_type, content, response, user_id):
-        """
-        Log a farmer's enquiry in the database.
-
-        Args:
-            enquiry_type (str): Type of the enquiry ('image' or 'text').
-            content (str): The content of the enquiry.
-            response (str): The response given by the system.
-            user_id (str): The ID of the user making the enquiry.
-        """
-        enquiry = {
-            "user_id": user_id,
-            "enquiry_type": enquiry_type,
-            "content": content,
-            "response": response,
-            "timestamp": datetime.utcnow()
-        }
-        logger.info(f"Logging enquiry: {enquiry}")
-        enquiries_collection.insert_one(enquiry)
-
-    def authenticate_user(self, username, password):
-        """
-        Authenticate a user with username and password.
-
-        Args:
-            username (str): Username of the user.
-            password (str): Password of the user.
-
-        Returns:
-            dict: User information if authentication is successful, None otherwise.
-        """
-        logger.info(f"Authenticating user: {username}")
-        user = users_collection.find_one({"username": username})
-        if user and check_password_hash(user['password'], password):
-            logger.info("Authentication successful.")
-            return user
-        logger.warning("Authentication failed.")
-        return None
-
-    def register_user(self, username, password):
-        """
-        Register a new user with username and password.
-
-        Args:
-            username (str): Username of the new user.
-            password (str): Password of the new user.
-
-        Returns:
-            bool: True if registration is successful, False otherwise.
-        """
-        logger.info(f"Registering user: {username}")
-        if users_collection.find_one({"username": username}):
-            logger.warning("Username already exists.")
-            return False
-        hashed_password = generate_password_hash(password)
-        users_collection.insert_one({"username": username, "password": hashed_password})
-        logger.info("User registration successful.")
-        return True
-
-
 # Initialize the bot instance
 logger.info("Initializing PoultryFarmBot instance.")
 bot = PoultryFarmBot(db)
@@ -297,41 +72,6 @@ if tokenizer.pad_token is None:
     tokenizer.add_special_tokens({'pad_token': '[PAD]'})
     model.resize_token_embeddings(len(tokenizer))
 
-
-# Llama 3 response generation
-def llama3_response(user_input):
-    """
-    Generate a response using the Llama 3.2 model.
-
-    Args:
-        user_input (str): Input prompt for the Llama 3.2 model.
-
-        Returns:
-            str: Generated response from the model.
-    """
-    try:
-        logger.info("Generating response using Llama 3.2 model.")
-        # Tokenize the input prompt
-        inputs = tokenizer(user_input, return_tensors="pt", truncation=True, max_length=150, padding=True)
-        # Generate a response using the Llama 3.2 model
-        outputs = model.generate(
-            inputs["input_ids"],
-            max_length=150,
-            do_sample=True,
-            temperature=0.7,
-            pad_token_id=tokenizer.pad_token_id,
-            attention_mask=inputs["attention_mask"]
-        )
-        # Decode the generated response
-        response = tokenizer.decode(outputs[0], skip_special_tokens=True)
-        logger.info("Response generation successful.")
-        return response
-    except Exception as e:
-        logger.error(f"Error generating response: {str(e)}")
-        return f"Error generating response: {str(e)}"
-
-
-# Main chatbot function
 def chatbot_response(image, text, username, password):
     """
     Handle user input and generate appropriate responses.
@@ -366,11 +106,10 @@ def chatbot_response(image, text, username, password):
     else:
         # Generate a response using Llama 3.2 for general text input
         logger.info("Text input detected. Generating response.")
-        response = llama3_response(text)
+        response = llama3_response(text, tokenizer, model)
         bot.log_enquiry("text", text, response, user_id)
         return response
 
-
 # Gradio interface
 def build_gradio_interface():
     """
@@ -382,12 +121,63 @@ def build_gradio_interface():
     logger.info("Building Gradio interface.")
     with gr.Blocks(theme=gr.themes.Base()):
         gr.Markdown("# 🐔 Poultry Management Chatbot")
-        gr.Markdown(
-            "Welcome! This chatbot helps you manage your poultry with ease. You can upload an image for disease diagnosis or ask any questions about poultry management.")
+        gr.Markdown("Welcome! This chatbot helps you manage your poultry with ease. You can upload an image for disease diagnosis or ask any questions about poultry management.")
 
         chat_history = gr.Chatbot()
         with gr.Row():
             with gr.Column(scale=1):
                 fecal_image = gr.Image(
                     label="Upload Image of Poultry Feces (Optional)",
+                    type="numpy",
+                    elem_id="image-upload",
+                    show_label=True,
+                )
+            with gr.Column(scale=2):
+                user_input = gr.Textbox(
+                    label="Ask a question",
+                    placeholder="Ask about poultry management...",
+                    lines=3,
+                    elem_id="user-input",
+                )
+                username = gr.Textbox(
+                    label="Username",
+                    placeholder="Enter your username",
+                    lines=1,
+                    elem_id="username-input",
+                )
+                password = gr.Textbox(
+                    label="Password",
+                    placeholder="Enter your password",
+                    type="password",
+                    lines=1,
+                    elem_id="password-input",
+                )
+
+        output_box = gr.Textbox(
+            label="Response",
+            placeholder="Response will appear here...",
+            interactive=False,
+            lines=10,
+            elem_id="output-box",
+        )
 
+        submit_button = gr.Button(
+            "Submit",
+            variant="primary",
+            elem_id="submit-button"
+        )
+        # Connect the submit button to the chatbot response function
+        submit_button.click(
+            fn=chatbot_response,
+            inputs=[fecal_image, user_input, username, password],
+            outputs=[output_box]
+        )
+    logger.info("Gradio interface built successfully.")
+    return chatbot_interface
+
+# Launch the Gradio interface
+if __name__ == "__main__":
+    logger.info("Launching Gradio interface.")
+    interface = build_gradio_interface()
+    # Launch the interface with queuing enabled for concurrent requests
+    interface.queue().launch(debug=True, share=True)

requirements.txt

@@ -12,16 +12,6 @@ accelerate
 # Image Processing Libraries
 opencv-python
 
-# Web Framework and MongoDB
-flask
-pymongo
-bcrypt
-PyJWT
-python-dotenv
-gunicorn
-
-# streamlit Interface for the Chatbot
-streamlit
 
 # Data Handling and Analysis
 pandas

utils.py

@@ -0,0 +1,215 @@
+# utils.py
+
+import tensorflow as tf
+from keras.models import load_model
+import cv2
+import logging
+import numpy as np
+from transformers import AutoModelForCausalLM, AutoTokenizer
+from datetime import datetime
+from werkzeug.security import generate_password_hash, check_password_hash
+
+logger = logging.getLogger(__name__)
+
+def load_model_with_device(model_path, device_name):
+    """
+    Load a Keras model on the specified device.
+
+    Args:
+        model_path (str): Path to the model file.
+        device_name (str): Device to load the model on ('/GPU:0' or '/CPU:0').
+
+    Returns:
+        Model object if successful, None otherwise.
+    """
+    try:
+        logger.info(f"Loading model from '{model_path}' on {device_name}.")
+        with tf.device(device_name):
+            model = load_model(model_path, compile=True)
+            logger.info(f"Model '{model_path}' loaded successfully on {device_name}.")
+            return model
+    except Exception as e:
+        logger.error(f"Error loading model '{model_path}': {e}")
+        return None
+
+def llama3_response(user_input, tokenizer, model):
+    """
+    Generate a response using the Llama 3.2 model.
+
+    Args:
+        user_input (str): Input prompt for the Llama 3.2 model.
+        tokenizer (AutoTokenizer): Tokenizer to preprocess the input.
+        model (AutoModelForCausalLM): Pre-trained language model for generating responses.
+
+    Returns:
+        str: Generated response from the model.
+    """
+    try:
+        logger.info("Generating response using Llama 3.2 model.")
+        # Tokenize the input prompt
+        inputs = tokenizer(user_input, return_tensors="pt", truncation=True, max_length=150, padding=True)
+        # Generate a response using the Llama 3.2 model
+        outputs = model.generate(
+            inputs["input_ids"],
+            max_length=150,
+            do_sample=True,
+            temperature=0.7,
+            pad_token_id=tokenizer.pad_token_id,
+            attention_mask=inputs["attention_mask"]
+        )
+        # Decode the generated response
+        response = tokenizer.decode(outputs[0], skip_special_tokens=True)
+        logger.info("Response generation successful.")
+        return response
+    except Exception as e:
+        logger.error(f"Error generating response: {str(e)}")
+        return f"Error generating response: {str(e)}"
+
+class PoultryFarmBot:
+    def __init__(self, db):
+        """
+        Initialize the PoultryFarmBot with a MongoDB database connection.
+
+        Args:
+            db (MongoClient): MongoDB database instance.
+        """
+        self.db = db  # MongoDB database for future use
+        logger.info("PoultryFarmBot initialized with MongoDB connection.")
+
+    def preprocess_image(self, image):
+        """
+        Preprocess the input image for disease detection.
+
+        Args:
+            image (numpy.ndarray): Input image to preprocess.
+
+        Returns:
+            numpy.ndarray: Preprocessed image ready for model input.
+        """
+        try:
+            logger.info("Preprocessing image for disease detection.")
+            # Resize the image to match the model input size (224, 224)
+            image_check = cv2.resize(image, (224, 224))
+            # Add a batch dimension to the image
+            image_check = np.expand_dims(image_check, axis=0)
+            logger.info("Image preprocessing successful.")
+            return image_check
+        except Exception as e:
+            logger.error(f"Error in image preprocessing: {e}")
+            return None
+
+    def predict_disease(self, image, model, name_disease, result, recommend):
+        """
+        Predict the disease from the given poultry fecal image.
+
+        Args:
+            image (numpy.ndarray): Input image to predict.
+            model (tf.keras.Model): Loaded Keras model for disease prediction.
+            name_disease (dict): Dictionary mapping class indices to disease names.
+            result (dict): Dictionary mapping class indices to status.
+            recommend (dict): Dictionary mapping class indices to recommendations.
+
+        Returns:
+            tuple: Detailed response, disease name, status, and recommendation.
+        """
+        logger.info("Starting disease prediction.")
+        image_check = self.preprocess_image(image)
+        if image_check is None:
+            logger.warning("Image preprocessing failed.")
+            return "Image preprocessing failed.", None, None, None
+
+        try:
+            logger.info("Running model prediction.")
+            indx = model.predict(image_check).argmax()
+            name = name_disease.get(indx, "Unknown disease")
+            status = result.get(indx, "unknown condition")
+            recom = recommend.get(indx, "no recommendation available")
+            logger.info(f"Disease prediction successful: {name}, Status: {status}, Recommendation: {recom}")
+            detailed_response = self.generate_disease_response(name, status, recom)
+            return detailed_response, name, status, recom
+        except Exception as e:
+            logger.error(f"Error during disease prediction: {e}")
+            return "Error during prediction.", None, None, None
+
+    def generate_disease_response(self, disease_name, status, recommendation):
+        """
+        Generate a detailed response about the detected disease.
+
+        Args:
+            disease_name (str): Name of the detected disease.
+            status (str): Status of the disease (e.g., critical or no issue).
+            recommendation (str): Recommended action.
+
+        Returns:
+            str: Detailed response generated by Llama 3.2 model.
+        """
+        logger.info(f"Generating detailed response for disease: {disease_name}")
+        prompt = (
+            f"The disease detected is {disease_name}, classified as {status}. "
+            f"Recommended action: {recommendation}. "
+            f"Here is some information about {disease_name}: causes, symptoms, and treatment methods "
+            "to effectively manage this condition on a poultry farm."
+        )
+        return prompt
+
+    def log_enquiry(self, enquiry_type, content, response, user_id, enquiries_collection):
+        """
+        Log a farmer's enquiry in the database.
+
+        Args:
+            enquiry_type (str): Type of the enquiry ('image' or 'text').
+            content (str): The content of the enquiry.
+            response (str): The response given by the system.
+            user_id (str): The ID of the user making the enquiry.
+            enquiries_collection (MongoClient): Collection to store farmer enquiries.
+        """
+        enquiry = {
+            "user_id": user_id,
+            "enquiry_type": enquiry_type,
+            "content": content,
+            "response": response,
+            "timestamp": datetime.utcnow()
+        }
+        logger.info(f"Logging enquiry: {enquiry}")
+        enquiries_collection.insert_one(enquiry)
+
+    def authenticate_user(self, username, password, users_collection):
+        """
+        Authenticate a user with username and password.
+
+        Args:
+            username (str): Username of the user.
+            password (str): Password of the user.
+            users_collection (MongoClient): Collection to store user credentials.
+
+        Returns:
+            dict: User information if authentication is successful, None otherwise.
+        """
+        logger.info(f"Authenticating user: {username}")
+        user = users_collection.find_one({"username": username})
+        if user and check_password_hash(user['password'], password):
+            logger.info("Authentication successful.")
+            return user
+        logger.warning("Authentication failed.")
+        return None
+
+    def register_user(self, username, password, users_collection):
+        """
+        Register a new user with username and password.
+
+        Args:
+            username (str): Username of the new user.
+            password (str): Password of the new user.
+            users_collection (MongoClient): Collection to store user credentials.
+
+        Returns:
+            bool: True if registration is successful, False otherwise.
+        """
+        logger.info(f"Registering user: {username}")
+        if users_collection.find_one({"username": username}):
+            logger.warning("Username already exists.")
+            return False
+        hashed_password = generate_password_hash(password)
+        users_collection.insert_one({"username": username, "password": hashed_password})
+        logger.info("User registration successful.")
+        return True