Code/docs/MODULES.md

Guide des modules FabNum

Documentation rapide des modules principaux du projet.

Table des matières

• Modules applicatifs
• Utilitaires
• Indices et métriques

---

Modules applicatifs

app/analyse - Analyse des chaînes

Fonction principale: Visualiser les chaînes d'approvisionnement et identifier les vulnérabilités.

Utilisation:

from app.analyse.interface import selectionner_minerais
from app.analyse.sankey import generer_sankey

# Sélectionner un niveau de départ et d'arrivée
minerais = selectionner_minerais(G, niveau_depart=0, niveau_arrivee=2)

# Générer le diagramme de Sankey
generer_sankey(G, chemins)

Données retournées:

• Chemins d'approvisionnement (listes de nœuds)
• Indices IHH, ICS, IVC par étape
• Visualisation Sankey interactive

---

app/fiches - Génération de fiches techniques

Fonction principale: Générer des fiches markdown pour chaque élément du graphe.

Utilisation:

from app.fiches.generer import generer_fiches_depuis_graph

# Générer toutes les fiches
generer_fiches_depuis_graph(G, output_dir="Documents")

Structure d'une fiche:

1. En-tête (titre, description)
2. Indicateurs de vulnérabilité (IHH, ISG)
3. Données de criticité (ICS, IVC pour minerais)
4. Chemins critiques
5. Recommandations

Intégration tickets:

from app.fiches.utils.tickets.core import creer_ticket_gitea, rechercher_tickets_gitea

# Rechercher les tickets existants pour une fiche
tickets = rechercher_tickets_gitea("Lithium")

# Créer un nouveau ticket
creer_ticket_gitea(
    titre="Vulnérabilité détectée: Lithium",
    corps="## Description\n...",
    labels=[1, 2]  # IDs des labels
)

---

app/plan_d_action - Analyse de criticité

Fonction principale: Calculer la criticité des chaînes et générer un plan d'action.

Utilisation:

from app.plan_d_action.utils.data.plan_d_action import (
    calcul_poids_chaine,
    analyser_chaines,
    tableau_de_bord
)

# Calculer la criticité d'une chaîne
criticite, niveaux, poids = calcul_poids_chaine(
    poids_A=2,  # Assemblage
    poids_F=3,  # Fabrication
    poids_T=2,  # Traitement
    poids_E=3,  # Extraction
    poids_M=1   # Substitution
)
# Résultat: ("Critique", {...}, 11)

# Analyser toutes les chaînes
chains = analyser_chaines(G, produit="Smartphone", mineraux_selectionnes=["Lithium", "Cobalt"])

# Afficher le tableau de bord interactif
tableau_de_bord(chains, produits, composants, mineraux, seuils)

Niveaux de criticité:

• Très critique: poids ≥ 12
• Critique: 9 ≤ poids < 12
• Moyenne: 6 ≤ poids < 9
• Faible: poids < 6

---

app/personnalisation - Personnalisation du graphe

Fonction principale: Ajouter/modifier des produits personnalisés dans le graphe.

Utilisation:

from app.personnalisation.utils.ajout import ajouter_produit
from app.personnalisation.utils.import_export import importer_exporter_graph

# Ajouter un produit personnalisé
G_modifie = ajouter_produit(G)

# Exporter/importer la configuration
G_export = importer_exporter_graph(G)

---

Utilitaires

utils/gitea.py - Intégration Gitea

API Gitea avec cache local:

from utils.gitea import (
    charger_instructions_depuis_gitea,
    charger_schema_depuis_gitea,
    charger_arborescence_fiches
)

# Charger un fichier depuis Gitea (avec cache)
contenu = charger_instructions_depuis_gitea("Instructions.md")

# Charger le graphe DOT
charger_schema_depuis_gitea("schema_temp.txt")

# Lister toutes les fiches
arbo = charger_arborescence_fiches()
# Retourne: {"Composants": [{"nom": "Processeur.md", "download_url": "..."}, ...]}

Fonctionnalités:

• ✓ Cache local avec vérification de timestamp
• ✓ Authentification automatique
• ✓ Fallback sur cache en cas d'erreur réseau
• ✓ 100% couverture de tests

