# Guide des modules FabNum Documentation rapide des modules principaux du projet. ## Table des matières - [Modules applicatifs](#modules-applicatifs) - [Utilitaires](#utilitaires) - [Indices et métriques](#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:** ```python 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:** ```python 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:** ```python 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:** ```python 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:** ```python 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:** ```python 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:** ```python 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:** ```python # 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 ```python 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 ```python 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 ```python 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 ```python 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:** ```python 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:** ```python 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:** ```python # 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:** ```python 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 ```python 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 ```python 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é ```python 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 ```python 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 ```python # ✓ 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 ```python # ✓ 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 ```python # ✓ 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 ```python # ✓ 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: ```python @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:** ```python 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`: ```python @st.cache_resource def charger_graphe_cached(): return charger_graphe("schema_temp.txt") ``` 2. Limiter les rerun Streamlit: ```python # Utiliser des clés stables pour les widgets st.selectbox("Produit", options, key="produit_select") ``` --- ## Références - [Documentation complète](ARCHITECTURE.md) - [Configuration pytest](../pyproject.toml) - [Convention Google docstrings](https://google.github.io/styleguide/pyguide.html#38-comments-and-docstrings) - [Streamlit documentation](https://docs.streamlit.io/) - [NetworkX documentation](https://networkx.org/documentation/stable/)