Ajoute l’analyse des catégories de pièces

lib/plots/part_categories.py

@@ -0,0 +1,130 @@
+"""Visualisations des parts par catégorie."""
+
+from pathlib import Path
+from typing import Dict, List, Sequence
+
+import matplotlib.pyplot as plt
+import numpy as np
+from matplotlib.colors import Normalize
+from matplotlib.cm import ScalarMappable
+from matplotlib.patches import Patch
+
+from lib.filesystem import ensure_parent_dir
+from lib.rebrickable.stats import read_rows
+
+
+def load_rows(path: Path) -> List[dict]:
+    """Charge un CSV en mémoire."""
+    return read_rows(path)
+
+
+def extract_years(rows: Sequence[dict]) -> List[str]:
+    """Récupère la liste ordonnée des années présentes."""
+    years = {row["year"] for row in rows}
+    return sorted(years, key=int)
+
+
+def build_shares_by_year(rows: Sequence[dict]) -> Dict[tuple[str, str], float]:
+    """Indexe share_non_spare par (year, category_id)."""
+    return {(row["year"], row["category_id"]): float(row["share_non_spare"]) for row in rows}
+
+
+def plot_top_part_categories_area(
+    categories_by_year_path: Path,
+    categories_global_path: Path,
+    destination_path: Path,
+    top_n: int = 8,
+) -> None:
+    """Trace l'évolution des catégories principales en parts empilées."""
+    yearly_rows = load_rows(categories_by_year_path)
+    global_rows = load_rows(categories_global_path)
+    years = extract_years(yearly_rows)
+    top_categories = global_rows[:top_n]
+    labels = [row["category_name"] for row in top_categories] + ["Autres"]
+    shares_lookup = build_shares_by_year(yearly_rows)
+    series: List[List[float]] = []
+    for top in top_categories:
+        series.append([shares_lookup.get((year, top["category_id"]), 0.0) for year in years])
+    other_series: List[float] = []
+    for year in years:
+        year_total = sum(value for (yr, _), value in shares_lookup.items() if yr == year)
+        top_sum = sum(values[years.index(year)] for values in series)
+        other_series.append(max(0.0, year_total - top_sum))
+    series.append(other_series)
+    x = np.arange(len(years))
+    fig, ax = plt.subplots(figsize=(12, 7))
+    colors = plt.get_cmap("tab20").colors
+    ax.stackplot(x, series, labels=labels, colors=colors[: len(labels)], alpha=0.9, linewidth=0.6)
+    ax.set_xticks(x)
+    ax.set_xticklabels(years, rotation=45, ha="right")
+    ax.set_ylabel("Part des pièces (hors rechanges)")
+    ax.set_title("Part des principales catégories de pièces (par année)")
+    ax.legend(loc="upper left", frameon=False, ncol=2)
+    ax.grid(axis="y", linestyle="--", alpha=0.35)
+
+    ensure_parent_dir(destination_path)
+    fig.tight_layout()
+    fig.savefig(destination_path, dpi=170)
+    plt.close(fig)
+
+
+def plot_part_categories_heatmap(categories_by_year_path: Path, destination_path: Path) -> None:
+    """Heatmap des parts par catégorie et par année."""
+    rows = load_rows(categories_by_year_path)
+    years = extract_years(rows)
+    totals: Dict[str, int] = {}
+    for row in rows:
+        totals[row["category_id"]] = totals.get(row["category_id"], 0) + int(row["quantity_non_spare"])
+    categories = sorted(totals.keys(), key=lambda cat_id: -totals[cat_id])
+    matrix = np.zeros((len(categories), len(years)))
+    lookup = {(row["year"], row["category_id"]): float(row["share_non_spare"]) for row in rows}
+    for i, cat_id in enumerate(categories):
+        for j, year in enumerate(years):
+            matrix[i, j] = lookup.get((year, cat_id), 0.0)
+    fig, ax = plt.subplots(figsize=(12, 10))
+    cmap = plt.get_cmap("viridis")
+    im = ax.imshow(matrix, aspect="auto", cmap=cmap, norm=Normalize(vmin=0, vmax=matrix.max()))
+    ax.set_xticks(np.arange(len(years)))
+    ax.set_xticklabels(years, rotation=45, ha="right")
+    labels = {row["category_id"]: row["category_name"] for row in rows}
+    ax.set_yticks(np.arange(len(categories)))
+    ax.set_yticklabels([labels[cat_id] for cat_id in categories])
+    ax.set_xlabel("Année")
+    ax.set_ylabel("Catégorie de pièce")
+    ax.set_title("Part des catégories de pièces par année")
+    cbar = fig.colorbar(ScalarMappable(norm=im.norm, cmap=cmap), ax=ax, fraction=0.025, pad=0.015)
+    cbar.ax.set_ylabel("Part des pièces", rotation=90)
+
+    ensure_parent_dir(destination_path)
+    fig.tight_layout()
+    fig.savefig(destination_path, dpi=170)
+    plt.close(fig)
+
+
+def plot_structural_share_timeline(categories_by_year_path: Path, destination_path: Path) -> None:
+    """Trace l'évolution de la part des catégories structurelles."""
+    rows = load_rows(categories_by_year_path)
+    years = extract_years(rows)
+    structural_share: Dict[str, float] = {}
+    for row in rows:
+        if row["is_structural"] != "true":
+            continue
+        share = structural_share.get(row["year"], 0.0)
+        structural_share[row["year"]] = share + float(row["share_non_spare"])
+    x = np.arange(len(years))
+    y = [structural_share.get(year, 0.0) for year in years]
+    fig, ax = plt.subplots(figsize=(11, 6))
+    ax.plot(x, y, color="#d62728", linewidth=2.2)
+    ax.fill_between(x, y, color="#d62728", alpha=0.18)
+    ax.set_xticks(x)
+    ax.set_xticklabels(years, rotation=45, ha="right")
+    ax.set_ylabel("Part des pièces structurelles")
+    ax.set_title("Evolution de la part des pièces structurelles")
+    ax.grid(True, linestyle="--", alpha=0.35)
+    legend = [Patch(facecolor="#d62728", edgecolor="none", alpha=0.6, label="Structurel / Technic")]
+    ax.legend(handles=legend, loc="upper right", frameon=False)
+
+    ensure_parent_dir(destination_path)
+    fig.tight_layout()
+    fig.savefig(destination_path, dpi=170)
+    plt.close(fig)