---

utils/graph_utils.py - Manipulation de graphes

Fonctions principales:

from utils.graph_utils import (
    charger_graphe,
    extraire_chemins_depuis,
    extraire_chemins_vers,
    recuperer_donnees,
    recuperer_donnees_2
)

# Charger le graphe depuis un fichier DOT
G = charger_graphe("schema_temp.txt")

# Extraire tous les chemins depuis un nœud
chemins = extraire_chemins_depuis(G, "Smartphone")
# Retourne: [["Smartphone", "Processeur", "Lithium"], ...]

# Extraire les chemins vers un nœud depuis un niveau
chemins_vers = extraire_chemins_vers(G, "Lithium", niveau_demande=0)

# Récupérer les données IHH/ICS pour des opérations-minerais
df = recuperer_donnees(G, ["Fabrication_Processeur", "Traitement_Lithium"])
# Colonnes: categorie, nom, ihh_pays, ihh_acteurs, ics_minerai, ics_cat

# Récupérer les données IVC/IHH pour minerais
donnees_minerais = recuperer_donnees_2(G, ["Lithium", "Cobalt"])
# Retourne: [{"nom": "Lithium", "ivc": 60, "ihh_extraction": 70, ...}, ...]

Structure du graphe NetworkX:

# Niveaux hiérarchiques
NIVEAU_PRODUIT = 0
NIVEAU_COMPOSANT = 1
NIVEAU_MINERAI = 2
NIVEAU_OPERATION = 10      # Assemblage, Fabrication, Traitement, Extraction, Réserves
NIVEAU_PAYS_OPERATION = 11
NIVEAU_PAYS_GEO = 99

# Attributs des nœuds
G.nodes["Lithium"]["niveau"] = 2
G.nodes["Lithium"]["ivc"] = 60
G.nodes["Extraction_Lithium"]["ihh_pays"] = 70
G.nodes["Extraction_Lithium"]["ihh_acteurs"] = 65

# Attributs des arêtes
G["Processeur"]["Lithium"]["ics"] = 0.8

---

utils/logger.py - Système de logging

from utils.logger import setup_logger, get_logger

# Configuration initiale
logger = setup_logger("mon_module", level="DEBUG", log_to_file=True)

# Utilisation
logger.info("Chargement du graphe...")
logger.warning("Minerai non trouvé: Uranium")
logger.error("Erreur API Gitea", exc_info=True)

# Récupération d'un logger existant
logger = get_logger("mon_module")

Fonctionnalités:

• ✓ Handler console + fichier optionnel
• ✓ Pas de duplication des handlers
• ✓ Format standardisé avec timestamp
• ✓ 94% couverture de tests

---

utils/persistance.py - Gestion de session Streamlit

from utils.persistance import (
    get_session_id,
    update_session_paths,
    get_champ_statut,
    maj_champ_statut,
    get_full_structure
)

# Récupérer l'ID de session
session_id = get_session_id()

# Stocker des chemins en session
update_session_paths(session_id, chemins)

# Lire/écrire dans st.session_state
valeur = get_champ_statut("graphe_charge")
maj_champ_statut("graphe_charge", True)

# Récupérer toute la structure en session
structure = get_full_structure()

---

utils/widgets.py - Widgets HTML personnalisés

from utils.widgets import html_expander

# Créer un expander HTML personnalisé (avec rendu markdown)
html_expander(
    titre="Détails de vulnérabilité",
    contenu="## IHH Pays\nConcentration élevée: 70%\n\n...",
    open_by_default=False,
    details_class="custom-details",
    summary_class="custom-summary"
)

Avantages vs. st.expander():

• Rendu markdown riche
• Classes CSS personnalisables
• IDs uniques générés automatiquement

---

utils/visualisation.py - Visualisations Altair

from utils.visualisation import (
    afficher_graphique_altair,
    creer_graphes,
    lancer_visualisation_ihh_ics,
    lancer_visualisation_ihh_ivc
)

