""" Serveur MCP pour l'analyse des données agricoles Expose les outils d'analyse et de visualisation via Gradio avec support MCP """ import os import json import traceback import pandas as pd import gradio as gr import plotly.express as px import plotly.graph_objects as go from plotly.subplots import make_subplots # Configuration pour désactiver les analytics Gradio os.environ["GRADIO_ANALYTICS_ENABLED"] = "False" # Import des modules locaux from data_loader import DataLoader from herbicide_analyzer import HerbicideAnalyzer from config import DATASET_ID, HF_TOKEN, PLOT_CONFIG class AgricultureMCPServer: """Serveur MCP pour l'analyse des données agricoles""" def __init__(self): self.data_loader = DataLoader() self.herbicide_analyzer = HerbicideAnalyzer() self.df = None self.is_data_loaded = False def load_agriculture_data(self) -> str: """ Charge les données agricoles depuis Hugging Face Returns: str: Message de statut du chargement """ try: status = self.data_loader.load_data() self.df = self.data_loader.get_data() if self.df is not None: self.herbicide_analyzer.set_data(self.df) self.is_data_loaded = True return f"✅ Données chargées: {len(self.df)} lignes, {len(self.df.columns)} colonnes" else: self.is_data_loaded = False return f"❌ Échec du chargement: {status}" except Exception as e: self.is_data_loaded = False return f"❌ Erreur lors du chargement: {str(e)}" def get_data_summary(self) -> str: """ Obtient un résumé des données chargées Returns: str: JSON avec les statistiques des données """ try: if not self.is_data_loaded or self.df is None: return json.dumps({"error": "Aucune donnée chargée. Utilisez load_agriculture_data() d'abord."}) summary = { "nombre_lignes": len(self.df), "nombre_colonnes": len(self.df.columns), "colonnes": list(self.df.columns), "annees_disponibles": sorted(self.df['millesime'].unique().tolist()) if 'millesime' in self.df.columns else [], "nombre_parcelles": self.df['numparcell'].nunique() if 'numparcell' in self.df.columns else 0, "familles_produits": self.df['familleprod'].unique().tolist() if 'familleprod' in self.df.columns else [] } return json.dumps(summary, indent=2, ensure_ascii=False) except Exception as e: return json.dumps({"error": f"Erreur lors du résumé: {str(e)}"}) def analyze_herbicide_usage(self, year: int = 2023, analysis_type: str = "statistics") -> str: """ Analyse l'utilisation des herbicides Args: year (int): Année à analyser analysis_type (str): Type d'analyse ("statistics", "top_parcels", "trends") Returns: str: JSON avec les résultats de l'analyse """ try: if not self.is_data_loaded: return json.dumps({"error": "Données non chargées. Utilisez load_agriculture_data() d'abord."}) if analysis_type == "statistics": stats, msg = self.herbicide_analyzer.get_herbicide_usage_statistics(year) if stats: return json.dumps({"status": msg, "data": stats}, indent=2, ensure_ascii=False) else: return json.dumps({"error": msg}) elif analysis_type == "top_parcels": parcels, msg = self.herbicide_analyzer.get_top_ift_parcels_by_year(year, 10) if parcels is not None: # Convertir DataFrame en dict pour JSON return json.dumps({ "status": msg, "data": parcels.to_dict('records') }, indent=2, ensure_ascii=False) else: return json.dumps({"error": msg}) elif analysis_type == "trends": trends, msg = self.herbicide_analyzer.analyze_herbicide_trends_by_region() if trends is not None: return json.dumps({ "status": msg, "data": trends.head(20).to_dict('records') # Limiter pour éviter des réponses trop longues }, indent=2, ensure_ascii=False) else: return json.dumps({"error": msg}) else: return json.dumps({"error": f"Type d'analyse '{analysis_type}' non supporté"}) except Exception as e: return json.dumps({"error": f"Erreur lors de l'analyse: {str(e)}"}) def create_herbicide_visualization(self, year: int = 2023, chart_type: str = "top_ift") -> go.Figure: """ Crée une visualisation des données herbicides Args: year (int): Année à analyser chart_type (str): Type de graphique ("top_ift", "regional_trends") Returns: plotly.graph_objects.Figure: Graphique Plotly """ try: if not self.is_data_loaded: return self._create_error_plot("Données non chargées", "Utilisez load_agriculture_data() d'abord") return self.herbicide_analyzer.create_herbicide_visualization( analysis_type=chart_type, year=year, n_parcels=10 ) except Exception as e: return self._create_error_plot("Erreur de visualisation", f"Erreur: {str(e)}") def query_database(self, query_type: str, **kwargs) -> str: """ Effectue une requête sur la base de données Args: query_type (str): Type de requête **kwargs: Paramètres de la requête Returns: str: JSON avec les résultats """ try: if not self.is_data_loaded: return json.dumps({"error": "Données non chargées"}) if query_type == "parcels_by_commune": commune = kwargs.get('commune', '') if 'nomcommune' in self.df.columns: results = self.df[self.df['nomcommune'].str.contains(commune, case=False, na=False)] return json.dumps({ "count": len(results), "data": results.head(50).to_dict('records') }, ensure_ascii=False) else: return json.dumps({"error": "Colonne 'nomcommune' non disponible"}) elif query_type == "products_by_year": year = kwargs.get('year', 2023) year_data = self.df[self.df['millesime'] == year] products = year_data['familleprod'].value_counts().to_dict() if 'familleprod' in year_data.columns else {} return json.dumps({ "year": year, "products": products }, ensure_ascii=False) else: return json.dumps({"error": f"Type de requête '{query_type}' non supporté"}) except Exception as e: return json.dumps({"error": f"Erreur de requête: {str(e)}"}) def _create_error_plot(self, title: str, message: str) -> go.Figure: """Crée un graphique d'erreur""" fig = go.Figure() fig.add_annotation( text=message, xref="paper", yref="paper", x=0.5, y=0.5, showarrow=False, font=dict(size=14, color="red") ) fig.update_layout(title=title, width=600, height=400) return fig # Initialisation du serveur MCP mcp_server = AgricultureMCPServer() # Interface Gradio avec support MCP with gr.Blocks(title="Agriculture Data MCP Server", theme=gr.themes.Soft()) as demo: gr.Markdown(""" # 🌾 Agriculture Data MCP Server Ce serveur MCP expose des outils d'analyse des données agricoles via le Model Context Protocol. Connectez un agent IA pour analyser les données et générer des rapports puissants. """) with gr.Tab("📊 Chargement des Données"): load_btn = gr.Button("Charger les données depuis Hugging Face", variant="primary") load_output = gr.Textbox(label="Statut du chargement", lines=3) summary_btn = gr.Button("Obtenir le résumé des données") summary_output = gr.JSON(label="Résumé des données") with gr.Tab("🌿 Analyse des Herbicides"): with gr.Row(): herb_year = gr.Number(value=2023, label="Année", precision=0) herb_analysis = gr.Dropdown( choices=["statistics", "top_parcels", "trends"], value="statistics", label="Type d'analyse" ) herb_analyze_btn = gr.Button("Analyser les herbicides") herb_output = gr.JSON(label="Résultats de l'analyse") with gr.Tab("📈 Visualisations"): with gr.Row(): viz_year = gr.Number(value=2023, label="Année", precision=0) viz_type = gr.Dropdown( choices=["top_ift", "regional_trends"], value="top_ift", label="Type de graphique" ) viz_btn = gr.Button("Créer la visualisation") viz_output = gr.Plot(label="Graphique") with gr.Tab("🔍 Requêtes Base de Données"): with gr.Row(): query_type = gr.Dropdown( choices=["parcels_by_commune", "products_by_year"], value="parcels_by_commune", label="Type de requête" ) query_param = gr.Textbox(label="Paramètre (commune ou année)", value="") query_btn = gr.Button("Exécuter la requête") query_output = gr.JSON(label="Résultats de la requête") # Connexions des événements load_btn.click( fn=mcp_server.load_agriculture_data, outputs=load_output ) summary_btn.click( fn=lambda: json.loads(mcp_server.get_data_summary()), outputs=summary_output ) herb_analyze_btn.click( fn=lambda year, analysis: json.loads(mcp_server.analyze_herbicide_usage(int(year), analysis)), inputs=[herb_year, herb_analysis], outputs=herb_output ) viz_btn.click( fn=mcp_server.create_herbicide_visualization, inputs=[viz_year, viz_type], outputs=viz_output ) query_btn.click( fn=lambda qtype, param: json.loads(mcp_server.query_database( qtype, commune=param if qtype == "parcels_by_commune" else None, year=int(param) if qtype == "products_by_year" and param.isdigit() else 2023 )), inputs=[query_type, query_param], outputs=query_output ) # Lancement du serveur avec support MCP if __name__ == "__main__": demo.launch( server_name="0.0.0.0", server_port=7860, share=True, mcp_server=True # Active le support MCP )