#app.py from flask import Flask, render_template, request, jsonify import os os.environ["TF_USE_LEGACY_KERAS"] = "1" import tensorflow as tf import tensorflow_hub as hub import numpy as np from PIL import Image import requests from googletrans import Translator import io import logging import asyncio app = Flask(__name__) translator = Translator() # Configure logging logging.basicConfig(level=logging.INFO) # Environment variables LM_STUDIOS_API_URL = "LM_STUDIOS_API_URL"# LM Studios API URL GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY") # Google API Key GOOGLE_CX = os.getenv("GOOGLE_CX") # Google Custom Search Engine ID # Define UPLOAD_FOLDER UPLOAD_FOLDER = 'uploads' app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER pesticide_recommendations = { 'Bacterial Blight': 'Copper-based fungicides, Streptomycin', 'Red Rot': 'Fungicides containing Mancozeb or Copper', 'Blight': 'Fungicides containing Chlorothalonil', 'Common_Rust': 'Fungicides containing Azoxystrobin or Propiconazole', 'Gray_Leaf_Spot,Healthy': 'Fungicides containing Azoxystrobin or Propiconazole', 'Bacterial blight': 'Copper-based fungicides, Streptomycin', 'curl_virus': 'Insecticides such as Imidacloprid or Pyrethroids', 'fussarium_wilt': 'Soil fumigants, Fungicides containing Thiophanate-methyl', 'Bacterial_blight': 'Copper-based fungicides, Streptomycin', 'Blast': 'Fungicides containing Tricyclazole or Propiconazole', 'Brownspot': 'Fungicides containing Azoxystrobin or Propiconazole', 'Tungro': 'Insecticides such as Neonicotinoids or Pyrethroids', 'septoria': 'Fungicides containing Azoxystrobin or Propiconazole', 'strip_rust': 'Fungicides containing Azoxystrobin or Propiconazole' } class_names = { 'sugarcane': ['Bacterial Blight', 'Healthy', 'Red Rot'], 'maize': ['Blight', 'Common_Rust', 'Gray_Leaf_Spot,Healthy'], 'cotton': ['Bacterial blight', 'curl_virus', 'fussarium_wilt', 'Healthy'], 'rice': ['Bacterial_blight', 'Blast', 'Brownspot', 'Tungro'], 'wheat': ['Healthy', 'septoria', 'strip_rust'], } def recommend_pesticide(predicted_class): if predicted_class == 'Healthy': return 'No need for any pesticide, plant is healthy' return pesticide_recommendations.get(predicted_class, "No recommendation available") def load_model(model_path): return tf.keras.models.load_model( model_path, custom_objects={"KerasLayer": hub.KerasLayer}, compile=False ) models = { 'sugarcane': load_model("models/sugercane_model.h5"), 'maize': load_model("models/maize_model.h5"), 'cotton': load_model("models/cotton_model.h5"), 'rice': load_model("models/rice.h5"), 'wheat': load_model("models/wheat_model.h5"), } def preprocess_image(image_file): try: image = Image.open(image_file).convert("RGB") image = image.resize((224, 224)) img_array = np.array(image).astype("float32") / 255.0 return np.expand_dims(img_array, axis=0) except Exception as e: logging.error("Error in preprocessing image: %s", str(e)) return None def classify_image(model_name, image_path): input_image = preprocess_image(image_path) if input_image is None: return None, None predictions = models[model_name].predict(input_image) predicted_index = np.argmax(predictions) predicted_class = class_names[model_name][predicted_index] recommended_pesticide = recommend_pesticide(predicted_class) return predicted_class, recommended_pesticide # Async translation helper async def batch_translate(texts, src='en', dest='mr'): tasks = [translator.translate(text, src=src, dest=dest) for text in texts] results = await asyncio.gather(*tasks) return [res.text for res in results] def translate(texts, src='en', dest='mr'): loop = asyncio.new_event_loop() # Create a new event loop asyncio.set_event_loop(loop) # Set it as the current event loop return loop.run_until_complete(batch_translate(texts, src, dest)) def get_web_pesticide_info(disease, plant_type="Unknown"): query = f"site:agrowon.esakal.com {disease} in {plant_type}" url = "https://www.googleapis.com/customsearch/v1" params = { "key": GOOGLE_API_KEY, "cx": GOOGLE_CX, "q": query, "num": 3 } try: response = requests.get(url, params=params) response.raise_for_status() data = response.json() if "items" in data and len(data["items"]) > 0: item = data["items"][0] return {"title": item.get("title", "No title available"), "link": item.get("link", "#"), "snippet": item.get("snippet", "No snippet available"), "summary": item.get("snippet", "No snippet available")} except Exception as e: print(f"Error retrieving web pesticide info: {e}") return None def get_more_web_info(query): url = "https://www.googleapis.com/customsearch/v1" params = { "key": GOOGLE_API_KEY, "cx": GOOGLE_CX, "q": query, "num": 3 } try: response = requests.get(url, params=params) response.raise_for_status() data = response.json() results = [] if "items" in data: for item in data["items"]: results.append({"title": item.get("title", "No title available"), "link": item.get("link", "#"), "snippet": item.get("snippet", "No snippet available")}) return results except Exception as e: print(f"Error retrieving additional articles: {e}") return [] def get_commercial_product_info(recommendation): indiamart_query = f"site:indiamart.com pesticide '{recommendation}'" krishi_query = f"site:krishisevakendra.in/products pesticide '{recommendation}'" indiamart_results = get_more_web_info(indiamart_query) krishi_results = get_more_web_info(krishi_query) return indiamart_results + krishi_results @app.route('/', methods=['GET', 'POST']) def index(): if request.method == 'POST': # Get form data plant_type = request.form.get('plant_type') language = request.form.get('language') uploaded_file = request.files.get('image') if uploaded_file: # Save the uploaded image try: image_path = os.path.join(app.config['UPLOAD_FOLDER'], uploaded_file.filename) uploaded_file.save(image_path) # Classify the image predicted_class, pesticide = classify_image(plant_type, image_path) if predicted_class: # Retrieve additional information plant_info = get_plant_info(predicted_class, plant_type) web_pesticide_info = get_web_pesticide_info(predicted_class, plant_type) commercial_products = get_commercial_product_info(pesticide) more_articles = get_more_web_info(f"{predicted_class} in {plant_type}") # Translate text based on the selected language translated_predicted_class = translate(predicted_class, language) translated_pesticide = translate(pesticide, language) translated_plant_info = {k: translate(v, language) for k, v in (plant_info.get("detailed_info") or {}).items()} #Added this so as to avoid error if no details are fetched from LM Studio translated_web_pesticide_info = {k: translate(v, language) for k, v in (web_pesticide_info or {}).items()} translated_commercial_products = [{k: translate(v, language) for k, v in item.items()} for item in commercial_products] translated_more_articles = [{k: translate(v, language) for k, v in item.items()} for item in more_articles] # Render the result template with the data return render_template('result.html', image_path=image_path, plant_type=plant_type, predicted_class=translated_predicted_class, pesticide=translated_pesticide, plant_info=plant_info, web_pesticide_info=translated_web_pesticide_info, commercial_products=translated_commercial_products, more_articles=translated_more_articles, language=language) else: return render_template('index.html', error_message=translate("Error in classification. Please try again.", language), language=language) except Exception as e: print(f"Error during file processing or classification: {e}") return render_template('index.html', error_message=translate("An error occurred during processing. Please try again.", language), language=language) else: return render_template('index.html', error_message=translate("Please upload an image.", language), language=language) # If it's a GET request or initial load return render_template('index.html', language="en") @app.route('/classify', methods=['POST']) def classify(): try: if 'file' not in request.files: logging.error('No file uploaded') return jsonify({'error': 'No file uploaded'}), 400 file = request.files['file'] plant_type = request.form.get('plant_type', 'sugarcane') language = request.form.get('language', 'English') if file.filename == '': logging.error('No file selected') return jsonify({'error': 'No file selected'}), 400 if plant_type not in models: logging.error(f'Model for {plant_type} not available') return jsonify({'error': f'Model for {plant_type} not available'}), 400 logging.info(f'Processing file: {file.filename}') input_image = preprocess_image(file) if input_image is None: logging.error('Error processing image') return jsonify({'error': 'Error processing image'}), 400 logging.info('Predicting using the model') predictions = models[plant_type].predict(input_image) predicted_index = np.argmax(predictions) predicted_class = class_names[plant_type][predicted_index] pesticide = pesticide_recommendations.get(predicted_class, "No recommendation available") # If language is set to Marathi, translate both texts in one batch. if language == 'Marathi': try: logging.info('Translating texts to Marathi') translated = asyncio.run(batch_translate([predicted_class, pesticide], src='en', dest='mr')) predicted_class, pesticide = translated except Exception as e: logging.error(f"Translation error in /classify: {e}") logging.info(f'Prediction: {predicted_class}, Pesticide: {pesticide}') return jsonify({ 'predicted_class': predicted_class, 'pesticide': pesticide }) except Exception as e: logging.error(f"Error in /classify endpoint: {e}") return jsonify({'error': 'An error occurred during classification'}), 500 async def async_translate_individual(texts, src='en', dest='mr'): """ Asynchronously translate a list of texts individually """ translated_texts = [] for text in texts: try: if text and isinstance(text, str): translated = await translator.translate(text, src=src, dest=dest) translated_texts.append(translated.text) else: translated_texts.append(text) except Exception as e: logging.error(f"Translation error for text '{text}': {e}") translated_texts.append(text) # Fall back to original text on error return translated_texts def translate_individual(texts, src='en', dest='mr'): """ Wrapper for async translation function """ try: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) return loop.run_until_complete(async_translate_individual(texts, src, dest)) except Exception as e: logging.error(f"Translation wrapper error: {e}") return texts # Return original texts if translation fails def get_plant_info(disease, plant_type="Unknown", language="en"): """ Get plant disease information and translate if needed """ # First, get the disease name in English if it's in Marathi try: if any(ord(c) > 127 for c in disease): # Check if contains non-ASCII characters # Translate disease name back to English for API query translated = translator.translate(disease, src='mr', dest='en') disease = translated.text except Exception as e: logging.error(f"Error translating disease name to English: {e}") # Continue with original disease name if translation fails prompt = f""" Disease Name: {disease} Plant Type: {plant_type} Explain this disease in a very simple and easy-to-understand way, as if you are talking to a farmer with no scientific background. Use simple words and avoid technical terms. Include the following details: - Symptoms: What signs will the farmer see on the plant? How will the leaves, stem, or fruit look? - Causes: Why does this disease happen? - Severity: How serious is this disease? Does it spread quickly? How much crop damage can it cause? - How It Spreads: How does this disease grow? What will happen if the farmer does nothing? - Treatment & Prevention: What pesticides or sprays should the farmer use and what steps can be taken to prevent the disease? """ try: API_URL = "https://api-inference.huggingface.co/models/meta-llama/Llama-3.2-1B-Instruct/v1/chat/completions" headers = {"Authorization": f"Bearer {os.getenv('HUGGINGFACE_API_TOKEN')}"} try: data = { "messages": [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": prompt}, ] } response = requests.post(API_URL, headers=headers, json=data) response.raise_for_status() print("Response:", data) except requests.exceptions.HTTPError as http_err: print("HTTP error occurred:", http_err) print("Response Content:", response.content) except Exception as e: print("Other error occurred:", str(e)) response_data = response.json() detailed_info = response_data.get("choices", [{}])[0].get("message", {}).get("content", "") # response = requests.post(LM_STUDIOS_API_URL, # json={"messages": [{"role": "user", "content": prompt}]}, # headers={"Content-Type": "application/json"}) # response.raise_for_status() # data = response.json() # detailed_info = data.get("choices", [{}])[0].get("message", {}).get("content", "") # Translate the response if language is Marathi if language == "Marathi" and detailed_info: try: translated = translator.translate(detailed_info, src='en', dest='mr') detailed_info = translated.text logging.info(f"Successfully translated plant info to Marathi") except Exception as e: logging.error(f"Error translating plant info to Marathi: {e}") # Continue with English content if translation fails return detailed_info except Exception as e: logging.error(f"Error in get_plant_info: {e}") error_msg = "माहिती आणण्यात त्रुटी आली आहे. कृपया पुन्हा प्रयत्न करा." if language == "Marathi" else "Unable to retrieve information. Please try again." return error_msg @app.route('/get_additional_info', methods=['POST']) def get_additional_info(): try: data = request.json disease = data.get('disease') plant_type = data.get('plant_type') language = data.get('language', 'English') # Get detailed info with language parameter detailed_info = get_plant_info(disease, plant_type, language) # Get additional info web_results = get_more_web_info(f"{disease} in {plant_type}") commercial_products = get_more_web_info(f"pesticide for {disease} in {plant_type}") if language == 'Marathi': try: # Translate web results if web_results: translated_results = [] for result in web_results: if isinstance(result, dict): translated_result = { 'title': translator.translate(result.get('title', ''), dest='mr').text, 'link': result.get('link', '#'), 'snippet': translator.translate(result.get('snippet', ''), dest='mr').text } translated_results.append(translated_result) web_results = translated_results # Translate commercial products if commercial_products: translated_products = [] for product in commercial_products: if isinstance(product, dict): translated_product = { 'title': translator.translate(product.get('title', ''), dest='mr').text, 'link': product.get('link', '#'), 'snippet': translator.translate(product.get('snippet', ''), dest='mr').text } translated_products.append(translated_product) commercial_products = translated_products logging.info("Successfully translated additional information to Marathi") except Exception as e: logging.error(f"Translation error in get_additional_info: {e}") # Continue with untranslated content if translation fails return jsonify({ 'detailed_info': detailed_info, 'web_results': web_results or [], 'commercial_products': commercial_products or [] }) except Exception as e: logging.error(f"Error in get_additional_info: {e}") error_msg = "माहिती आणण्यात त्रुटी आली आहे." if language == "Marathi" else "An error occurred while retrieving information." return jsonify({ 'error': error_msg, 'detailed_info': error_msg, 'web_results': [], 'commercial_products': [] }), 500 if __name__ == '__main__': app = Flask(__name__) # Your route definitions and other code here app.run(host='0.0.0.0', port=7860, debug=False,use_reloader=False)