Spaces:
Sleeping
Sleeping
| """MCP Server for Agricultural Weed Pressure Analysis""" | |
| import gradio as gr | |
| import pandas as pd | |
| import numpy as np | |
| import plotly.express as px | |
| from data_loader import AgriculturalDataLoader | |
| import warnings | |
| warnings.filterwarnings('ignore') | |
| class WeedPressureAnalyzer: | |
| """Analyze weed pressure and recommend plots for sensitive crops.""" | |
| def __init__(self): | |
| self.data_loader = AgriculturalDataLoader() | |
| self.data_cache = None | |
| def load_data(self): | |
| if self.data_cache is None: | |
| self.data_cache = self.data_loader.load_all_files() | |
| return self.data_cache | |
| def calculate_herbicide_ift(self, years=None): | |
| """Calculate IFT for herbicides by plot and year.""" | |
| df = self.load_data() | |
| if years: | |
| df = df[df['year'].isin(years)] | |
| herbicide_df = df[df['is_herbicide'] == True].copy() | |
| if len(herbicide_df) == 0: | |
| return pd.DataFrame() | |
| ift_summary = herbicide_df.groupby(['plot_name', 'year', 'crop_type']).agg({ | |
| 'produit': 'count', | |
| 'plot_surface': 'first', | |
| 'quantitetot': 'sum' | |
| }).reset_index() | |
| ift_summary['ift_herbicide'] = ift_summary['produit'] / ift_summary['plot_surface'] | |
| return ift_summary | |
| def predict_weed_pressure(self, target_years=[2025, 2026, 2027]): | |
| """Predict weed pressure for future years.""" | |
| ift_data = self.calculate_herbicide_ift() | |
| if len(ift_data) == 0: | |
| return pd.DataFrame() | |
| predictions = [] | |
| for plot in ift_data['plot_name'].unique(): | |
| plot_data = ift_data[ift_data['plot_name'] == plot].sort_values('year') | |
| if len(plot_data) < 2: | |
| continue | |
| years = plot_data['year'].values | |
| ift_values = plot_data['ift_herbicide'].values | |
| if len(years) > 1: | |
| slope = np.polyfit(years, ift_values, 1)[0] | |
| intercept = np.polyfit(years, ift_values, 1)[1] | |
| for target_year in target_years: | |
| predicted_ift = slope * target_year + intercept | |
| predicted_ift = max(0, predicted_ift) | |
| if predicted_ift < 1.0: | |
| risk_level = "Faible" | |
| elif predicted_ift < 2.0: | |
| risk_level = "Modéré" | |
| else: | |
| risk_level = "Élevé" | |
| predictions.append({ | |
| 'plot_name': plot, | |
| 'year': target_year, | |
| 'predicted_ift': predicted_ift, | |
| 'risk_level': risk_level, | |
| 'recent_crops': ', '.join(plot_data['crop_type'].tail(3).unique()), | |
| 'historical_avg_ift': plot_data['ift_herbicide'].mean() | |
| }) | |
| return pd.DataFrame(predictions) | |
| # Initialize analyzer | |
| analyzer = WeedPressureAnalyzer() | |
| # MCP Resources Implementation | |
| # Custom MCP resource decorator implementation for Gradio compatibility | |
| # This simulates @gr.mcp.resource functionality until it's officially available | |
| class MCPResourceRegistry: | |
| """Registry for MCP resources with URI patterns""" | |
| def __init__(self): | |
| self.resources = {} | |
| def resource(self, uri_pattern: str): | |
| """Decorator to register MCP resources""" | |
| def decorator(func): | |
| self.resources[uri_pattern] = { | |
| 'function': func, | |
| 'uri_pattern': uri_pattern, | |
| 'description': func.__doc__ or "No description available" | |
| } | |
| return func | |
| return decorator | |
| def get_resource(self, uri: str): | |
| """Get resource by URI pattern match""" | |
| for pattern, resource_info in self.resources.items(): | |
| if self._match_pattern(pattern, uri): | |
| return resource_info | |
| return None | |
| def _match_pattern(self, pattern: str, uri: str): | |
| """Simple pattern matching for URI templates""" | |
| if pattern == uri: | |
| return True | |
| # Handle {param} patterns - simple string replacement approach | |
| if '{' in pattern and '}' in pattern: | |
| # Extract the base pattern and check if URI starts with it | |
| base_pattern = pattern.split('{')[0] | |
| return uri.startswith(base_pattern) | |
| return False | |
| def list_resources(self): | |
| """List all registered resources""" | |
| return self.resources | |
| # Global MCP resource registry | |
| mcp_registry = MCPResourceRegistry() | |
| # MCP resource decorator | |
| def mcp_resource(uri_pattern: str): | |
| """Decorator to register MCP resources""" | |
| return mcp_registry.resource(uri_pattern) | |
| def get_available_plots_resource() -> str: | |
| """Get list of all available agricultural plots from the dataset""" | |
| try: | |
| plots = get_available_plots() | |
| return f"Available plots ({len(plots)-1}): " + ", ".join(plots[1:6]) + f" ... and {len(plots)-6} more" | |
| except Exception as e: | |
| return f"Error loading plots: {str(e)}" | |
| def get_available_crops_resource() -> str: | |
| """Get list of all crop types in the dataset""" | |
| try: | |
| crops = get_available_crops() | |
| return f"Available crops ({len(crops)-1}): " + ", ".join(crops[1:6]) + f" ... and {len(crops)-6} more" | |
| except Exception as e: | |
| return f"Error loading crops: {str(e)}" | |
| def get_available_years_resource() -> str: | |
| """Get range of years available in the dataset""" | |
| try: | |
| df = analyzer.load_data() | |
| years = sorted(df['year'].dropna().unique()) | |
| return f"Available years: {min(years)}-{max(years)} ({len(years)} years total)" | |
| except Exception as e: | |
| return f"Error loading years: {str(e)}" | |
| def get_dataset_info() -> str: | |
| """Get comprehensive information about the agricultural dataset""" | |
| try: | |
| df = analyzer.load_data() | |
| herbicide_data = df[df['is_herbicide'] == True] | |
| info = f""" | |
| Agricultural Dataset Information: | |
| - Total records: {len(df):,} | |
| - Years covered: {df['year'].min()}-{df['year'].max()} | |
| - Number of plots: {df['plot_name'].nunique()} | |
| - Number of crop types: {df['crop_type'].nunique()} | |
| - Number of intervention types: {df['intervention_type'].nunique()} | |
| - Herbicide applications: {len(herbicide_data):,} | |
| - Average IFT (herbicides): {herbicide_data.groupby(['plot_name', 'year']).size().mean():.2f} | |
| - Data source: Station Expérimentale de Kerguéhennec | |
| - Last updated: 2025 | |
| """ | |
| return info.strip() | |
| except Exception as e: | |
| return f"Error loading dataset info: {str(e)}" | |
| def get_plot_info(plot_name: str) -> str: | |
| """Get detailed information about a specific agricultural plot""" | |
| try: | |
| df = analyzer.load_data() | |
| plot_data = df[df['plot_name'] == plot_name] | |
| if len(plot_data) == 0: | |
| return f"Plot '{plot_name}' not found in dataset" | |
| herbicide_data = plot_data[plot_data['is_herbicide'] == True] | |
| years = sorted(plot_data['year'].unique()) | |
| info = f""" | |
| Plot Information: {plot_name} | |
| - Total interventions: {len(plot_data):,} | |
| - Years active: {min(years)}-{max(years)} ({len(years)} years) | |
| - Herbicide applications: {len(herbicide_data):,} | |
| - Average IFT: {herbicide_data.groupby('year').size().mean():.2f} | |
| - Surface: {plot_data['plot_surface'].iloc[0]:.2f} hectares | |
| - Main crops: {', '.join(plot_data['crop_type'].value_counts().head(3).index.tolist())} | |
| - Main interventions: {', '.join(plot_data['intervention_type'].value_counts().head(3).index.tolist())} | |
| """ | |
| return info.strip() | |
| except Exception as e: | |
| return f"Error loading plot info: {str(e)}" | |
| def get_crop_info(crop_type: str) -> str: | |
| """Get information about a specific crop type and its cultivation patterns""" | |
| try: | |
| df = analyzer.load_data() | |
| crop_data = df[df['crop_type'] == crop_type] | |
| if len(crop_data) == 0: | |
| return f"Crop type '{crop_type}' not found in dataset" | |
| herbicide_data = crop_data[crop_data['is_herbicide'] == True] | |
| years = sorted(crop_data['year'].unique()) | |
| plots = crop_data['plot_name'].nunique() | |
| info = f""" | |
| Crop Information: {crop_type} | |
| - Total interventions: {len(crop_data):,} | |
| - Years cultivated: {min(years)}-{max(years)} ({len(years)} years) | |
| - Number of plots: {plots} | |
| - Herbicide applications: {len(herbicide_data):,} | |
| - Average IFT: {herbicide_data.groupby(['plot_name', 'year']).size().mean():.2f} | |
| - Main plots: {', '.join(crop_data['plot_name'].value_counts().head(3).index.tolist())} | |
| - Main interventions: {', '.join(crop_data['intervention_type'].value_counts().head(3).index.tolist())} | |
| """ | |
| return info.strip() | |
| except Exception as e: | |
| return f"Error loading crop info: {str(e)}" | |
| def get_year_summary(year: int) -> str: | |
| """Get summary of agricultural activities for a specific year""" | |
| try: | |
| df = analyzer.load_data() | |
| year_data = df[df['year'] == year] | |
| if len(year_data) == 0: | |
| return f"No data available for year {year}" | |
| herbicide_data = year_data[year_data['is_herbicide'] == True] | |
| plots = year_data['plot_name'].nunique() | |
| crops = year_data['crop_type'].nunique() | |
| info = f""" | |
| Year Summary: {year} | |
| - Total interventions: {len(year_data):,} | |
| - Active plots: {plots} | |
| - Crop types: {crops} | |
| - Herbicide applications: {len(herbicide_data):,} | |
| - Average IFT: {herbicide_data.groupby('plot_name').size().mean():.2f} | |
| - Most active plot: {year_data['plot_name'].value_counts().index[0]} ({year_data['plot_name'].value_counts().iloc[0]} interventions) | |
| - Most common crop: {year_data['crop_type'].value_counts().index[0]} ({year_data['crop_type'].value_counts().iloc[0]} interventions) | |
| - Most common intervention: {year_data['intervention_type'].value_counts().index[0]} ({year_data['intervention_type'].value_counts().iloc[0]} interventions) | |
| """ | |
| return info.strip() | |
| except Exception as e: | |
| return f"Error loading year summary: {str(e)}" | |
| def get_herbicide_usage_summary() -> str: | |
| """Get comprehensive summary of herbicide usage patterns""" | |
| try: | |
| df = analyzer.load_data() | |
| herbicide_data = df[df['is_herbicide'] == True] | |
| if len(herbicide_data) == 0: | |
| return "No herbicide data available" | |
| # Calculate IFT by plot and year | |
| ift_data = herbicide_data.groupby(['plot_name', 'year']).size().reset_index(name='applications') | |
| ift_data['ift'] = ift_data['applications'] / herbicide_data.groupby(['plot_name', 'year'])['plot_surface'].first().values | |
| avg_ift = ift_data['ift'].mean() | |
| max_ift = ift_data['ift'].max() | |
| min_ift = ift_data['ift'].min() | |
| # Risk distribution | |
| low_risk = len(ift_data[ift_data['ift'] < 1.0]) | |
| moderate_risk = len(ift_data[(ift_data['ift'] >= 1.0) & (ift_data['ift'] < 2.0)]) | |
| high_risk = len(ift_data[ift_data['ift'] >= 2.0]) | |
| info = f""" | |
| Herbicide Usage Summary: | |
| - Total applications: {len(herbicide_data):,} | |
| - Plots with herbicides: {herbicide_data['plot_name'].nunique()} | |
| - Years with data: {herbicide_data['year'].nunique()} | |
| - Average IFT: {avg_ift:.2f} | |
| - IFT range: {min_ift:.2f} - {max_ift:.2f} | |
| - Risk distribution: | |
| * Low risk (IFT < 1.0): {low_risk} plot-years ({low_risk/len(ift_data)*100:.1f}%) | |
| * Moderate risk (1.0 ≤ IFT < 2.0): {moderate_risk} plot-years ({moderate_risk/len(ift_data)*100:.1f}%) | |
| * High risk (IFT ≥ 2.0): {high_risk} plot-years ({high_risk/len(ift_data)*100:.1f}%) | |
| - Most used herbicides: {', '.join(herbicide_data['produit'].value_counts().head(3).index.tolist())} | |
| """ | |
| return info.strip() | |
| except Exception as e: | |
| return f"Error loading herbicide usage summary: {str(e)}" | |
| def get_predictions_summary() -> str: | |
| """Get summary of weed pressure predictions for 2025-2027""" | |
| try: | |
| predictions = analyzer.predict_weed_pressure() | |
| if len(predictions) == 0: | |
| return "No predictions available - insufficient historical data" | |
| low_risk = len(predictions[predictions['risk_level'] == 'Faible']) | |
| moderate_risk = len(predictions[predictions['risk_level'] == 'Modéré']) | |
| high_risk = len(predictions[predictions['risk_level'] == 'Élevé']) | |
| avg_ift = predictions['predicted_ift'].mean() | |
| info = f""" | |
| Weed Pressure Predictions 2025-2027: | |
| - Total predictions: {len(predictions)} | |
| - Average predicted IFT: {avg_ift:.2f} | |
| - Risk distribution: | |
| * Low risk (IFT < 1.0): {low_risk} predictions ({low_risk/len(predictions)*100:.1f}%) | |
| * Moderate risk (1.0 ≤ IFT < 2.0): {moderate_risk} predictions ({moderate_risk/len(predictions)*100:.1f}%) | |
| * High risk (IFT ≥ 2.0): {high_risk} predictions ({high_risk/len(predictions)*100:.1f}%) | |
| - Best plots (lowest IFT): {', '.join(predictions.nsmallest(3, 'predicted_ift')['plot_name'].tolist())} | |
| - Method: Linear regression on historical IFT data | |
| """ | |
| return info.strip() | |
| except Exception as e: | |
| return f"Error loading predictions summary: {str(e)}" | |
| def get_recommendations_summary() -> str: | |
| """Get summary of plot recommendations for sensitive crops (pois, haricot)""" | |
| try: | |
| predictions = analyzer.predict_weed_pressure() | |
| suitable_plots = predictions[predictions['risk_level'] == "Faible"].copy() | |
| if len(suitable_plots) == 0: | |
| return "No plots recommended for sensitive crops - all plots have high predicted weed pressure" | |
| suitable_plots['recommendation_score'] = 100 - (suitable_plots['predicted_ift'] * 30) | |
| suitable_plots = suitable_plots.sort_values('recommendation_score', ascending=False) | |
| top_plots = suitable_plots.head(5) | |
| avg_score = suitable_plots['recommendation_score'].mean() | |
| info = f""" | |
| Sensitive Crop Recommendations (Pois, Haricot): | |
| - Suitable plots: {len(suitable_plots)} | |
| - Average recommendation score: {avg_score:.1f}/100 | |
| - Top 5 recommended plots: | |
| """ | |
| for i, (_, plot) in enumerate(top_plots.iterrows(), 1): | |
| info += f" {i}. {plot['plot_name']} - Score: {plot['recommendation_score']:.1f}, IFT: {plot['predicted_ift']:.2f}\n" | |
| info += f"- Criteria: IFT < 1.0 (low weed pressure)\n" | |
| info += f"- Score formula: 100 - (predicted_ift × 30)\n" | |
| info += f"- Method: Based on 2025-2027 predictions" | |
| return info.strip() | |
| except Exception as e: | |
| return f"Error loading recommendations summary: {str(e)}" | |
| def get_plot_predictions(plot_name: str) -> str: | |
| """Get weed pressure predictions for a specific plot""" | |
| try: | |
| predictions = analyzer.predict_weed_pressure() | |
| plot_predictions = predictions[predictions['plot_name'] == plot_name] | |
| if len(plot_predictions) == 0: | |
| return f"No predictions available for plot '{plot_name}' - insufficient historical data" | |
| plot_predictions = plot_predictions.sort_values('year') | |
| info = f""" | |
| Predictions for {plot_name}: | |
| """ | |
| for _, pred in plot_predictions.iterrows(): | |
| info += f"- {pred['year']}: IFT {pred['predicted_ift']:.2f} ({pred['risk_level']} risk)\n" | |
| info += f"- Historical average IFT: {plot_predictions['historical_avg_ift'].iloc[0]:.2f}\n" | |
| info += f"- Recent crops: {plot_predictions['recent_crops'].iloc[0]}\n" | |
| info += f"- Recommendation: {'Suitable for sensitive crops' if plot_predictions['risk_level'].iloc[0] == 'Faible' else 'Not recommended for sensitive crops'}" | |
| return info.strip() | |
| except Exception as e: | |
| return f"Error loading plot predictions: {str(e)}" | |
| def list_mcp_resources() -> str: | |
| """List all available MCP resources with their URIs and descriptions""" | |
| try: | |
| resources = mcp_registry.list_resources() | |
| if not resources: | |
| return "No MCP resources available" | |
| info = "## Available MCP Resources\n\n" | |
| for uri_pattern, resource_info in resources.items(): | |
| info += f"### `{uri_pattern}`\n" | |
| info += f"**Description:** {resource_info['description']}\n\n" | |
| return info | |
| except Exception as e: | |
| return f"Error listing resources: {str(e)}" | |
| def analyze_herbicide_trends(year_start, year_end, plot_filter): | |
| """ | |
| Analyze herbicide usage trends over time by calculating IFT (Treatment Frequency Index). | |
| This tool calculates the IFT (Indice de Fréquence de Traitement) for herbicides, which represents | |
| the number of herbicide applications per hectare. It provides visualizations and statistics to | |
| understand weed pressure evolution over time. | |
| Args: | |
| year_start (int): Starting year for analysis (2014-2025) | |
| year_end (int): Ending year for analysis (2014-2025) | |
| plot_filter (str): Specific plot name or "Toutes" for all plots | |
| Returns: | |
| tuple: (plotly_figure, markdown_summary) | |
| - plotly_figure: Interactive line chart showing IFT evolution by plot and year | |
| - markdown_summary: Detailed statistics including mean/max IFT, risk distribution | |
| """ | |
| try: | |
| # Créer la liste des années à partir des deux sliders | |
| start_year = int(year_start) | |
| end_year = int(year_end) | |
| # S'assurer que start <= end | |
| if start_year > end_year: | |
| start_year, end_year = end_year, start_year | |
| years = list(range(start_year, end_year + 1)) | |
| ift_data = analyzer.calculate_herbicide_ift(years=years) | |
| if len(ift_data) == 0: | |
| return None, "Aucune donnée d'herbicides trouvée pour la période sélectionnée." | |
| # Filtrage par parcelle si nécessaire | |
| if plot_filter and plot_filter != "Toutes": | |
| ift_data = ift_data[ift_data['plot_name'] == plot_filter] | |
| if len(ift_data) == 0: | |
| return None, f"Aucune donnée trouvée pour la parcelle '{plot_filter}' sur la période {years[0]}-{years[-1]}." | |
| # Création du graphique | |
| fig = px.line(ift_data, | |
| x='year', | |
| y='ift_herbicide', | |
| color='plot_name', | |
| title=f'Évolution de l\'IFT Herbicides ({years[0]}-{years[-1]})', | |
| labels={'ift_herbicide': 'IFT Herbicides', 'year': 'Année'}, | |
| markers=True) | |
| fig.update_layout( | |
| height=500, | |
| xaxis_title="Année", | |
| yaxis_title="IFT Herbicides", | |
| legend_title="Parcelle" | |
| ) | |
| # Ajout d'une ligne de référence IFT = 2.0 | |
| fig.add_hline(y=2.0, line_dash="dash", line_color="red", | |
| annotation_text="Seuil IFT élevé (2.0)", annotation_position="top right") | |
| fig.add_hline(y=1.0, line_dash="dash", line_color="orange", | |
| annotation_text="Seuil IFT modéré (1.0)", annotation_position="bottom right") | |
| # Calcul des statistiques | |
| ift_mean = ift_data['ift_herbicide'].mean() | |
| ift_max = ift_data['ift_herbicide'].max() | |
| ift_min = ift_data['ift_herbicide'].min() | |
| n_plots = ift_data['plot_name'].nunique() | |
| n_records = len(ift_data) | |
| # Classification des niveaux de risque | |
| low_risk = len(ift_data[ift_data['ift_herbicide'] < 1.0]) | |
| moderate_risk = len(ift_data[(ift_data['ift_herbicide'] >= 1.0) & (ift_data['ift_herbicide'] < 2.0)]) | |
| high_risk = len(ift_data[ift_data['ift_herbicide'] >= 2.0]) | |
| summary = f""" | |
| 📊 **Analyse de l'IFT Herbicides ({years[0]}-{years[-1]})** | |
| **Période analysée:** {years[0]} à {years[-1]} | |
| **Parcelle(s):** {plot_filter if plot_filter != "Toutes" else "Toutes les parcelles"} | |
| **Statistiques globales:** | |
| - IFT moyen: {ift_mean:.2f} | |
| - IFT minimum: {ift_min:.2f} | |
| - IFT maximum: {ift_max:.2f} | |
| - Nombre de parcelles: {n_plots} | |
| - Nombre d'observations: {n_records} | |
| **Répartition des niveaux de pression:** | |
| - 🟢 Faible (IFT < 1.0): {low_risk} observations ({low_risk/n_records*100:.1f}%) | |
| - 🟡 Modérée (1.0 ≤ IFT < 2.0): {moderate_risk} observations ({moderate_risk/n_records*100:.1f}%) | |
| - 🔴 Élevée (IFT ≥ 2.0): {high_risk} observations ({high_risk/n_records*100:.1f}%) | |
| **Interprétation:** | |
| - IFT < 1.0: Pression adventices faible ✅ | |
| - 1.0 ≤ IFT < 2.0: Pression adventices modérée ⚠️ | |
| - IFT ≥ 2.0: Pression adventices élevée ❌ | |
| """ | |
| return fig, summary | |
| except Exception as e: | |
| import traceback | |
| error_msg = f"Erreur dans l'analyse: {str(e)}\n{traceback.format_exc()}" | |
| print(error_msg) | |
| return None, error_msg | |
| def predict_future_weed_pressure(): | |
| """ | |
| Predict weed pressure for the next 3 years (2025-2027) using linear regression on historical IFT data. | |
| This tool uses historical herbicide IFT data to predict future weed pressure. It applies linear | |
| regression to each plot's IFT evolution over time and extrapolates to 2025-2027. Risk levels are | |
| classified as: Faible (IFT < 1.0), Modéré (1.0 ≤ IFT < 2.0), Élevé (IFT ≥ 2.0). | |
| Prediction Method: | |
| 1. Calculate historical IFT for each plot/year combination | |
| 2. Apply linear regression: IFT = slope × year + intercept | |
| 3. Extrapolate to target years 2025-2027 | |
| 4. Classify risk levels based on predicted IFT values | |
| 5. Include recent crop history and average historical IFT for context | |
| Returns: | |
| tuple: (plotly_figure, markdown_summary) | |
| - plotly_figure: Bar chart showing predicted IFT by plot and year with risk color coding | |
| - markdown_summary: Risk distribution statistics and interpretation | |
| """ | |
| try: | |
| predictions = analyzer.predict_weed_pressure() | |
| if len(predictions) == 0: | |
| return None, "Impossible de générer des prédictions." | |
| fig = px.bar(predictions, | |
| x='plot_name', | |
| y='predicted_ift', | |
| color='risk_level', | |
| facet_col='year', | |
| title='Prédiction Pression Adventices (2025-2027)', | |
| color_discrete_map={'Faible': 'green', 'Modéré': 'orange', 'Élevé': 'red'}) | |
| low_risk = len(predictions[predictions['risk_level'] == 'Faible']) | |
| moderate_risk = len(predictions[predictions['risk_level'] == 'Modéré']) | |
| high_risk = len(predictions[predictions['risk_level'] == 'Élevé']) | |
| summary = f""" | |
| 🔮 **Prédictions 2025-2027** | |
| **Répartition des risques:** | |
| - ✅ Risque faible: {low_risk} prédictions | |
| - ⚠️ Risque modéré: {moderate_risk} prédictions | |
| - ❌ Risque élevé: {high_risk} prédictions | |
| """ | |
| return fig, summary | |
| except Exception as e: | |
| return None, f"Erreur: {str(e)}" | |
| def recommend_sensitive_crop_plots(): | |
| """ | |
| Recommend plots suitable for sensitive crops (pois, haricot) based on predicted weed pressure. | |
| This tool identifies plots with low predicted weed pressure (IFT < 1.0) and calculates a | |
| recommendation score to rank them for sensitive crop cultivation. | |
| Recommendation Method: | |
| 1. Get predicted IFT for 2025-2027 from predict_future_weed_pressure() | |
| 2. Filter plots with risk_level = "Faible" (IFT < 1.0) | |
| 3. Calculate recommendation_score = 100 - (predicted_ift × 30) | |
| 4. Sort plots by recommendation score (higher = better) | |
| 5. Include recent crop history and historical average IFT for context | |
| Recommendation Score: | |
| - 100-70: Excellent for sensitive crops | |
| - 70-50: Good for sensitive crops with monitoring | |
| - 50-0: Requires careful management | |
| Returns: | |
| tuple: (plotly_figure, markdown_summary) | |
| - plotly_figure: Scatter plot showing predicted IFT vs recommendation score | |
| - markdown_summary: Top recommended plots with scores and criteria | |
| """ | |
| try: | |
| predictions = analyzer.predict_weed_pressure() | |
| if len(predictions) == 0: | |
| return None, "Aucune recommandation disponible." | |
| suitable_plots = predictions[predictions['risk_level'] == "Faible"].copy() | |
| if len(suitable_plots) > 0: | |
| suitable_plots['recommendation_score'] = 100 - (suitable_plots['predicted_ift'] * 30) | |
| suitable_plots = suitable_plots.sort_values('recommendation_score', ascending=False) | |
| top_recommendations = suitable_plots.head(10)[['plot_name', 'year', 'predicted_ift', 'recommendation_score']] | |
| summary = f""" | |
| 🌱 **Recommandations Cultures Sensibles** | |
| **Top parcelles recommandées:** | |
| {top_recommendations.to_string(index=False)} | |
| **Critères:** IFT prédit < 1.0 (faible pression adventices) | |
| """ | |
| fig = px.scatter(suitable_plots, | |
| x='predicted_ift', | |
| y='recommendation_score', | |
| color='year', | |
| hover_data=['plot_name'], | |
| title='Parcelles Recommandées pour Cultures Sensibles') | |
| return fig, summary | |
| else: | |
| return None, "Aucune parcelle à faible risque identifiée." | |
| except Exception as e: | |
| return None, f"Erreur: {str(e)}" | |
| def explore_raw_data(year_start, year_end, plot_filter, crop_filter, intervention_filter): | |
| """ | |
| Explore raw agricultural intervention data with filtering capabilities. | |
| This tool provides access to the raw dataset from the Station Expérimentale de Kerguéhennec | |
| (2014-2025) with filtering options to explore specific subsets of data. | |
| Args: | |
| year_start (int): Starting year for filtering (2014-2025) | |
| year_end (int): Ending year for filtering (2014-2025) | |
| plot_filter (str): Specific plot name or "Toutes" for all plots | |
| crop_filter (str): Specific crop type or "Toutes" for all crops | |
| intervention_filter (str): Specific intervention type or "Toutes" for all interventions | |
| Returns: | |
| tuple: (plotly_figure, markdown_summary) | |
| - plotly_figure: Interactive data table or visualization | |
| - markdown_summary: Data summary with statistics and filtering info | |
| """ | |
| try: | |
| # Charger les données | |
| df = analyzer.load_data() | |
| # Appliquer les filtres | |
| if year_start and year_end: | |
| df = df[(df['year'] >= year_start) & (df['year'] <= year_end)] | |
| if plot_filter and plot_filter != "Toutes": | |
| df = df[df['plot_name'] == plot_filter] | |
| if crop_filter and crop_filter != "Toutes": | |
| df = df[df['crop_type'] == crop_filter] | |
| if intervention_filter and intervention_filter != "Toutes": | |
| df = df[df['intervention_type'] == intervention_filter] | |
| if len(df) == 0: | |
| return None, "Aucune donnée trouvée avec les filtres sélectionnés." | |
| # Créer un résumé des données | |
| summary = f""" | |
| 📊 **Exploration des Données Brutes** | |
| **Filtres appliqués:** | |
| - Période: {year_start}-{year_end} | |
| - Parcelle: {plot_filter} | |
| - Culture: {crop_filter} | |
| - Type d'intervention: {intervention_filter} | |
| **Statistiques:** | |
| - Nombre total d'enregistrements: {len(df):,} | |
| - Nombre de parcelles: {df['plot_name'].nunique()} | |
| - Nombre d'années: {df['year'].nunique()} | |
| - Types de cultures: {df['crop_type'].nunique()} | |
| - Types d'interventions: {df['intervention_type'].nunique()} | |
| **Répartition par année:** | |
| {df['year'].value_counts().sort_index().to_string()} | |
| **Top 10 parcelles:** | |
| {df['plot_name'].value_counts().head(10).to_string()} | |
| **Top 10 cultures:** | |
| {df['crop_type'].value_counts().head(10).to_string()} | |
| **Top 10 interventions:** | |
| {df['intervention_type'].value_counts().head(10).to_string()} | |
| """ | |
| # Créer une visualisation des données | |
| if len(df) > 0: | |
| # Graphique des interventions par année | |
| yearly_counts = df.groupby('year').size().reset_index(name='count') | |
| fig = px.bar(yearly_counts, x='year', y='count', | |
| title=f'Nombre d\'interventions par année ({year_start}-{year_end})', | |
| labels={'count': 'Nombre d\'interventions', 'year': 'Année'}) | |
| fig.update_layout(height=400) | |
| return fig, summary | |
| else: | |
| return None, summary | |
| except Exception as e: | |
| return None, f"Erreur lors de l'exploration des données: {str(e)}" | |
| def get_available_plots(): | |
| """Get available plots.""" | |
| try: | |
| df = analyzer.load_data() | |
| plots = sorted(df['plot_name'].dropna().unique().tolist()) | |
| return ["Toutes"] + plots | |
| except Exception as e: | |
| print(f"Erreur lors du chargement des parcelles: {e}") | |
| return ["Toutes", "Champ ferme Bas", "Etang Milieu", "Lann Chebot"] | |
| def get_available_crops(): | |
| """Get available crop types.""" | |
| try: | |
| df = analyzer.load_data() | |
| crops = sorted(df['crop_type'].dropna().unique().tolist()) | |
| return ["Toutes"] + crops | |
| except Exception as e: | |
| print(f"Erreur lors du chargement des cultures: {e}") | |
| return ["Toutes", "blé tendre hiver", "pois de conserve", "haricot mange-tout industrie"] | |
| def get_available_interventions(): | |
| """Get available intervention types.""" | |
| try: | |
| df = analyzer.load_data() | |
| interventions = sorted(df['intervention_type'].dropna().unique().tolist()) | |
| return ["Toutes"] + interventions | |
| except Exception as e: | |
| print(f"Erreur lors du chargement des interventions: {e}") | |
| return ["Toutes", "Traitement et protection des cultures", "Fertilisation", "Travail et Entretien du sol"] | |
| # Create Gradio Interface | |
| def create_mcp_interface(): | |
| with gr.Blocks(title="🚜 Analyse Pression Adventices", theme=gr.themes.Soft()) as demo: | |
| gr.Markdown(""" | |
| # 🚜 Analyse Pression Adventices - CRA Bretagne | |
| Anticiper et réduire la pression des adventices pour optimiser les cultures sensibles (pois, haricot). | |
| """) | |
| with gr.Tabs(): | |
| with gr.Tab("📈 Analyse Tendances"): | |
| gr.Markdown("### Analyser l'évolution de l'IFT herbicides par parcelle et période") | |
| gr.Markdown(""" | |
| **Calcul de l'IFT (Indice de Fréquence de Traitement) :** | |
| - IFT = Nombre d'applications herbicides / Surface de la parcelle | |
| - Seuils d'interprétation : | |
| - 🟢 Faible : IFT < 1.0 (pression adventices faible) | |
| - 🟡 Modéré : 1.0 ≤ IFT < 2.0 (pression modérée) | |
| - 🔴 Élevé : IFT ≥ 2.0 (pression élevée) | |
| """) | |
| with gr.Row(): | |
| with gr.Column(): | |
| with gr.Row(): | |
| year_start = gr.Slider( | |
| minimum=2014, | |
| maximum=2025, | |
| value=2020, | |
| step=1, | |
| label="Année de début" | |
| ) | |
| year_end = gr.Slider( | |
| minimum=2014, | |
| maximum=2025, | |
| value=2025, | |
| step=1, | |
| label="Année de fin" | |
| ) | |
| plot_dropdown = gr.Dropdown( | |
| choices=get_available_plots(), | |
| value="Toutes", | |
| label="Filtrer par parcelle", | |
| info="Choisissez une parcelle spécifique ou toutes" | |
| ) | |
| analyze_btn = gr.Button("🔍 Analyser les Tendances", variant="primary", size="lg") | |
| with gr.Row(): | |
| with gr.Column(scale=2): | |
| trends_plot = gr.Plot(label="Graphique d'évolution") | |
| with gr.Column(scale=1): | |
| trends_summary = gr.Markdown(label="Résumé statistique") | |
| analyze_btn.click( | |
| analyze_herbicide_trends, | |
| inputs=[year_start, year_end, plot_dropdown], | |
| outputs=[trends_plot, trends_summary] | |
| ) | |
| with gr.Tab("🔮 Prédictions"): | |
| gr.Markdown("### Prédiction de la pression adventices 2025-2027") | |
| gr.Markdown(""" | |
| **Méthode de prédiction :** | |
| 1. Calcul de l'IFT historique par parcelle et année | |
| 2. Régression linéaire : IFT = pente × année + ordonnée_origine | |
| 3. Extrapolation aux années 2025-2027 | |
| 4. Classification des risques : | |
| - 🟢 Faible : IFT < 1.0 | |
| - 🟡 Modéré : 1.0 ≤ IFT < 2.0 | |
| - 🔴 Élevé : IFT ≥ 2.0 | |
| """) | |
| predict_btn = gr.Button("🎯 Prédire 2025-2027", variant="primary") | |
| with gr.Row(): | |
| predictions_plot = gr.Plot() | |
| predictions_summary = gr.Markdown() | |
| predict_btn.click(predict_future_weed_pressure, outputs=[predictions_plot, predictions_summary]) | |
| with gr.Tab("🌱 Recommandations"): | |
| gr.Markdown("### Recommandations pour cultures sensibles (pois, haricot)") | |
| gr.Markdown(""" | |
| **Méthode de recommandation :** | |
| 1. Prédiction IFT 2025-2027 par régression linéaire | |
| 2. Filtrage des parcelles à faible risque (IFT < 1.0) | |
| 3. Calcul du score de recommandation : 100 - (IFT_prédit × 30) | |
| 4. Classement par score (plus élevé = meilleur) | |
| """) | |
| recommend_btn = gr.Button("🎯 Recommander Parcelles", variant="primary") | |
| with gr.Row(): | |
| recommendations_plot = gr.Plot() | |
| recommendations_summary = gr.Markdown() | |
| recommend_btn.click(recommend_sensitive_crop_plots, outputs=[recommendations_plot, recommendations_summary]) | |
| with gr.Tab("📊 Exploration Données"): | |
| gr.Markdown("### Explorer les données brutes de la Station Expérimentale de Kerguéhennec") | |
| with gr.Row(): | |
| with gr.Column(): | |
| data_year_start = gr.Slider( | |
| minimum=2014, | |
| maximum=2025, | |
| value=2020, | |
| step=1, | |
| label="Année de début" | |
| ) | |
| data_year_end = gr.Slider( | |
| minimum=2014, | |
| maximum=2025, | |
| value=2025, | |
| step=1, | |
| label="Année de fin" | |
| ) | |
| data_plot_filter = gr.Dropdown( | |
| choices=get_available_plots(), | |
| value="Toutes", | |
| label="Filtrer par parcelle" | |
| ) | |
| data_crop_filter = gr.Dropdown( | |
| choices=get_available_crops(), | |
| value="Toutes", | |
| label="Filtrer par culture" | |
| ) | |
| data_intervention_filter = gr.Dropdown( | |
| choices=get_available_interventions(), | |
| value="Toutes", | |
| label="Filtrer par type d'intervention" | |
| ) | |
| explore_btn = gr.Button("🔍 Explorer les Données", variant="primary") | |
| with gr.Row(): | |
| data_plot = gr.Plot() | |
| data_summary = gr.Markdown() | |
| explore_btn.click( | |
| explore_raw_data, | |
| inputs=[data_year_start, data_year_end, data_plot_filter, data_crop_filter, data_intervention_filter], | |
| outputs=[data_plot, data_summary] | |
| ) | |
| with gr.Tab("🔧 Resources MCP"): | |
| gr.Markdown("### Resources MCP disponibles pour les LLM") | |
| gr.Markdown(""" | |
| Ces resources fournissent un accès structuré aux données agricoles pour les LLM via le protocole MCP. | |
| """) | |
| with gr.Row(): | |
| with gr.Column(): | |
| gr.Markdown("#### Resources statiques") | |
| static_btn1 = gr.Button("📋 Plots disponibles", variant="secondary") | |
| static_btn2 = gr.Button("🌾 Cultures disponibles", variant="secondary") | |
| static_btn3 = gr.Button("📅 Années disponibles", variant="secondary") | |
| static_btn4 = gr.Button("📊 Info dataset", variant="secondary") | |
| static_btn5 = gr.Button("🧪 Usage herbicides", variant="secondary") | |
| static_btn6 = gr.Button("🔮 Prédictions 2025-2027", variant="secondary") | |
| static_btn7 = gr.Button("🌱 Recommandations", variant="secondary") | |
| static_btn8 = gr.Button("📋 Liste Resources", variant="primary") | |
| with gr.Column(): | |
| gr.Markdown("#### Resources paramétrées") | |
| with gr.Row(): | |
| plot_input = gr.Textbox(label="Nom de parcelle", placeholder="Champ ferme W du sol") | |
| plot_btn = gr.Button("🏞️ Info parcelle", variant="secondary") | |
| with gr.Row(): | |
| crop_input = gr.Textbox(label="Type de culture", placeholder="blé tendre hiver") | |
| crop_btn = gr.Button("🌾 Info culture", variant="secondary") | |
| with gr.Row(): | |
| year_input = gr.Number(label="Année", value=2023, precision=0) | |
| year_btn = gr.Button("📅 Résumé année", variant="secondary") | |
| with gr.Row(): | |
| pred_plot_input = gr.Textbox(label="Parcelle pour prédictions", placeholder="Etang 5") | |
| pred_plot_btn = gr.Button("🔮 Prédictions parcelle", variant="secondary") | |
| with gr.Row(): | |
| resource_output = gr.Markdown(label="Résultat de la resource") | |
| # Connexions des boutons | |
| static_btn1.click(lambda: get_available_plots_resource(), outputs=[resource_output]) | |
| static_btn2.click(lambda: get_available_crops_resource(), outputs=[resource_output]) | |
| static_btn3.click(lambda: get_available_years_resource(), outputs=[resource_output]) | |
| static_btn4.click(lambda: get_dataset_info(), outputs=[resource_output]) | |
| static_btn5.click(lambda: get_herbicide_usage_summary(), outputs=[resource_output]) | |
| static_btn6.click(lambda: get_predictions_summary(), outputs=[resource_output]) | |
| static_btn7.click(lambda: get_recommendations_summary(), outputs=[resource_output]) | |
| static_btn8.click(lambda: list_mcp_resources(), outputs=[resource_output]) | |
| plot_btn.click(lambda plot: get_plot_info(plot), inputs=[plot_input], outputs=[resource_output]) | |
| crop_btn.click(lambda crop: get_crop_info(crop), inputs=[crop_input], outputs=[resource_output]) | |
| year_btn.click(lambda year: get_year_summary(int(year)), inputs=[year_input], outputs=[resource_output]) | |
| pred_plot_btn.click(lambda plot: get_plot_predictions(plot), inputs=[pred_plot_input], outputs=[resource_output]) | |
| return demo | |