File size: 3,428 Bytes
adf4c1c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 |
# Import Hugging Face datasets
from datasets import load_dataset
from colorama import Fore
from mcp.server.fastmcp import FastMCP
import chromadb
# Create server
mcp = FastMCP("croptimizeserver")
# Load crop optimization dataset
dataset = load_dataset("DARJYO/sawotiQ29_crop_optimization")
# Initialize ChromaDB for crop data
chroma_client = chromadb.PersistentClient(path="crop_db")
collection = chroma_client.get_collection(name="crop_data")
# Add prompt function for crop recommendations
@mcp.prompt()
def crop_recommendation(crop_data: str) -> str:
"""Prompt template for generating crop recommendations"""
return f"""You are an agricultural expert assistant designed to provide crop optimization advice.
Analyze the following crop data and provide recommendations for optimal cultivation:
{crop_data}"""
# Resource for searching crop information
@mcp.resource("crops://search/{query}")
def search_crops(query: str) -> str:
"""Search for crops based on growing conditions or characteristics"""
results = collection.query(
query_texts=[query],
n_results=3,
include=["documents", "metadatas"]
)
return str(results)
# Tool for getting crop details
@mcp.tool()
def crop_details(crop_name: str) -> str:
"""Get detailed information about a specific crop"""
filtered_data = dataset['train'].filter(lambda x: x['crop_name'].lower() == crop_name.lower())
if not filtered_data:
return f"No information found for {crop_name}"
return str(filtered_data[0])
# Tool for optimal growing conditions
@mcp.tool()
def optimal_conditions(crop_name: str) -> str:
"""Get optimal growing conditions for a specific crop"""
crop_data = dataset['train'].filter(lambda x: x['crop_name'].lower() == crop_name.lower())
if not crop_data:
return f"No data available for {crop_name}"
conditions = {
'temperature': crop_data[0]['optimal_temperature'],
'rainfall': crop_data[0]['annual_rainfall'],
'soil_type': crop_data[0]['preferred_soil'],
'altitude': crop_data[0]['optimal_altitude']
}
return str(conditions)
# Tool for yield prediction
@mcp.tool()
def yield_prediction(crop_name: str, region: str) -> str:
"""Predict yield for a crop in a specific region"""
region_data = dataset['train'].filter(lambda x:
(x['region'].lower() == region.lower()) and
(x['crop_name'].lower() == crop_name.lower())
)
if not region_data:
return f"No yield data available for {crop_name} in {region}"
prediction = {
'crop': crop_name,
'region': region,
'expected_yield': region_data[0]['average_yield'],
'optimal_season': region_data[0]['best_season']
}
return str(prediction)
# Tool for pest/disease information
@mcp.tool()
def crop_protection(crop_name: str) -> str:
"""Get common pests and diseases for a crop"""
crop_data = dataset['train'].filter(lambda x: x['crop_name'].lower() == crop_name.lower())
if not crop_data:
return f"No protection data available for {crop_name}"
protection_info = {
'common_pests': crop_data[0]['common_pests'],
'common_diseases': crop_data[0]['common_diseases'],
'prevention_methods': crop_data[0]['prevention_methods']
}
return str(protection_info)
if __name__ == "__main__":
mcp.run(transport="stdio")
|