268 lines
8.9 KiB
Python
268 lines
8.9 KiB
Python
import networkx as nx
|
|
import pandas as pd
|
|
import logging
|
|
import streamlit as st
|
|
import json
|
|
import yaml
|
|
import pathlib
|
|
from networkx.drawing.nx_agraph import read_dot
|
|
|
|
# Configuration Gitea
|
|
from config import DOT_FILE
|
|
from utils.gitea import (
|
|
charger_schema_depuis_gitea
|
|
)
|
|
|
|
|
|
def extraire_chemins_depuis(G, source):
|
|
chemins = []
|
|
stack = [(source, [source])]
|
|
while stack:
|
|
(node, path) = stack.pop()
|
|
voisins = list(G.successors(node))
|
|
if not voisins:
|
|
chemins.append(path)
|
|
else:
|
|
for voisin in voisins:
|
|
if voisin not in path:
|
|
stack.append((voisin, path + [voisin]))
|
|
return chemins
|
|
|
|
|
|
def extraire_chemins_vers(G, target, niveau_demande):
|
|
chemins = []
|
|
reverse_G = G.reverse()
|
|
niveaux = nx.get_node_attributes(G, "niveau")
|
|
stack = [(target, [target])]
|
|
|
|
while stack:
|
|
(node, path) = stack.pop()
|
|
voisins = list(reverse_G.successors(node))
|
|
if not voisins:
|
|
chemin_inverse = list(reversed(path))
|
|
contient_niveau = any(
|
|
int(niveaux.get(n, -1)) == niveau_demande for n in chemin_inverse
|
|
)
|
|
if contient_niveau:
|
|
chemins.append(chemin_inverse)
|
|
else:
|
|
for voisin in voisins:
|
|
if voisin not in path:
|
|
stack.append((voisin, path + [voisin]))
|
|
|
|
return chemins
|
|
|
|
|
|
def recuperer_donnees(graph, noeuds):
|
|
donnees = []
|
|
ics = {}
|
|
|
|
for noeud in noeuds:
|
|
try:
|
|
operation, minerai = noeud.split('_', 1)
|
|
except ValueError:
|
|
logging.warning(f"Nom de nœud inattendu : {noeud}")
|
|
continue
|
|
|
|
if operation == "Traitement":
|
|
try:
|
|
fabrications = list(graph.predecessors(minerai))
|
|
valeurs = [
|
|
int(float(graph.get_edge_data(f, minerai)[0].get('ics', 0)) * 100)
|
|
for f in fabrications
|
|
if graph.get_edge_data(f, minerai)
|
|
]
|
|
if valeurs:
|
|
ics[minerai] = round(sum(valeurs) / len(valeurs))
|
|
except Exception as e:
|
|
logging.warning(f"Erreur criticité pour {noeud} : {e}")
|
|
ics[minerai] = 50
|
|
|
|
for noeud in noeuds:
|
|
try:
|
|
operation, minerai = noeud.split('_', 1)
|
|
ihh_pays = int(graph.nodes[noeud].get('ihh_pays', 0))
|
|
ihh_acteurs = int(graph.nodes[noeud].get('ihh_acteurs', 0))
|
|
ics_val = ics.get(minerai, 50)
|
|
ics_cat = 1 if ics_val <= 33 else (2 if ics_val <= 66 else 3)
|
|
|
|
donnees.append({
|
|
'categorie': operation,
|
|
'nom': minerai,
|
|
'ihh_pays': ihh_pays,
|
|
'ihh_acteurs': ihh_acteurs,
|
|
'ics_minerai': ics_val,
|
|
'ics_cat': ics_cat
|
|
})
|
|
except Exception as e:
|
|
logging.error(f"Erreur sur le nœud {noeud} : {e}", exc_info=True)
|
|
|
|
return pd.DataFrame(donnees)
|
|
|
|
|
|
def recuperer_donnees_2(graph, noeuds_2):
|
|
donnees = []
|
|
for minerai in noeuds_2:
|
|
try:
|
|
missing = []
|
|
if not graph.has_node(minerai):
|
|
missing.append(minerai)
|
|
if not graph.has_node(f"Extraction_{minerai}"):
|
|
missing.append(f"Extraction_{minerai}")
|
|
if not graph.has_node(f"Reserves_{minerai}"):
|
|
missing.append(f"Reserves_{minerai}")
|
|
|
|
if missing:
|
|
print(f"⚠️ Nœuds manquants pour {minerai} : {', '.join(missing)} — Ignoré.")
|
|
continue
|
|
|
|
ivc = int(graph.nodes[minerai].get('ivc', 0))
|
|
ihh_extraction_pays = int(graph.nodes[f"Extraction_{minerai}"].get('ihh_pays', 0))
|
|
ihh_reserves_pays = int(graph.nodes[f"Reserves_{minerai}"].get('ihh_pays', 0))
|
|
|
|
donnees.append({
|
|
'nom': minerai,
|
|
'ivc': ivc,
|
|
'ihh_extraction': ihh_extraction_pays,
|
|
'ihh_reserves': ihh_reserves_pays
|
|
})
|
|
except Exception as e:
|
|
print(f"Erreur avec le nœud {minerai} : {e}")
|
|
return donnees
|
|
|
|
|
|
def load_seuils_config(path: str = "assets/config.yaml") -> dict:
|
|
"""
|
|
Charge les seuils depuis le fichier de configuration YAML.
|
|
|
|
Args:
|
|
path (str): Chemin vers le fichier de configuration.
|
|
|
|
Returns:
|
|
dict: Dictionnaire contenant les seuils pour chaque indice.
|
|
"""
|
|
try:
|
|
data = yaml.safe_load(pathlib.Path(path).read_text(encoding="utf-8"))
|
|
return data.get("seuils", {})
|
|
except Exception as e:
|
|
logging.warning(f"Erreur lors du chargement des seuils depuis {path}: {e}")
|
|
# Valeurs par défaut en cas d'erreur
|
|
return {
|
|
"ISG": {"vert": {"max": 40}, "orange": {"min": 40, "max": 70}, "rouge": {"min": 70}},
|
|
"IHH": {"vert": {"max": 15}, "orange": {"min": 15, "max": 25}, "rouge": {"min": 25}},
|
|
"IVC": {"vert": {"max": 15}, "orange": {"min": 15, "max": 60}, "rouge": {"min": 60}}
|
|
}
|
|
|
|
|
|
def determiner_couleur_par_seuil(valeur: int, seuils_indice: dict) -> str:
|
|
"""
|
|
Détermine la couleur en fonction de la valeur et des seuils configurés.
|
|
Logique alignée avec determine_threshold_color du projet.
|
|
|
|
Args:
|
|
valeur (int): Valeur de l'indice à évaluer.
|
|
seuils_indice (dict): Seuils pour cet indice depuis la configuration.
|
|
|
|
Returns:
|
|
str: Couleur correspondante ("darkgreen", "orange", "darkred", "gray").
|
|
"""
|
|
if valeur < 0:
|
|
return "gray"
|
|
|
|
# Vérifier d'abord le seuil rouge (priorité la plus haute)
|
|
if "rouge" in seuils_indice and "min" in seuils_indice["rouge"]:
|
|
if valeur >= seuils_indice["rouge"]["min"]:
|
|
return "darkred"
|
|
|
|
# Ensuite le seuil orange
|
|
if "orange" in seuils_indice and "min" in seuils_indice["orange"] and "max" in seuils_indice["orange"]:
|
|
if valeur >= seuils_indice["orange"]["min"] and valeur < seuils_indice["orange"]["max"]:
|
|
return "orange"
|
|
|
|
# Seuil vert (valeurs inférieures au seuil orange)
|
|
if "vert" in seuils_indice and "max" in seuils_indice["vert"]:
|
|
if valeur < seuils_indice["vert"]["max"]:
|
|
return "darkgreen"
|
|
|
|
# Par défaut orange si on ne trouve pas de correspondance exacte
|
|
return "orange"
|
|
|
|
|
|
def couleur_noeud(n: str, niveaux: dict, G: nx.DiGraph) -> str:
|
|
"""
|
|
Détermine la couleur d'un nœud en fonction de son niveau et de ses attributs.
|
|
Utilise les seuils définis dans le fichier de configuration.
|
|
|
|
Args:
|
|
n (str): Nom du nœud.
|
|
niveaux (dict): Dictionnaire associant chaque nœud à son niveau.
|
|
G (nx.DiGraph): Graphe NetworkX contenant les données.
|
|
|
|
Returns:
|
|
str: Couleur du nœud.
|
|
"""
|
|
niveau = niveaux.get(n, 99)
|
|
attrs = G.nodes[n]
|
|
seuils = load_seuils_config()
|
|
|
|
# Niveau 99 : pays géographique avec isg
|
|
if niveau == 99:
|
|
isg = int(attrs.get("isg", -1))
|
|
if isg >= 0 and "ISG" in seuils:
|
|
return determiner_couleur_par_seuil(isg, seuils["ISG"])
|
|
return "gray"
|
|
|
|
# Niveau 11 ou 12 connecté à un pays géographique
|
|
if niveau in (11, 12, 1011, 1012):
|
|
for succ in G.successors(n):
|
|
if niveaux.get(succ) == 99:
|
|
isg = int(G.nodes[succ].get("isg", -1))
|
|
if isg >= 0 and "ISG" in seuils:
|
|
return determiner_couleur_par_seuil(isg, seuils["ISG"])
|
|
return "gray"
|
|
|
|
# Logique existante pour IHH / IVC
|
|
if niveau in (10, 1010) and attrs.get("ihh_pays"):
|
|
ihh = int(attrs["ihh_pays"])
|
|
if "IHH" in seuils:
|
|
return determiner_couleur_par_seuil(ihh, seuils["IHH"])
|
|
# Fallback vers les anciennes valeurs
|
|
return (
|
|
"darkgreen" if ihh <= 15 else
|
|
"orange" if ihh <= 25 else
|
|
"darkred"
|
|
)
|
|
elif niveau == 2 and attrs.get("ivc"):
|
|
ivc = int(attrs["ivc"])
|
|
if "IVC" in seuils:
|
|
return determiner_couleur_par_seuil(ivc, seuils["IVC"])
|
|
# Fallback vers les anciennes valeurs
|
|
return (
|
|
"darkgreen" if ivc <= 15 else
|
|
"orange" if ivc <= 30 else
|
|
"darkred"
|
|
)
|
|
|
|
return "lightblue"
|
|
|
|
def charger_graphe():
|
|
if "G_temp" not in st.session_state:
|
|
try:
|
|
if charger_schema_depuis_gitea(DOT_FILE):
|
|
st.session_state["G_temp"] = read_dot(DOT_FILE)
|
|
st.session_state["G_temp_ivc"] = st.session_state["G_temp"].copy()
|
|
dot_file_path = True
|
|
else:
|
|
dot_file_path = False
|
|
except Exception as e:
|
|
st.error(f"Erreur de lecture du fichier DOT : {e}")
|
|
dot_file_path = False
|
|
else:
|
|
dot_file_path = True
|
|
|
|
if dot_file_path:
|
|
return dot_file_path
|
|
else:
|
|
st.error("Impossible de charger le graphe pour cet onglet.")
|
|
return dot_file_path
|