Update app.py

app.py

@@ -7,11 +7,9 @@ import cv2
 import numpy as np
 from huggingface_hub import login
 from pymongo import MongoClient
-from datetime import datetime
-import json
-import requests
+from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
 
-# Ensure the HF token is set
+# Ensure the Hugging Face token is set
 tok = os.getenv('HF_Token')
 if tok:
     login(token=tok, add_to_git_credential=True)
@@ -21,7 +19,7 @@ else:
 # Set your OpenAI API key
 openai.api_key = os.getenv("OPENAI_API_KEY")
 
-# MongoDB Setup
+# MongoDB Setup (for inventory, record-keeping, etc.)
 MONGO_URI = os.getenv("MONGO_URI")
 client = MongoClient(MONGO_URI)
 db = client.poultry_farm  # Database
@@ -31,9 +29,8 @@ print("TensorFlow version:", tf.__version__)
 print("Eager execution:", tf.executing_eagerly())
 print("TensorFlow GPU Available:", tf.config.list_physical_devices('GPU'))
 
-# Set TensorFlow to use mixed precision to leverage the T4 GPU's capabilities when available
+# Set TensorFlow to use mixed precision with available GPU
 from tensorflow.keras import mixed_precision
-
 if len(tf.config.list_physical_devices('GPU')) > 0:
     policy = mixed_precision.Policy('mixed_float16')
     mixed_precision.set_global_policy(policy)
@@ -44,19 +41,27 @@ else:
 # Load TensorFlow/Keras models with GPU support if available, otherwise use CPU
 try:
     device_name = '/GPU:0' if len(tf.config.list_physical_devices('GPU')) > 0 else '/CPU:0'
-    with tf.device(device_name):  # Use GPU if available, otherwise CPU
-        my_model = load_model('models/Final_Chicken_disease_model.h5', compile=True)
-        auth_model = load_model('models/auth_model.h5', compile=True)
-        print(f"Models loaded successfully on {device_name}.")
+    with tf.device(device_name):
+        fecal_model = load_model('models/Fecal_Disease_Model.h5', compile=True)
+        print(f"Fecal disease detection model loaded successfully on {device_name}.")
 except Exception as e:
-    print(f"Error loading models: {e}")
+    print(f"Error loading fecal disease model: {e}")
     raise
 
+# Disease names and recommendations based on fecal analysis
+name_disease = {0: "Coccidiosis", 1: "Salmonella", 2: "Worm Infestation"}  # Update with actual classes
+result = {0: "critical", 1: "serious", 2: "moderate"}  # Update with relevant statuses
+recommend = {
+    0: "Use anticoccidial medication and improve litter management.",
+    1: "Administer antibiotics as prescribed by a veterinarian.",
+    2: "Deworm chickens using an appropriate anti-parasitic drug."
+}  # Update with relevant recommendations
+
 class PoultryFarmBot:
     def __init__(self):
-        self.db = db  # MongoDB database
+        self.db = db  # MongoDB database for future use
 
-    # Health Monitoring and Disease Diagnosis
+    # Image Preprocessing for Fecal Disease Detection
     def preprocess_image(self, image):
         try:
             image_check = cv2.resize(image, (224, 224))
@@ -66,187 +71,118 @@ class PoultryFarmBot:
             print(f"Error in image preprocessing: {e}")
             return None
 
+    # Predict Disease from Fecal Image
     def predict(self, image):
         image_check = self.preprocess_image(image)
         if image_check is None:
             return "Image preprocessing failed.", None, None, None
 
-        indx = auth_model.predict(image_check).argmax()
+        # Predict using the fecal disease detection model
+        indx = fecal_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")
 
-        if indx == 0:  # If the image is recognized as a chicken disease image
-            indx = my_model.predict(image_check).argmax()
-            name = name_disease.get(indx)
-            status = result.get(indx)
-            recom = recommend.get(indx)
-
-            diagnosis = f"The chicken is in a {status} condition, diagnosed with {name}. The recommended medication is {recom}."
-            return diagnosis, name, status, recom
-        else:  # If the image is not recognized as a chicken disease image
-            return (
-                "The uploaded image is not recognized as a chicken or does not appear to be related to any known chicken diseases. "
-                "Please ensure the image is clear and shows a chicken or its symptoms to receive a proper diagnosis."
-            ), None, None, None
+        diagnosis = f"The fecal sample indicates the poultry animal is in a {status} condition, diagnosed with {name}. The recommended action is {recom}."
+        return diagnosis, name, status, recom
 
+    # Diagnose Disease Using Fecal Image or Symptoms
     def diagnose_disease(self, image=None, symptoms=None):
         if image is not None and image.size > 0:  # Ensure image is valid and has elements
             return self.predict(image)
         elif symptoms:
