data_loader.py

"""
Chargement et indexation des données CEREMA pour le serveur MCP.

Optimisations mémoire :
- DV3F : seules les ~60 colonnes utiles sont chargées (sur 619), avec types optimisés
- Cartofriches : chargement GeoPandas standard (112 Mo en RAM)

Les données sont chargées une seule fois au démarrage et indexées pour des requêtes rapides.
Les friches sont pré-agrégées à chaque maille territoriale (commune, EPCI, département,
région, national) au chargement pour des réponses instantanées.
"""

import os
import pandas as pd
import geopandas as gpd
import numpy as np
import logging

logger = logging.getLogger(__name__)

# --- Répertoire des données ---
DATA_DIR = os.path.join(os.path.dirname(__file__), "data")

# --- Colonnes DV3F à charger ---
# Base
BASE_COLS = ["annee", "echelle", "code", "libelle"]

# Transactions et prix globaux
GLOBAL_COLS = [
    "nbtrans_cod1",           # Total mutations
    "nbtrans_cod11",          # Biens bâtis
    "nbtrans_cod111",         # Maisons
    "nbtrans_cod121",         # Appartements
    "nbtrans_cod2",           # Non bâti (terrains)
    "valeurfonc_sum_cod1",    # Valeur foncière totale
    "valeurfonc_sum_cod111",  # Valeur foncière maisons
    "valeurfonc_sum_cod121",  # Valeur foncière appartements
]

# Maisons - statistiques détaillées
MAISON_COLS = [
    "valeurfonc_median_cod111", "valeurfonc_q25_cod111", "valeurfonc_q75_cod111",
    "pxm2_median_cod111", "pxm2_q25_cod111", "pxm2_q75_cod111",
    "sbati_median_cod111", "sbati_sum_cod111",
]

# Appartements - statistiques détaillées
APPART_COLS = [
    "valeurfonc_median_cod121", "valeurfonc_q25_cod121", "valeurfonc_q75_cod121",
    "pxm2_median_cod121", "pxm2_q25_cod121", "pxm2_q75_cod121",
    "sbati_median_cod121", "sbati_sum_cod121",
]

# Maisons par période de construction
MAISON_PERIODE_COLS = []
for p in ["mp1", "mp2", "mp3", "mp4", "mp5", "mpx"]:
    MAISON_PERIODE_COLS.extend([
        f"nbtrans_{p}", f"valeurfonc_median_{p}", f"pxm2_median_{p}", f"sbati_median_{p}"
    ])

# Appartements par période de construction
APPART_PERIODE_COLS = []
for p in ["ap1", "ap2", "ap3", "ap4", "ap5", "apx"]:
    APPART_PERIODE_COLS.extend([
        f"nbtrans_{p}", f"valeurfonc_median_{p}", f"pxm2_median_{p}", f"sbati_median_{p}"
    ])

ALL_DV3F_COLS = BASE_COLS + GLOBAL_COLS + MAISON_COLS + APPART_COLS + MAISON_PERIODE_COLS + APPART_PERIODE_COLS

# --- Nomenclature des périodes de construction ---
PERIODES_CONSTRUCTION = {
    "mp1": "avant 1914", "mp2": "1914–1947", "mp3": "1948–1969",
    "mp4": "1970–1989", "mp5": "1990 et après", "mpx": "période inconnue",
    "ap1": "avant 1914", "ap2": "1914–1947", "ap3": "1948–1969",
    "ap4": "1970–1989", "ap5": "1990 et après", "apx": "période inconnue",
}

# --- Noms des régions ---
REG_NAMES = {
    "01": "Guadeloupe", "02": "Martinique", "03": "Guyane",
    "04": "La Réunion", "06": "Mayotte",
    "11": "Île-de-France", "24": "Centre-Val de Loire",
    "27": "Bourgogne-Franche-Comté", "28": "Normandie",
    "32": "Hauts-de-France", "44": "Grand Est",
    "52": "Pays de la Loire", "53": "Bretagne",
    "75": "Nouvelle-Aquitaine", "76": "Occitanie",
    "84": "Auvergne-Rhône-Alpes",
    "93": "Provence-Alpes-Côte d'Azur", "94": "Corse",
}

