Code/utils/visualisation.py

from collections import Counter

import altair as alt
import numpy as np
import pandas as pd
import streamlit as st

from utils.translations import _


def afficher_graphique_altair(df):
    """Affiche des graphiques Altair de concentration IHH par categorie d'operation.

    Cree un graphique par categorie (assemblage, fabrication, traitement, extraction)
    montrant la concentration IHH pays/acteurs avec indicateur ICS par taille/couleur.
    Applique un offset aux points superposes pour ameliorer la lisibilite.

    Args:
        df: DataFrame avec colonnes categorie, nom, ihh_pays, ihh_acteurs, ics_cat.
    """
    ordre_personnalise = [
        str(_("pages.visualisations.categories.assembly")),
        str(_("pages.visualisations.categories.manufacturing")),
        str(_("pages.visualisations.categories.processing")),
        str(_("pages.visualisations.categories.extraction"))
    ]
    categories = [cat for cat in ordre_personnalise if cat in df['categorie'].unique()]
    for cat in categories:
        st.markdown(f"### {str(cat)}")
        df_cat = df[df['categorie'] == cat].copy()

        # Convertir les colonnes en float pour éviter les warnings de compatibilité
        df_cat = df_cat.astype({'ihh_pays': float, 'ihh_acteurs': float})

        coord_pairs = list(zip(df_cat['ihh_pays'].round(1), df_cat['ihh_acteurs'].round(1), strict=False))
        counts = Counter(coord_pairs)

        offset_x = []
        offset_y = {}
        seen = Counter()
        for pair in coord_pairs:
            rank = seen[pair]
            seen[pair] += 1
            if counts[pair] > 1:
                angle = rank * 1.5
                radius = 0.8 + 0.4 * rank
                offset_x.append(radius * np.cos(angle))
                offset_y[pair] = radius * np.sin(angle)
            else:
                offset_x.append(0)
                offset_y[pair] = 0

        df_cat.loc[:, 'ihh_pays'] = df_cat['ihh_pays'] + offset_x
        df_cat.loc[:, 'ihh_acteurs'] = df_cat['ihh_acteurs'] + [offset_y[p] for p in coord_pairs]
        df_cat.loc[:, 'ihh_pays_text'] = df_cat['ihh_pays'] + 0.5
        df_cat.loc[:, 'ihh_acteurs_text'] = df_cat['ihh_acteurs'] + 0.5

        base = alt.Chart(df_cat).encode(
            x=alt.X('ihh_pays:Q', title=str(_("pages.visualisations.axis_titles.ihh_countries"))),
            y=alt.Y('ihh_acteurs:Q', title=str(_("pages.visualisations.axis_titles.ihh_actors"))),
            size=alt.Size('ics_cat:Q', scale=alt.Scale(domain=[1, 2, 3], range=[50, 500, 1000]), legend=None),
            color=alt.Color('ics_cat:N', scale=alt.Scale(domain=[1, 2, 3], range=['darkgreen', 'orange', 'darkred']))
        )

        points = base.mark_circle(opacity=0.6)
        lines = alt.Chart(df_cat).mark_rule(strokeWidth=0.5, color='gray').encode(
            x='ihh_pays:Q', x2='ihh_pays_text:Q',
            y='ihh_acteurs:Q', y2='ihh_acteurs_text:Q'
        )

        labels = alt.Chart(df_cat).mark_text(
            align='left', dx=3, dy=-3, fontSize=8, font='Arial', angle=335
        ).encode(
            x='ihh_pays_text:Q',
            y='ihh_acteurs_text:Q',
            text='nom:N'
        )

        hline_15 = alt.Chart(df_cat).mark_rule(strokeDash=[2,2], color='green').encode(y=alt.datum(15))
        hline_25 = alt.Chart(df_cat).mark_rule(strokeDash=[2,2], color='red').encode(y=alt.datum(25))
        vline_15 = alt.Chart(df_cat).mark_rule(strokeDash=[2,2], color='green').encode(x=alt.datum(15))
        vline_25 = alt.Chart(df_cat).mark_rule(strokeDash=[2,2], color='red').encode(x=alt.datum(25))

        chart = (points + lines + labels + hline_15 + hline_25 + vline_15 + vline_25).properties(
            width=500,
            height=400,
            title=str(_("pages.visualisations.chart_titles.concentration_criticality")).format(str(cat))
        ).interactive()

        st.altair_chart(chart, use_container_width=True)