-            # Generate diagnosis using ChatGPT based on the provided symptoms
+            # Generate diagnosis using GPT-based model based on the provided symptoms
             context = f"Symptoms: {symptoms}."
             response = openai.ChatCompletion.create(
-                model="gpt-3.5-turbo",  # Use GPT-4 if you have access
+                model="gpt-3.5-turbo",
                 messages=[
-                    {"role": "system",
-                     "content": "You are an advanced poultry farm management system, helping poultry farmers manage their flocks efficiently."},
+                    {"role": "system", "content": "You are an advanced poultry farm management system."},
                     {"role": "user", "content": context}
                 ],
                 max_tokens=150
             )
             diagnosis = response['choices'][0]['message']['content'].strip()
             return diagnosis, None, None, None
-        return "Please provide an image or describe the symptoms.", None, None, None
-
-    # Inventory Management
-    def track_inventory(self, item, usage):
-        collection = self.db.inventory
-        inventory_item = collection.find_one({"item": item})
-        if inventory_item:
-            new_quantity = inventory_item["quantity"] - usage
-            collection.update_one({"item": item}, {"$set": {"quantity": new_quantity}})
-            if new_quantity < 10:
-                return f"{item} inventory is low, please reorder."
-            return f"{item} inventory updated. Current inventory: {new_quantity} units."
-        else:
-            return f"Item {item} not recognized in inventory."
-
-    def add_inventory_item(self, item, quantity):
-        collection = self.db.inventory
-        if collection.find_one({"item": item}):
-            collection.update_one({"item": item}, {"$inc": {"quantity": quantity}})
-        else:
-            collection.insert_one({"item": item, "quantity": quantity})
-        return f"Added {quantity} units of {item} to the inventory."
-
-    def view_inventory(self):
-        collection = self.db.inventory
-        inventory_list = list(collection.find({}))
-        if not inventory_list:
-            return "Inventory is empty."
-        return json.dumps(inventory_list, indent=4)
-
-    # Chicken and Egg Management
-    def add_chicken(self, chicken_id, breed, age):
-        collection = self.db.chickens
-        collection.insert_one({"chicken_id": chicken_id, "breed": breed, "age": age})
-        return f"Chicken {chicken_id} added to the database."
-
-    def update_chicken(self, chicken_id, update_data):
-        collection = self.db.chickens
-        collection.update_one({"chicken_id": chicken_id}, {"$set": update_data})
-        return f"Chicken {chicken_id} updated."
-
-    def add_eggs(self, quantity):
-        collection = self.db.eggs
-        collection.insert_one({"date": datetime.now(), "quantity": quantity})
-        return f"Added {quantity} eggs to the database."
-
-    # Reporting and Analytics
-    def generate_report(self):
-        report = {
-            "date": str(datetime.now()),
-            "feed_inventory": list(self.db.inventory.find({})),
-            "chickens": list(self.db.chickens.find({})),
-            "eggs_collected": list(self.db.eggs.find({})),
-            "health_reports": list(self.db.health_reports.find({}))
-        }
-        self.db.reports.insert_one(report)
-        return json.dumps(report, indent=4)
-
-    # Integration with External Systems
-    def integrate_with_external_system(self, system_url, data):
-        try:
-            response = requests.post(system_url, json=data)
-            if response.status_code == 200:
-                return "Data successfully sent to external system."
-            else:
-                return f"Failed to send data. Status code: {response.status_code}"
-
-        except Exception as e:
-            return f"Integration failed with error: {str(e)}"
-
-    # Emergency Handling
-    def handle_emergency(self, emergency_type):
-        if emergency_type == "disease_outbreak":
-            return "Disease outbreak detected. Isolate affected chickens and contact a veterinarian immediately."
-        elif emergency_type == "equipment_failure":
-            return "Equipment failure detected. Check the equipment immediately and perform necessary repairs."
-        else:
-            return "Unknown emergency type."
-
+        return "Please provide an image of poultry fecal matter or describe the symptoms.", None, None, None
 
 # Initialize the bot instance
 bot = PoultryFarmBot()
 
