Code/app/plan_d_action/utils/data/plan_d_action.py

import streamlit as st
import matplotlib.pyplot as plt

from app.plan_d_action.utils.data import (
    PRECONISATIONS,
    INDICATEURS,
    poids_operation,
    parse_chains_md,
    set_vulnerability,
    colorer_couleurs,
    afficher_bloc_ihh_isg,
    afficher_description,
    afficher_caracteristiques_minerai,
    initialiser_seuils
)

def calcul_poids_chaine(poids_A: int, poids_F: int, poids_T: int, poids_E: int, poids_M: int) -> tuple[str,dict,int]:
    poids_total = (\
        poids_A * poids_operation["Assemblage"] + \
        poids_F * poids_operation["Fabrication"] + \
        poids_T * poids_operation["Traitement"] + \
        poids_E * poids_operation["Extraction"] + \
        poids_M * poids_operation["Substitution"] \
    ) / sum(poids_operation.values())

    if poids_total < 3:
        criticite_chaine = "Modérée"
        niveau_criticite = {"Facile"}
    elif poids_total < 6:
        criticite_chaine = "Élevée"
        niveau_criticite = {"Facile", "Modérée"}
    else:
        criticite_chaine = "Critique"
        niveau_criticite = {"Facile", "Modérée", "Difficile"}

    return criticite_chaine, niveau_criticite, poids_total

def analyser_chaines(chaines: list[dict], produits: dict, composants: dict, mineraux: dict, seuils: dict, top_n: int = 0) -> list[tuple[str, str, int]]:
    resultats = []

    for chaine in chaines:
        sel_prod = chaine["produit"]
        sel_comp = chaine["composant"]
        sel_miner = chaine["minerai"]

        poids_A, *_ = set_vulnerability(produits[sel_prod]["IHH_Assemblage"], produits[sel_prod]["ISG_Assemblage"], "IHH", "ISG", seuils)
        poids_F, *_ = set_vulnerability(composants[sel_comp]["IHH_Fabrication"], composants[sel_comp]["ISG_Fabrication"], "IHH", "ISG", seuils)
        poids_T, *_ = set_vulnerability(mineraux[sel_miner]["IHH_Traitement"], mineraux[sel_miner]["ISG_Traitement"], "IHH", "ISG", seuils)
        poids_E, *_ = set_vulnerability(mineraux[sel_miner]["IHH_Extraction"], mineraux[sel_miner]["ISG_Extraction"], "IHH", "ISG", seuils)
        poids_M, *_ = set_vulnerability(mineraux[sel_miner]["ICS"], mineraux[sel_miner]["IVC"], "ICS", "IVC", seuils)

        criticite_chaine, niveau_criticite, poids_total = calcul_poids_chaine(
            poids_A, poids_F, poids_T, poids_E, poids_M
        )

        resultats.append({
            "chaine": chaine,
            "criticite_chaine": criticite_chaine,
            "niveau_criticite": niveau_criticite,
            "poids_total": poids_total
        })

    # Tri décroissant
    resultats.sort(key=lambda x: x["poids_total"], reverse=True)

    # Si top_n n'est pas spécifié, tout est retourné
    if top_n == 0 or top_n >= len(resultats):
        return resultats

    # Déterminer le seuil de coupure
    seuil_poids = resultats[top_n - 1]["poids_total"]

    # Inclure tous ceux dont le poids est égal au seuil
    top_resultats = [r for r in resultats if r["poids_total"] >= seuil_poids]

    return top_resultats

