Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |
| ) | |