""" 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" )