Upload folder using huggingface_hub

.gitattributes

@@ -33,3 +33,5 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+data/cartofriches.geojson filter=lfs diff=lfs merge=lfs -text
+data/prix_volumes.csv filter=lfs diff=lfs merge=lfs -text

README.md

@@ -1,13 +1,88 @@
 ---
 title: MCP Data Foncier
-emoji: 🦀
-colorFrom: blue
-colorTo: green
+emoji: 🏗️
+colorFrom: green
+colorTo: blue
 sdk: gradio
-sdk_version: 6.5.1
+sdk_version: 5.29.0
 app_file: app.py
 pinned: false
-short_description: Accès aux données DV3F et Cartofriches
+tags:
+  - mcp-server-track
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# 🏗️ MCP Data Foncier — Données foncières & friches pour la transition écologique
+
+Serveur MCP (Model Context Protocol) exposant les données du CEREMA pour la plateforme **Sofia V2** (ADEME / CEREMA / CGDD).
+
+Accès aux données DV3F (transactions immobilières) et Cartofriches (inventaire national des friches).
+
+## 🛠️ Tools MCP disponibles
+
+| Tool | Description |
+|---|---|
+| `rechercher_friches` | Recherche de friches par commune, département, type, surface, statut |
+| `statistiques_prix_foncier` | Prix et volumes de transactions immobilières par territoire et année |
+| `evolution_prix` | Évolution temporelle des prix de 2010 à 2024 |
+| `statistiques_friches` | Statistiques agrégées multi-échelle (commune → national) |
+| `diagnostic_foncier_territoire` | Diagnostic complet croisant friches + marché foncier (pour le ZAN) |
+
+## 📊 Données exposées
+
+### Cartofriches
+Inventaire national des friches (28 373 sites géolocalisés) :
+- Type, surface, statut, pollution, zonage urbanisme
+
+### DV3F (Demande de Valeurs Foncières)
+Statistiques de transactions immobilières (2010–2024) :
+- Prix médians, prix au m², volumes, par type de bien et période de construction
+- Multi-échelle : commune, EPCI, département, région
+
+## 🔌 Connexion MCP
+
+### Claude Desktop
+
+```json
+{
+  "mcpServers": {
+    "data-foncier": {
+      "url": "https://qdonnars-mcp-data-foncier.hf.space/gradio_api/mcp/sse"
+    }
+  }
+}
+```
+
+### Cursor
+
+```json
+{
+  "data-foncier": {
+    "url": "https://qdonnars-mcp-data-foncier.hf.space/gradio_api/mcp/sse"
+  }
+}
+```
+
+## 🌱 Cas d'usage transition écologique
+
+- **Zéro Artificialisation Nette (ZAN)** : identifier les friches mobilisables
+- **Planification territoriale** : contextualiser le marché foncier local (PLU, SCOT)
+- **Rénovation énergétique** : distinguer le parc ancien du neuf via les segments DV3F
+- **Croisement multi-MCP** : combiner avec indicateurs TE et Base Carbone
+
+## 📁 Structure
+
+```
+├── app.py              # Point d'entrée Gradio + MCP
+├── server.py           # Fonctions-tools MCP
+├── data_loader.py      # Chargement optimisé des données
+├── requirements.txt
+└── data/
+    ├── prix_volumes.csv
+    ├── cartofriches.geojson
+    └── mapping_communes.csv
+```
+
+## Sources
+
+- **Cartofriches** : https://cartofriches.cerema.fr — CEREMA, licence ouverte
+- **DV3F** : https://datafoncier.cerema.fr — CEREMA/DGFiP, données ouvertes DVF

app.py