def tableau_de_bord(chains: list[dict], produits: dict, composants: dict, mineraux: dict, seuils: dict):
    col_left, col_right = st.columns([2, 3], gap="small", border=True)
    with col_left:
        st.markdown("**<u>Panneau de sélection</u>**", unsafe_allow_html=True)

        produits_disponibles = sorted({c["produit"] for c in chains})
        sel_prod = st.selectbox("Produit", produits_disponibles, index=produits_disponibles.index(st.session_state.sel_prod) if st.session_state.sel_prod else 0)
        if sel_prod != st.session_state.sel_prod:
            st.session_state.sel_prod = sel_prod
            st.session_state.sel_comp = None
            st.session_state.sel_miner = None
            st.rerun()

        composants_dispo = sorted({c["composant"] for c in chains if c["produit"] == sel_prod})
        sel_comp = st.selectbox("Composant", composants_dispo, index=composants_dispo.index(st.session_state.sel_comp) if st.session_state.sel_comp else 0)
        if sel_comp != st.session_state.sel_comp:
            st.session_state.sel_comp = sel_comp
            st.session_state.sel_miner = None
            st.rerun()

        mineraux_dispo = sorted({c["minerai"] for c in chains if c["produit"] == sel_prod and c["composant"] == sel_comp})
        sel_miner = st.selectbox("Minerai", mineraux_dispo, index=mineraux_dispo.index(st.session_state.sel_miner) if st.session_state.sel_miner else 0)
        if sel_miner != st.session_state.sel_miner:
            st.session_state.sel_miner = sel_miner
            st.rerun()

    with col_right:
        top_chains = analyser_chaines(chains, produits, composants, mineraux, seuils, top_n=5)
        st.markdown("**<u>Top chaînes critiques pour sélection rapide</u>**", unsafe_allow_html=True)
        for i, entry in enumerate(top_chains):
            ch = entry["chaine"]
            poids = entry["poids_total"]
            criticite = entry["criticite_chaine"]
            if st.button(f"**{ch['produit']} <-> {ch['composant']} <-> {ch['minerai']}** : {poids:.2f} → {criticite}", key=f"select_{i}"):
                st.session_state.sel_prod = ch["produit"]
                st.session_state.sel_comp = ch["composant"]
                st.session_state.sel_miner = ch["minerai"]
                st.rerun()

    c1, c2 = st.columns([3, 2], gap="small", border=True, vertical_alignment='center')
    with c1:
        st.markdown("**<u>Synthèse des criticités</u>**", unsafe_allow_html=True)
        poids_A, couleur_A, couleur_A_ihh, couleur_A_isg = set_vulnerability(produits[sel_prod]["IHH_Assemblage"], produits[sel_prod]["ISG_Assemblage"], "IHH", "ISG", seuils)
        poids_F, couleur_F, couleur_F_ihh, couleur_F_isg = set_vulnerability(composants[sel_comp]["IHH_Fabrication"], composants[sel_comp]["ISG_Fabrication"], "IHH", "ISG", seuils)
        poids_T, couleur_T, couleur_T_ihh, couleur_T_isg = set_vulnerability(mineraux[sel_miner]["IHH_Traitement"], mineraux[sel_miner]["ISG_Traitement"], "IHH", "ISG", seuils)
        poids_E, couleur_E, couleur_E_ihh, couleur_E_isg = set_vulnerability(mineraux[sel_miner]["IHH_Extraction"], mineraux[sel_miner]["ISG_Extraction"], "IHH", "ISG", seuils)
        poids_M, couleur_M, couleur_M_ics, couleur_M_ivc = set_vulnerability(mineraux[sel_miner]["ICS"], mineraux[sel_miner]["IVC"], "ICS", "IVC", seuils)

        st.markdown(f"* **{sel_prod} - Assemblage** : {colorer_couleurs(couleur_A)} ({poids_A})")
        st.markdown(f"* **{sel_comp} - Fabrication** : {colorer_couleurs(couleur_F)} ({poids_F})")
        st.markdown(f"* **{sel_miner} - Traitement** : {colorer_couleurs(couleur_T)} ({poids_T})")
        st.markdown(f"* **{sel_miner} - Extraction** : {colorer_couleurs(couleur_E)} ({poids_E})")
        st.markdown(f"* **{sel_miner} - Minerai** : {colorer_couleurs(couleur_M)} ({poids_M})")

        criticite_chaine, niveau_criticite, poids_total = calcul_poids_chaine(poids_A, poids_F, poids_T, poids_E, poids_M)
    with c2:
        st.error(f"**Criticité globale : {criticite_chaine} ({poids_total})**")

    return (
    sel_prod, sel_comp, sel_miner, niveau_criticite,
    couleur_A, poids_A, couleur_F, poids_F, couleur_T, poids_T, couleur_E, poids_E, couleur_M, poids_M,
    couleur_A_ihh, couleur_A_isg, couleur_F_ihh, couleur_F_isg, couleur_T_ihh, couleur_T_isg,couleur_E_ihh, couleur_E_isg, couleur_M_ics, couleur_M_ivc
    )