# --- Échelles territoriales ---
ECHELLES = ["commune", "epci", "departement", "region", "national"]

# --- Variables globales pour les données chargées ---
_dv3f: pd.DataFrame | None = None
_friches: gpd.GeoDataFrame | None = None
_mapping: pd.DataFrame | None = None
_friches_agg: dict[str, pd.DataFrame] = {}  # Agrégations pré-calculées par échelle


def load_mapping() -> pd.DataFrame:
    """Charge la table de correspondance commune → EPCI → département → région (INSEE)."""
    global _mapping
    if _mapping is not None:
        return _mapping

    filepath = os.path.join(DATA_DIR, "mapping_communes.csv")
    logger.info(f"Chargement du mapping territorial depuis {filepath}...")
    _mapping = pd.read_csv(filepath, dtype=str)
    _mapping["reg_nom"] = _mapping["reg_nom"].fillna("")
    _mapping["epci_nom"] = _mapping["epci_nom"].fillna("")
    logger.info(f"Mapping chargé : {len(_mapping)} communes, "
                f"{_mapping['epci'].nunique()} EPCI, "
                f"{_mapping['dep'].nunique()} départements, "
                f"{_mapping['reg'].nunique()} régions")
    return _mapping


def load_dv3f() -> pd.DataFrame:
    """Charge les données DV3F avec optimisation mémoire."""
    global _dv3f
    if _dv3f is not None:
        return _dv3f

    filepath = os.path.join(DATA_DIR, "prix_volumes.csv")
    logger.info(f"Chargement de DV3F depuis {filepath}...")

    _dv3f = pd.read_csv(
        filepath,
        sep=";",
        usecols=ALL_DV3F_COLS,
        dtype={
            "annee": "int16",
            "echelle": "category",
            "code": "str",
            "libelle": "str",
        },
        low_memory=False,
    )

    # Convertir les colonnes numériques en float32 pour économiser la mémoire
    numeric_cols = [c for c in _dv3f.columns if c not in BASE_COLS]
    for col in numeric_cols:
        _dv3f[col] = pd.to_numeric(_dv3f[col], errors="coerce").astype("float32")

    # Créer un index pour des lookups rapides
    _dv3f.set_index(["echelle", "code", "annee"], inplace=True)
    _dv3f.sort_index(inplace=True)

    mem_mb = _dv3f.memory_usage(deep=True).sum() / 1e6
    logger.info(f"DV3F chargé : {len(_dv3f)} lignes, {mem_mb:.0f} Mo en mémoire")
    return _dv3f


def load_friches() -> gpd.GeoDataFrame:
    """Charge les données Cartofriches et enrichit avec le mapping territorial."""
    global _friches
    if _friches is not None:
        return _friches

    filepath = os.path.join(DATA_DIR, "cartofriches.geojson")
    logger.info(f"Chargement de Cartofriches depuis {filepath}...")

    _friches = gpd.read_file(filepath)

    # Nettoyer les colonnes utiles
    _friches["site_surface_num"] = pd.to_numeric(
        _friches["site_surface"], errors="coerce"
    )
    _friches["site_surface_ha"] = _friches["site_surface_num"] / 10000

    # Enrichir avec le mapping territorial (EPCI, région)
    mapping = load_mapping()
    _friches = _friches.merge(
        mapping[["comm_insee", "epci", "epci_nom", "reg", "reg_nom"]],
        on="comm_insee",
        how="left",
        suffixes=("", "_map"),
    )

    mem_mb = _friches.memory_usage(deep=True).sum() / 1e6
    logger.info(f"Cartofriches chargé et enrichi : {len(_friches)} friches, {mem_mb:.0f} Mo en mémoire")
    return _friches


