Ajoute le total de minifigs aux statistiques

README.md

@@ -208,3 +208,11 @@ Le script lit `data/intermediate/minifig_heads_by_year.csv` et produit `figures/
 3. `python -m scripts.plot_global_minifig_skin_tones`
 
 Ces scripts lisent les CSV bruts du catalogue complet (`data/raw/inventories.csv`, `inventory_parts.csv`, `parts.csv`, `colors.csv`, `sets.csv`), extraient les têtes de minifigs via `part_cat_id=59`, agrègent les couleurs par année dans `data/intermediate/global_minifig_heads_by_year.csv`, puis tracent `figures/step17/global_minifig_heads_yellow_share.png` montrant la part annuelle de la couleur Yellow comparée au reste, jalons inclus.
+
+### Étape 19 : total de minifigs des sets filtrés
+
+1. `source .venv/bin/activate`
+2. `python -m scripts.compute_minifig_stats`
+
+Le script relit les sets (`data/raw/themes.csv`, `data/raw/sets.csv`, `data/intermediate/sets_filtered.csv`, `data/intermediate/sets_enriched.csv`) ainsi que les inventaires (`data/raw/inventories.csv`, `data/raw/inventory_minifigs.csv`), recalcule toutes les statistiques de base puis régénère `data/final/stats.csv` en y ajoutant le libellé « Nombre total de minifigs (thèmes filtrés) ».
+Cette étape se lance après le téléchargement des données d'inventaire (étape 8) et doit être rejouée si les sets filtrés ou les inventaires sont mis à jour.

lib/rebrickable/minifig_stats.py

@@ -0,0 +1,55 @@
+"""Statistiques liées aux minifigs pour les sets filtrés."""
+
+from pathlib import Path
+from typing import Dict, Iterable, List, Sequence, Tuple
+
+from lib.rebrickable.parts_inventory import index_inventory_minifigs_by_inventory, select_latest_inventories
+from lib.rebrickable.stats import read_rows
+
+
+MINIFIG_TOTAL_LABEL = "Nombre total de minifigs (thèmes filtrés)"
+TOTAL_PARTS_LABEL = "Total de pièces pour les thèmes filtrés"
+
+
+def load_filtered_sets(path: Path) -> List[dict]:
+    """Charge les sets filtrés depuis un CSV."""
+    return read_rows(path)
+
+
+def compute_total_minifigs(
+    filtered_sets: Iterable[dict],
+    inventories: Dict[str, dict],
+    minifigs_by_inventory: Dict[str, List[dict]],
+) -> int:
+    """Additionne les minifigs présentes dans les inventaires des sets filtrés."""
+    total = 0
+    for set_row in filtered_sets:
+        inventory = inventories[set_row["set_num"]]
+        for minifig_row in minifigs_by_inventory.get(inventory["id"], []):
+            total += int(minifig_row["quantity"])
+    return total
+
+
+def compute_filtered_minifig_total(
+    filtered_sets: Iterable[dict],
+    inventories_path: Path,
+    inventory_minifigs_path: Path,
+) -> int:
+    """Calcule le total de minifigs pour les sets filtrés à partir des CSV bruts."""
+    inventories = select_latest_inventories(inventories_path)
+    minifigs_by_inventory = index_inventory_minifigs_by_inventory(inventory_minifigs_path)
+    return compute_total_minifigs(filtered_sets, inventories, minifigs_by_inventory)
+
+
+def merge_minifig_stat(stats: Sequence[Tuple[str, str]], total_minifigs: int) -> List[Tuple[str, str]]:
+    """Insère le total de minifigs en évitant les doublons et en préservant l'ordre."""
+    filtered_stats = [(label, value) for label, value in stats if label != MINIFIG_TOTAL_LABEL]
+    insertion_index = next(
+        (index for index, (label, _) in enumerate(filtered_stats) if label == TOTAL_PARTS_LABEL),
+        len(filtered_stats),
+    )
+    return (
+        filtered_stats[: insertion_index + 1]
+        + [(MINIFIG_TOTAL_LABEL, str(total_minifigs))]
+        + filtered_stats[insertion_index + 1 :]
+    )

scripts/compute_minifig_stats.py

@@ -0,0 +1,31 @@
+"""Ajoute le total de minifigs aux statistiques principales."""
+
+from pathlib import Path
+
+from lib.rebrickable.minifig_stats import compute_filtered_minifig_total, merge_minifig_stat
+from lib.rebrickable.stats import compute_basic_stats, read_rows, write_stats_csv
+
+
+THEMES_PATH = Path("data/raw/themes.csv")
+ALL_SETS_PATH = Path("data/raw/sets.csv")
+FILTERED_SETS_PATH = Path("data/intermediate/sets_filtered.csv")
+ENRICHED_SETS_PATH = Path("data/intermediate/sets_enriched.csv")
+INVENTORIES_PATH = Path("data/raw/inventories.csv")
+INVENTORY_MINIFIGS_PATH = Path("data/raw/inventory_minifigs.csv")
+DESTINATION_PATH = Path("data/final/stats.csv")
+
+
+def main() -> None:
+    """Recalcule les statistiques de base et ajoute le total de minifigs."""
+    themes = read_rows(THEMES_PATH)
+    all_sets = read_rows(ALL_SETS_PATH)
+    filtered_sets = read_rows(FILTERED_SETS_PATH)
+    enriched_sets = read_rows(ENRICHED_SETS_PATH)
+    base_stats = compute_basic_stats(themes, all_sets, filtered_sets, enriched_sets)
+    minifig_total = compute_filtered_minifig_total(filtered_sets, INVENTORIES_PATH, INVENTORY_MINIFIGS_PATH)
+    stats = merge_minifig_stat(base_stats, minifig_total)
+    write_stats_csv(DESTINATION_PATH, stats)
+
+
+if __name__ == "__main__":
+    main()

tests/test_minifig_stats.py

@@ -0,0 +1,56 @@
+"""Tests des statistiques liées aux minifigs."""
+
+from lib.rebrickable.minifig_stats import (
+    MINIFIG_TOTAL_LABEL,
+    TOTAL_PARTS_LABEL,
+    compute_filtered_minifig_total,
+    merge_minifig_stat,
+)
+
+
+def test_compute_filtered_minifig_total_counts_latest_inventory(tmp_path) -> None:
+    """Additionne les minifigs en utilisant la dernière version d'inventaire."""
+    inventories_path = tmp_path / "inventories.csv"
+    inventories_path.write_text(
+        "id,version,set_num\n"
+        "1,1,123-1\n"
+        "2,2,123-1\n"
+        "3,1,124-1\n"
+    )
+    inventory_minifigs_path = tmp_path / "inventory_minifigs.csv"
+    inventory_minifigs_path.write_text(
+        "inventory_id,fig_num,quantity\n"
+        "1,fig-01,1\n"
+        "2,fig-02,3\n"
+        "3,fig-03,2\n"
+    )
+    filtered_sets = [{"set_num": "123-1"}, {"set_num": "124-1"}]
+
+    total = compute_filtered_minifig_total(filtered_sets, inventories_path, inventory_minifigs_path)
+
+    assert total == 5
+
+
+def test_merge_minifig_stat_inserts_after_total_parts_and_replaces_existing() -> None:
+    """Insère l'entrée minifigs après le total de pièces et remplace l'ancienne valeur."""
+    base_stats = [
+        ("A", "1"),
+        (TOTAL_PARTS_LABEL, "10"),
+        ("B", "2"),
+    ]
+
+    with_minifigs = merge_minifig_stat(base_stats, 7)
+    refreshed = merge_minifig_stat(with_minifigs, 8)
+
+    assert with_minifigs == [
+        ("A", "1"),
+        (TOTAL_PARTS_LABEL, "10"),
+        (MINIFIG_TOTAL_LABEL, "7"),
+        ("B", "2"),
+    ]
+    assert refreshed == [
+        ("A", "1"),
+        (TOTAL_PARTS_LABEL, "10"),
+        (MINIFIG_TOTAL_LABEL, "8"),
+        ("B", "2"),
+    ]