def afficher_criticites(produits: dict, composants: dict, mineraux: dict, sel_prod: str, sel_comp: str, sel_miner: str, seuils: dict) -> None:
    with st.expander("Vue d’ensemble des criticités", expanded=True):
        st.markdown("## Vue d’ensemble des criticités", unsafe_allow_html=True)

        col_left, col_right = st.columns([1, 1], gap="small", border=True)

        with col_left:
            fig1, ax1 = plt.subplots(figsize=(2, 2.4))
            ax1.scatter([produits[sel_prod]["ISG_Assemblage"]], [produits[sel_prod]["IHH_Assemblage"]], label="Assemblage", s=5)
            ax1.scatter([composants[sel_comp]["ISG_Fabrication"]], [composants[sel_comp]["IHH_Fabrication"]], label="Fabrication", s=5)
            ax1.scatter([mineraux[sel_miner]["ISG_Extraction"]], [mineraux[sel_miner]["IHH_Extraction"]], label="Extraction", s=5)
            ax1.scatter([mineraux[sel_miner]["ISG_Traitement"]], [mineraux[sel_miner]["IHH_Traitement"]], label="Traitement", s=5)

            # Seuils ISG (vertical)
            ax1.axvline(seuils["ISG"]["vert"]["max"], linestyle='--', color='green', alpha=0.7, linewidth=0.5)  # Seuil vert-orange
            ax1.axvline(seuils["ISG"]["rouge"]["min"], linestyle='--', color='red', alpha=0.7, linewidth=0.5)    # Seuil orange-rouge

            # Seuils IHH (horizontal)
            ax1.axhline(seuils["IHH"]["vert"]["max"], linestyle='--', color='green', alpha=0.7, linewidth=0.5)  # Seuil vert-orange
            ax1.axhline(seuils["IHH"]["rouge"]["min"], linestyle='--', color='red', alpha=0.7, linewidth=0.5)    # Seuil orange-rouge

            ax1.set_xlim(0, 100)
            ax1.set_ylim(0, 100)
            ax1.set_xlabel("ISG", fontsize=4)
            ax1.set_ylabel("IHH", fontsize=4)
            ax1.tick_params(axis='both', which='major', labelsize=4)
            ax1.legend(bbox_to_anchor=(0.5, -0.25), loc='upper center', fontsize=4)
            plt.tight_layout()
            st.pyplot(fig1)

        with col_right:
            fig2, ax2 = plt.subplots(figsize=(2, 2.1))
            ax2.scatter([mineraux[sel_miner]["IVC"]], [mineraux[sel_miner]["ICS"]], color='green', s=5, label=sel_miner)

            # Seuils IVC (vertical)
            ax2.axvline(seuils["IVC"]["vert"]["max"], linestyle='--', color='green', alpha=0.7, linewidth=0.5)  # Seuil vert-orange
            ax2.axvline(seuils["IVC"]["rouge"]["min"], linestyle='--', color='red', alpha=0.7, linewidth=0.5)    # Seuil orange-rouge

            # Seuils ICS (horizontal)
            ax2.axhline(seuils["ICS"]["vert"]["max"], linestyle='--', color='green', alpha=0.7, linewidth=0.5)  # Seuil vert-orange
            ax2.axhline(seuils["ICS"]["rouge"]["min"], linestyle='--', color='red', alpha=0.7, linewidth=0.5)    # Seuil orange-rouge

            ax2.set_xlim(0, max(100, mineraux[sel_miner]["IVC"]))
            ax2.set_ylim(0, 1)
            ax2.set_xlabel("IVC", fontsize=4)
            ax2.set_ylabel("ICS", fontsize=4)
            ax2.tick_params(axis='both', which='major', labelsize=4)
            ax2.legend(bbox_to_anchor=(0.5, -0.25), loc='upper center', fontsize=4)
            plt.tight_layout()
            st.pyplot(fig2)

        st.markdown(f"""
            Les lignes pointillées en {colorer_couleurs("vert")} ou {colorer_couleurs("rouge")} représentent les seuils des indices concernés.\n
            Les indices ISG (stabilité géopolitique) et IVC (concurrence intersectorielle) influent sur la probabilité de survenance d'un risque.\n
            Les indices IHH (concentration géographique) et ICS (capacité de substitution) influent sur le niveau d'impact d'un risque.\n
            Une opération se trouvant au-dessus des deux seuils a donc une forte probabilité d'être impactée avec un niveau élevé sur l'incapacité à continuer la production.
        """)