def _aggregate_friches_for_group(group: pd.DataFrame) -> dict:
    """Calcule les statistiques agrégées pour un groupe de friches."""
    surfaces = group["site_surface_num"].dropna()
    statuts = group["site_statut"].value_counts()
    types = group["site_type"].value_counts()

    # Pollution avérée ou supposée
    pollution = group["sol_pollution_existe"].fillna("inconnu")
    nb_polluees = pollution.str.contains("avérée|supposée", case=False, na=False).sum()

    # En zone U
    nb_zone_u = (group["urba_zone_type"] == "U").sum()

    # Mobilisables (sans projet + potentielles)
    mobilisables = group[group["site_statut"].isin(["friche sans projet", "friche potentielle"])]
    surface_mobilisable = mobilisables["site_surface_num"].sum() / 10000

    return {
        "nb_friches": len(group),
        "surface_totale_ha": surfaces.sum() / 10000,
        "surface_mediane_ha": surfaces.median() / 10000 if len(surfaces) > 0 else 0,
        "surface_moyenne_ha": surfaces.mean() / 10000 if len(surfaces) > 0 else 0,
        "surface_min_ha": surfaces.min() / 10000 if len(surfaces) > 0 else 0,
        "surface_max_ha": surfaces.max() / 10000 if len(surfaces) > 0 else 0,
        "nb_sans_projet": int(statuts.get("friche sans projet", 0)),
        "nb_avec_projet": int(statuts.get("friche avec projet", 0)),
        "nb_potentielle": int(statuts.get("friche potentielle", 0)),
        "nb_reconvertie": int(statuts.get("friche reconvertie", 0)),
        "nb_mobilisables": len(mobilisables),
        "surface_mobilisable_ha": surface_mobilisable,
        "nb_polluees": int(nb_polluees),
        "nb_zone_u": int(nb_zone_u),
        "pct_zone_u": round(100 * nb_zone_u / len(group), 1) if len(group) > 0 else 0,
        "top_types": types.head(5).to_dict(),
        "nb_communes": group["comm_insee"].nunique() if "comm_insee" in group.columns else 0,
    }


def precompute_friches_aggregations():
    """Pré-calcule les agrégations de friches à chaque maille territoriale."""
    global _friches_agg
    gdf = load_friches()

    logger.info("Pré-agrégation des friches par commune...")
    commune_groups = gdf.groupby("comm_insee")
    agg_commune = {}
    for code, group in commune_groups:
        stats = _aggregate_friches_for_group(group)
        stats["libelle"] = group.iloc[0]["comm_nom"] if len(group) > 0 else ""
        agg_commune[code] = stats
    _friches_agg["commune"] = pd.DataFrame(agg_commune).T
    logger.info(f"  Communes : {len(agg_commune)} territoires")

    logger.info("Pré-agrégation des friches par EPCI...")
    epci_groups = gdf[gdf["epci"].notna() & (gdf["epci"] != "ZZZZZZZZZ")].groupby("epci")
    agg_epci = {}
    for code, group in epci_groups:
        stats = _aggregate_friches_for_group(group)
        stats["libelle"] = group.iloc[0].get("epci_nom", "") or ""
        agg_epci[code] = stats
    _friches_agg["epci"] = pd.DataFrame(agg_epci).T
    logger.info(f"  EPCI : {len(agg_epci)} territoires")

    logger.info("Pré-agrégation des friches par département...")
    dep_groups = gdf[gdf["dep"].notna()].groupby("dep")
    agg_dep = {}
    for code, group in dep_groups:
        stats = _aggregate_friches_for_group(group)
        # Trouver le nom du département dans DV3F
        stats["libelle"] = ""
        agg_dep[code] = stats
    _friches_agg["departement"] = pd.DataFrame(agg_dep).T
    logger.info(f"  Départements : {len(agg_dep)} territoires")

    logger.info("Pré-agrégation des friches par région...")
    reg_groups = gdf[gdf["reg"].notna()].groupby("reg")
    agg_reg = {}
    for code, group in reg_groups:
        stats = _aggregate_friches_for_group(group)
        stats["libelle"] = REG_NAMES.get(code, "")
        agg_reg[code] = stats
    _friches_agg["region"] = pd.DataFrame(agg_reg).T
    logger.info(f"  Régions : {len(agg_reg)} territoires")

    logger.info("Pré-agrégation des friches au niveau national...")
    national_stats = _aggregate_friches_for_group(gdf)
    national_stats["libelle"] = "France entière"
    _friches_agg["national"] = pd.DataFrame({"france": national_stats}).T
    logger.info(f"  National : 1 territoire")

    logger.info("Pré-agrégation terminée.")


