Spaces:
Sleeping
Sleeping
| """ | |
| Module de visualisation des données agricoles | |
| """ | |
| import pandas as pd | |
| import plotly.express as px | |
| import plotly.graph_objects as go | |
| from config import RISK_COLORS, PLOT_CONFIG | |
| class AgricultureVisualizer: | |
| """Classe responsable de la création des visualisations""" | |
| def __init__(self, data=None, risk_analysis=None): | |
| self.df = data | |
| self.risk_analysis = risk_analysis | |
| def set_data(self, data, risk_analysis=None): | |
| """Définit les données à visualiser""" | |
| self.df = data | |
| if risk_analysis is not None: | |
| self.risk_analysis = risk_analysis | |
| def _create_error_plot(self, title, message): | |
| """Crée un graphique d'erreur standardisé""" | |
| try: | |
| fig = px.scatter(title=title) | |
| 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( | |
| width=PLOT_CONFIG.get("width", 700), | |
| height=PLOT_CONFIG.get("height", 400) | |
| ) | |
| return fig | |
| except Exception as e: | |
| print(f"❌ Erreur lors de la création du graphique d'erreur: {e}") | |
| # Retourner un graphique minimal en cas d'échec total | |
| return go.Figure().add_annotation(text="Erreur critique", x=0.5, y=0.5) | |
| def create_risk_visualization(self): | |
| """Crée la visualisation des risques avec gestion d'erreur robuste""" | |
| try: | |
| if self.risk_analysis is None or len(self.risk_analysis) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune donnée d'analyse des risques disponible", | |
| "Veuillez charger et analyser les données d'abord" | |
| ) | |
| try: | |
| risk_df = self.risk_analysis.reset_index() | |
| # Vérifier les colonnes requises | |
| required_cols = ['surfparc', 'IFT_herbicide_approx', 'Risque_adventice', 'Nb_herbicides'] | |
| missing_cols = [col for col in required_cols if col not in risk_df.columns] | |
| if missing_cols: | |
| return self._create_error_plot( | |
| f"❌ Colonnes manquantes: {missing_cols}", | |
| "Les données ne contiennent pas toutes les colonnes nécessaires" | |
| ) | |
| # Vérifier quelles colonnes sont disponibles pour hover_data | |
| available_hover_cols = [] | |
| for col in ['nomparc', 'libelleusag']: | |
| if col in risk_df.columns: | |
| available_hover_cols.append(col) | |
| # Nettoyer les données pour éviter les erreurs de plotting | |
| risk_df = risk_df.dropna(subset=required_cols) | |
| if len(risk_df) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune donnée valide après nettoyage", | |
| "Toutes les données contiennent des valeurs manquantes" | |
| ) | |
| fig = px.scatter( | |
| risk_df, | |
| x='surfparc', | |
| y='IFT_herbicide_approx', | |
| color='Risque_adventice', | |
| size='Nb_herbicides', | |
| hover_data=available_hover_cols if available_hover_cols else None, | |
| color_discrete_map=RISK_COLORS, | |
| title="🎯 Analyse du Risque Adventice par Parcelle", | |
| labels={ | |
| 'surfparc': 'Surface de la parcelle (ha)', | |
| 'IFT_herbicide_approx': 'IFT Herbicide (approximatif)', | |
| 'Risque_adventice': 'Niveau de risque' | |
| } | |
| ) | |
| fig.update_layout( | |
| width=PLOT_CONFIG["width"], | |
| height=PLOT_CONFIG["height"], | |
| title_font_size=PLOT_CONFIG["title_font_size"] | |
| ) | |
| return fig | |
| except Exception as e: | |
| print(f"❌ Erreur lors de la création du graphique de risque: {e}") | |
| return self._create_error_plot( | |
| "❌ Erreur lors de la création du graphique", | |
| f"Erreur technique: {str(e)[:100]}..." | |
| ) | |
| except Exception as e: | |
| print(f"❌ Erreur critique dans create_risk_visualization: {e}") | |
| return self._create_error_plot( | |
| "❌ Erreur critique", | |
| "Impossible de créer la visualisation des risques" | |
| ) | |
| def create_culture_analysis(self): | |
| """Analyse par type de culture avec gestion d'erreur robuste""" | |
| try: | |
| if self.df is None or len(self.df) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune donnée disponible", | |
| "Veuillez charger les données d'abord" | |
| ) | |
| if 'libelleusag' not in self.df.columns: | |
| return self._create_error_plot( | |
| "❌ Colonne 'libelleusag' non disponible", | |
| "Les données de culture ne sont pas disponibles" | |
| ) | |
| try: | |
| # Nettoyer les données de culture | |
| culture_data = self.df['libelleusag'].dropna() | |
| if len(culture_data) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune donnée de culture valide", | |
| "Toutes les valeurs de culture sont manquantes" | |
| ) | |
| culture_counts = culture_data.value_counts() | |
| if len(culture_counts) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune culture détectée", | |
| "Aucune donnée de culture trouvée après nettoyage" | |
| ) | |
| fig = px.pie( | |
| values=culture_counts.values, | |
| names=culture_counts.index, | |
| title="🌱 Répartition des Cultures" | |
| ) | |
| fig.update_layout(width=700, height=500) | |
| return fig | |
| except Exception as e: | |
| print(f"❌ Erreur lors de la création du graphique de culture: {e}") | |
| return self._create_error_plot( | |
| "❌ Erreur lors de la création du graphique", | |
| f"Erreur technique: {str(e)[:100]}..." | |
| ) | |
| except Exception as e: | |
| print(f"❌ Erreur critique dans create_culture_analysis: {e}") | |
| return self._create_error_plot( | |
| "❌ Erreur critique", | |
| "Impossible de créer l'analyse des cultures" | |
| ) | |
| def create_risk_distribution(self): | |
| """Distribution des niveaux de risque avec gestion d'erreur robuste""" | |
| try: | |
| if self.risk_analysis is None or len(self.risk_analysis) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune analyse des risques disponible", | |
| "Veuillez charger et analyser les données d'abord" | |
| ) | |
| if 'Risque_adventice' not in self.risk_analysis.columns: | |
| return self._create_error_plot( | |
| "❌ Colonne 'Risque_adventice' manquante", | |
| "L'analyse des risques est incomplète" | |
| ) | |
| try: | |
| # Nettoyer les données de risque | |
| risk_data = self.risk_analysis['Risque_adventice'].dropna() | |
| if len(risk_data) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune donnée de risque valide", | |
| "Toutes les valeurs de risque sont manquantes" | |
| ) | |
| risk_counts = risk_data.value_counts() | |
| if len(risk_counts) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucun niveau de risque détecté", | |
| "Aucune donnée de risque trouvée après nettoyage" | |
| ) | |
| fig = px.bar( | |
| x=risk_counts.index, | |
| y=risk_counts.values, | |
| color=risk_counts.index, | |
| color_discrete_map=RISK_COLORS, | |
| title="📊 Distribution des Niveaux de Risque Adventice", | |
| labels={'x': 'Niveau de risque', 'y': 'Nombre de parcelles'} | |
| ) | |
| fig.update_layout(width=700, height=500, showlegend=False) | |
| return fig | |
| except Exception as e: | |
| print(f"❌ Erreur lors de la création du graphique de distribution: {e}") | |
| return self._create_error_plot( | |
| "❌ Erreur lors de la création du graphique", | |
| f"Erreur technique: {str(e)[:100]}..." | |
| ) | |
| except Exception as e: | |
| print(f"❌ Erreur critique dans create_risk_distribution: {e}") | |
| return self._create_error_plot( | |
| "❌ Erreur critique", | |
| "Impossible de créer la distribution des risques" | |
| ) | |
| def create_herbicide_timeline(self): | |
| """Crée un graphique de l'évolution temporelle des herbicides avec gestion d'erreur""" | |
| try: | |
| if self.df is None or len(self.df) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune donnée disponible", | |
| "Veuillez charger les données d'abord" | |
| ) | |
| required_cols = ['millesime', 'familleprod'] | |
| missing_cols = [col for col in required_cols if col not in self.df.columns] | |
| if missing_cols: | |
| return self._create_error_plot( | |
| f"❌ Colonnes manquantes: {missing_cols}", | |
| "Les données temporelles ne sont pas disponibles" | |
| ) | |
| try: | |
| # Filtrer les herbicides et grouper par année | |
| herbicides_df = self.df[self.df['familleprod'] == 'Herbicides'] | |
| if len(herbicides_df) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune donnée d'herbicide disponible", | |
| "Aucune intervention herbicide trouvée dans les données" | |
| ) | |
| agg_dict = {'numparcell': 'nunique'} | |
| if 'quantitetot' in herbicides_df.columns: | |
| agg_dict['quantitetot'] = 'sum' | |
| yearly_herbicides = herbicides_df.groupby('millesime').agg(agg_dict).reset_index() | |
| if len(yearly_herbicides) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune donnée temporelle valide", | |
| "Impossible de grouper les données par année" | |
| ) | |
| y_col = 'quantitetot' if 'quantitetot' in yearly_herbicides.columns else 'numparcell' | |
| y_label = 'Quantité totale d\'herbicides' if y_col == 'quantitetot' else 'Nombre de parcelles traitées' | |
| fig = px.line( | |
| yearly_herbicides, | |
| x='millesime', | |
| y=y_col, | |
| title="📈 Évolution de l'Usage des Herbicides par Année", | |
| labels={ | |
| 'millesime': 'Année', | |
| y_col: y_label | |
| } | |
| ) | |
| fig.update_layout(width=700, height=400) | |
| return fig | |
| except Exception as e: | |
| print(f"❌ Erreur lors de la création du graphique temporel: {e}") | |
| return self._create_error_plot( | |
| "❌ Erreur lors de la création du graphique", | |
| f"Erreur technique: {str(e)[:100]}..." | |
| ) | |
| except Exception as e: | |
| print(f"❌ Erreur critique dans create_herbicide_timeline: {e}") | |
| return self._create_error_plot( | |
| "❌ Erreur critique", | |
| "Impossible de créer l'évolution temporelle" | |
| ) | |
| def create_surface_analysis(self): | |
| """Analyse de la distribution des surfaces avec gestion d'erreur""" | |
| try: | |
| if self.df is None or len(self.df) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune donnée disponible", | |
| "Veuillez charger les données d'abord" | |
| ) | |
| if 'surfparc' not in self.df.columns: | |
| return self._create_error_plot( | |
| "❌ Colonne 'surfparc' manquante", | |
| "Les données de surface ne sont pas disponibles" | |
| ) | |
| try: | |
| # Nettoyer les données de surface | |
| surface_data = self.df['surfparc'].dropna() | |
| if len(surface_data) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune donnée de surface valide", | |
| "Toutes les valeurs de surface sont manquantes" | |
| ) | |
| fig = px.histogram( | |
| x=surface_data, | |
| nbins=20, | |
| title="📏 Distribution des Surfaces de Parcelles", | |
| labels={ | |
| 'x': 'Surface (ha)', | |
| 'count': 'Nombre de parcelles' | |
| } | |
| ) | |
| fig.update_layout(width=700, height=400) | |
| return fig | |
| except Exception as e: | |
| print(f"❌ Erreur lors de la création du graphique de surface: {e}") | |
| return self._create_error_plot( | |
| "❌ Erreur lors de la création du graphique", | |
| f"Erreur technique: {str(e)[:100]}..." | |
| ) | |
| except Exception as e: | |
| print(f"❌ Erreur critique dans create_surface_analysis: {e}") | |
| return self._create_error_plot( | |
| "❌ Erreur critique", | |
| "Impossible de créer l'analyse des surfaces" | |
| ) | |
| def get_available_years(self): | |
| """Retourne la liste des années disponibles dans les données""" | |
| try: | |
| if self.df is None or len(self.df) == 0: | |
| return [] | |
| if 'millesime' not in self.df.columns: | |
| return [] | |
| years = sorted(self.df['millesime'].dropna().unique()) | |
| return [int(year) for year in years if str(year).isdigit()] | |
| except Exception as e: | |
| print(f"❌ Erreur lors de la récupération des années: {e}") | |
| return [] | |
| def create_data_table_by_year(self, year=None): | |
| """Crée un tableau des données pour une année donnée""" | |
| try: | |
| if self.df is None or len(self.df) == 0: | |
| return None, "❌ Aucune donnée disponible" | |
| if 'millesime' not in self.df.columns: | |
| return None, "❌ Colonne 'millesime' manquante" | |
| # Filtrer par année si spécifiée | |
| if year is not None: | |
| try: | |
| filtered_df = self.df[self.df['millesime'] == year].copy() | |
| if len(filtered_df) == 0: | |
| return None, f"❌ Aucune donnée pour l'année {year}" | |
| except Exception as e: | |
| return None, f"❌ Erreur lors du filtrage par année: {e}" | |
| else: | |
| filtered_df = self.df.copy() | |
| # Sélectionner les colonnes les plus importantes pour l'affichage | |
| display_cols = [] | |
| important_cols = [ | |
| 'millesime', 'numparcell', 'nomparc', 'surfparc', | |
| 'libelleusag', 'libevenem', 'familleprod', 'produit', | |
| 'quantitetot', 'unite', 'datedebut', 'datefin' | |
| ] | |
| for col in important_cols: | |
| if col in filtered_df.columns: | |
| display_cols.append(col) | |
| if not display_cols: | |
| return None, "❌ Aucune colonne importante trouvée" | |
| # Limiter le nombre de lignes pour l'affichage | |
| max_rows = 1000 | |
| if len(filtered_df) > max_rows: | |
| display_df = filtered_df[display_cols].head(max_rows) | |
| info_msg = f"📊 Affichage de {max_rows} premières lignes sur {len(filtered_df)} total" | |
| else: | |
| display_df = filtered_df[display_cols] | |
| info_msg = f"📊 {len(display_df)} enregistrements au total" | |
| return display_df, info_msg | |
| except Exception as e: | |
| print(f"❌ Erreur dans create_data_table_by_year: {e}") | |
| return None, f"❌ Erreur lors de la création du tableau: {str(e)[:100]}..." | |
| def create_yearly_summary_chart(self, year=None): | |
| """Crée un graphique de résumé pour une année donnée""" | |
| try: | |
| if self.df is None or len(self.df) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune donnée disponible", | |
| "Veuillez charger les données d'abord" | |
| ) | |
| # Filtrer par année si spécifiée | |
| if year is not None: | |
| try: | |
| year_data = self.df[self.df['millesime'] == year].copy() | |
| if len(year_data) == 0: | |
| return self._create_error_plot( | |
| f"❌ Aucune donnée pour {year}", | |
| f"L'année {year} ne contient aucun enregistrement" | |
| ) | |
| title_suffix = f" - Année {year}" | |
| except Exception as e: | |
| return self._create_error_plot( | |
| "❌ Erreur de filtrage", | |
| f"Impossible de filtrer par année: {str(e)[:50]}..." | |
| ) | |
| else: | |
| year_data = self.df.copy() | |
| title_suffix = " - Toutes années" | |
| try: | |
| # Créer un graphique en barres des types d'interventions | |
| if 'libevenem' in year_data.columns: | |
| event_counts = year_data['libevenem'].value_counts().head(10) | |
| if len(event_counts) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucun événement trouvé", | |
| "Aucune intervention trouvée dans les données" | |
| ) | |
| fig = px.bar( | |
| x=event_counts.values, | |
| y=event_counts.index, | |
| orientation='h', | |
| title=f"📊 Top 10 des Types d'Interventions{title_suffix}", | |
| labels={ | |
| 'x': 'Nombre d\'interventions', | |
| 'y': 'Type d\'intervention' | |
| } | |
| ) | |
| fig.update_layout( | |
| width=800, | |
| height=500, | |
| yaxis={'categoryorder': 'total ascending'} | |
| ) | |
| return fig | |
| else: | |
| return self._create_error_plot( | |
| "❌ Colonne manquante", | |
| "La colonne 'libevenem' n'est pas disponible" | |
| ) | |
| except Exception as e: | |
| print(f"❌ Erreur lors de la création du graphique annuel: {e}") | |
| return self._create_error_plot( | |
| "❌ Erreur lors de la création du graphique", | |
| f"Erreur technique: {str(e)[:100]}..." | |
| ) | |
| except Exception as e: | |
| print(f"❌ Erreur critique dans create_yearly_summary_chart: {e}") | |
| return self._create_error_plot( | |
| "❌ Erreur critique", | |
| "Impossible de créer le résumé annuel" | |
| ) | |
| def create_monthly_activity_chart(self, year=None): | |
| """Crée un graphique d'activité mensuelle pour une année""" | |
| try: | |
| if self.df is None or len(self.df) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune donnée disponible", | |
| "Veuillez charger les données d'abord" | |
| ) | |
| if 'datedebut' not in self.df.columns: | |
| return self._create_error_plot( | |
| "❌ Colonne 'datedebut' manquante", | |
| "Les données de date ne sont pas disponibles" | |
| ) | |
| try: | |
| # Filtrer par année si spécifiée | |
| if year is not None: | |
| year_data = self.df[self.df['millesime'] == year].copy() | |
| title_suffix = f" - Année {year}" | |
| else: | |
| year_data = self.df.copy() | |
| title_suffix = " - Toutes années" | |
| if len(year_data) == 0: | |
| return self._create_error_plot( | |
| f"❌ Aucune donnée pour {year}", | |
| f"L'année {year} ne contient aucun enregistrement" | |
| ) | |
| # Convertir les dates et extraire les mois | |
| year_data['datedebut_parsed'] = pd.to_datetime(year_data['datedebut'], errors='coerce') | |
| year_data = year_data.dropna(subset=['datedebut_parsed']) | |
| if len(year_data) == 0: | |
| return self._create_error_plot( | |
| "❌ Aucune date valide", | |
| "Aucune date d'intervention valide trouvée" | |
| ) | |
| year_data['mois'] = year_data['datedebut_parsed'].dt.month | |
| monthly_counts = year_data['mois'].value_counts().sort_index() | |
| # Noms des mois | |
| month_names = { | |
| 1: 'Janvier', 2: 'Février', 3: 'Mars', 4: 'Avril', | |
| 5: 'Mai', 6: 'Juin', 7: 'Juillet', 8: 'Août', | |
| 9: 'Septembre', 10: 'Octobre', 11: 'Novembre', 12: 'Décembre' | |
| } | |
| months = [month_names.get(m, f'Mois {m}') for m in monthly_counts.index] | |
| fig = px.bar( | |
| x=months, | |
| y=monthly_counts.values, | |
| title=f"📅 Activité Mensuelle{title_suffix}", | |
| labels={ | |
| 'x': 'Mois', | |
| 'y': 'Nombre d\'interventions' | |
| } | |
| ) | |
| fig.update_layout(width=800, height=400) | |
| return fig | |
| except Exception as e: | |
| print(f"❌ Erreur lors de la création du graphique mensuel: {e}") | |
| return self._create_error_plot( | |
| "❌ Erreur lors de la création du graphique", | |
| f"Erreur technique: {str(e)[:100]}..." | |
| ) | |
| except Exception as e: | |
| print(f"❌ Erreur critique dans create_monthly_activity_chart: {e}") | |
| return self._create_error_plot( | |
| "❌ Erreur critique", | |
| "Impossible de créer l'activité mensuelle" | |
| ) | |