def afficher_explications_et_details(
    couleur_A, poids_A, couleur_F, poids_F, couleur_T, poids_T, couleur_E, poids_E, couleur_M, poids_M,
    produits, composants, mineraux, sel_prod, sel_comp, sel_miner,
    couleur_A_ihh, couleur_A_isg, couleur_F_ihh, couleur_F_isg, couleur_T_ihh, couleur_T_isg,couleur_E_ihh, couleur_E_isg, couleur_M_ics, couleur_M_ivc, ui = True) -> str|None:
    with st.expander("Explications et détails", expanded = True):
        from collections import Counter
        couleurs = [couleur_A, couleur_F, couleur_T, couleur_E, couleur_M]
        compte = Counter(couleurs)
        nb_rouge = compte["Rouge"]
        nb_orange = compte["Orange"]
        nb_vert = compte["Vert"]

        contenu_md = f"""
            Pour cette chaîne :blue-background[**{sel_prod} <-> {sel_comp} <-> {sel_miner}**], avec {nb_rouge} criticité(s) de niveau {colorer_couleurs("Rouge")}, {nb_orange} {colorer_couleurs("Orange")} et {nb_vert} {colorer_couleurs("Vert")}, les indices individuels par opération sont :

            * **{sel_prod} - Assemblage** : {colorer_couleurs(couleur_A)} ({poids_A})
                * IHH = {produits[sel_prod]["IHH_Assemblage"]} ({colorer_couleurs(couleur_A_ihh)}) <-> ISG = {produits[sel_prod]["ISG_Assemblage"]} ({colorer_couleurs(couleur_A_isg)})
                * pondération de l'Assemblage dans le calcul de la criticité globale : 1,5
                * se référer à **{sel_prod} et Assemblage** plus bas pour le détail complet
            * **{sel_comp} - Fabrication** : {colorer_couleurs(couleur_F)} ({poids_F})
                * IHH = {composants[sel_comp]["IHH_Fabrication"]} ({colorer_couleurs(couleur_F_ihh)}) <-> ISG = {composants[sel_comp]["ISG_Fabrication"]} ({colorer_couleurs(couleur_F_isg)})
                * pondération de la Fabrication dans le calcul de la criticité globale : 2
                * se référer à **{sel_comp} et Fabrication** plus bas pour le détail complet
            * **{sel_miner} - Traitement** : {colorer_couleurs(couleur_A)} ({poids_A})
                * IHH = {mineraux[sel_miner]["IHH_Traitement"]} ({colorer_couleurs(couleur_T_ihh)}) <-> ISG = {mineraux[sel_miner]["ISG_Traitement"]} ({colorer_couleurs(couleur_T_isg)})
                * pondération du Traitement dans le calcul de la criticité globale : 1,5
                * se référer à **{sel_miner} — Vue globale** plus bas pour le détail complet de l'ensemble du minerai
            * **{sel_miner} - Extraction** : {colorer_couleurs(couleur_E)} ({poids_E})
                * IHH = {mineraux[sel_miner]["IHH_Extraction"]} ({colorer_couleurs(couleur_E_ihh)}) <-> ISG = {mineraux[sel_miner]["ISG_Extraction"]} ({colorer_couleurs(couleur_E_isg)})
                * pondération de l'Extraction dans le calcul de la criticité globale : 1
            * **{sel_miner} - Minerai** : {colorer_couleurs(couleur_M)} ({poids_M})
                * ICS = {mineraux[sel_miner]["ICS"]} ({colorer_couleurs(couleur_M_ics)}) <-> IVC = {mineraux[sel_miner]["IVC"]} ({colorer_couleurs(couleur_M_ivc)})
                * pondération de la Substitution dans le calcul de la criticité globale : 2
        """
        contenu_md += "\n"
        if ui:
            st.markdown(contenu_md)
            return None
        else:
            return contenu_md