def get_friches_agg(echelle: str, code: str = "") -> dict | None:
    """Récupère les statistiques agrégées de friches pour un territoire donné.

    Args:
        echelle: "commune", "epci", "departement", "region" ou "national"
        code: Code du territoire (INSEE, SIREN EPCI, etc.). Ignoré pour "national".

    Returns:
        Dict avec les statistiques agrégées, ou None si non trouvé.
    """
    if echelle not in _friches_agg:
        return None
    if echelle == "national":
        return _friches_agg["national"].iloc[0].to_dict()
    if code in _friches_agg[echelle].index:
        return _friches_agg[echelle].loc[code].to_dict()
    return None


def get_epci_for_commune(code_commune: str) -> tuple[str, str] | None:
    """Retourne (code_epci, nom_epci) pour une commune donnée."""
    mapping = load_mapping()
    row = mapping[mapping["comm_insee"] == code_commune]
    if len(row) == 0:
        return None
    epci_code = row.iloc[0]["epci"]
    epci_nom = row.iloc[0]["epci_nom"]
    if epci_code == "ZZZZZZZZZ" or pd.isna(epci_code):
        return None
    return (epci_code, epci_nom if pd.notna(epci_nom) else "")


def get_region_for_commune(code_commune: str) -> tuple[str, str] | None:
    """Retourne (code_region, nom_region) pour une commune donnée."""
    mapping = load_mapping()
    row = mapping[mapping["comm_insee"] == code_commune]
    if len(row) == 0:
        return None
    reg = row.iloc[0]["reg"]
    reg_nom = row.iloc[0]["reg_nom"]
    return (reg, reg_nom if pd.notna(reg_nom) else REG_NAMES.get(reg, ""))


def get_region_for_departement(code_dep: str) -> tuple[str, str] | None:
    """Retourne (code_region, nom_region) pour un département donné."""
    mapping = load_mapping()
    dep_rows = mapping[mapping["dep"] == code_dep]
    if len(dep_rows) == 0:
        return None
    reg = dep_rows.iloc[0]["reg"]
    reg_nom = dep_rows.iloc[0]["reg_nom"]
    return (reg, reg_nom if pd.notna(reg_nom) else REG_NAMES.get(reg, ""))


def get_departement_from_commune(code_commune: str) -> str:
    """Extrait le code département à partir d'un code INSEE de commune."""
    if code_commune.startswith("97") or code_commune.startswith("98"):
        return code_commune[:3]  # DOM-TOM
    return code_commune[:2]


def get_all_echelles_for_commune(code_commune: str) -> dict:
    """Retourne tous les codes territoriaux pour une commune.

    Returns:
        Dict avec commune, epci, departement, region et leurs libellés.
    """
    result = {"commune": code_commune, "commune_nom": ""}
    mapping = load_mapping()
    row = mapping[mapping["comm_insee"] == code_commune]
    if len(row) > 0:
        r = row.iloc[0]
        result["commune_nom"] = r.get("comm_nom", "")
        result["epci"] = r.get("epci", "")
        result["epci_nom"] = r.get("epci_nom", "")
        result["departement"] = r.get("dep", "")
        result["region"] = r.get("reg", "")
        result["region_nom"] = r.get("reg_nom", REG_NAMES.get(r.get("reg", ""), ""))
    else:
        dep = get_departement_from_commune(code_commune)
        result["departement"] = dep
        result["epci"] = ""
        result["epci_nom"] = ""
        reg_info = get_region_for_departement(dep)
        if reg_info:
            result["region"] = reg_info[0]
            result["region_nom"] = reg_info[1]
        else:
            result["region"] = ""
            result["region_nom"] = ""
    return result


def init_all():
    """Charge toutes les données au démarrage et pré-calcule les agrégations."""
    logger.info("Initialisation des données...")
    load_mapping()
    load_dv3f()
    load_friches()
    precompute_friches_aggregations()
    logger.info("Toutes les données sont chargées et pré-agrégées.")