data / app.py
Tracy André
updated
2eaa9ec
raw
history blame
26.8 kB
import os
os.environ["GRADIO_ANALYTICS_ENABLED"] = "False"
import gradio as gr
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
import plotly.express as px
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import warnings
from datasets import load_dataset
import pandas as pd
from huggingface_hub import HfApi
import urllib.parse
warnings.filterwarnings('ignore')
# Configuration Hugging Face
hf_token = os.environ.get("HF_TOKEN")
dataset_id = "HackathonCRA/2024"
# Configuration des graphiques
plt.style.use('default')
sns.set_palette("husl")
class AgricultureAnalyzer:
def __init__(self):
self.df = None
self.risk_analysis = None
def load_data(self, file_path=None):
"""Charge les données agricoles UNIQUEMENT depuis Hugging Face"""
try:
print(f"🤗 Chargement du dataset Hugging Face: {dataset_id}")
# Chargement du dataset avec configuration CSV robuste
try:
print(f"🔧 Tentative avec configuration CSV sécurisée...")
# Configuration pour forcer le chargement en string
if hf_token:
dataset = load_dataset(
dataset_id,
token=hf_token,
data_files="*.csv", # Seulement les CSV
sep=",",
encoding="utf-8",
dtype=str, # Force tout en string
na_filter=False, # Pas de conversion NaN automatique
keep_default_na=False # Pas de valeurs NA par défaut
)
print(f"🔑 Chargement sécurisé avec token réussi")
else:
dataset = load_dataset(
dataset_id,
data_files="*.csv",
sep=",",
encoding="utf-8",
dtype=str,
na_filter=False,
keep_default_na=False
)
print(f"🔑 Chargement sécurisé sans token réussi")
except Exception as parse_error:
print(f"⚠️ Erreur avec configuration sécurisée: {str(parse_error)[:100]}...")
print(f"🔄 Tentative de chargement standard...")
# Fallback: chargement standard
try:
if hf_token:
dataset = load_dataset(dataset_id, token=hf_token)
print(f"🔑 Chargement standard avec token réussi")
else:
dataset = load_dataset(dataset_id)
print(f"🔑 Chargement standard sans token réussi")
except Exception as standard_error:
print(f"⚠️ Erreur de chargement standard: {str(standard_error)[:100]}...")
print(f"🔄 Tentative avec chargement CSV manuel...")
# Forcer tous les types en string pour éviter les erreurs de parsing
# Chargement avec configuration CSV personnalisée
from datasets import DatasetDict
import pandas as pd
# Alternative: utiliser l'API HuggingFace pour lister les fichiers CSV
try:
api = HfApi(token=hf_token)
all_files = api.list_repo_files(dataset_id, repo_type="dataset")
# Filtrer pour ne garder que les CSV récents
csv_files = [f for f in all_files if f.endswith('.csv') and any(year in f for year in ['2020', '2021', '2022', '2023', '2024', '2025'])]
csv_files.sort() # Trier par ordre alphabétique
print(f"📁 Fichiers CSV détectés: {len(csv_files)}")
for f in csv_files:
print(f" - {f}")
except Exception as api_error:
print(f"⚠️ Erreur API HF: {api_error}")
# Fallback avec noms corrects
csv_files = [
"Interventions-(sortie-excel)-Station_Expérimentale_de_Kerguéhennec-2020.csv",
"Interventions-(sortie-excel)-Station_Expérimentale_de_Kerguéhennec-2021.csv",
"Interventions-(sortie-excel)-Station_Expérimentale_de_Kerguéhennec-2022.csv",
"Interventions-(sortie-excel)-Station_Expérimentale_de_Kerguéhennec-2023.csv",
"Interventions-(sortie-excel)-Station_Expérimentale_de_Kerguéhennec-2024.csv",
"Interventions-(sortie-excel)-Station_Expérimentale_de_Kerguéhennec-2025.csv"
]
print(f"📊 Chargement alternatif: fichiers CSV individuels...")
# Charger chaque fichier avec pandas et concaténer
all_dataframes = []
for csv_file in csv_files:
try:
# URL directe vers le fichier avec encodage URL correct
encoded_filename = urllib.parse.quote(csv_file, safe='-()_.')
file_url = f"https://huggingface.co/datasets/{dataset_id}/resolve/main/{encoded_filename}"
print(f" ⚙️ Chargement: {csv_file}")
# Charger avec pandas en forçant tout en string et encodage UTF-8
df_temp = pd.read_csv(file_url, dtype=str, na_filter=False, encoding='utf-8')
df_temp['source_file'] = csv_file # Ajouter la source
all_dataframes.append(df_temp)
print(f" ✅ Succès: {len(df_temp)} lignes")
except Exception as file_error:
print(f" ⚠️ Erreur pour {csv_file}: {str(file_error)[:100]}...")
# Essayer avec un autre encodage
try:
print(f" 🔄 Tentative avec encodage latin-1...")
df_temp = pd.read_csv(file_url, dtype=str, na_filter=False, encoding='latin-1')
df_temp['source_file'] = csv_file
all_dataframes.append(df_temp)
print(f" ✅ Succès avec latin-1: {len(df_temp)} lignes")
except Exception as second_error:
print(f" ❌ Échec définitif: {str(second_error)[:50]}...")
continue
if all_dataframes:
# Concaténer tous les DataFrames
df_combined = pd.concat(all_dataframes, ignore_index=True)
print(f"✅ Chargement alternatif réussi: {len(df_combined)} lignes")
# Convertir en format Dataset
from datasets import Dataset
dataset = DatasetDict({
'train': Dataset.from_pandas(df_combined)
})
else:
raise Exception("Aucun fichier CSV n'a pu être chargé")
except Exception as csv_error:
print(f"❌ Échec du chargement CSV manuel: {str(csv_error)[:100]}...")
raise standard_error # Relancer l'erreur précédente
available_splits = list(dataset.keys())
print(f"📊 Splits disponibles: {available_splits}")
# Déterminer quel split utiliser
split_to_use = None
if 'train' in available_splits:
split_to_use = 'train'
elif len(available_splits) > 0:
split_to_use = available_splits[0] # Prendre le premier split disponible
else:
raise Exception("Aucun split trouvé dans le dataset")
print(f"🎯 Utilisation du split: '{split_to_use}'")
# Convertir en DataFrame pandas
df_raw = dataset[split_to_use].to_pandas()
print(f"✅ Dataset chargé: {len(df_raw)} lignes, {len(df_raw.columns)} colonnes")
# Afficher quelques colonnes pour debug
print(f"🏷️ Colonnes: {list(df_raw.columns)[:10]}{'...' if len(df_raw.columns) > 10 else ''}")
# Filtrer pour exclure les fichiers XLSX
# Vérifier les colonnes 'file' ou 'source_file'
file_column = None
if 'file' in df_raw.columns:
file_column = 'file'
elif 'source_file' in df_raw.columns:
file_column = 'source_file'
if file_column:
print(f"📁 Types de fichiers détectés: {df_raw[file_column].unique()[:5]}")
# Ne garder que les fichiers CSV (exclure XLSX)
csv_mask = df_raw[file_column].str.endswith('.csv', na=False)
csv_data = df_raw[csv_mask]
print(f"📊 Avant filtrage CSV: {len(df_raw)} lignes")
if len(csv_data) > 0:
df_raw = csv_data
print(f"🗂️ Après filtrage CSV: {len(df_raw)} lignes restantes")
else:
print(f"⚠️ Aucun fichier CSV trouvé dans la colonne '{file_column}', conservation de toutes les données")
else:
print(f"⚠️ Pas de colonne de fichier détectée, on garde toutes les données")
# Filtrer par année si disponible
if 'millesime' in df_raw.columns:
# Convertir la colonne millesime en numérique si elle est en string
try:
df_raw['millesime'] = pd.to_numeric(df_raw['millesime'], errors='coerce')
# Supprimer les lignes avec millesime invalide
df_raw = df_raw.dropna(subset=['millesime'])
df_raw['millesime'] = df_raw['millesime'].astype(int)
except Exception as e:
print(f"⚠️ Problème conversion millesime: {e}")
years = sorted(df_raw['millesime'].unique())
print(f"📅 Années disponibles: {years}")
# Prendre les données récentes (2020+)
recent_data = df_raw[df_raw['millesime'] >= 2020]
if len(recent_data) > 0:
self.df = recent_data
print(f"✅ Données filtrées (2020+): {len(self.df)} lignes")
else:
self.df = df_raw
print(f"✅ Toutes les données utilisées: {len(self.df)} lignes")
else:
self.df = df_raw
print(f"✅ Données chargées: {len(self.df)} lignes (pas de colonne millesime)")
if len(self.df) == 0:
raise Exception("Aucune donnée disponible après filtrage")
return self.analyze_data()
except Exception as e:
print(f"❌ ERREUR lors du chargement du dataset HuggingFace:")
print(f" {str(e)[:200]}...")
print(f"💡 Solutions:")
print(f" 1. Vérifiez l'URL: https://huggingface.co/datasets/{dataset_id}")
print(f" 2. Configurez votre token: export HF_TOKEN='votre_token'")
print(f" 3. Vérifiez vos permissions d'accès")
print(f" 4. Problème de parsing: données avec types incohérents")
raise Exception(f"Dataset HuggingFace requis: {dataset_id} - Erreur: {str(e)[:100]}...")
def create_sample_data(self):
"""Méthode désactivée - utilisation exclusive du dataset HF"""
raise Exception("Cette application nécessite le dataset HuggingFace HackathonCRA/2024")
def analyze_data(self):
"""Analyse des données et calcul des risques"""
if self.df is None:
return "Erreur: Aucune donnée chargée"
# Analyse générale
general_stats = {
'total_parcelles': self.df['numparcell'].nunique(),
'total_interventions': len(self.df),
'surface_totale': self.df['surfparc'].sum(),
'surface_moyenne': self.df['surfparc'].mean(),
'periode': f"{self.df['millesime'].min()} - {self.df['millesime'].max()}"
}
# Analyse des herbicides
herbicides_df = self.df[self.df['familleprod'] == 'Herbicides'].copy()
herbicide_stats = {
'nb_interventions_herbicides': len(herbicides_df),
'pourcentage_herbicides': (len(herbicides_df) / len(self.df)) * 100,
'parcelles_traitees': herbicides_df['numparcell'].nunique()
}
# Calcul de l'analyse des risques
self.calculate_risk_analysis()
return general_stats, herbicide_stats
def calculate_risk_analysis(self):
"""Calcule l'analyse des risques par parcelle"""
# Groupement des données par parcelle
risk_analysis = self.df.groupby(['numparcell', 'nomparc', 'libelleusag', 'surfparc']).agg({
'familleprod': lambda x: (x == 'Herbicides').sum(), # Nb traitements herbicides
'libevenem': lambda x: len(x.unique()), # Diversité des événements
'produit': lambda x: len(x.unique()), # Diversité des produits
'quantitetot': 'sum' # Quantité totale
}).round(2)
# Quantités d'herbicides spécifiques
herbicide_quantities = self.df[self.df['familleprod'] == 'Herbicides'].groupby(
['numparcell', 'nomparc', 'libelleusag', 'surfparc'])['quantitetot'].sum().fillna(0)
risk_analysis['Quantite_herbicides'] = herbicide_quantities.reindex(risk_analysis.index, fill_value=0)
risk_analysis.columns = ['Nb_herbicides', 'Diversite_evenements', 'Diversite_produits',
'Quantite_totale', 'Quantite_herbicides']
# Calcul de l'IFT approximatif
risk_analysis['IFT_herbicide_approx'] = (risk_analysis['Quantite_herbicides'] /
risk_analysis.index.get_level_values('surfparc')).round(2)
# Classification du risque
def classify_risk(row):
ift = row['IFT_herbicide_approx']
nb_herb = row['Nb_herbicides']
if ift == 0 and nb_herb == 0:
return 'TRÈS FAIBLE'
elif ift < 1 and nb_herb <= 1:
return 'FAIBLE'
elif ift < 3 and nb_herb <= 3:
return 'MODÉRÉ'
elif ift < 5 and nb_herb <= 5:
return 'ÉLEVÉ'
else:
return 'TRÈS ÉLEVÉ'
risk_analysis['Risque_adventice'] = risk_analysis.apply(classify_risk, axis=1)
# Tri par risque
risk_order = ['TRÈS FAIBLE', 'FAIBLE', 'MODÉRÉ', 'ÉLEVÉ', 'TRÈS ÉLEVÉ']
risk_analysis['Risk_Score'] = risk_analysis['Risque_adventice'].map({r: i for i, r in enumerate(risk_order)})
self.risk_analysis = risk_analysis.sort_values(['Risk_Score', 'IFT_herbicide_approx'])
def get_summary_stats(self):
"""Retourne les statistiques de résumé"""
if self.df is None:
return "Aucune donnée disponible"
stats_text = f"""
## 📊 Statistiques Générales
- **Nombre total de parcelles**: {self.df['numparcell'].nunique()}
- **Nombre d'interventions**: {len(self.df):,}
- **Surface totale**: {self.df['surfparc'].sum():.2f} hectares
- **Surface moyenne par parcelle**: {self.df['surfparc'].mean():.2f} hectares
- **Période**: {self.df['millesime'].min()} - {self.df['millesime'].max()}
## 🧪 Analyse Herbicides
"""
herbicides_df = self.df[self.df['familleprod'] == 'Herbicides']
if len(herbicides_df) > 0:
stats_text += f"""
- **Interventions herbicides**: {len(herbicides_df)} ({(len(herbicides_df)/len(self.df)*100):.1f}%)
- **Parcelles traitées**: {herbicides_df['numparcell'].nunique()}
- **Produits herbicides différents**: {herbicides_df['produit'].nunique()}
"""
if self.risk_analysis is not None:
risk_distribution = self.risk_analysis['Risque_adventice'].value_counts()
stats_text += f"""
## 🎯 Répartition des Risques Adventices
"""
for risk_level in ['TRÈS FAIBLE', 'FAIBLE', 'MODÉRÉ', 'ÉLEVÉ', 'TRÈS ÉLEVÉ']:
if risk_level in risk_distribution:
count = risk_distribution[risk_level]
pct = (count / len(self.risk_analysis)) * 100
stats_text += f"- **{risk_level}**: {count} parcelles ({pct:.1f}%)\n"
return stats_text
def get_low_risk_recommendations(self):
"""Retourne les recommandations pour les parcelles à faible risque"""
if self.risk_analysis is None:
return "Analyse des risques non disponible"
low_risk = self.risk_analysis[
self.risk_analysis['Risque_adventice'].isin(['TRÈS FAIBLE', 'FAIBLE'])
].head(10)
recommendations = "## 🌾 TOP 10 - Parcelles Recommandées pour Cultures Sensibles (Pois, Haricot)\n\n"
for idx, row in low_risk.iterrows():
parcelle, nom, culture, surface = idx
recommendations += f"""
**Parcelle {parcelle}** ({nom})
- Culture actuelle: {culture}
- Surface: {surface:.2f} ha
- Niveau de risque: {row['Risque_adventice']}
- IFT herbicide: {row['IFT_herbicide_approx']:.2f}
- Nombre d'herbicides: {row['Nb_herbicides']}
---
"""
return recommendations
def create_risk_visualization(self):
"""Crée la visualisation des risques"""
if self.risk_analysis is None:
return None
risk_df = self.risk_analysis.reset_index()
fig = px.scatter(risk_df,
x='surfparc',
y='IFT_herbicide_approx',
color='Risque_adventice',
size='Nb_herbicides',
hover_data=['nomparc', 'libelleusag'],
color_discrete_map={
'TRÈS FAIBLE': 'green',
'FAIBLE': 'lightgreen',
'MODÉRÉ': 'orange',
'ÉLEVÉ': 'red',
'TRÈS ÉLEVÉ': 'darkred'
},
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=800, height=600, title_font_size=16)
return fig
def create_culture_analysis(self):
"""Analyse par type de culture"""
if self.df is None:
return None
culture_counts = self.df['libelleusag'].value_counts()
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
def create_risk_distribution(self):
"""Distribution des niveaux de risque"""
if self.risk_analysis is None:
return None
risk_counts = self.risk_analysis['Risque_adventice'].value_counts()
fig = px.bar(x=risk_counts.index,
y=risk_counts.values,
color=risk_counts.index,
color_discrete_map={
'TRÈS FAIBLE': 'green',
'FAIBLE': 'lightgreen',
'MODÉRÉ': 'orange',
'ÉLEVÉ': 'red',
'TRÈS ÉLEVÉ': 'darkred'
},
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
# Initialisation de l'analyseur
analyzer = AgricultureAnalyzer()
analyzer.load_data()
# Interface Gradio
def create_interface():
with gr.Blocks(title="🌾 Analyse Adventices Agricoles CRA", theme=gr.themes.Soft()) as demo:
gr.Markdown("""
# 🌾 Analyse des Adventices Agricoles - CRA Bretagne
**Objectif**: Anticiper et réduire la pression des adventices dans les parcelles agricoles bretonnes
Cette application analyse les données historiques pour identifier les parcelles les plus adaptées
à la culture de plantes sensibles comme le pois ou le haricot.
""")
with gr.Tabs():
with gr.TabItem("📊 Vue d'ensemble"):
gr.Markdown("## Statistiques générales des données agricoles")
stats_output = gr.Markdown(analyzer.get_summary_stats())
with gr.Row():
culture_plot = gr.Plot(analyzer.create_culture_analysis())
risk_dist_plot = gr.Plot(analyzer.create_risk_distribution())
with gr.TabItem("🎯 Analyse des Risques"):
gr.Markdown("## Cartographie des risques adventices par parcelle")
risk_plot = gr.Plot(analyzer.create_risk_visualization())
gr.Markdown("""
**Interprétation du graphique**:
- **Axe X**: Surface de la parcelle (hectares)
- **Axe Y**: IFT Herbicide approximatif
- **Couleur**: Niveau de risque adventice
- **Taille**: Nombre d'herbicides utilisés
Les parcelles vertes (risque faible) sont idéales pour les cultures sensibles.
""")
with gr.TabItem("🌾 Recommandations"):
gr.Markdown(analyzer.get_low_risk_recommendations())
gr.Markdown("""
## 💡 Conseils pour la gestion des adventices
### Parcelles à Très Faible Risque (Vertes)
- ✅ **Idéales pour pois et haricot**
- ✅ Historique d'usage herbicide minimal
- ✅ Pression adventice faible attendue
### Parcelles à Faible Risque (Vert clair)
- ⚠️ Surveillance légère recommandée
- ✅ Conviennent aux cultures sensibles avec précautions
### Parcelles à Risque Modéré/Élevé (Orange/Rouge)
- ❌ Éviter pour cultures sensibles
- 🔍 Rotation nécessaire avant implantation
- 📈 Surveillance renforcée des adventices
### Stratégies alternatives
- **Rotation longue**: 3-4 ans avant cultures sensibles
- **Cultures intermédiaires**: CIPAN pour réduire la pression
- **Techniques mécaniques**: Hersage, binage
- **Biostimulants**: Renforcement naturel des cultures
""")
with gr.TabItem("ℹ️ À propos"):
gr.Markdown("""
## 🎯 Méthodologie
Cette analyse se base sur :
### Calcul de l'IFT (Indice de Fréquence de Traitement)
- **IFT ≈ Quantité appliquée / Surface de parcelle**
- Indicateur de l'intensité des traitements herbicides
### Classification des risques
- **TRÈS FAIBLE**: IFT = 0, aucun herbicide
- **FAIBLE**: IFT < 1, usage minimal
- **MODÉRÉ**: IFT < 3, usage modéré
- **ÉLEVÉ**: IFT < 5, usage important
- **TRÈS ÉLEVÉ**: IFT ≥ 5, usage intensif
### Données analysées
- **Source**: Station Expérimentale de Kerguéhennec
- **Période**: Campagne 2025
- **Variables**: Interventions, produits, quantités, surfaces
---
**Développé pour le Hackathon CRA Bretagne** 🏆
*Application d'aide à la décision pour une agriculture durable*
""")
# Bouton de rafraîchissement
refresh_btn = gr.Button("🔄 Actualiser les données", variant="secondary")
def refresh_data():
analyzer.load_data()
return (
analyzer.get_summary_stats(),
analyzer.create_culture_analysis(),
analyzer.create_risk_distribution(),
analyzer.create_risk_visualization(),
analyzer.get_low_risk_recommendations()
)
refresh_btn.click(
refresh_data,
outputs=[stats_output, culture_plot, risk_dist_plot, risk_plot]
)
return demo
# Lancement de l'application
if __name__ == "__main__":
demo = create_interface()
# Configuration pour Hugging Face Spaces
demo.launch(
server_name="0.0.0.0",
server_port=7860,
share=False # Pas besoin de share sur HF Spaces
)