def afficher_preconisations_et_indicateurs_generiques(niveau_criticite: dict, poids_A: int, poids_F: int, poids_T: int, poids_E: int, poids_M: int, ui: bool = True) -> tuple[str|None,str|None]:
    contenu_md_left = "### Préconisations :\n\n"
    contenu_md_left += "Mise en œuvre : \n"

    for niveau, contenu in PRECONISATIONS.items():
        if niveau in niveau_criticite:
            contenu_md_left += f"* {colorer_couleurs(niveau)}\n"
            for p in PRECONISATIONS[niveau]:
                contenu_md_left += f"    - {p}\n"

    contenu_md_right = "### Indicateurs :\n\n"
    contenu_md_right += "Mise en œuvre : \n"

    for niveau, contenu in INDICATEURS.items():
        if niveau in niveau_criticite:
            contenu_md_right += f"* {colorer_couleurs(niveau)}\n"
            for p in INDICATEURS[niveau]:
                contenu_md_right += f"    - {p}\n"

    if ui:
        with st.expander("Préconisations et indicateurs génériques"):
            col_left, col_right = st.columns([1, 1], gap="small", border=True)
            with col_left:
                st.markdown(contenu_md_left)
            with col_right:
                st.markdown(contenu_md_right)
        return None, None
    else:
        return contenu_md_left, contenu_md_right

def afficher_preconisations_specifiques(operation: str, niveau_criticite_operation: dict) -> str:
    contenu_md = "#### Préconisations :\n\n"
    contenu_md += "Mise en œuvre : \n"
    for niveau, contenu in PRECONISATIONS[operation].items():
        if niveau in niveau_criticite_operation[operation]:
            contenu_md += f"* {colorer_couleurs(niveau)}\n"
            for p in PRECONISATIONS[operation][niveau]:
                contenu_md += f"    - {p}\n"
    return(contenu_md)

def afficher_indicateurs_specifiques(operation: str, niveau_criticite_operation: dict) -> str:
    contenu_md = "#### Indicateurs :\n\n"
    contenu_md += "Mise en œuvre : \n"
    for niveau, contenu in INDICATEURS[operation].items():
        if niveau in niveau_criticite_operation[operation]:
            contenu_md += f"* {colorer_couleurs(niveau)}\n"
            for p in INDICATEURS[operation][niveau]:
                contenu_md += f"    - {p}\n"
    return(contenu_md)

def afficher_preconisations_et_indicateurs_specifiques(sel_prod: str, sel_comp: str, sel_miner: str, niveau_criticite_operation: dict) -> None:
    for operation in ["Assemblage", "Fabrication", "Traitement", "Extraction"]:
        if operation == "Assemblage":
            item = sel_prod
        elif operation == "Fabrication":
            item = sel_comp
        else:
            item = sel_miner
        with st.expander(f"Préconisations et indicateurs spécifiques - {operation}"):
            st.markdown(f"### {operation} -> :blue-background[{item}]")
            col_left, col_right = st.columns([1, 1], gap="small", border=True)
            with col_left:
                st.markdown(afficher_preconisations_specifiques(operation, niveau_criticite_operation))
            with col_right:
                st.markdown(afficher_indicateurs_specifiques(operation, niveau_criticite_operation))

def afficher_preconisations_et_indicateurs(niveau_criticite: dict, sel_prod: str, sel_comp: str, sel_miner: str, poids_A: int, poids_F: int, poids_T: int, poids_E: int, poids_M: int) -> None:
    st.markdown("## Préconisations et indicateurs")

    afficher_preconisations_et_indicateurs_generiques(niveau_criticite, poids_A, poids_F, poids_T, poids_E, poids_M)

    def affectation_poids(poids_operation):
        if poids_operation < 3:
            niveau_criticite = {"Facile"}
        elif poids_operation < 6:
            niveau_criticite = {"Facile", "Modérée"}
        else:
            niveau_criticite = {"Facile", "Modérée", "Difficile"}
        return niveau_criticite

    niveau_criticite_operation = {}
    niveau_criticite_operation["Assemblage"] = affectation_poids(poids_A)
    niveau_criticite_operation["Fabrication"] = affectation_poids(poids_F)
    niveau_criticite_operation["Traitement"] = affectation_poids(poids_T)
    niveau_criticite_operation["Extraction"] = affectation_poids(poids_E)

    afficher_preconisations_et_indicateurs_specifiques(sel_prod, sel_comp, sel_miner, niveau_criticite_operation)

