241 lines
9.6 KiB
Python
241 lines
9.6 KiB
Python
from typing import List, Dict, Optional, Any
|
|
import networkx as nx
|
|
import streamlit as st
|
|
import altair as alt
|
|
import numpy as np
|
|
from collections import Counter
|
|
import pandas as pd
|
|
from utils.translations import _
|
|
|
|
|
|
def afficher_graphique_altair(df: pd.DataFrame) -> None:
|
|
"""
|
|
Affiche un graphique Altair pour les données d'IHH.
|
|
|
|
Args:
|
|
df (pd.DataFrame): DataFrame contenant les données de IHH.
|
|
|
|
Notes:
|
|
Cette fonction crée un graphique interactif pour visualiser les
|
|
données d'IHH selon différentes catégories et niveaux de gravité.
|
|
"""
|
|
# Définir les catégories originales (en français) et leur ordre
|
|
categories_fr = ["Assemblage", "Fabrication", "Traitement", "Extraction"]
|
|
|
|
# Créer un dictionnaire de mappage entre les catégories originales et leurs traductions
|
|
mappage_categories = {
|
|
"Assemblage": str(_("pages.visualisations.categories.assembly")),
|
|
"Fabrication": str(_("pages.visualisations.categories.manufacturing")),
|
|
"Traitement": str(_("pages.visualisations.categories.processing")),
|
|
"Extraction": str(_("pages.visualisations.categories.extraction"))
|
|
}
|
|
|
|
# Filtrer les catégories qui existent dans les données
|
|
categories_fr_filtrees = [cat for cat in categories_fr if cat in df['categorie'].unique()]
|
|
|
|
# Parcourir les catégories dans l'ordre défini
|
|
for cat_fr in categories_fr_filtrees:
|
|
# Obtenir le nom traduit de la catégorie pour l'affichage
|
|
cat_traduit = mappage_categories[cat_fr]
|
|
st.markdown(f"### {cat_traduit}")
|
|
# Mais filtrer sur le nom original dans les données
|
|
df_cat = df[df['categorie'] == cat_fr].copy()
|
|
|
|
coord_pairs = list(zip(df_cat['ihh_pays'].round(1), df_cat['ihh_acteurs'].round(1)))
|
|
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['ihh_pays'] += offset_x
|
|
df_cat['ihh_acteurs'] += [offset_y[p] for p in coord_pairs]
|
|
df_cat['ihh_pays_text'] = df_cat['ihh_pays'] + 0.5
|
|
df_cat['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))
|
|
hline_100 = alt.Chart(df_cat).mark_rule(strokeDash=[2,2], color='white').encode(y=alt.datum(100))
|
|
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))
|
|
vline_100 = alt.Chart(df_cat).mark_rule(strokeDash=[2,2], color='white').encode(x=alt.datum(100))
|
|
|
|
chart = (points + lines + labels + hline_15 + hline_25 + hline_100 + vline_15 + vline_25 + vline_100).properties(
|
|
width=500,
|
|
height=400,
|
|
title=str(_("pages.visualisations.chart_titles.concentration_criticality")).format(cat_traduit)
|
|
).interactive()
|
|
|
|
st.altair_chart(chart, use_container_width=True)
|
|
|
|
|
|
def creer_graphes(donnees: Optional[List[Dict[str, Any]]]) -> None:
|
|
"""
|
|
Crée un graphique Altair pour les données d'IVC.
|
|
|
|
Args:
|
|
donnees (Optional[List[Dict[str, Any]]]): Liste des données d'IVC.
|
|
|
|
Returns:
|
|
None.
|
|
|
|
Notes:
|
|
Cette fonction traite les données d'IVC et crée un graphique
|
|
interactif pour visualiser la concentration des ressources.
|
|
"""
|
|
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))
|
|
|
|
from collections import Counter
|
|
coord_pairs = list(zip(df['ihh_extraction'].round(1), df['ihh_reserves'].round(1)))
|
|
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['ihh_extraction'] += offset_x
|
|
df['ihh_reserves'] += [offset_y[p] for p in coord_pairs]
|
|
df['ihh_extraction_text'] = df['ihh_extraction'] + 0.5
|
|
df['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))
|
|
hline_100 = alt.Chart(df).mark_rule(strokeDash=[2,2], color='red').encode(y=alt.datum(100))
|
|
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))
|
|
vline_100 = alt.Chart(df).mark_rule(strokeDash=[2,2], color='red').encode(x=alt.datum(100))
|
|
|
|
chart = (points + lines + labels + hline_15 + hline_25 + hline_100 + vline_15 + vline_25 + vline_100).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: nx.DiGraph) -> None:
|
|
"""
|
|
Lance une visualisation Altair pour les données d'IHH critique.
|
|
|
|
Args:
|
|
graph (nx.DiGraph): Le graphe NetworkX contenant les données de IHH.
|
|
|
|
Notes:
|
|
Cette fonction traite le graphe et crée un graphique Altair
|
|
pour visualiser les données d'IHH critique.
|
|
"""
|
|
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: nx.DiGraph) -> None:
|
|
"""
|
|
Lance une visualisation Altair pour les données d'IVC.
|
|
|
|
Args:
|
|
graph (Annx.Graphy): Le graphe NetworkX contenant les données de IV C.
|
|
|
|
Notes:
|
|
Cette fonction traite le graphe et crée un graphique Altair
|
|
pour visualiser les données d'IV C.
|
|
"""
|
|
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}")
|