def creer_graphes(donnees):
    """Cree un graphique Altair IHH extraction/reserves vs IVC pour les minerais.

    Visualise la concentration des ressources minieres (IHH extraction et reserves)
    en fonction de la vulnerabilite competitive (IVC). Applique un offset aux points
    superposes.

    Args:
        donnees: Liste de dictionnaires avec cles nom, ivc, ihh_extraction, ihh_reserves.
    """
    if not donnees:
        st.warning(str(_("pages.visualisations.no_data")))
        return

    try:
        df = pd.DataFrame(donnees)
        df['ivc_cat'] = df['ivc'].apply(lambda x: 1 if x <= 15 else (2 if x <= 30 else 3))

        # Convertir les colonnes en float pour éviter les warnings de compatibilité
        df = df.astype({'ihh_extraction': float, 'ihh_reserves': float})

        from collections import Counter
        coord_pairs = list(zip(df['ihh_extraction'].round(1), df['ihh_reserves'].round(1), strict=False))
        counts = Counter(coord_pairs)

        offset_x, offset_y = [], {}
        seen = Counter()
        for pair in coord_pairs:
            rank = seen[pair]
            seen[pair] += 1
            if counts[pair] > 1:
                angle = rank * 1.5
                radius = 0.8 + 0.4 * rank
                offset_x.append(radius * np.cos(angle))
                offset_y[pair] = radius * np.sin(angle)
            else:
                offset_x.append(0)
                offset_y[pair] = 0

        df.loc[:, 'ihh_extraction'] = df['ihh_extraction'] + offset_x
        df.loc[:, 'ihh_reserves'] = df['ihh_reserves'] + [offset_y[p] for p in coord_pairs]
        df.loc[:, 'ihh_extraction_text'] = df['ihh_extraction'] + 0.5
        df.loc[:, 'ihh_reserves_text'] = df['ihh_reserves'] + 0.5

        base = alt.Chart(df).encode(
            x=alt.X('ihh_extraction:Q', title=str(_("pages.visualisations.axis_titles.ihh_extraction"))),
            y=alt.Y('ihh_reserves:Q', title=str(_("pages.visualisations.axis_titles.ihh_reserves"))),
            size=alt.Size('ivc_cat:Q', scale=alt.Scale(domain=[1, 2, 3], range=[50, 500, 1000]), legend=None),
            color=alt.Color('ivc_cat:N', scale=alt.Scale(domain=[1, 2, 3], range=['darkgreen', 'orange', 'darkred'])),
            tooltip=['nom:N', 'ivc:Q', 'ihh_extraction:Q', 'ihh_reserves:Q']
        )

        points = base.mark_circle(opacity=0.6)
        lines = alt.Chart(df).mark_rule(strokeWidth=0.5, color='gray').encode(
            x='ihh_extraction:Q', x2='ihh_extraction_text:Q',
            y='ihh_reserves:Q', y2='ihh_reserves_text:Q'
        )

        labels = alt.Chart(df).mark_text(
            align='left', dx=10, dy=-10, fontSize=10, font='Arial', angle=335
        ).encode(
            x='ihh_extraction_text:Q',
            y='ihh_reserves_text:Q',
            text='nom:N'
        )

        hline_15 = alt.Chart(df).mark_rule(strokeDash=[2,2], color='green').encode(y=alt.datum(15))
        hline_25 = alt.Chart(df).mark_rule(strokeDash=[2,2], color='red').encode(y=alt.datum(25))
        vline_15 = alt.Chart(df).mark_rule(strokeDash=[2,2], color='green').encode(x=alt.datum(15))
        vline_25 = alt.Chart(df).mark_rule(strokeDash=[2,2], color='red').encode(x=alt.datum(25))

        chart = (points + lines + labels + hline_15 + hline_25 + vline_15 + vline_25).properties(
            width=600,
            height=500,
            title=str(_("pages.visualisations.chart_titles.concentration_resources"))
        ).interactive()

        st.altair_chart(chart, use_container_width=True)

    except Exception as e:
        st.error(f"{str(_('errors.graph_creation_error'))} {e}")


def lancer_visualisation_ihh_ics(graph):
    """Lance la visualisation IHH/ICS pour les operations (niveau 10).

    Filtre les noeuds de niveau 10 (operations), recupere leurs donnees IHH/ICS,
    et affiche les graphiques par categorie d'operation.

    Args:
        graph: Graphe NetworkX contenant les attributs niveau, ihh_pays, ihh_acteurs, ics.
    """
    try:
        import networkx as nx

        from utils.graph_utils import recuperer_donnees

        niveaux = nx.get_node_attributes(graph, "niveau")
        noeuds = [n for n, v in niveaux.items() if v == "10" and "Reserves" not in n]
        noeuds.sort()

        df = recuperer_donnees(graph, noeuds)
        if df.empty:
            st.warning(str(_("pages.visualisations.no_data")))
        else:
            afficher_graphique_altair(df)
    except Exception as e:
        st.error(f"{str(_('errors.ihh_criticality_error'))} {e}")


def lancer_visualisation_ihh_ivc(graph):
    """Lance la visualisation IHH/IVC pour les minerais (niveau 2).

    Filtre les minerais (niveau 2) ayant un attribut IVC, recupere leurs donnees
    IHH extraction/reserves et IVC, et affiche le graphique de concentration.

    Args:
        graph: Graphe NetworkX contenant les attributs niveau, ivc, ihh_pays.
    """
    try:
        from utils.graph_utils import recuperer_donnees_2
        noeuds_niveau_2 = [
            n for n, data in graph.nodes(data=True)
            if data.get("niveau") == "2" and "ivc" in data
        ]
        if not noeuds_niveau_2:
            return
        data = recuperer_donnees_2(graph, noeuds_niveau_2)
        creer_graphes(data)
    except Exception as e:
        st.error(f"{str(_('errors.ihh_ivc_error'))} {e}")