# Créer un graphique de concentration IHH
df = pd.DataFrame({
    "categorie": ["Fabrication", "Traitement"],
    "nom": ["Processeur", "Lithium"],
    "ihh_pays": [30, 70],
    "ihh_acteurs": [25, 65],
    "ics_cat": [0.5, 0.8]
})
afficher_graphique_altair(df)

# Lancer la visualisation complète IHH-ICS
lancer_visualisation_ihh_ics(G)

# Lancer la visualisation IHH-IVC (minerais)
lancer_visualisation_ihh_ivc(G)

---

Indices et métriques

IHH - Indice de Herfindahl-Hirschman

Définition: Mesure la concentration géographique ou par acteurs.

Formule: IHH = Σ (part_marché_i)²

Seuils:

• Vert: < 15 (faible concentration)
• Orange: 15-25 (concentration modérée)
• Rouge: > 25 (forte concentration)

Utilisation dans le code:

ihh_pays = G.nodes["Extraction_Lithium"]["ihh_pays"]  # ex: 70
ihh_acteurs = G.nodes["Extraction_Lithium"]["ihh_acteurs"]  # ex: 65

---

ISG - Indice de Stabilité Géopolitique

Définition: Mesure l'instabilité politique/économique d'un pays.

Seuils:

• Vert: < 40 (stable)
• Orange: 40-70 (risques modérés)
• Rouge: > 70 (instable)

Utilisation:

isg = G.nodes["Chine_geographique"]["isg"]  # ex: 54

---

ICS - Indice de Criticité Supply-Side

Définition: Risque lié à l'approvisionnement d'un minerai pour un composant.

Formule: Basé sur la concentration de l'offre et les contraintes géopolitiques.

Seuils:

• Vert: < 15
• Orange: 15-60
• Rouge: > 60

Utilisation:

# ICS stocké sur les arêtes composant → minerai
ics = G["Processeur"]["Lithium"]["ics"]  # ex: 0.8 (80%)

# Calcul ICS moyen pour un minerai
ics_moyen = df[df["nom"] == "Lithium"]["ics_minerai"].mean()

---

IVC - Indice de Vulnérabilité (Demand-Side)

Définition: Risque lié à la demande et à la substituabilité d'un minerai.

Seuils:

• Vert: < 15 (faible vulnérabilité)
• Orange: 15-60 (vulnérabilité modérée)
• Rouge: > 60 (haute vulnérabilité)

Utilisation:

ivc = G.nodes["Lithium"]["ivc"]  # ex: 60

# Récupérer IVC avec données d'extraction/réserves
donnees = recuperer_donnees_2(G, ["Lithium"])
# Retourne: [{"nom": "Lithium", "ivc": 60, "ihh_extraction": 70, "ihh_reserves": 80}]

---

Exemples d'utilisation courants

1. Analyser un produit complet

from utils.graph_utils import charger_graphe, extraire_chemins_depuis
from app.plan_d_action.utils.data.plan_d_action import analyser_chaines

# Charger le graphe
G = charger_graphe("schema_temp.txt")

# Extraire tous les chemins depuis le produit
chemins = extraire_chemins_depuis(G, "Smartphone")

# Analyser les chaînes
chains = analyser_chaines(G, produit="Smartphone", mineraux_selectionnes=["Lithium", "Cobalt", "Tantale"])

# Afficher les chaînes critiques
chains_critiques = [c for c in chains if c["criticite"] in ["Très critique", "Critique"]]
for chain in chains_critiques:
    print(f"{chain['minerai']}: {chain['criticite']} (poids: {chain['poids_total']})")

2. Générer et publier des fiches

from app.fiches.generer import generer_fiches_depuis_graph
from utils.gitea import charger_arborescence_fiches

# Générer toutes les fiches
generer_fiches_depuis_graph(G, output_dir="Documents")

# Vérifier les fiches créées
arbo = charger_arborescence_fiches()
print(f"Fiches Composants: {len(arbo.get('Composants', []))}")
print(f"Fiches Minerais: {len(arbo.get('Minerais', []))}")

3. Créer un ticket pour une vulnérabilité