-# Gradio Interface for adding, updating, and viewing inventory
-def manage_inventory(action, item, quantity):
-    if action == "Add":
-        return bot.add_inventory_item(item, int(quantity))
-    elif action == "Update":
-        return bot.track_inventory(item, int(quantity))
-    elif action == "View":
-        return bot.view_inventory()
-    else:
-        return "Invalid action. Please select either 'Add', 'Update', or 'View'."
+# Load the Mistral model and tokenizer for text generation
+model_name = "mistralai/mistral-7b-v0.1"
+tokenizer = AutoTokenizer.from_pretrained(model_name)
+model = AutoModelForCausalLM.from_pretrained(model_name)
+generator = pipeline("text-generation", model=model, tokenizer=tokenizer)
+
+# Define Mistral-based response generation
+def mistral_response(user_input):
+    try:
+        responses = generator(
+            user_input,
+            max_length=150,
+            num_return_sequences=1,
+            temperature=0.7
+        )
+        return responses[0]["generated_text"]
+    except Exception as e:
+        return f"Error generating response: {str(e)}"
 
-# Function to generate a response using OpenAI's GPT model
+# Gradio interface for disease diagnosis and chatbot response
 def generate_combined_response(image, text):
     diagnosis, name, status, recom = bot.diagnose_disease(image=image, symptoms=text)
-
-    if name and status and recom:  # If the disease is recognized
-        context = f"The chicken is in a {status} condition, diagnosed with {name}. The recommended medication is {recom}. "
+    if name and status and recom:
+        context = f"The fecal sample indicates the poultry animal is in a {status} condition, diagnosed with {name}. The recommended action is {recom}. "
         if text:
             context += f"Additionally, the user asked: '{text}'"
-
-        # Use OpenAI's GPT model to generate additional advice
-        response = openai.ChatCompletion.create(
-            model="gpt-3.5-turbo",  # Use GPT-4 or gpt-3.5-turbo based on your API access
-            messages=[
-                {"role": "system", "content": "You are an expert poultry farm management assistant."},
-                {"role": "user", "content": context}
-            ],
-            max_tokens=150
-        )
-        advice = response['choices'][0]['message']['content'].strip()
+        advice = mistral_response(context)
         return diagnosis + "\n\nAdditional Advice: " + advice
     else:
         return diagnosis
 
-# Gradio Interface
-inventory_interface = gr.Interface(
-    fn=manage_inventory,
-    inputs=[
-        gr.Radio(label="Action", choices=["Add", "Update", "View"]),
-        gr.Textbox(label="Item Name"),
-        gr.Number(label="Quantity", value=0)
-    ],
-    outputs=gr.Textbox(label="Inventory Management Response"),
-    title="Poultry Farm Inventory Management",
-    description="Use this interface to add, update, or view the inventory for your poultry farm."
-)
-
-diagnosis_interface = gr.Interface(
-    fn=generate_combined_response,
-    inputs=[gr.Image(label='Upload Image'), gr.Textbox(label='Describe symptoms or ask a question')],
-    outputs=gr.Textbox(label="Diagnosis and Advice"),
-    title="Poultry Health Diagnostic Assistant",
-    description="Describe your chicken's symptoms and get a diagnosis and advice."
-)
-
-# Combine the inventory management and diagnostic interfaces
-combined_interface = gr.TabbedInterface(
-    [inventory_interface, diagnosis_interface],
-    ["Inventory Management", "Health Diagnosis"]
-)
-
-# Launch the interface
-combined_interface.launch(debug=True, share=True)
+# Gradio interface styling and layout
+with gr.Blocks(theme=gr.themes.Soft(primary_hue="green", neutral_hue="slate")) as diagnosis_interface:
+    gr.Markdown("# 🐔 Poultry Fecal Diagnostic Assistant")
+    gr.Markdown(
+        "Upload an image of poultry fecal matter or describe symptoms to get a diagnosis and advice."
+    )
+
+    with gr.Row():
+        with gr.Column(scale=1):
+            fecal_image = gr.Image(
+                label="Upload Image of Poultry Feces",
+                type="numpy",
+                tool="editor",
+                elem_id="image-upload",
+                show_label=True,
+            )
+        with gr.Column(scale=2):
+            symptom_text = gr.Textbox(
+                label="Describe Symptoms or Ask a Question",
+                placeholder="Enter symptoms of the poultry...",
+                lines=3,
+                elem_id="symptom-textbox",
+            )
+
+    output_box = gr.Textbox(
+        label="Diagnosis and Advice",
+        placeholder="The diagnosis and advice will appear here...",
+        interactive=False,
+        lines=6,
+        elem_id="output-box",
+    )
+
+    submit_button = gr.Button(
+        "Submit",
+        variant="primary",
+        elem_id="submit-button",
+        style={"width": "100%", "margin-top": "20px"},
+    )
+    submit_button.click(
+        fn=generate_combined_response,
+        inputs=[fecal_image, symptom_text],
+        outputs=[output_box],
+    )
+
+# Launch the Gradio interface
+if __name__ == "__main__":
+    diagnosis_interface.launch(debug=True, share=True)