@@ -0,0 +1,150 @@
+"""
+MCP Server CEREMA — Données foncières et friches pour la transition écologique.
+
+Point d'entrée Gradio exposant les tools MCP pour Sofia V2.
+Les données (DV3F + Cartofriches) sont chargées en mémoire au démarrage.
+Les friches sont pré-agrégées à toutes les mailles (commune, EPCI, département, région, national).
+
+Endpoint MCP : http://localhost:7860/gradio_api/mcp/sse
+Schema : http://localhost:7860/gradio_api/mcp/schema
+"""
+
+import logging
+import gradio as gr
+
+from data_loader import init_all
+from server import (
+    rechercher_friches,
+    statistiques_prix_foncier,
+    evolution_prix,
+    statistiques_friches,
+    diagnostic_foncier_territoire,
+)
+
+# Configuration du logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+# --- Chargement des données au démarrage ---
+logger.info("Démarrage du serveur MCP CEREMA...")
+init_all()
+logger.info("Données chargées. Construction de l'interface Gradio...")
+
+# --- Interfaces Gradio pour chaque tool ---
+
+# Tool 1 : Recherche de friches
+iface_friches = gr.Interface(
+    fn=rechercher_friches,
+    inputs=[
+        gr.Textbox(label="Code INSEE commune", placeholder="ex: 13055, 59350, 75056"),
+        gr.Textbox(label="Code département", placeholder="ex: 13, 59, 75"),
+        gr.Dropdown(
+            label="Type de friche",
+            choices=["", "industrielle", "habitat", "commerciale", "ferroviaire",
+                     "militaire", "hospitalière", "logistique", "agro-industrielle",
+                     "équipement public", "carrière ou mine"],
+            value="",
+        ),
+        gr.Number(label="Surface minimale (m²)", value=0),
+        gr.Dropdown(
+            label="Statut",
+            choices=["", "sans projet", "avec projet", "potentielle", "reconvertie"],
+            value="",
+        ),
+    ],
+    outputs=gr.Markdown(label="Résultats"),
+    title="Recherche de friches",
+    description="Recherche des friches disponibles sur un territoire (base Cartofriches du CEREMA).",
+    api_name="rechercher_friches",
+)
+
+# Tool 2 : Statistiques de prix foncier
+iface_prix = gr.Interface(
+    fn=statistiques_prix_foncier,
+    inputs=[
+        gr.Textbox(label="Code INSEE commune", placeholder="ex: 13055, 59350"),
+        gr.Textbox(label="Code département", placeholder="ex: 13, 59"),
+        gr.Dropdown(
+            label="Type de bien",
+            choices=["tous", "maison", "appartement"],
+            value="tous",
+        ),
+        gr.Textbox(label="Année", value="2024", placeholder="2010 à 2024"),
+    ],
+    outputs=gr.Markdown(label="Résultats"),
+    title="Prix foncier",
+    description="Statistiques de prix et volumes de transactions immobilières (DV3F, CEREMA/DGFiP).",
+    api_name="statistiques_prix_foncier",
+)
+
+# Tool 3 : Évolution des prix
+iface_evolution = gr.Interface(
+    fn=evolution_prix,
+    inputs=[
+        gr.Textbox(label="Code INSEE commune", placeholder="ex: 13055"),
+        gr.Textbox(label="Code département", placeholder="ex: 13"),
+        gr.Dropdown(
+            label="Type de bien",
+            choices=["maison", "appartement"],
+            value="maison",
+        ),
+    ],
+    outputs=gr.Markdown(label="Résultats"),
+    title="Évolution des prix",
+    description="Évolution temporelle des prix fonciers de 2010 à 2024 (DV3F, CEREMA/DGFiP).",
+    api_name="evolution_prix",
+)
+
+# Tool 4 : Statistiques agrégées de friches (multi-échelle)
+iface_stats_friches = gr.Interface(
+    fn=statistiques_friches,
+    inputs=[
+        gr.Textbox(label="Code INSEE commune", placeholder="ex: 13055 — retourne aussi EPCI, département, région"),
+        gr.Textbox(label="Code SIREN EPCI", placeholder="ex: 200054807 (Aix-Marseille-Provence)"),
+        gr.Textbox(label="Code département", placeholder="ex: 13, 59"),
+        gr.Textbox(label="Code région", placeholder="ex: 93 (PACA), 32 (Hauts-de-France)"),
+        gr.Dropdown(
+            label="Échelle",
+            choices=["", "commune", "epci", "departement", "region", "national"],
+            value="",
+        ),
+    ],
+    outputs=gr.Markdown(label="Résultats"),
+    title="Statistiques friches (multi-échelle)",
+    description="Statistiques agrégées des friches : commune, EPCI, département, région ou national. "
+                "Si une commune est fournie, les stats aux échelles supérieures sont aussi affichées.",
+    api_name="statistiques_friches",
+)
+
+# Tool 5 : Diagnostic foncier territorial (multi-échelle)
+iface_diagnostic = gr.Interface(
+    fn=diagnostic_foncier_territoire,
+    inputs=[
+        gr.Textbox(label="Code INSEE commune", placeholder="ex: 13055"),
+        gr.Textbox(label="Code SIREN EPCI", placeholder="ex: 200054807"),
+        gr.Textbox(label="Code département", placeholder="ex: 13"),
+        gr.Textbox(label="Code région", placeholder="ex: 93 (PACA)"),
+    ],
+    outputs=gr.Markdown(label="Résultats"),
+    title="Diagnostic foncier",
+    description="Diagnostic complet croisant friches et marché foncier à toutes les échelles (Cartofriches + DV3F).",
+    api_name="diagnostic_foncier_territoire",
+)
+
+# --- Application Gradio combinée ---
+demo = gr.TabbedInterface(
+    [iface_friches, iface_prix, iface_evolution, iface_stats_friches, iface_diagnostic],
+    tab_names=["🏗️ Friches", "💰 Prix foncier", "📈 Évolution prix",
+               "📊 Stats friches", "🔍 Diagnostic territorial"],
+    title="MCP CEREMA — Données foncières & friches pour la transition écologique",
+)
+
+if __name__ == "__main__":
+    demo.launch(
+        mcp_server=True,
+        server_name="0.0.0.0",
+        server_port=7860,
+    )

data/cartofriches.geojson

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:508be66a72d3ebdc7040a252237dc80264ef73a9d8d070a6c7645092e560785a
+size 18848774

data/prix_volumes.csv

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:38647dfa09a7f7ccce319b675de17a17ed65e171817ad3ba2e12873de7394f6e
+size 15502944

data_loader.py