from app.fiches.utils.tickets.core import (
    get_labels_existants,
    creer_ticket_gitea,
    construire_corps_ticket_markdown
)

# Récupérer les labels existants
labels = get_labels_existants()  # {"Extraction": 1, "Minerai": 2, ...}

# Construire le corps du ticket
corps = construire_corps_ticket_markdown({
    "Description": "Concentration élevée pour l'extraction de Lithium",
    "IHH Pays": "70 (seuil critique dépassé)",
    "Recommandation": "Diversifier les sources d'approvisionnement"
})

# Créer le ticket
succes = creer_ticket_gitea(
    titre="[CRITIQUE] Vulnérabilité Lithium - Concentration extraction",
    corps=corps,
    labels=[labels["Extraction"], labels["Minerai"]]
)

4. Visualiser les concentrations

from utils.graph_utils import recuperer_donnees
from utils.visualisation import afficher_graphique_altair

# Récupérer les données pour plusieurs opérations
noeuds = [
    "Fabrication_Processeur",
    "Fabrication_Memoire",
    "Traitement_Lithium",
    "Extraction_Cobalt"
]
df = recuperer_donnees(G, noeuds)

# Afficher le graphique interactif
afficher_graphique_altair(df)

---

Bonnes pratiques

1. Gestion du graphe

# ✓ BON: Charger une seule fois et stocker en session
if "graphe" not in st.session_state:
    st.session_state["graphe"] = charger_graphe()
G = st.session_state["graphe"]

# ✗ MAUVAIS: Recharger à chaque interaction
G = charger_graphe()  # Trop lent

2. Gestion des erreurs Gitea

# ✓ BON: Fallback sur cache local
from utils.gitea import charger_instructions_depuis_gitea

instructions = charger_instructions_depuis_gitea("Instructions.md")
if instructions is None:
    st.warning("Utilisation du cache local (pas de connexion Gitea)")

# ✗ MAUVAIS: Crasher si Gitea indisponible
instructions = charger_instructions_depuis_gitea("Instructions.md")
contenu = instructions.split("\n")[0]  # Crash si None

3. Logging

# ✓ BON: Utiliser le logger du module
from utils.logger import setup_logger
logger = setup_logger(__name__)

logger.info("Traitement démarré")
logger.warning(f"Minerai non trouvé: {minerai}")

# ✗ MAUVAIS: Utiliser print()
print("Traitement démarré")  # Pas de niveau, timestamp, etc.

4. Tests

# ✓ BON: Mocker les dépendances externes
@patch("app.fiches.utils.tickets.core.gitea_request")
def test_rechercher_tickets(mock_gitea):
    mock_gitea.return_value = Mock(json=lambda: [...])
    resultat = rechercher_tickets_gitea("Processeur")
    assert len(resultat) > 0

# ✗ MAUVAIS: Appeler l'API réelle dans les tests
def test_rechercher_tickets():
    resultat = rechercher_tickets_gitea("Processeur")  # Appel API réel

---

Dépannage

Erreur: "Missing ScriptRunContext"

Cause: Streamlit n'est pas initialisé (appel hors contexte Streamlit).

Solution: Utiliser des mocks dans les tests:

@patch("app.fiches.utils.tickets.core.st")
def test_ma_fonction(mock_st):
    # Test code here
    pass

Erreur: "Graphe non chargé"

Cause: Le fichier DOT n'est pas disponible.

Solution:

from utils.gitea import charger_schema_depuis_gitea

# Télécharger le schéma depuis Gitea
charger_schema_depuis_gitea("schema_temp.txt")

# Puis charger le graphe
G = charger_graphe("schema_temp.txt")

Performances lentes

Symptômes: Interface Streamlit lente, rechargement fréquent.

Solutions:

1. Utiliser st.cache_data ou st.cache_resource:

@st.cache_resource
def charger_graphe_cached():
    return charger_graphe("schema_temp.txt")

2. Limiter les rerun Streamlit:

# Utiliser des clés stables pour les widgets
st.selectbox("Produit", options, key="produit_select")

---

Références

• Documentation complète
• Configuration pytest
• Convention Google docstrings
• Streamlit documentation
• NetworkX documentation