def afficher_details_operations(produits, composants, mineraux, sel_prod, sel_comp, sel_miner, details_sections) -> None:
    st.markdown("## Détails des opérations")

    with st.expander(f"{sel_prod} et Assemblage"):
        assemblage_details = details_sections.get(f"{sel_prod}_assemblage", "")

        afficher_description(f"{sel_prod} et Assemblage", assemblage_details)
        afficher_bloc_ihh_isg("Assemblage", produits[sel_prod]["IHH_Assemblage"], produits[sel_prod]["ISG_Assemblage"], assemblage_details)

    with st.expander(f"{sel_comp} et Fabrication"):
        fabrication_details = details_sections.get(f"{sel_comp}_fabrication", "")
        afficher_description(f"{sel_comp} et Fabrication", fabrication_details)
        afficher_bloc_ihh_isg("Fabrication", composants[sel_comp]["IHH_Fabrication"], composants[sel_comp]["ISG_Fabrication"], fabrication_details)

    with st.expander(f"{sel_miner} — Vue globale"):
        minerai_general = details_sections.get(f"{sel_miner}_general", "")
        afficher_description(f"{sel_miner} — Vue globale", minerai_general)

        extraction_details = details_sections.get(f"{sel_miner}_extraction", "")
        afficher_bloc_ihh_isg("Extraction", mineraux[sel_miner]["IHH_Extraction"], mineraux[sel_miner]["ISG_Extraction"], extraction_details)

        traitement_details = details_sections.get(f"{sel_miner}_traitement", "").removesuffix("\n---\n")
        afficher_bloc_ihh_isg("Traitement", mineraux[sel_miner]["IHH_Traitement"], mineraux[sel_miner]["ISG_Traitement"], traitement_details)

        afficher_caracteristiques_minerai(sel_miner, mineraux[sel_miner], minerai_general)

def initialiser_interface(filepath: str, config_path: str = "assets/config.yaml") -> None:

    produits, composants, mineraux, chains, descriptions, details_sections = parse_chains_md(filepath)

    if not chains:
        st.warning("Aucune chaîne critique trouvée dans le fichier.")
        return

    seuils = initialiser_seuils(config_path)

    sel_prod, sel_comp, sel_miner, niveau_criticite, \
    couleur_A, poids_A, couleur_F, poids_F, couleur_T, poids_T, couleur_E, poids_E, couleur_M, poids_M, \
    couleur_A_ihh, couleur_A_isg, couleur_F_ihh, couleur_F_isg, couleur_T_ihh, couleur_T_isg,couleur_E_ihh, couleur_E_isg, couleur_M_ics, couleur_M_ivc \
    = tableau_de_bord(chains, produits, composants, mineraux, seuils)

    afficher_criticites(produits, composants, mineraux, sel_prod, sel_comp, sel_miner, seuils)

    afficher_explications_et_details(
        couleur_A, poids_A, couleur_F, poids_F, couleur_T, poids_T, couleur_E, poids_E, couleur_M, poids_M,
        produits, composants, mineraux, sel_prod, sel_comp, sel_miner,
        couleur_A_ihh, couleur_A_isg, couleur_F_ihh, couleur_F_isg, couleur_T_ihh, couleur_T_isg,couleur_E_ihh, couleur_E_isg, couleur_M_ics, couleur_M_ivc)

    afficher_preconisations_et_indicateurs(niveau_criticite, sel_prod, sel_comp, sel_miner, poids_A, poids_F, poids_T, poids_E, poids_M)

    afficher_details_operations(produits, composants, mineraux, sel_prod, sel_comp, sel_miner, details_sections)