"""
app.py - Application Streamlit pour la détermination de l'impact carbone
des matières premières pour aliments composés.
Lancement : streamlit run app.py
"""
import re
import sys
import os
import streamlit as st
import pandas as pd
# Ajouter le dossier src/ au path pour les imports
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "src"))
from flowchart_engine import evaluate_carbon_impact, CarbonResult
import llm_service
import config
import proxy_log
# ============================================================================
# Fonctions utilitaires
# ============================================================================
def get_country_list() -> list[str]:
"""Renvoie la liste des pays connus (clés du mapping FR→ISO)."""
return sorted([p.title() for p in config.PAYS_FR_TO_ISO.keys()])
def _format_action_line(line: str) -> str:
"""Convertit les impacts affichés en kg CO2 eq / t produit."""
m = re.search(r"=\s*([0-9]+(?:\.[0-9]+)?)\s*kg\s*CO2\s*eq\s*/\s*t", line)
if m:
val = float(m.group(1))
return re.sub(
r"=\s*[0-9]+(?:\.[0-9]+)?\s*kg\s*CO2\s*eq\s*/\s*t",
f"= {val:.2f} kg CO2 eq / t", line,
)
m = re.search(r"=\s*([0-9]+(?:\.[0-9]+)?)\s*kg\s*CO2\s*eq\s*/\s*kg", line)
if m:
val = float(m.group(1)) * 1000.0
return re.sub(
r"=\s*[0-9]+(?:\.[0-9]+)?\s*kg\s*CO2\s*eq\s*/\s*kg",
f"= {val:.2f} kg CO2 eq / t", line,
)
return line
def _get_transport_surcharge(
pays_production: str | None = None,
pays_transformation: str | None = None,
type_mp: str = "vegetal_animal",
source_db: str | None = None,
classification: str = "brut",
) -> tuple[float, str]:
"""Retourne (forfait_kg_co2_t, zone) selon nouvelles règles de transport.
Nouvelles règles :
- MP Végétales & Soja :
* ECOALIM → 10 kg CO2 eq / t
* GFLI :
- Transformée → référence pays_transformation
- Brute → référence pays_production
- Minéraux : référence pays_transformation (ECOALIM & GFLI)
Calcul par pays :
- France → +10 kg CO2 eq / t
- Europe (hors FR) → +100 kg CO2 eq / t
- Autre / inconnu → +300 kg CO2 eq / t
"""
# Cas ECOALIM pour végétal/soja : forfait fixe
if source_db == "ECOALIM" and type_mp in ("vegetal_animal", "soja"):
return 10.0, "ECOALIM (forfait fixe)"
# Déterminer le pays de référence selon le type de MP et la classification
pays_ref = None
if type_mp == "mineral":
# Minéraux : toujours utiliser pays_transformation
pays_ref = pays_transformation or pays_production
elif type_mp in ("vegetal_animal", "soja"):
# Végétal/Soja GFLI : classification détermine le pays
if classification == "transforme":
pays_ref = pays_transformation or pays_production
else: # brut
pays_ref = pays_production or pays_transformation
else:
pays_ref = pays_production
# Si pas de pays de référence, utiliser la valeur "autre"
if not pays_ref or not pays_ref.strip():
return config.TRANSPORT_SURCHARGE["autre"], "Autre (inconnu)"
pays_lower = pays_ref.strip().lower()
# France
if pays_lower == "france":
return config.TRANSPORT_SURCHARGE["france"], "France"
# Europe (via le set de noms FR ou via code ISO)
iso = config.PAYS_FR_TO_ISO.get(pays_lower, pays_ref.upper())
if pays_lower in config.EUROPEAN_COUNTRIES_FR or iso in config.EUROPEAN_COUNTRIES_ISO:
return config.TRANSPORT_SURCHARGE["europe"], "Europe"
return config.TRANSPORT_SURCHARGE["autre"], "Autre"
def _display_4_alternatives(
alts: dict,
title: str = "🎯 4 Alternatives proposées",
selection_key: str = "alt_selection",
) -> dict | None:
"""Affiche les 4 colonnes d'alternatives avec possibilité de sélection.
Retourne l'alternative sélectionnée si l'utilisateur valide.
"""
st.subheader(title)
st.info(
"Quand aucune matière exacte n'est trouvée, voici 4 propositions. "
"Sélectionnez celle que vous préférez puis validez."
)
col1, col2, col3, col4 = st.columns(4)
labels = [
("itinerary", "### 🔄 Itinéraire", col1),
("locality", "### 📍 Localité", col2),
("form", "### 🌱 Forme structurelle", col3),
("combined", "### ✨ Compromis", col4),
]
radio_options: list[str] = []
radio_map: dict[str, dict] = {}
for key, header, col in labels:
with col:
alt = alts.get(key)
st.markdown(header)
if alt:
st.markdown(f"**{alt['name']}**")
st.metric("Impact", f"{alt['impact']:.2f}")
st.caption(f"kg CO2 eq/t | Source: {alt['source']}")
with st.expander("Raison"):
st.markdown(alt["reasoning"])
label = f"{alt['name']} ({alt['impact']:.2f} kg CO2 eq/t — {alt['source']})"
radio_options.append(label)
radio_map[label] = alt
else:
st.caption("Non disponible")
selected = None
if radio_options:
chosen = st.radio(
"Sélectionnez l'alternative à utiliser :",
options=radio_options,
index=0,
key=f"radio_{selection_key}",
horizontal=True,
)
if st.button("✅ Valider cette alternative", key=f"btn_validate_{selection_key}"):
selected = radio_map.get(chosen)
if selected:
st.session_state[f"selected_{selection_key}"] = selected
st.success(
f"✅ Alternative sélectionnée : **{selected['name']}** — "
f"{selected['impact']:.2f} kg CO2 eq/t"
)
st.divider()
return st.session_state.get(f"selected_{selection_key}") or selected
# ============================================================================
# Configuration de la page
# ============================================================================
st.set_page_config(
page_title="POC GAIA - Impact Carbone MP",
page_icon="🌿",
layout="wide",
)
st.title("🌿 POC GAIA — Impact Carbone des Matières Premières")
st.markdown("""
**Logigramme d'aide pour faciliter l'application du Guide de calcul de l'impact carbone
des aliments composés** *(GT Carbone)*
""")
st.divider()
# ============================================================================
# ONGLETS PRINCIPAUX
# ============================================================================
tab_formulation, tab_single, tab_stats = st.tabs(
["📊 Calcul par liste de MP", "🔍 Calcul unitaire", "📈 Statistiques proxies"]
)
# ============================================================================
# TAB 1 : FORMULATION PRODUIT
# ============================================================================
with tab_formulation:
st.subheader("📊 Calcul d'impact par liste de matières premières")
st.markdown("""
Remplissez le tableau ci-dessous avec les matières premières.
- **Code MP** : code interne de la matière première
- **Matière première** : nom usuel
- **Type MP** : Végétale/Animale, Soja ou Minérale (détermine le logigramme)
- **Pays production** / **Pays transformation** : provenance (laisser vide si inconnu)
- **% Appro origine** : pourcentage d'approvisionnement depuis cette origine spécifique (doit totaliser 100% par Code MP)
""")
# --- Initialiser le DataFrame éditable dans session_state ---
_EMPTY_ROW = {
"Type MP": "vegetal_animal",
"Code MP": "",
"Matière première": "",
"Pays production": "",
"Pays transformation": "",
"% Appro origine": 100.0,
"Extrusion": False,
}
if "formulation_df" not in st.session_state:
st.session_state["formulation_df"] = pd.DataFrame([_EMPTY_ROW.copy() for _ in range(5)])
# --- Boutons d'action ---
col_add, col_del, col_import, col_clear, _ = st.columns([1, 1, 1, 1, 2])
with col_add:
if st.button("➕ Ajouter une ligne", key="btn_add_row"):
st.session_state["formulation_df"] = pd.concat(
[st.session_state["formulation_df"], pd.DataFrame([_EMPTY_ROW.copy()])],
ignore_index=True,
)
st.rerun()
with col_del:
if st.button("➖ Retirer dernière ligne", key="btn_del_row"):
if len(st.session_state["formulation_df"]) > 1:
st.session_state["formulation_df"] = (
st.session_state["formulation_df"].iloc[:-1].reset_index(drop=True)
)
st.rerun()
else:
st.warning("⚠️ Le tableau doit contenir au moins une ligne.")
with col_import:
_import_file = st.file_uploader(
"📥 Importer tableau Excel",
type=["xlsx", "xls"],
key="btn_import_formulation",
label_visibility="collapsed",
)
if _import_file is not None:
try:
_df_imp = pd.read_excel(_import_file)
# Mapper les colonnes connues
_col_map = {}
for _c in _df_imp.columns:
_cl = str(_c).lower()
if "code" in _cl:
_col_map["Code MP"] = _c
elif "matière" in _cl or "matiere" in _cl or ("mp" in _cl and "code" not in _cl):
_col_map["Matière première"] = _c
elif "type" in _cl:
_col_map["Type MP"] = _c
elif "production" in _cl or ("pays" in _cl and "transf" not in _cl):
_col_map["Pays production"] = _c
elif "transf" in _cl:
_col_map["Pays transformation"] = _c
elif "extru" in _cl:
_col_map["Extrusion"] = _c
elif "appro" in _cl or "%" in _cl:
_col_map["% Appro origine"] = _c
_new_df = pd.DataFrame([_EMPTY_ROW.copy() for _ in range(len(_df_imp))])
for target, src in _col_map.items():
_new_df[target] = _df_imp[src].astype(str).fillna("")
if "% Appro origine" in _col_map:
_new_df["% Appro origine"] = pd.to_numeric(
_df_imp[_col_map["% Appro origine"]], errors="coerce"
).fillna(100.0)
if "Extrusion" in _col_map:
_new_df["Extrusion"] = _df_imp[_col_map["Extrusion"]].astype(bool)
st.session_state["formulation_df"] = _new_df
st.session_state.pop("formulation_results", None)
st.success(f"✅ {len(_new_df)} lignes importées.")
st.rerun()
except Exception as _e:
st.error(f"Erreur d'import : {_e}")
with col_clear:
if st.button("🗑️ Réinitialiser", key="btn_clear_form"):
st.session_state.pop("formulation_df", None)
st.session_state.pop("formulation_results", None)
st.rerun()
# --- Tableau éditable ---
edited_df = st.data_editor(
st.session_state["formulation_df"],
num_rows="dynamic",
width='stretch',
key="formulation_editor",
column_config={
"Type MP": st.column_config.SelectboxColumn(
"Type MP",
options=["vegetal_animal", "soja", "mineral"],
width="small",
help="Végétale/Animale (hors soja), Dérivé du soja, Minérale/Micro-ingrédient/Additif",
),
"Code MP": st.column_config.TextColumn("Code MP", width="small"),
"Matière première": st.column_config.TextColumn("Matière première", width="medium"),
"Pays production": st.column_config.TextColumn("Pays production", width="medium"),
"Pays transformation": st.column_config.TextColumn("Pays transformation", width="medium"),
"% Appro origine": st.column_config.NumberColumn("% Appro origine", min_value=0, max_value=100, step=0.1, format="%.1f"),
"Extrusion": st.column_config.CheckboxColumn("Extrusion", help="Cocher si la MP subit une extrusion (+56,77 kg CO2 eq/t)", width="small"),
},
)
# Synchroniser les éditions - faire une copie explicite pour éviter les bugs
if not edited_df.equals(st.session_state["formulation_df"]):
st.session_state["formulation_df"] = edited_df.copy()
# Réinitialiser les résultats si le tableau a changé
st.session_state.pop("formulation_results", None)
if st.button("🚀 Calculer l'impact de la formulation", type="primary", width='stretch', key="btn_calc_formulation"):
# Filtrer les lignes valides
rows_to_eval = edited_df[edited_df["Matière première"].astype(str).str.strip() != ""].copy()
if len(rows_to_eval) == 0:
st.error("❌ Aucune matière première renseignée.")
else:
results_list = []
progress = st.progress(0)
total = len(rows_to_eval)
for idx, (_, row) in enumerate(rows_to_eval.iterrows()):
code_mp = str(row["Code MP"]).strip()
mp_name = str(row["Matière première"]).strip()
type_mp_val = str(row.get("Type MP", "vegetal_animal")).strip() or "vegetal_animal"
pays_p = str(row["Pays production"]).strip() if str(row["Pays production"]).strip() else None
pays_t = str(row["Pays transformation"]).strip() if str(row["Pays transformation"]).strip() else None
pct_appro = float(row["% Appro origine"]) if row["% Appro origine"] else 100.0
# Nettoyage
if pays_p and pays_p.lower() in ("", "nan", "none"):
pays_p = None
if pays_t and pays_t.lower() in ("", "nan", "none"):
pays_t = None
with st.spinner(f"Évaluation de {mp_name} ({pays_p or '?'})…"):
res = evaluate_carbon_impact(mp_name, pays_p, pays_t, type_mp=type_mp_val)
# Calcul impact en kg CO2 eq / t
if res.impact_kg_co2_eq is not None:
if "tonne" in (res.unite_source or ""):
impact_kg_t = res.impact_kg_co2_eq
else:
impact_kg_t = res.impact_kg_co2_eq * 1000.0
else:
impact_kg_t = None
# Forfait transport - nouvelle logique (dépend du type MP, classification, BD)
transport_val, transport_zone = _get_transport_surcharge(
pays_production=pays_p,
pays_transformation=pays_t,
type_mp=type_mp_val,
source_db=res.source_db or None,
classification=res.classification or "brut",
)
# Forfait extrusion
is_extrusion = bool(row.get("Extrusion", False))
extrusion_val = config.FORFAIT_EXTRUSION if is_extrusion else 0.0
# Impact total = impact unitaire + transport + extrusion
if impact_kg_t is not None:
impact_avec_transport = impact_kg_t + transport_val + extrusion_val
else:
impact_avec_transport = None
# Impact pondéré = (impact + transport + extrusion) × (% appro / 100)
if impact_avec_transport is not None:
impact_pondere = impact_avec_transport * (pct_appro / 100.0)
else:
impact_pondere = None
# --- Enregistrer le proxy choisi (formulation) ---
if res.intrant_utilise:
proxy_log.log_selection(
matiere_recherchee=mp_name,
proxy_choisi=res.intrant_utilise,
scenario=res.node_resultat or "inconnu",
impact_kg_co2_t=impact_kg_t,
source_db=res.source_db or "",
match_exact=res.match_exact,
pays_production=pays_p or "",
pays_transformation=pays_t or "",
type_mp=type_mp_val,
)
results_list.append({
"Type MP": type_mp_val,
"Code MP": code_mp,
"Matière première": mp_name,
"Pays production": pays_p or "",
"Pays transformation": pays_t or "",
"% Appro origine": pct_appro,
"Impact unitaire (kg CO2 eq/t)": round(impact_kg_t, 2) if impact_kg_t else None,
"Zone transport": transport_zone,
"Forfait transport (kg CO2 eq/t)": transport_val,
"Forfait extrusion (kg CO2 eq/t)": extrusion_val if is_extrusion else 0.0,
"Impact total (kg CO2 eq/t)": round(impact_avec_transport, 2) if impact_avec_transport else None,
"Impact pondéré (kg CO2 eq/t)": round(impact_pondere, 2) if impact_pondere else None,
"Proxy utilisé": res.intrant_utilise or "",
"Source": res.source_db or "",
"Scénario (node)": res.node_resultat or "",
"Match exact": "✅" if res.match_exact else "⚠️",
"Extrusion": "✅" if is_extrusion else "",
"Erreur": res.erreur or "",
})
progress.progress((idx + 1) / total)
st.session_state["formulation_results"] = results_list
# --- Affichage des résultats ---
if "formulation_results" in st.session_state:
results_list = st.session_state["formulation_results"]
df_results = pd.DataFrame(results_list)
st.divider()
st.subheader("📊 Résultats de la formulation")
# Tableau coloré
st.dataframe(
df_results,
width='stretch',
column_config={
"Impact unitaire (kg CO2 eq/t)": st.column_config.NumberColumn(format="%.2f"),
"Forfait transport (kg CO2 eq/t)": st.column_config.NumberColumn(format="%.0f"),
"Forfait extrusion (kg CO2 eq/t)": st.column_config.NumberColumn(format="%.2f"),
"Impact total (kg CO2 eq/t)": st.column_config.NumberColumn(format="%.2f"),
"Impact pondéré (kg CO2 eq/t)": st.column_config.NumberColumn(format="%.2f"),
"% Appro origine": st.column_config.NumberColumn(format="%.1f"),
},
)
# --- Bouton pour proposer des scénarios sur les lignes en erreur ---
df_errors = df_results[df_results["Erreur"].astype(str).str.strip() != ""]
if len(df_errors) > 0:
st.warning(
f"⚠️ **{len(df_errors)} ligne(s)** n'ont pas pu être évaluées. "
"Vous pouvez lancer une recherche d'alternatives pour ces matières."
)
if st.button(
"🔍 Proposer des scénarios pour les lignes en erreur",
key="btn_scenario_errors",
type="primary",
):
for _idx, _err_row in df_errors.iterrows():
_mp_err = _err_row["Matière première"]
_pays_hint = _err_row.get("Pays production", "") or _err_row.get("Pays transformation", "")
st.markdown(f"#### 🔎 Alternatives pour **{_mp_err}**")
with st.spinner(f"Recherche d'alternatives pour {_mp_err}…"):
_alts = llm_service.find_alternative_materials(
_mp_err,
db_name="GFLI",
country_hint=_pays_hint if _pays_hint else None,
)
if _alts:
_sel = _display_4_alternatives(
_alts,
title=f"🎯 Alternatives pour {_mp_err}",
selection_key=f"form_err_{_idx}",
)
if _sel:
_sel_impact = _sel["impact"]
_sel_name = _sel["name"]
_sel_source = _sel["source"]
_sel_reasoning = _sel.get("reasoning", "")
# Mettre à jour TOUS les résultats qui correspondent à cette matière
for _r in st.session_state["formulation_results"]:
if _r["Matière première"] == _mp_err and _r.get("Erreur"):
_r["Impact unitaire (kg CO2 eq/t)"] = round(_sel_impact, 2)
# Déterminer la classification : brut si pas de transfo, sinon transformé
_classification = "transforme" if _r.get("Pays transformation") else "brut"
transport_v, _ = _get_transport_surcharge(
pays_production=_r.get("Pays production"),
pays_transformation=_r.get("Pays transformation"),
type_mp=_r.get("Type MP", "vegetal_animal"),
source_db=_r.get("Source"),
classification=_classification,
)
extr_v = config.FORFAIT_EXTRUSION if _r.get("Extrusion") == "✅" else 0.0
_r["Impact total (kg CO2 eq/t)"] = round(_sel_impact + transport_v + extr_v, 2)
pct = _r.get("% Appro origine", 100.0)
_r["Impact pondéré (kg CO2 eq/t)"] = round(
(_sel_impact + transport_v + extr_v) * pct / 100.0, 2
)
_r["Proxy utilisé"] = _sel_name
_r["Source"] = _sel_source
_r["Match exact"] = "⚠️"
_r["Erreur"] = ""
st.rerun()
else:
st.info(f"Aucune alternative trouvée pour {_mp_err}.")
# --- Résumé par Code MP ---
st.subheader("📋 Résumé par matière première")
df_res = df_results.copy()
summary_rows = []
for code_mp, group in df_res.groupby("Code MP", sort=False):
mp_name = group["Matière première"].iloc[0]
# Somme des impacts pondérés de toutes les origines pour ce code
total_impact_pondere = group["Impact pondéré (kg CO2 eq/t)"].sum()
# Impact moyen pondéré par appro
impacts_unitaires = group["Impact unitaire (kg CO2 eq/t)"].dropna()
appro_vals = group.loc[impacts_unitaires.index, "% Appro origine"]
if len(impacts_unitaires) > 0 and appro_vals.sum() > 0:
impact_moyen = (impacts_unitaires * appro_vals).sum() / appro_vals.sum()
else:
impact_moyen = None
summary_rows.append({
"Code MP": code_mp,
"Matière première": mp_name,
"Nb origines": len(group),
"Impact moyen pondéré appro (kg CO2 eq/t)": round(impact_moyen, 2) if impact_moyen else None,
"Contribution (kg CO2 eq/t produit)": round(total_impact_pondere, 2) if total_impact_pondere else None,
})
df_summary = pd.DataFrame(summary_rows)
st.dataframe(df_summary, width='stretch')
# --- Erreurs & Téléchargement ---
nb_errors = df_results["Erreur"].astype(str).str.strip().ne("").sum()
st.divider()
col_errors, col_download = st.columns(2)
with col_errors:
if nb_errors > 0:
st.warning(f"⚠️ {nb_errors} ligne(s) en erreur (sans impact calculé)")
else:
st.success("✅ Toutes les lignes ont un impact calculé")
with col_download:
csv = df_results.to_csv(index=False).encode("utf-8")
st.download_button(
label="📥 Télécharger les résultats (CSV)",
data=csv,
file_name="formulation_impact_carbone.csv",
mime="text/csv",
key="btn_download_formulation",
)
# ============================================================================
# TAB 2 : RECHERCHE UNITAIRE
# ============================================================================
with tab_single:
st.markdown(
"Renseignez les informations sur votre matière première ci-dessous, "
"puis lancez l'évaluation."
)
col_form, col_info = st.columns([2, 1])
with col_form:
st.subheader("📝 Formulaire de saisie")
# --- Type de matière première ---
type_mp = st.radio(
"Type de matière première",
options=[
"🌾 Végétale / Animale (hors soja)",
"🫘 Dérivé du soja",
"🧪 Minérale / Micro-ingrédient / Additif",
],
horizontal=True,
key="input_type_mp",
help="Choisissez le type de MP pour appliquer le logigramme adapté.",
)
# Mapper vers le code interne
TYPE_MAP = {
"🌾 Végétale / Animale (hors soja)": "vegetal_animal",
"🫘 Dérivé du soja": "soja",
"🧪 Minérale / Micro-ingrédient / Additif": "mineral",
}
type_mp_code = TYPE_MAP[type_mp]
matiere = st.text_input(
"Nom de la matière première",
key="input_matiere",
placeholder="Ex : BLE, T.TNSL DEC., ORGE, T. COLZA, LUZERNE, T. SOJA, CARBONATE DE CALCIUM…",
help="Entrez le nom usuel de la matière première.",
)
provenance_connue = st.radio(
"Connaissez-vous la provenance de la matière première ?",
options=["Oui", "Non"],
horizontal=True,
)
pays_production = None
pays_transformation = None
if provenance_connue == "Oui":
if type_mp_code == "mineral":
# Pour les minéraux, on demande uniquement le pays de transformation
pays_transformation = st.text_input(
"Pays de transformation",
key="input_pays_transfo",
placeholder="Ex : France, Allemagne, Pays-Bas…",
help="Pays où la transformation du minéral a eu lieu.",
)
else:
pays_production = st.text_input(
"Pays de production primaire / origine de la graine",
key="input_pays_prod",
placeholder="Ex : France, Brésil, Ukraine…",
help="Pays de production de la MP brute (ou pays d'origine de la graine pour le soja).",
)
pays_transformation = st.text_input(
"Pays de transformation (laisser vide si MP brute)",
key="input_pays_transfo",
placeholder="Ex : France, Allemagne… (vide si non transformée)",
help="Pays où la transformation a eu lieu. Laissez vide si la matière est brute.",
)
st.markdown("---")
extrusion_single = st.checkbox(
"🔧 Forfait extrusion (+56,77 kg CO2 eq/t)",
key="input_extrusion",
help="Cocher si la matière première subit une extrusion.",
)
run_button = st.button(
"🚀 Évaluer l'impact carbone", type="primary", width='stretch'
)
with col_info:
st.subheader("ℹ️ Règles générales")
st.info("""
- Le choix des valeurs par défaut relève de la responsabilité des entreprises.
- Il faut attribuer un facteur d'émission à **tous** les intrants.
- Si provenance inconnue → valeur la **plus défavorable**.
- Si provenance connue mais donnée inexistante → donnée générique la **plus pertinente**.
- Bases : **GFLI** et **ECOALIM** en priorité.
""")
# Afficher le logigramme actif
if type_mp_code == "vegetal_animal":
st.success("📋 **Logigramme actif** : MP Végétale / Animale (hors soja)")
st.warning(
"⚠️ **Hors produits dérivés du soja** — "
"Pour les dérivés de soja, sélectionnez le type correspondant."
)
elif type_mp_code == "soja":
st.success("📋 **Logigramme actif** : MP Dérivée du soja")
st.markdown(
"Ce logigramme est spécifique aux **graines de soja** et à leurs "
"**produits dérivés** (tourteaux, huiles, lécithines…)."
)
else:
st.success("📋 **Logigramme actif** : MP Minérale / Micro-ingrédient / Additif")
st.markdown(
"Ce logigramme couvre les **minéraux**, **micro-ingrédients**, "
"**vitamines** et **additifs**."
)
# ========================================================================
# Exécution et stockage des résultats
# ========================================================================
if run_button:
if not matiere:
st.error("❌ Veuillez saisir un nom de matière première.")
st.stop()
pays_prod = (
pays_production.strip()
if pays_production and pays_production.strip()
else None
)
pays_transfo = (
pays_transformation.strip()
if pays_transformation and pays_transformation.strip()
else None
)
with st.spinner(
"🔄 Évaluation en cours… (classification LLM + recherche dans les bases de données)"
):
result: CarbonResult = evaluate_carbon_impact(
matiere_premiere=matiere.strip(),
pays_production=pays_prod,
pays_transformation=pays_transfo,
type_mp=type_mp_code,
)
st.session_state["last_result"] = result
st.session_state["last_matiere"] = matiere.strip()
st.session_state.pop("searched_alternatives", None)
# --- Enregistrer le proxy choisi ---
if result.intrant_utilise:
_impact_for_log = None
if result.impact_kg_co2_eq is not None:
_impact_for_log = (
result.impact_kg_co2_eq
if "tonne" in (result.unite_source or "")
else result.impact_kg_co2_eq * 1000.0
)
proxy_log.log_selection(
matiere_recherchee=matiere.strip(),
proxy_choisi=result.intrant_utilise,
scenario=result.node_resultat or "inconnu",
impact_kg_co2_t=_impact_for_log,
source_db=result.source_db or "",
match_exact=result.match_exact,
pays_production=result.pays_production or "",
pays_transformation=result.pays_transformation or "",
type_mp=type_mp_code,
)
# ========================================================================
# Affichage des résultats (depuis session_state — persiste entre reruns)
# ========================================================================
if "last_result" in st.session_state:
result = st.session_state["last_result"]
st.divider()
# Section 0b : 4 alternatives (fallback)
if result.alternatives_combined or result.alternatives_itinerary:
_fallback_sel = _display_4_alternatives(
{
"itinerary": result.alternatives_itinerary,
"locality": result.alternatives_locality,
"form": result.alternatives_form,
"combined": result.alternatives_combined,
},
title="🎯 4 Alternatives proposées (absence de correspondance)",
selection_key="single_fallback",
)
if _fallback_sel:
result.impact_kg_co2_eq = _fallback_sel["impact"]
result.impact_tonne_co2_eq = _fallback_sel["impact"] / 1000.0
result.unite_source = "kg CO2 eq / tonne de produit"
result.source_db = _fallback_sel["source"]
result.intrant_utilise = _fallback_sel["name"]
result.match_exact = False
result.justification_alternative = _fallback_sel.get("reasoning", "")
result.erreur = None
st.session_state["last_result"] = result
# Main layout: Results and Candidates side-by-side
if result.erreur:
st.error(f"❌ {result.erreur}")
else:
# Create two columns for side-by-side display
col_result, col_candidates = st.columns([1.2, 1])
# Calculer l'impact pour l'affichage
if result.impact_kg_co2_eq is not None:
if "tonne" in (result.unite_source or ""):
impact_kg_t = result.impact_kg_co2_eq
else:
impact_kg_t = result.impact_kg_co2_eq * 1000.0
# --- Forfait transport ---
transport_val, transport_zone = _get_transport_surcharge(
pays_production=result.pays_production,
pays_transformation=result.pays_transformation,
type_mp=type_mp_code,
source_db=result.source_db,
classification=result.classification,
)
# --- Forfait extrusion ---
extrusion_val = config.FORFAIT_EXTRUSION if extrusion_single else 0.0
impact_total = impact_kg_t + transport_val + extrusion_val
# ========================================================================
# LEFT COLUMN: Résultat de l'impact carbone
# ========================================================================
with col_result:
with st.container():
st.markdown("### 📊 Résultat de l'impact carbone")
st.metric(
label="🌍 Impact carbone Total (MP + Transport)",
value=f"{impact_total:.2f}",
delta="kg CO2 eq / t produit",
)
# Display Source and Intrant utilisé right next to Impact metric
st.markdown(f"Source: {result.source_db}
", unsafe_allow_html=True)
st.markdown(f"Intrant utilisé: {result.intrant_utilise}
", unsafe_allow_html=True)
# Match exact and Classification
st.markdown(f"**Match exact:** {'✅ Oui' if result.match_exact else '⚠️ Non'}")
st.markdown(f"**Classification:** {'🏭 Transformé' if result.classification == 'transforme' else '🌾 Brut'}")
# Pays information if available
if result.pays_production:
st.markdown(f"**Pays production:** {result.pays_production}")
if result.pays_transformation:
st.markdown(f"**Pays transformation:** {result.pays_transformation}")
with st.expander("Détails techniques"):
st.markdown(f"**Node résultat :** `{result.node_resultat}`")
st.markdown("---")
# Affichage des impacts
st.markdown("#### 🎯 Impacts calculés")
# Impact de base
st.metric(
label="Impact carbone (MP)",
value=f"{impact_kg_t:.2f} kg CO2 eq / t",
)
# Forfaits additionnels en colonnes
forfait_cols = st.columns(2)
with forfait_cols[0]:
st.metric(
label=f"🚚 Transport ({transport_zone})",
value=f"+{transport_val:.0f} kg CO2 eq / t",
)
with forfait_cols[1]:
if extrusion_single:
st.metric(
label="🔧 Extrusion",
value=f"+{extrusion_val:.2f} kg CO2 eq / t",
)
# ========================================================================
# RIGHT COLUMN: Produits candidats
# ========================================================================
with col_candidates:
with st.container():
st.markdown("### 📋 Autres Alternatives")
if result.candidats_alternatifs:
if not result.match_exact:
st.warning("⚠️ Pas de correspondance exacte. Sélectionnez une alternative ci-dessous.")
else:
st.info("ℹ️ Autres produits disponibles pour votre recherche.")
# Affichage sous forme de tableau amélioré
candidates_data = []
candidates_map = {}
for idx, cand in enumerate(result.candidats_alternatifs):
nom = cand.get("nom", "")
impact = cand.get("impact", 0)
unite = str(cand.get("unite", ""))
source = cand.get("source", "")
source_upper = source.upper()
is_gfli = "GFLI" in source_upper
if "tonne" in unite or is_gfli:
impact_kg_t = impact
else:
impact_kg_t = impact * 1000.0
candidates_data.append({
"Intrant": nom,
"Impact (kg CO2 eq/t)": round(impact_kg_t, 2),
"Base": source if source else "—"
})
candidates_map[idx] = {
"nom": nom,
"impact": impact_kg_t,
"source": source,
"unite": unite
}
df_candidates = pd.DataFrame(candidates_data)
# Sélection d'une ligne dans le dataframe
st.markdown("**Cliquez sur une ligne pour sélectionner :**")
event = st.dataframe(
df_candidates,
use_container_width=True,
hide_index=True,
on_select="rerun",
selection_mode="single-row",
key="candidate_dataframe",
column_config={
"Intrant": st.column_config.TextColumn("Intrant", width="large"),
"Impact (kg CO2 eq/t)": st.column_config.NumberColumn(
"Impact (kg CO2 eq/t)",
format="%.2f",
width="medium"
),
"Base": st.column_config.TextColumn("Base", width="small"),
}
)
# Récupérer la sélection
selected_rows = event.selection.rows if hasattr(event, 'selection') else []
if selected_rows:
selected_idx = selected_rows[0]
selected_cand = candidates_map[selected_idx]
st.markdown("---")
st.info(f"**Sélectionné :** {selected_cand['nom']} — {selected_cand['impact']:.2f} kg CO2 eq/t")
if st.button("✅ Valider cette alternative", key="apply_candidate", type="primary", use_container_width=True):
result.impact_kg_co2_eq = selected_cand["impact"]
result.impact_tonne_co2_eq = selected_cand["impact"] / 1000.0
result.unite_source = "kg CO2 eq / tonne de produit"
result.source_db = selected_cand["source"]
result.intrant_utilise = selected_cand["nom"]
result.match_exact = False
st.session_state["last_result"] = result
st.success(f"✅ Alternative appliquée : **{selected_cand['nom']}** ({selected_cand['impact']:.2f} kg CO2 eq/t)")
st.rerun()
else:
st.caption("👆 Sélectionnez une ligne dans le tableau ci-dessus")
else:
st.info("Aucun produit candidat disponible.")
# Section 3 : Bouton "Chercher une alternative" si match non exact
if not result.match_exact and result.impact_kg_co2_eq is not None:
st.divider()
st.info("💡 La correspondance n'est pas exacte. Vous pouvez chercher d'autres alternatives.")
_c1, _c2, _c3 = st.columns([1, 2, 1])
with _c2:
if st.button(
"🔍 Chercher une alternative plus proche",
width='stretch',
key="btn_find_alternative",
):
matiere_search = st.session_state.get("last_matiere", "")
with st.spinner("Recherche des 4 alternatives en cours..."):
db_name = "GFLI" if "GFLI" in (result.source_db or "") else "ECOALIM"
country_hint = result.pays_production or result.pays_transformation
alternatives = llm_service.find_alternative_materials(
matiere_search,
db_name=db_name,
country_hint=country_hint,
)
if alternatives:
st.session_state["searched_alternatives"] = {
"itinerary": alternatives.get("itinerary"),
"locality": alternatives.get("locality"),
"form": alternatives.get("form"),
"combined": alternatives.get("combined"),
}
st.rerun()
else:
st.error("❌ Pas d'alternatives trouvées.")
# Afficher les alternatives trouvées via le bouton
if "searched_alternatives" in st.session_state:
_searched_sel = _display_4_alternatives(
st.session_state["searched_alternatives"],
title="🎯 Alternatives recherchées",
selection_key="single_searched",
)
if _searched_sel:
result.impact_kg_co2_eq = _searched_sel["impact"]
result.impact_tonne_co2_eq = _searched_sel["impact"] / 1000.0
result.unite_source = "kg CO2 eq / tonne de produit"
result.source_db = _searched_sel["source"]
result.intrant_utilise = _searched_sel["name"]
result.match_exact = False
result.justification_alternative = _searched_sel.get("reasoning", "")
result.erreur = None
st.session_state["last_result"] = result
# Section 4 : Parcours du logigramme
st.divider()
st.subheader("🔀 Parcours du logigramme")
for i, step in enumerate(result.parcours):
with st.expander(f"Étape {i+1} — {step.node_id}", expanded=(i == 0)):
if step.question:
st.markdown(f"**Question :** {step.question}")
if step.answer:
st.markdown(f"**Réponse :** {step.answer}")
if step.action:
st.markdown(f"**Action :** {step.action}")
# Section 5 : Actions appliquées
st.subheader("🔍 Détail des recherches effectuées")
for action in result.actions_appliquees:
line = _format_action_line(action)
if line.startswith(" →"):
st.success(line)
else:
st.markdown(f"- {line}")
# Section 6 : Justification si valeur alternative
if result.justification_alternative:
st.subheader("💡 Justification du choix de valeur")
st.info(result.justification_alternative)
# Section 7 : Classification détaillée
with st.expander("📋 Détail de la classification brut/transformé"):
st.markdown(f"**Classification :** {result.classification}")
st.markdown(f"**Justification :** {result.classification_justification}")
# ========================================================================
# Mode batch (import fichier)
# ========================================================================
st.divider()
st.subheader("📁 Évaluation par lot (import fichier)")
st.markdown(
"Importez un fichier Excel avec les colonnes : "
"`Matière première`, `Pays de la production primaire de la MP brute`, "
"`Pays de la transformation`"
)
uploaded_file = st.file_uploader("Choisir un fichier Excel", type=["xlsx", "xls"])
if uploaded_file is not None:
try:
df_input = pd.read_excel(uploaded_file)
st.dataframe(df_input, width='stretch')
col_mapping = {}
for col in df_input.columns:
col_lower = str(col).lower()
if "matière" in col_lower or "matiere" in col_lower or "mp" in col_lower:
col_mapping["matiere"] = col
elif ("production" in col_lower or "pays" in col_lower) and "transf" not in col_lower:
if "matiere" not in col_lower:
col_mapping["pays_prod"] = col
elif "transf" in col_lower:
col_mapping["pays_transfo"] = col
if "matiere" not in col_mapping:
st.warning("⚠️ Colonne 'Matière première' non détectée. Vérifiez les noms de colonnes.")
else:
if st.button("🚀 Lancer l'évaluation par lot", type="primary"):
results_list = []
progress_bar = st.progress(0)
for idx, row in df_input.iterrows():
mp = str(row.get(col_mapping["matiere"], "")).strip()
if not mp or mp == "nan":
continue
pays_p = str(row.get(col_mapping.get("pays_prod", ""), "")).strip()
pays_t = str(row.get(col_mapping.get("pays_transfo", ""), "")).strip()
pays_p = pays_p if pays_p and pays_p != "nan" else None
pays_t = pays_t if pays_t and pays_t != "nan" else None
with st.spinner(f"Évaluation de {mp}..."):
res = evaluate_carbon_impact(mp, pays_p, pays_t)
# --- Enregistrer le proxy choisi (batch) ---
if res.intrant_utilise:
_batch_impact = None
if res.impact_kg_co2_eq is not None:
_batch_impact = (
res.impact_kg_co2_eq
if "tonne" in (res.unite_source or "")
else res.impact_kg_co2_eq * 1000.0
)
proxy_log.log_selection(
matiere_recherchee=mp,
proxy_choisi=res.intrant_utilise,
scenario=res.node_resultat or "inconnu",
impact_kg_co2_t=_batch_impact,
source_db=res.source_db or "",
match_exact=res.match_exact,
pays_production=pays_p or "",
pays_transformation=pays_t or "",
)
results_list.append({
"Matière première": mp,
"Pays production": pays_p or "",
"Pays transformation": pays_t or "",
"Classification": res.classification,
"Impact (kg CO2 eq / t)": (
res.impact_kg_co2_eq
if res.impact_kg_co2_eq is None
else (
res.impact_kg_co2_eq
if "tonne" in (res.unite_source or "")
else res.impact_kg_co2_eq * 1000.0
)
),
"Unité": "kg CO2 eq / t produit",
"Source": res.source_db,
"Intrant utilisé": res.intrant_utilise,
"Match exact": "Oui" if res.match_exact else "Non",
"Node résultat": res.node_resultat,
"Justification": res.justification_alternative or "",
"Erreur": res.erreur or "",
})
progress_bar.progress((idx + 1) / len(df_input))
if results_list:
df_results = pd.DataFrame(results_list)
st.subheader("📊 Résultats du lot")
st.dataframe(df_results, width='stretch')
csv = df_results.to_csv(index=False).encode("utf-8")
st.download_button(
label="📥 Télécharger les résultats (CSV)",
data=csv,
file_name="resultats_impact_carbone.csv",
mime="text/csv",
)
except Exception as e:
st.error(f"Erreur lors de la lecture du fichier : {e}")
# ============================================================================
# TAB 3 : STATISTIQUES PROXIES
# ============================================================================
with tab_stats:
st.subheader("📈 Statistiques des proxies sélectionnés")
st.markdown(
"Ce tableau recense les **intrants (proxies)** choisis lors des évaluations, "
"avec le nombre de fois qu'ils ont été sélectionnés."
)
# --- Filtre temporel ---
col_filter, _ = st.columns([1, 3])
with col_filter:
period = st.selectbox(
"Période",
options=["Tout", "7 derniers jours", "30 derniers jours", "90 derniers jours"],
index=0,
key="stats_period",
)
days_map = {"Tout": None, "7 derniers jours": 7, "30 derniers jours": 30, "90 derniers jours": 90}
selected_days = days_map[period]
# --- Top proxies ---
st.markdown("### 🏆 Top proxies les plus choisis")
df_top_proxies = proxy_log.top_proxies(n=30, days=selected_days)
if df_top_proxies.empty:
st.info("Aucune sélection enregistrée pour le moment.")
else:
st.dataframe(
df_top_proxies,
width='stretch',
column_config={
"proxy_choisi": st.column_config.TextColumn("Intrant / Proxy", width="large"),
"nb_selections": st.column_config.NumberColumn("Nb sélections", format="%d"),
"dernière_utilisation": st.column_config.DatetimeColumn(
"Dernière utilisation", format="DD/MM/YYYY HH:mm"
),
},
)
st.divider()
# --- Top scénarios ---
st.markdown("### 🔀 Top scénarios (nodes résultat)")
df_top_scenarios = proxy_log.top_scenarios(n=20, days=selected_days)
if df_top_scenarios.empty:
st.info("Aucune sélection enregistrée pour le moment.")
else:
st.dataframe(
df_top_scenarios,
width='stretch',
column_config={
"scenario": st.column_config.TextColumn("Scénario (node)", width="medium"),
"nb_selections": st.column_config.NumberColumn("Nb sélections", format="%d"),
"dernière_utilisation": st.column_config.DatetimeColumn(
"Dernière utilisation", format="DD/MM/YYYY HH:mm"
),
},
)
st.divider()
# --- Journal complet ---
with st.expander("📋 Journal complet des sélections", expanded=False):
df_full = proxy_log.load_log()
if df_full.empty:
st.info("Aucune sélection enregistrée.")
else:
st.dataframe(df_full.sort_values("timestamp", ascending=False), width='stretch')
csv_log = df_full.to_csv(index=False).encode("utf-8")
st.download_button(
label="📥 Télécharger le journal (CSV)",
data=csv_log,
file_name="proxy_selections_log.csv",
mime="text/csv",
key="btn_download_proxy_log",
)
# ============================================================================
# Footer
# ============================================================================
st.divider()
st.caption(
"POC GAIA — Outil d'aide à la détermination de l'impact carbone des matières premières "
"pour aliments composés. Basé sur le Guide GT Carbone, les bases ECOALIM v9 et GFLI 2.0. "
"Classification brut/transformé assistée par Mistral AI et le Catalogue UE des Matières Premières."
)