@@ -0,0 +1,379 @@
+"""
+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.")

requirements.txt

@@ -0,0 +1,5 @@
+gradio[mcp]>=5.27.1,<6.0.0
+pandas>=2.0
+geopandas>=0.14
+pyogrio
+numpy

server.py

@@ -0,0 +1,941 @@
+"""
+Serveur MCP CEREMA — Tools pour les données foncières et friches.
+
+Ce module définit les 5 fonctions-tools exposées via Gradio MCP :
+1. rechercher_friches : recherche de friches sur un territoire
+2. statistiques_prix_foncier : prix et volumes de transactions immobilières
+3. evolution_prix : évolution temporelle des prix fonciers
+4. statistiques_friches : statistiques agrégées de friches multi-échelle
+5. diagnostic_foncier_territoire : vision combinée friches + marché foncier
+"""
+
+import pandas as pd
+from data_loader import (
+    load_dv3f,
+    load_friches,
+    get_departement_from_commune,
+    get_friches_agg,
+    get_all_echelles_for_commune,
+    get_epci_for_commune,
+    get_region_for_commune,
+    get_region_for_departement,
+    REG_NAMES,
+    PERIODES_CONSTRUCTION,
+)
+
+
+# =============================================================================
+# Helpers
+# =============================================================================
+
+def _format_prix(val: float | None) -> str:
+    """Formate un prix en euros lisible."""
+    if val is None or pd.isna(val):
+        return "non disponible"
+    if val >= 1_000_000:
+        return f"{val/1_000_000:,.2f} M€".replace(",", " ")
+    if val >= 1_000:
+        return f"{val:,.0f} €".replace(",", " ")
+    return f"{val:.0f} €"
+
+
+def _format_pxm2(val: float | None) -> str:
+    """Formate un prix au m²."""
+    if val is None or pd.isna(val):
+        return "non disponible"
+    return f"{val:,.0f} €/m²".replace(",", " ")
+
+
+def _format_surface(val: float | None, unite: str = "m²") -> str:
+    """Formate une surface."""
+    if val is None or pd.isna(val):
+        return "non disponible"
+    return f"{val:,.0f} {unite}".replace(",", " ")
+
+
+def _safe_int(val) -> int | None:
+    """Convertit en int si possible, None sinon."""
+    try:
+        if pd.isna(val):
+            return None
+        return int(val)
+    except (ValueError, TypeError):
+        return None
+
+
+def _safe_float(val) -> float | None:
+    """Convertit en float si possible, None sinon."""
+    try:
+        if pd.isna(val):
+            return None
+        return float(val)
+    except (ValueError, TypeError):
+        return None
+
+
+def _get_dv3f_row(echelle: str, code: str, annee: int) -> pd.Series | None:
+    """Récupère une ligne DV3F par index."""
+    df = load_dv3f()
+    try:
+        row = df.loc[(echelle, code, annee)]
+        if isinstance(row, pd.DataFrame):
+            row = row.iloc[0]
+        return row
+    except KeyError:
+        return None
+
+
+def _get_dv3f_series(echelle: str, code: str) -> pd.DataFrame | None:
+    """Récupère toutes les années pour un territoire."""
+    df = load_dv3f()
+    try:
+        result = df.loc[(echelle, code)]
+        if isinstance(result, pd.Series):
+            result = result.to_frame().T
+        return result
+    except KeyError:
+        return None
+
+
+def _find_best_echelle(commune: str = "", departement: str = "", annee: int = 2024):
+    """Trouve la meilleure échelle disponible avec fallback.
+
+    Retourne (echelle, code, libelle, row) ou None.
+    """
+    df = load_dv3f()
+
+    # 1. Essayer la commune
+    if commune:
+        row = _get_dv3f_row("communes", commune, annee)
+        if row is not None:
+            nb = _safe_int(row.get("nbtrans_cod111"))
+            if nb is not None and nb >= 5:
+                return ("communes", commune, row.get("libelle", commune), row)
+            # Commune trouvée mais pas assez de transactions pour les prix
+            # On retourne quand même la commune mais on signalera le fallback si besoin
+            dep = departement or get_departement_from_commune(commune)
+            row_dep = _get_dv3f_row("departements", dep, annee)
+            if row_dep is not None:
+                return ("departements", dep, row_dep.get("libelle", dep), row_dep,
+                        "communes", commune, row)
+        else:
+            # Commune non trouvée, essayer le département
+            dep = departement or get_departement_from_commune(commune)
+            row_dep = _get_dv3f_row("departements", dep, annee)
+            if row_dep is not None:
+                return ("departements", dep, row_dep.get("libelle", dep), row_dep)
+
+    # 2. Essayer le département
+    if departement:
+        row = _get_dv3f_row("departements", departement, annee)
+        if row is not None:
+            return ("departements", departement, row.get("libelle", departement), row)
+
+    return None
+
+
+# =============================================================================
+# Tool 1 : Recherche de friches
+# =============================================================================
+
+def rechercher_friches(
+    commune: str = "",
+    departement: str = "",
+    type_friche: str = "",
+    surface_min: float = 0,
+    statut: str = "",
+) -> str:
+    """Recherche des friches disponibles sur un territoire donné.
+
+    Interroge la base Cartofriches du CEREMA (inventaire national des friches) pour
+    trouver les friches correspondant aux critères. Utile pour la politique de Zéro
+    Artificialisation Nette (ZAN) : identifier les terrains déjà artificialisés qui
+    peuvent être réhabilités plutôt que de consommer de nouveaux espaces naturels.
+
+    Args:
+        commune: Code INSEE de la commune (ex: "13055" pour Marseille, "75056" pour Paris). Optionnel.
+        departement: Code du département (ex: "13", "75", "59"). Optionnel.
+        type_friche: Type de friche à rechercher. Valeurs possibles : "industrielle", "habitat",
+                     "commerciale", "ferroviaire", "militaire", "hospitalière", "logistique",
+                     "agro-industrielle", "équipement public", "carrière ou mine". Optionnel.
+        surface_min: Surface minimale de la friche en mètres carrés (ex: 5000 pour 0.5 ha). Défaut : 0.
+        statut: Statut de la friche. Valeurs possibles : "sans projet", "avec projet",
+                "potentielle", "reconvertie". Optionnel.
+
+    Returns:
+        Texte structuré décrivant les friches trouvées, avec pour chacune : nom, type, surface,
+        statut, pollution, zonage urbanisme, et commune. Si aucun résultat, un message explicite.
+    """
+    gdf = load_friches()
+    mask = pd.Series([True] * len(gdf), index=gdf.index)
+
+    # Filtres
+    if commune:
+        mask &= gdf["comm_insee"] == commune
+    if departement:
+        mask &= gdf["dep"] == departement
+    if type_friche:
+        type_mapping = {
+            "industrielle": "friche industrielle",
+            "habitat": "friche d'habitat",
+            "commerciale": "friche commerciale",
+            "ferroviaire": "friche ferroviaire",
+            "militaire": "friche militaire",
+            "hospitalière": "friche hospitalière",
+            "logistique": "friche logistique",
+            "agro-industrielle": "friche agro-industrielle",
+            "équipement public": "friche d'équipement public",
+            "carrière ou mine": "friche carrière ou mine",
+        }
+        type_val = type_mapping.get(type_friche.lower(), type_friche)
+        mask &= gdf["site_type"].str.lower() == type_val.lower()
+    if surface_min > 0:
+        mask &= pd.to_numeric(gdf["site_surface"], errors="coerce") >= surface_min
+    if statut:
+        statut_mapping = {
+            "sans projet": "friche sans projet",
+            "avec projet": "friche avec projet",
+            "potentielle": "friche potentielle",
+            "reconvertie": "friche reconvertie",
+        }
+        statut_val = statut_mapping.get(statut.lower(), statut)
+        mask &= gdf["site_statut"].str.lower() == statut_val.lower()
+
+    results = gdf[mask]
+
+    if len(results) == 0:
+        filters_desc = []
+        if commune:
+            filters_desc.append(f"commune {commune}")
+        if departement:
+            filters_desc.append(f"département {departement}")
+        if type_friche:
+            filters_desc.append(f"type '{type_friche}'")
+        if surface_min > 0:
+            filters_desc.append(f"surface ≥ {surface_min:,.0f} m²")
+        if statut:
+            filters_desc.append(f"statut '{statut}'")
+        return f"Aucune friche trouvée avec les critères : {', '.join(filters_desc)}. Essayez d'élargir la recherche (retirer un filtre ou chercher au niveau département)."
+
+    # Construire la réponse
+    total = len(results)
+    lines = []
+    territory = ""
+    if commune:
+        commune_name = results.iloc[0]["comm_nom"] if len(results) > 0 else commune
+        territory = f"la commune de {commune_name} ({commune})"
+    elif departement:
+        territory = f"le département {departement}"
+    else:
+        territory = "le territoire recherché"
+
+    lines.append(f"## Friches trouvées sur {territory}")
+    lines.append(f"**{total} friche(s) trouvée(s)**" + (f" (les 30 premières sont affichées)" if total > 30 else ""))
+    lines.append("")
+
+    # Statistiques résumées (sur TOUS les résultats, pas seulement les 30 premiers)
+    surfaces = pd.to_numeric(results["site_surface"], errors="coerce")
+    lines.append(f"**Surface totale** : {surfaces.sum()/10000:,.1f} hectares")
+    lines.append(f"**Surface médiane** : {surfaces.median()/10000:,.2f} hectares")
+    lines.append("")
+
+    # Répartition par statut (sur TOUS les résultats)
+    statut_counts = results["site_statut"].value_counts()
+    lines.append("**Répartition par statut** :")
+    for s, c in statut_counts.items():
+        lines.append(f"- {s} : {c}")
+    lines.append("")
+
+    # Limiter à 30 résultats pour le détail
+    results_display = results.head(30)
+
+    # Liste détaillée
+    lines.append("### Détail des friches")
+    lines.append("")
+
+    for _, row in results_display.iterrows():
+        surface_val = _safe_float(row.get("site_surface"))
+        surface_ha = f"{surface_val/10000:,.2f} ha" if surface_val else "non renseignée"
+        surface_m2 = f"{surface_val:,.0f} m²" if surface_val else ""
+
+        pollution = row.get("sol_pollution_existe", "inconnu")
+        if pd.isna(pollution):
+            pollution = "inconnu"
+
+        zonage = row.get("urba_zone_type", "non renseigné")
+        if pd.isna(zonage):
+            zonage = "non renseigné"
+
+        bati_etat = row.get("bati_etat", "inconnu")
+        if pd.isna(bati_etat):
+            bati_etat = "inconnu"
+
+        site_type = row.get("site_type", "inconnu")
+        if pd.isna(site_type):
+            site_type = "inconnu"
+
+        lines.append(f"**{row.get('site_nom', 'Sans nom')}**")
+        lines.append(f"- Commune : {row.get('comm_nom', '?')} ({row.get('comm_insee', '?')})")
+        lines.append(f"- Type : {site_type}")
+        lines.append(f"- Surface : {surface_ha} ({surface_m2})")
+        lines.append(f"- Statut : {row.get('site_statut', '?')}")
+        lines.append(f"- Pollution sol : {pollution}")
+        lines.append(f"- Zonage urbanisme : {zonage}")
+        lines.append(f"- État du bâti : {bati_etat}")
+        lines.append("")
+
+    return "\n".join(lines)
+
+
+# =============================================================================
+# Tool 2 : Statistiques de prix foncier
+# =============================================================================
+
+def statistiques_prix_foncier(
+    commune: str = "",
+    departement: str = "",
+    type_bien: str = "tous",
+    annee: str = "2024",
+) -> str:
+    """Récupère les statistiques de prix foncier (transactions immobilières) sur un territoire.
+
+    Utilise les données DV3F du CEREMA (Demande de Valeurs Foncières) qui compilent
+    l'ensemble des transactions immobilières en France depuis 2010. Fournit les prix
+    médians, prix au m², et volumes de transactions pour les maisons et appartements.
+
+    Si la commune est trop petite (moins de 5 transactions), les statistiques du département
+    sont automatiquement fournies en complément.
+
+    Args:
+        commune: Code INSEE de la commune (ex: "13055" pour Marseille). Optionnel.
+        departement: Code du département (ex: "13", "75"). Optionnel. Si ni commune ni département
+                     ne sont fournis, un message d'erreur est retourné.
+        type_bien: Type de bien immobilier. Valeurs : "tous" (défaut), "maison", "appartement".
+        annee: Année des statistiques (de 2010 à 2024). Défaut : "2024".
+
+    Returns:
+        Texte structuré avec le nombre de transactions, le prix médian, le prix au m²
+        (quartiles Q25, médian, Q75), et la surface médiane. Inclut aussi la ventilation
+        par période de construction (avant 1914, 1914-1947, etc.) pour identifier le
+        potentiel de rénovation énergétique (biens anciens vs neufs).
+    """
+    if not commune and not departement:
+        return "Veuillez fournir au moins un code INSEE de commune ou un code département pour obtenir des statistiques de prix foncier."
+
+    try:
+        annee_int = int(annee)
+    except ValueError:
+        return f"L'année '{annee}' n'est pas valide. Veuillez fournir une année entre 2010 et 2024."
+
+    if annee_int < 2010 or annee_int > 2024:
+        return f"L'année {annee_int} est hors de la période couverte (2010–2024)."
+
+    result = _find_best_echelle(commune, departement, annee_int)
+    if result is None:
+        return f"Aucune donnée trouvée pour le territoire demandé (commune={commune}, département={departement}, année={annee})."
+
+    # Gérer le cas du fallback avec données commune
+    commune_row = None
+    commune_code = None
+    if len(result) == 7:
+        echelle, code, libelle, row, _, commune_code, commune_row = result
+        fallback = True
+    else:
+        echelle, code, libelle, row = result
+        fallback = echelle == "departements" and commune != ""
+
+    lines = []
+
+    # Titre
+    if echelle == "communes":
+        lines.append(f"## Prix foncier à {libelle} ({code}) — {annee_int}")
+    elif echelle == "departements":
+        lines.append(f"## Prix foncier dans le département {libelle} ({code}) — {annee_int}")
+        if commune and commune_row is not None:
+            nb_trans = _safe_int(commune_row.get("nbtrans_cod1"))
+            lines.append(f"\n> **Note** : La commune {commune} a seulement {nb_trans or 0} transaction(s) en {annee_int}, ce qui est insuffisant pour des statistiques de prix fiables. Les données du département sont présentées en complément.")
+        elif commune:
+            lines.append(f"\n> **Note** : La commune {commune} n'a pas été trouvée dans les données. Les données du département sont présentées.")
+    else:
+        lines.append(f"## Prix foncier — {libelle} ({code}) — {annee_int}")
+
+    lines.append("")
+
+    # Nombre total de transactions
+    nb_total = _safe_int(row.get("nbtrans_cod1"))
+    nb_maisons = _safe_int(row.get("nbtrans_cod111"))
+    nb_apparts = _safe_int(row.get("nbtrans_cod121"))
+    nb_terrains = _safe_int(row.get("nbtrans_cod2"))
+
+    lines.append("### Volume de transactions")
+    lines.append(f"- Total mutations : **{nb_total:,}**".replace(",", " ") if nb_total else "- Total mutations : non disponible")
+    lines.append(f"- Maisons : **{nb_maisons:,}**".replace(",", " ") if nb_maisons else "- Maisons : 0")
+    lines.append(f"- Appartements : **{nb_apparts:,}**".replace(",", " ") if nb_apparts else "- Appartements : 0")
+    lines.append(f"- Terrains (non bâti) : **{nb_terrains:,}**".replace(",", " ") if nb_terrains else "- Terrains : 0")
+    lines.append("")
+
+    # Maisons
+    if type_bien in ("tous", "maison"):
+        lines.append("### Maisons")
+        prix_med = _safe_float(row.get("valeurfonc_median_cod111"))
+        prix_q25 = _safe_float(row.get("valeurfonc_q25_cod111"))
+        prix_q75 = _safe_float(row.get("valeurfonc_q75_cod111"))
+        pxm2_med = _safe_float(row.get("pxm2_median_cod111"))
+        pxm2_q25 = _safe_float(row.get("pxm2_q25_cod111"))
+        pxm2_q75 = _safe_float(row.get("pxm2_q75_cod111"))
+        surf_med = _safe_float(row.get("sbati_median_cod111"))
+
+        lines.append(f"- Prix médian : **{_format_prix(prix_med)}** (Q25: {_format_prix(prix_q25)}, Q75: {_format_prix(prix_q75)})")
+        lines.append(f"- Prix au m² médian : **{_format_pxm2(pxm2_med)}** (Q25: {_format_pxm2(pxm2_q25)}, Q75: {_format_pxm2(pxm2_q75)})")
+        lines.append(f"- Surface médiane : **{_format_surface(surf_med)}**")
+        lines.append("")
+
+        # Par période
+        lines.append("**Transactions par période de construction** :")
+        for p in ["mp1", "mp2", "mp3", "mp4", "mp5", "mpx"]:
+            nb = _safe_int(row.get(f"nbtrans_{p}"))
+            if nb and nb > 0:
+                prix = _safe_float(row.get(f"valeurfonc_median_{p}"))
+                pxm2 = _safe_float(row.get(f"pxm2_median_{p}"))
+                periode_label = PERIODES_CONSTRUCTION[p]
+                lines.append(f"- {periode_label} : {nb} transactions, prix médian {_format_prix(prix)}, {_format_pxm2(pxm2)}")
+        lines.append("")
+
+    # Appartements
+    if type_bien in ("tous", "appartement"):
+        lines.append("### Appartements")
+        prix_med = _safe_float(row.get("valeurfonc_median_cod121"))
+        prix_q25 = _safe_float(row.get("valeurfonc_q25_cod121"))
+        prix_q75 = _safe_float(row.get("valeurfonc_q75_cod121"))
+        pxm2_med = _safe_float(row.get("pxm2_median_cod121"))
+        pxm2_q25 = _safe_float(row.get("pxm2_q25_cod121"))
+        pxm2_q75 = _safe_float(row.get("pxm2_q75_cod121"))
+        surf_med = _safe_float(row.get("sbati_median_cod121"))
+
+        if nb_apparts and nb_apparts > 0:
+            lines.append(f"- Prix médian : **{_format_prix(prix_med)}** (Q25: {_format_prix(prix_q25)}, Q75: {_format_prix(prix_q75)})")
+            lines.append(f"- Prix au m² médian : **{_format_pxm2(pxm2_med)}** (Q25: {_format_pxm2(pxm2_q25)}, Q75: {_format_pxm2(pxm2_q75)})")
+            lines.append(f"- Surface médiane : **{_format_surface(surf_med)}**")
+            lines.append("")
+
+            # Par période
+            lines.append("**Transactions par période de construction** :")
+            for p in ["ap1", "ap2", "ap3", "ap4", "ap5", "apx"]:
+                nb = _safe_int(row.get(f"nbtrans_{p}"))
+                if nb and nb > 0:
+                    prix = _safe_float(row.get(f"valeurfonc_median_{p}"))
+                    pxm2 = _safe_float(row.get(f"pxm2_median_{p}"))
+                    periode_label = PERIODES_CONSTRUCTION[p]
+                    lines.append(f"- {periode_label} : {nb} transactions, prix médian {_format_prix(prix)}, {_format_pxm2(pxm2)}")
+        else:
+            lines.append("Pas de transactions d'appartements sur ce territoire pour cette année.")
+        lines.append("")
+
+    # Source
+    lines.append("---")
+    lines.append(f"*Source : DV3F (CEREMA/DGFiP), échelle {echelle}, année {annee_int}.*")
+
+    return "\n".join(lines)
+
+
+# =============================================================================
+# Tool 3 : Évolution des prix
+# =============================================================================
+
+def evolution_prix(
+    commune: str = "",
+    departement: str = "",
+    type_bien: str = "maison",
+) -> str:
+    """Retourne l'évolution des prix fonciers sur un territoire de 2010 à 2024.
+
+    Permet de visualiser la tendance du marché immobilier sur 15 ans : évolution du prix
+    médian, du prix au m², et du volume de transactions. Utile pour les études de marché
+    et l'analyse des dynamiques territoriales liées à l'artificialisation.
+
+    Args:
+        commune: Code INSEE de la commune (ex: "13055"). Optionnel.
+        departement: Code du département (ex: "13"). Optionnel. Au moins un des deux est requis.
+        type_bien: Type de bien : "maison" (défaut) ou "appartement".
+
+    Returns:
+        Tableau année par année avec le nombre de transactions, le prix médian, et le
+        prix au m² médian. Inclut le calcul de l'évolution entre la première et la dernière
+        année disponible.
+    """
+    if not commune and not departement:
+        return "Veuillez fournir au moins un code INSEE de commune ou un code département."
+
+    # Déterminer l'échelle
+    echelle = "communes" if commune else "departements"
+    code = commune if commune else departement
+
+    data = _get_dv3f_series(echelle, code)
+    if data is None:
+        if commune:
+            dep = departement or get_departement_from_commune(commune)
+            data = _get_dv3f_series("departements", dep)
+            if data is None:
+                return f"Aucune donnée trouvée pour la commune {commune} ni le département {dep}."
+            echelle = "departements"
+            code = dep
+        else:
+            return f"Aucune donnée trouvée pour le département {departement}."
+
+    # Récupérer le libellé
+    libelle = data.iloc[0].get("libelle", code) if "libelle" in data.columns else code
+
+    # Colonnes selon le type de bien
+    if type_bien.lower() == "appartement":
+        col_nb = "nbtrans_cod121"
+        col_prix = "valeurfonc_median_cod121"
+        col_pxm2 = "pxm2_median_cod121"
+        bien_label = "Appartements"
+    else:
+        col_nb = "nbtrans_cod111"
+        col_prix = "valeurfonc_median_cod111"
+        col_pxm2 = "pxm2_median_cod111"
+        bien_label = "Maisons"
+
+    lines = []
+    if echelle == "communes":
+        lines.append(f"## Évolution des prix — {bien_label} à {libelle} ({code})")
+    else:
+        lines.append(f"## Évolution des prix — {bien_label} dans le département {libelle} ({code})")
+    lines.append("")
+
+    # Construire le tableau
+    lines.append("| Année | Nb transactions | Prix médian | Prix/m² médian |")
+    lines.append("|-------|-----------------|-------------|----------------|")
+
+    first_prix = None
+    last_prix = None
+    first_pxm2 = None
+    last_pxm2 = None
+
+    # data index is annee
+    for annee_val in sorted(data.index):
+        row = data.loc[annee_val]
+        if isinstance(row, pd.DataFrame):
+            row = row.iloc[0]
+
+        nb = _safe_int(row.get(col_nb))
+        prix = _safe_float(row.get(col_prix))
+        pxm2 = _safe_float(row.get(col_pxm2))
+
+        if prix is not None and first_prix is None:
+            first_prix = prix
+            first_pxm2 = pxm2
+        if prix is not None:
+            last_prix = prix
+            last_pxm2 = pxm2
+
+        nb_str = f"{nb:,}".replace(",", " ") if nb is not None else "-"
+        prix_str = _format_prix(prix) if prix else "-"
+        pxm2_str = _format_pxm2(pxm2) if pxm2 else "-"
+
+        lines.append(f"| {annee_val} | {nb_str} | {prix_str} | {pxm2_str} |")
+
+    lines.append("")
+
+    # Évolution
+    if first_prix and last_prix:
+        evol_prix = ((last_prix - first_prix) / first_prix) * 100
+        lines.append(f"**Évolution du prix médian** : {evol_prix:+.1f}% sur la période")
+    if first_pxm2 and last_pxm2:
+        evol_pxm2 = ((last_pxm2 - first_pxm2) / first_pxm2) * 100
+        lines.append(f"**Évolution du prix/m²** : {evol_pxm2:+.1f}% sur la période")
+
+    lines.append("")
+    lines.append("---")
+    lines.append(f"*Source : DV3F (CEREMA/DGFiP), échelle {echelle}.*")
+
+    return "\n".join(lines)
+
+
+# =============================================================================
+# Tool 4 : Statistiques agrégées de friches (multi-échelle)
+# =============================================================================
+
+def _format_friches_agg(stats: dict, echelle_label: str, territoire_label: str) -> str:
+    """Formate les statistiques agrégées de friches en texte lisible."""
+    lines = []
+    lines.append(f"## Statistiques des friches — {territoire_label}")
+    lines.append(f"*Échelle : {echelle_label}*")
+    lines.append("")
+
+    nb = int(stats.get("nb_friches", 0))
+    if nb == 0:
+        lines.append("**Aucune friche recensée** dans la base Cartofriches pour ce territoire.")
+        return "\n".join(lines)
+
+    # Chiffres clés
+    lines.append("### Chiffres clés")
+    lines.append(f"- **Nombre de friches** : {nb:,}".replace(",", " "))
+    nb_communes = stats.get("nb_communes", 0)
+    if isinstance(nb_communes, (int, float)) and nb_communes > 1:
+        lines.append(f"- **Communes concernées** : {int(nb_communes):,}".replace(",", " "))
+    surf_tot = stats.get("surface_totale_ha", 0)
+    lines.append(f"- **Surface totale** : {surf_tot:,.1f} hectares ({surf_tot/100:,.2f} km²)".replace(",", " "))
+    lines.append(f"- **Surface médiane** : {stats.get('surface_mediane_ha', 0):,.2f} hectares".replace(",", " "))
+    lines.append(f"- **Surface moyenne** : {stats.get('surface_moyenne_ha', 0):,.2f} hectares".replace(",", " "))
+    lines.append("")
+
+    # Statuts
+    lines.append("### Répartition par statut")
+    for statut_key, statut_label in [
+        ("nb_potentielle", "Friche potentielle"),
+        ("nb_sans_projet", "Friche sans projet"),
+        ("nb_avec_projet", "Friche avec projet"),
+        ("nb_reconvertie", "Friche reconvertie"),
+    ]:
+        count = int(stats.get(statut_key, 0))
+        pct = round(100 * count / nb, 1) if nb > 0 else 0
+        lines.append(f"- {statut_label} : **{count}** ({pct}%)")
+    lines.append("")
+
+    # Mobilisables
+    nb_mob = int(stats.get("nb_mobilisables", 0))
+    surf_mob = stats.get("surface_mobilisable_ha", 0)
+    lines.append("### Potentiel mobilisable (ZAN)")
+    lines.append(f"- **Friches mobilisables** (sans projet + potentielles) : **{nb_mob}** friches")
+    lines.append(f"- **Surface mobilisable** : **{surf_mob:,.1f} hectares**".replace(",", " "))
+    lines.append("")
+
+    # Qualité
+    nb_poll = int(stats.get("nb_polluees", 0))
+    nb_u = int(stats.get("nb_zone_u", 0))
+    pct_u = stats.get("pct_zone_u", 0)
+    lines.append("### Caractéristiques")
+    lines.append(f"- Pollution avérée ou supposée : {nb_poll} ({round(100*nb_poll/nb, 1) if nb else 0}%)")
+    lines.append(f"- En zone urbanisée (U) : {nb_u} ({pct_u}%) — requalifiables sans artificialisation")
+    lines.append("")
+
+    # Types principaux
+    top_types = stats.get("top_types", {})
+    if top_types and isinstance(top_types, dict):
+        lines.append("### Principaux types de friches")
+        for type_name, count in sorted(top_types.items(), key=lambda x: -x[1]):
+            lines.append(f"- {type_name} : {int(count)}")
+        lines.append("")
+
+    lines.append("---")
+    lines.append("*Source : Cartofriches (CEREMA), statistiques pré-agrégées.*")
+    return "\n".join(lines)
+
+
+def statistiques_friches(
+    commune: str = "",
+    epci: str = "",
+    departement: str = "",
+    region: str = "",
+    echelle: str = "",
+) -> str:
+    """Fournit les statistiques agrégées des friches à différentes échelles territoriales.
+
+    Retourne le nombre de friches, la surface totale et mobilisable, la répartition par
+    statut et type, le taux de pollution et le pourcentage en zone urbanisée. Les données
+    sont pré-agrégées pour des réponses instantanées.
+
+    Fonctionne à 5 échelles : commune, EPCI (intercommunalité), département, région, national.
+    Si une commune est fournie, l'outil retourne aussi automatiquement les statistiques
+    aux échelles supérieures (EPCI, département, région) pour permettre la comparaison.
+
+    Args:
+        commune: Code INSEE de la commune (ex: "13055" pour Marseille). Optionnel.
+        epci: Code SIREN de l'EPCI / intercommunalité (ex: "200054807" pour Aix-Marseille-Provence).
+              Optionnel.
+        departement: Code du département (ex: "13", "59"). Optionnel.
+        region: Code de la région (ex: "93" pour PACA, "32" pour Hauts-de-France). Optionnel.
+        echelle: Forcer une échelle spécifique : "commune", "epci", "departement", "region",
+                 "national". Si vide, l'échelle est déduite des paramètres fournis. Mettre
+                 "national" pour les statistiques France entière.
+
+    Returns:
+        Statistiques structurées : nombre de friches, surface totale et mobilisable,
+        répartition par statut/type, pollution, zone urbanisée. Si une commune est fournie,
+        inclut aussi les statistiques EPCI, département et région pour comparaison.
+    """
+    # Cas national
+    if echelle == "national" or (not commune and not epci and not departement and not region):
+        if not echelle:
+            return ("Veuillez fournir au moins un territoire (commune, EPCI, département, région) "
+                    "ou préciser echelle='national' pour les statistiques France entière.")
+        stats = get_friches_agg("national")
+        if stats:
+            return _format_friches_agg(stats, "National", "France entière")
+        return "Erreur : données nationales non disponibles."
+
+    # Si commune fournie → retourner multi-échelle pour comparaison
+    if commune:
+        infos = get_all_echelles_for_commune(commune)
+        lines_parts = []
+
+        # Commune
+        stats_comm = get_friches_agg("commune", commune)
+        if stats_comm:
+            lines_parts.append(_format_friches_agg(
+                stats_comm, "Commune",
+                f"{infos.get('commune_nom', '')} ({commune})"
+            ))
+        else:
+            lines_parts.append(f"## Friches — {infos.get('commune_nom', commune)} ({commune})\n\nAucune friche recensée dans cette commune.")
+
+        # EPCI
+        epci_code = infos.get("epci", "")
+        if epci_code and epci_code != "ZZZZZZZZZ":
+            stats_epci = get_friches_agg("epci", epci_code)
+            if stats_epci:
+                lines_parts.append(_format_friches_agg(
+                    stats_epci, "EPCI (intercommunalité)",
+                    f"{infos.get('epci_nom', epci_code)}"
+                ))
+
+        # Département
+        dep = infos.get("departement", "")
+        if dep:
+            stats_dep = get_friches_agg("departement", dep)
+            if stats_dep:
+                lines_parts.append(_format_friches_agg(
+                    stats_dep, "Département",
+                    f"Département {dep}"
+                ))
+
+        # Région
+        reg = infos.get("region", "")
+        if reg:
+            stats_reg = get_friches_agg("region", reg)
+            if stats_reg:
+                lines_parts.append(_format_friches_agg(
+                    stats_reg, "Région",
+                    f"{infos.get('region_nom', reg)}"
+                ))
+
+        return "\n\n---\n\n".join(lines_parts)
+
+    # EPCI seul
+    if epci:
+        stats = get_friches_agg("epci", epci)
+        if stats:
+            return _format_friches_agg(stats, "EPCI (intercommunalité)",
+                                       stats.get("libelle", epci))
+        return f"Aucune donnée de friches pour l'EPCI {epci}."
+
+    # Département seul
+    if departement:
+        stats = get_friches_agg("departement", departement)
+        if stats:
+            return _format_friches_agg(stats, "Département",
+                                       f"Département {departement}")
+        return f"Aucune donnée de friches pour le département {departement}."
+
+    # Région seule
+    if region:
+        stats = get_friches_agg("region", region)
+        if stats:
+            reg_nom = REG_NAMES.get(region, region)
+            return _format_friches_agg(stats, "Région", reg_nom)
+        return f"Aucune donnée de friches pour la région {region}."
+
+    return "Paramètres insuffisants. Fournissez un code commune, EPCI, département ou région."
+
+
+# =============================================================================
+# Tool 5 : Diagnostic foncier territorial (multi-échelle)
+# =============================================================================
+
+def diagnostic_foncier_territoire(
+    commune: str = "",
+    epci: str = "",
+    departement: str = "",
+    region: str = "",
+) -> str:
+    """Fournit un diagnostic foncier complet d'un territoire en croisant friches et marché immobilier.
+
+    Outil de synthèse qui combine les données Cartofriches (inventaire des friches) et
+    DV3F (transactions immobilières) pour donner une vision globale de la situation
+    foncière d'un territoire. Particulièrement utile dans le cadre du Zéro Artificialisation
+    Nette (ZAN) pour identifier le potentiel de requalification et contextualiser avec
+    le marché local.
+
+    Fonctionne à toutes les échelles : commune, EPCI, département, région.
+
+    Args:
+        commune: Code INSEE de la commune (ex: "13055" pour Marseille). Optionnel.
+        epci: Code SIREN de l'EPCI (ex: "200054807"). Optionnel.
+        departement: Code du département (ex: "13"). Optionnel.
+        region: Code de la région (ex: "93" pour PACA). Optionnel.
+
+    Returns:
+        Diagnostic structuré en 3 parties :
+        1. Inventaire des friches (nombre, surface, types, statuts)
+        2. Marché foncier local (prix, volumes, tendance)
+        3. Analyse croisée (contextualisation friches / marché)
+    """
+    if not commune and not epci and not departement and not region:
+        return "Veuillez fournir au moins un territoire (commune, EPCI, département ou région)."
+
+    lines = []
+
+    # --- Déterminer l'échelle et le territoire ---
+    if commune:
+        friches_echelle, friches_code = "commune", commune
+        dv3f_echelle, dv3f_code = "communes", commune
+        infos = get_all_echelles_for_commune(commune)
+        nom_territoire = f"{infos.get('commune_nom', commune)} ({commune})"
+        dep_code = infos.get("departement", get_departement_from_commune(commune))
+    elif epci:
+        friches_echelle, friches_code = "epci", epci
+        dv3f_echelle, dv3f_code = "epci", epci
+        stats_temp = get_friches_agg("epci", epci)
+        nom_territoire = stats_temp.get("libelle", epci) if stats_temp else epci
+        dep_code = ""
+    elif departement:
+        friches_echelle, friches_code = "departement", departement
+        dv3f_echelle, dv3f_code = "departements", departement
+        nom_territoire = f"Département {departement}"
+        dep_code = departement
+    else:  # region
+        friches_echelle, friches_code = "region", region
+        dv3f_echelle, dv3f_code = "regions", region
+        nom_territoire = REG_NAMES.get(region, f"Région {region}")
+        dep_code = ""
+
+    lines.append(f"## Diagnostic foncier — {nom_territoire}")
+    lines.append("")
+
+    # --- Partie 1 : Friches (depuis les agrégations pré-calculées) ---
+    lines.append("### 1. Inventaire des friches (Cartofriches)")
+    lines.append("")
+
+    stats = get_friches_agg(friches_echelle, friches_code)
+
+    if stats is None or int(stats.get("nb_friches", 0)) == 0:
+        lines.append("**Aucune friche recensée** dans la base Cartofriches pour ce territoire.")
+        lines.append("> Cela ne signifie pas qu'il n'y a pas de friches, mais qu'aucune n'a été inventoriée.")
+        nb_friches = 0
+    else:
+        nb_friches = int(stats["nb_friches"])
+        nb_communes = stats.get("nb_communes", 0)
+        lines.append(f"- **Nombre de friches** : {nb_friches:,}".replace(",", " "))
+        if isinstance(nb_communes, (int, float)) and nb_communes > 1:
+            lines.append(f"- **Communes concernées** : {int(nb_communes):,}".replace(",", " "))
+        lines.append(f"- **Surface totale** : {stats['surface_totale_ha']:,.1f} hectares".replace(",", " "))
+        lines.append(f"- **Surface médiane** : {stats['surface_mediane_ha']:,.2f} hectares".replace(",", " "))
+        lines.append("")
+
+        # Par statut
+        lines.append("**Par statut** :")
+        for key, label in [("nb_potentielle", "friche potentielle"), ("nb_sans_projet", "friche sans projet"),
+                           ("nb_avec_projet", "friche avec projet"), ("nb_reconvertie", "friche reconvertie")]:
+            count = int(stats.get(key, 0))
+            if count > 0:
+                pct = round(100 * count / nb_friches, 0)
+                lines.append(f"- {label} : {count} ({pct:.0f}%)")
+        lines.append("")
+
+        # Types
+        top_types = stats.get("top_types", {})
+        if top_types and isinstance(top_types, dict):
+            lines.append("**Principaux types** :")
+            for t, c in sorted(top_types.items(), key=lambda x: -x[1])[:5]:
+                lines.append(f"- {t} : {int(c)}")
+            lines.append("")
+
+        # Pollution et zone U
+        nb_poll = int(stats.get("nb_polluees", 0))
+        nb_u = int(stats.get("nb_zone_u", 0))
+        pct_u = stats.get("pct_zone_u", 0)
+        lines.append(f"**Pollution** : {nb_poll} friche(s) avec pollution avérée ou supposée ({round(100*nb_poll/nb_friches)}%)")
+        lines.append(f"**En zone urbanisée (U)** : {nb_u} ({pct_u}%) — requalifiables sans artificialisation")
+
+    lines.append("")
+
+    # --- Partie 2 : Marché foncier ---
+    lines.append("### 2. Marché foncier (DV3F)")
+    lines.append("")
+
+    annee = 2024
+    pxm2_maisons = None
+
+    if commune:
+        result = _find_best_echelle(commune, dep_code, annee)
+    elif departement:
+        result = _find_best_echelle("", departement, annee)
+    elif epci:
+        row = _get_dv3f_row("epci", epci, annee)
+        result = ("epci", epci, nom_territoire, row) if row is not None else None
+    elif region:
+        row = _get_dv3f_row("regions", region, annee)
+        result = ("regions", region, nom_territoire, row) if row is not None else None
+    else:
+        result = None
+
+    if result is None:
+        lines.append("Aucune donnée de marché foncier disponible pour ce territoire.")
+    else:
+        if len(result) == 7:
+            ech, ech_code, ech_libelle, row, _, _, _ = result
+        else:
+            ech, ech_code, ech_libelle, row = result
+
+        if ech == "departements" and commune:
+            lines.append(f"> Données au niveau département ({ech_libelle}) — commune trop petite pour des statistiques fiables.")
+            lines.append("")
+
+        nb_total = _safe_int(row.get("nbtrans_cod1"))
+        nb_maisons = _safe_int(row.get("nbtrans_cod111"))
+        nb_apparts = _safe_int(row.get("nbtrans_cod121"))
+
+        lines.append(f"**Année {annee}** :")
+        lines.append(f"- Transactions totales : {nb_total:,}".replace(",", " ") if nb_total else "- Transactions totales : non disponible")
+        lines.append("")
+
+        pxm2_maisons = _safe_float(row.get("pxm2_median_cod111"))
+        prix_maisons = _safe_float(row.get("valeurfonc_median_cod111"))
+        if nb_maisons and nb_maisons > 0:
+            lines.append(f"**Maisons** ({nb_maisons:,} transactions) :".replace(",", " "))
+            lines.append(f"- Prix médian : {_format_prix(prix_maisons)}")
+            lines.append(f"- Prix/m² médian : {_format_pxm2(pxm2_maisons)}")
+
+        pxm2_apparts = _safe_float(row.get("pxm2_median_cod121"))
+        prix_apparts = _safe_float(row.get("valeurfonc_median_cod121"))
+        if nb_apparts and nb_apparts > 0:
+            lines.append(f"\n**Appartements** ({nb_apparts:,} transactions) :".replace(",", " "))
+            lines.append(f"- Prix médian : {_format_prix(prix_apparts)}")
+            lines.append(f"- Prix/m² médian : {_format_pxm2(pxm2_apparts)}")
+
+        # Tendance
+        row_2020 = _get_dv3f_row(ech, ech_code, 2020)
+        if row_2020 is not None:
+            pxm2_2020 = _safe_float(row_2020.get("pxm2_median_cod111"))
+            if pxm2_2020 and pxm2_maisons:
+                evol = ((pxm2_maisons - pxm2_2020) / pxm2_2020) * 100
+                lines.append(f"\n**Tendance** : prix/m² maisons {evol:+.1f}% entre 2020 et 2024")
+
+    lines.append("")
+
+    # --- Partie 3 : Analyse croisée ---
+    lines.append("### 3. Analyse croisée pour le ZAN")
+    lines.append("")
+
+    if stats and nb_friches > 0 and result is not None:
+        nb_mob = int(stats.get("nb_mobilisables", 0))
+        surf_mob = stats.get("surface_mobilisable_ha", 0)
+        lines.append(f"- **Gisement de friches mobilisables** (sans projet + potentielles) : {nb_mob} friches, {surf_mob:,.1f} ha".replace(",", " "))
+        if nb_mob > 0:
+            if pxm2_maisons:
+                lines.append(f"- **Contexte marché** : le prix/m² local ({_format_pxm2(pxm2_maisons)} pour les maisons) permet de contextualiser le coût de réhabilitation")
+            lines.append(f"- **Potentiel ZAN** : ces {nb_mob} friches représentent du foncier déjà artificialisé, mobilisable pour éviter la consommation de nouveaux espaces naturels ou agricoles")
+        else:
+            lines.append("- Toutes les friches du territoire ont déjà un projet ou ont été reconverties.")
+    elif nb_friches == 0:
+        lines.append("Pas de friches inventoriées sur ce territoire — le diagnostic ZAN nécessiterait un inventaire local complémentaire.")
+    else:
+        lines.append("Données de marché foncier indisponibles — le croisement friches/marché n'est pas possible.")
+
+    lines.append("")
+    lines.append("---")
+    lines.append("*Sources : Cartofriches (CEREMA) + DV3F (CEREMA/DGFiP).*")
+
+    return "\n".join(lines)