from __future__ import annotations import json import os import re import unicodedata from pathlib import Path from typing import Any, Optional from dotenv import load_dotenv from fastapi import FastAPI, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles try: import psycopg except Exception: # pragma: no cover psycopg = None try: from open_data.aggregator import OpenDataBroker except Exception: # pragma: no cover OpenDataBroker = None try: from external import artificialisation, geo_api, georisques, insee_melodi, insee_socio, sitadel from external.cache import status as cache_status from external.config import agency_epci_codes, load_scope from external.http_client import ExternalApiError from external.sources import source_catalog except Exception: # pragma: no cover artificialisation = geo_api = georisques = insee_melodi = insee_socio = sitadel = None ExternalApiError = RuntimeError def cache_status(names: list[str]) -> dict[str, Any]: return {name: {"exists": False, "fresh": False} for name in names} def agency_epci_codes() -> list[str]: return ["246200364", "246200299", "200072460", "200069672", "200044030"] def load_scope() -> dict[str, Any]: return {"name": "Territoire AULA", "epci_codes": agency_epci_codes()} def source_catalog() -> list[dict[str, str]]: return [ {"id": "api_geo", "label": "API Géo", "description": "Référentiel communes / EPCI"}, {"id": "insee", "label": "INSEE", "description": "Socio-démo à mettre en cache"}, {"id": "sitadel", "label": "Sitadel", "description": "Autorisations d'urbanisme"}, {"id": "artificialisation", "label": "Artificialisation", "description": "Consommation d'espaces"}, {"id": "georisques", "label": "Géorisques", "description": "Risques par commune"}, ] load_dotenv(override=True) BASE_DIR = Path(__file__).resolve().parent STATIC_DIR = BASE_DIR / "static" DATA_DIR = BASE_DIR / "data" START_YEAR = int(os.environ.get("FONCIER_START_YEAR", "1970")) END_YEAR = int(os.environ.get("FONCIER_END_YEAR", "2025")) SOCIODEMO_DATA_YEAR = int(os.environ.get("FONCIER_SOCIODEMO_DATA_YEAR", "2022")) DEFAULT_LEVEL = os.environ.get("FONCIER_DEFAULT_LEVEL", "commune") SCHEMA = os.environ.get("FONCIER_SCHEMA", "foncier_poc") BUILDINGS_VIEW = os.environ.get("FONCIER_BUILDINGS_VIEW", "v_batiments_historique") TERRITORIES_VIEW = os.environ.get("FONCIER_TERRITORIES_VIEW", "v_territoires") GEOM_COL = os.environ.get("FONCIER_GEOM_COL", "geom") COUNT_COL = os.environ.get("FONCIER_COUNT_COL", "").strip() BDNB_VIEW = os.environ.get("FONCIER_BDNB_VIEW", BUILDINGS_VIEW).strip() BDNB_HEIGHT_COL = os.environ.get("FONCIER_BDNB_HEIGHT_COL", "hauteur_m").strip() BDNB_YEAR_COL = os.environ.get("FONCIER_BDNB_YEAR_COL", "annee_apparition").strip() SOCIODEMO_VIEW = os.environ.get("FONCIER_SOCIODEMO_VIEW", "v_sociodemo_insee").strip() SOCIODEMO_USE_INSEE = os.environ.get("FONCIER_SOCIODEMO_USE_INSEE", "true").lower() not in {"0", "false", "no", "non"} CONSUMPTION_VIEW = os.environ.get("FONCIER_CONSUMPTION_VIEW", "v_conso_enaf_cerema").strip() CONSUMPTION_TYPOLOGY_VIEW = os.environ.get("FONCIER_CONSUMPTION_TYPOLOGY_VIEW", "v_conso_enaf_cerema_typologies").strip() CONSUMPTION_USE_CEREMA = os.environ.get("FONCIER_CONSUMPTION_USE_CEREMA", "true").lower() not in {"0", "false", "no", "non"} CONSUMPTION_DEFAULT_RECENT_YEARS = int(os.environ.get("FONCIER_CONSUMPTION_RECENT_YEARS", "5")) CONSUMPTION_SOURCE_MODE = os.environ.get("FONCIER_CONSUMPTION_SOURCE", "postgis").lower() open_data_broker = OpenDataBroker(BASE_DIR) if OpenDataBroker else None app = FastAPI(title="Foncier narratif - core", version="0.7.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") def pg_configured() -> bool: required = ["PGHOST", "PGDATABASE", "PGUSER", "PGPASSWORD"] return psycopg is not None and all(os.environ.get(key) for key in required) def qident(name: str) -> str: return '"' + str(name).replace('"', '""') + '"' def qtable(schema: str, table: str) -> str: return f"{qident(schema)}.{qident(table)}" def count_sql_expr() -> str: """Return the SQL expression used to count buildings/objects. With real building geometries this is simply count(*). With parcel-level Fichiers fonciers pnb10_parcelle, set FONCIER_COUNT_COL=poids_bati so timeline KPIs use nbat without duplicating parcel geometries. """ if COUNT_COL: return f"coalesce(sum(greatest(coalesce({qident(COUNT_COL)}, 0), 0)), 0)::bigint" return "count(*)::bigint" def feature_weight_property() -> str: if COUNT_COL: return f", '{COUNT_COL}', {qident(COUNT_COL)}" return "" def get_conn(): if not pg_configured(): raise RuntimeError("PostGIS is not configured. Using demo mode.") return psycopg.connect( host=os.environ["PGHOST"], port=int(os.environ.get("PGPORT", "5432")), dbname=os.environ["PGDATABASE"], user=os.environ["PGUSER"], password=os.environ["PGPASSWORD"], sslmode=os.environ.get("PGSSLMODE", "disable"), ) def rows(sql: str, params: tuple[Any, ...] = ()): # pragma: no cover - depends on local PG with get_conn() as conn, conn.cursor() as cur: cur.execute(sql, params) return cur.fetchall() def sample_features() -> list[dict[str, Any]]: path = DATA_DIR / "sample_buildings.geojson" with path.open("r", encoding="utf-8") as f: return json.load(f)["features"] def feature_year(feature: dict[str, Any]) -> Optional[int]: value = feature.get("properties", {}).get("annee_apparition") try: return int(value) except Exception: return None def normalize_level(niveau: str | None) -> str: return "commune" if str(niveau or DEFAULT_LEVEL).lower() == "commune" else "epci" def scope_column(niveau: str | None) -> str: return "code_insee" if normalize_level(niveau) == "commune" else "code_territoire" def scope_name_column(niveau: str | None) -> str: return "commune" if normalize_level(niveau) == "commune" else "nom_territoire" def demo_territories(niveau: str | None = None) -> list[dict[str, str]]: level = normalize_level(niveau) features = sample_features() values: dict[str, str] = {} for feature in features: props = feature.get("properties", {}) if level == "commune": code = str(props.get("code_insee") or props.get("commune") or "commune_demo") name = str(props.get("commune") or "Commune démo") else: code = str(props.get("territoire") or props.get("code_territoire") or "demo") name = str(props.get("nom_territoire") or props.get("nom_epci") or "Territoire démo") values[code] = name return [{"code": code, "nom": name, "niveau": level} for code, name in sorted(values.items(), key=lambda item: item[1])] def filter_demo_features(territoire: str, niveau: str | None = None) -> list[dict[str, Any]]: level = normalize_level(niveau) col = "code_insee" if level == "commune" else "territoire" return [f for f in sample_features() if str(f.get("properties", {}).get(col)) == str(territoire)] def timeline_for_features(features: list[dict[str, Any]]) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] cum = 0 surf_cum = 0.0 for year in range(START_YEAR, END_YEAR + 1): current = [f for f in features if feature_year(f) == year] n = len(current) surf = sum(float(f.get("properties", {}).get("surface_emprise_m2") or 0) for f in current) cum += n surf_cum += surf result.append({ "year": year, "batiments_nouveaux": n, "batiments_cumules": cum, "surface_nouvelle_m2": surf, "surface_cumulee_m2": surf_cum, }) return result def usage_stats_for_demo(territoire: str, niveau: str | None = None) -> list[dict[str, Any]]: by_usage: dict[str, dict[str, Any]] = {} for feature in filter_demo_features(territoire, niveau): props = feature.get("properties", {}) usage = str(props.get("usage") or "non qualifié") if usage not in by_usage: by_usage[usage] = {"label": usage, "value": 0, "surface_m2": 0.0} by_usage[usage]["value"] += 1 by_usage[usage]["surface_m2"] += float(props.get("surface_emprise_m2") or 0) return sorted(by_usage.values(), key=lambda item: item["value"], reverse=True) def usage_stats_for_pg(territoire: str, niveau: str | None = None) -> list[dict[str, Any]]: col = scope_column(niveau) sql = f""" SELECT coalesce(usage::text, 'non qualifié') AS usage, {count_sql_expr()} AS nb, coalesce(sum(surface_emprise_m2), 0)::double precision AS surface_m2 FROM {qtable(SCHEMA, BUILDINGS_VIEW)} WHERE {qident(col)} = %s GROUP BY coalesce(usage::text, 'non qualifié') ORDER BY nb DESC """ return [{"label": u, "value": int(n), "surface_m2": float(s)} for u, n, s in rows(sql, (territoire,))] def decade_stats(timeline: list[dict[str, Any]]) -> list[dict[str, Any]]: buckets: dict[int, dict[str, Any]] = {} for item in timeline: year = int(item["year"]) decade = (year // 10) * 10 if decade not in buckets: buckets[decade] = { "decade": decade, "label": f"{decade}s", "batiments_nouveaux": 0, "surface_nouvelle_m2": 0.0, } buckets[decade]["batiments_nouveaux"] += int(item.get("batiments_nouveaux") or 0) buckets[decade]["surface_nouvelle_m2"] += float(item.get("surface_nouvelle_m2") or 0) return list(sorted(buckets.values(), key=lambda item: item["decade"])) def _numeric(value: Any, default: float = 0) -> float: try: if value is None: return default return float(value) except Exception: return default def _int_or_none(value: Any) -> int | None: try: if value is None: return None return int(round(float(value))) except Exception: return None def _household_size(population: Any, households: Any) -> float | None: pop = _numeric(population, 0) men = _numeric(households, 0) if pop <= 0 or men <= 0: return None return round(pop / men, 2) def demo_sociodemo(territoire: str, niveau: str | None = None) -> dict[str, Any]: # Valeurs fictives : le bloc est prêt pour INSEE RP / cache Melodi, mais ne remplace pas les valeurs officielles. level = normalize_level(niveau) profiles = { "demo": {"population_current": 354_800, "households": 161_500, "density": 1088, "delta": -0.006}, "62001": {"population_current": 19_450, "households": 8_920, "density": 1420, "delta": -0.004}, "62002": {"population_current": 12_840, "households": 5_780, "density": 780, "delta": -0.003}, "62003": {"population_current": 8_960, "households": 4_160, "density": 610, "delta": -0.005}, } p = profiles.get(str(territoire), profiles["demo"]) current = int(p["population_current"]) pop_series = [ {"year": 1975, "population": round(current * 1.18), "households": round(p["households"] * 0.62), "type": "observed"}, {"year": 1982, "population": round(current * 1.12), "households": round(p["households"] * 0.70), "type": "observed"}, {"year": 1990, "population": round(current * 1.06), "households": round(p["households"] * 0.78), "type": "observed"}, {"year": 1999, "population": round(current * 1.02), "households": round(p["households"] * 0.86), "type": "observed"}, {"year": 2009, "population": round(current * 1.01), "households": round(p["households"] * 0.94), "type": "observed"}, {"year": 2014, "population": round(current * 1.015), "households": round(p["households"] * 0.97), "type": "observed"}, {"year": SOCIODEMO_DATA_YEAR, "population": current, "households": int(p["households"]), "type": "observed"}, ] observed_only = [row for row in pop_series if row.get("type") == "observed"] first_pop = observed_only[0]["population"] if observed_only else current population_change_pct = round(((current - first_pop) / first_pop) * 100, 1) if first_pop else None return { "available": False, "official": False, "source_mode": "demo_fallback", "source": "Démo locale - à remplacer par INSEE RP / cache Melodi", "millesime": str(SOCIODEMO_DATA_YEAR), "data_year": SOCIODEMO_DATA_YEAR, "population_current": current, "population_label": "Population municipale" if level == "commune" else "Population EPCI", "population_change_pct": population_change_pct, "population_change_since": observed_only[0]["year"] if observed_only else None, "population_projection": None, "projection_year": None, "households": int(p["households"]), "housing_current": None, "jobs_current": None, "households_change_pct": round(((int(p["households"]) - pop_series[0]["households"]) / pop_series[0]["households"]) * 100, 1), "avg_household_size": _household_size(current, p["households"]), "density": int(p["density"]), "series": pop_series, "methodology": { "observed_source": f"INSEE RP - millésime {SOCIODEMO_DATA_YEAR} à brancher", "projection_source": "Projections non affichées dans ce chapitre", "projection_scale": "Commune" if level == "commune" else "EPCI", "projection_direct": False, "warning": "Ce bloc doit rester sur les données observées. Les projections Omphale sont traitées séparément dans le chapitre Anticiper." }, "sources": [ {"label": "Population municipale / EPCI", "source": "INSEE RP", "scale": "commune / EPCI", "status": "à brancher"}, {"label": "Ménages", "source": "INSEE RP", "scale": "commune / EPCI", "status": "à brancher"}, {"label": "Densité", "source": "INSEE + surface communale/EPCI", "scale": "commune / EPCI", "status": "à brancher"}, ], } def _relation_has_column(relation_name: str, column_name: str) -> bool: if not pg_configured() or not relation_name: return False sql = """ SELECT 1 FROM information_schema.columns WHERE table_schema = %s AND table_name = %s AND column_name = %s LIMIT 1 """ try: return bool(rows(sql, (SCHEMA, relation_name, column_name))) except Exception: return False def sociodemo_from_view(territoire: str, niveau: str | None = None) -> dict[str, Any] | None: """Lecture optionnelle d'une vue PostGIS/cache INSEE. Vue attendue : FONCIER_SCHEMA.FONCIER_SOCIODEMO_VIEW avec au minimum : niveau, code_territoire, year, population, households, density, source, millesime. Si la vue expose housing et jobs, l'application les récupère aussi. """ level = normalize_level(niveau) if not pg_configured() or not SOCIODEMO_VIEW: return None has_housing = _relation_has_column(SOCIODEMO_VIEW, 'housing') has_jobs = _relation_has_column(SOCIODEMO_VIEW, 'jobs') housing_expr = "housing::double precision" if has_housing else "NULL::double precision" jobs_expr = "jobs::double precision" if has_jobs else "NULL::double precision" sql = f""" SELECT year::integer, population::double precision, households::double precision, density::double precision, {housing_expr} AS housing, {jobs_expr} AS jobs, coalesce(source::text, 'INSEE RP') AS source, coalesce(millesime::text, '') AS millesime FROM {qtable(SCHEMA, SOCIODEMO_VIEW)} WHERE niveau = %s AND code_territoire::text = %s ORDER BY year::integer """ try: data = rows(sql, (level, territoire)) except Exception: return None if not data: return None series = [] for year, population, households, density, housing, jobs, source, millesime in data: series.append({ "year": int(year), "population": _int_or_none(population), "households": _int_or_none(households), "density": _numeric(density, 0) or None, "housing": _int_or_none(housing), "jobs": _int_or_none(jobs), "type": "observed", }) observed = [item for item in series if item.get("population") is not None] last = observed[-1] if observed else series[-1] first = observed[0] if observed else series[0] population_current = _int_or_none(last.get("population")) households_current = _int_or_none(last.get("households")) density_current = _numeric(last.get("density"), 0) or None housing_current = _int_or_none(last.get("housing")) jobs_current = _int_or_none(last.get("jobs")) population_change_pct = None if first.get("population") and population_current: population_change_pct = round(((population_current - first["population"]) / first["population"]) * 100, 1) households_change_pct = None if first.get("households") and households_current: households_change_pct = round(((households_current - first["households"]) / first["households"]) * 100, 1) return { "available": True, "official": True, "source_mode": "postgis_cache", "source": data[-1][6] or "INSEE RP", "millesime": data[-1][7] or str(last.get("year") or SOCIODEMO_DATA_YEAR), "data_year": int(last.get("year") or SOCIODEMO_DATA_YEAR), "population_current": population_current, "population_label": "Population municipale" if level == "commune" else "Population EPCI", "population_change_pct": population_change_pct, "population_change_since": first.get("year"), "population_projection": None, "projection_year": None, "households": households_current, "households_change_pct": households_change_pct, "housing_current": housing_current, "jobs_current": jobs_current, "avg_household_size": _household_size(population_current, households_current), "density": round(density_current, 1) if density_current else None, "series": series, "methodology": { "observed_source": f"{data[-1][6] or 'Recensements de population INSEE'}{(' - ' + data[-1][7]) if data[-1][7] else ''}", "warning": "Données observées issues des recensements de population INSEE." }, "sources": [ {"label": "Population", "source": data[-1][6] or "INSEE RP", "scale": level, "status": "cache"}, {"label": "Ménages", "source": data[-1][6] or "INSEE RP", "scale": level, "status": "cache"}, {"label": "Logements", "source": data[-1][6] or "INSEE RP", "scale": level, "status": "cache"}, {"label": "Densité", "source": "INSEE + surface du territoire", "scale": level, "status": "cache"}, ], } def official_or_demo_sociodemo(territoire: str, niveau: str | None = None) -> dict[str, Any]: cached = sociodemo_from_view(territoire, niveau) if cached: return cached if SOCIODEMO_USE_INSEE and insee_socio is not None: try: official = insee_socio.sociodemo_for_territory(territoire, normalize_level(niveau), base_dir=BASE_DIR) if official: return official except Exception: pass return demo_sociodemo(territoire, niveau) def consumption_from_view(territoire: str, niveau: str | None = None) -> dict[str, Any] | None: """Lecture simple de la consommation ENAF Cerema depuis PostGIS. Vue attendue : foncier_poc.v_conso_enaf_cerema Colonnes minimales : niveau, code_territoire, nom_territoire, year, conso_enaf_ha. Colonnes recommandées : period_label, source, millesime, updated_at. """ if not CONSUMPTION_USE_CEREMA or not CONSUMPTION_VIEW or not pg_configured(): return None level = normalize_level(niveau) try: sql = f""" SELECT year::integer AS year, coalesce(conso_enaf_ha, 0)::double precision AS conso_enaf_ha, coalesce(period_label::text, (year::text || '-' || (year::integer + 1)::text)) AS period_label, coalesce(source::text, 'Cerema - Portail national de l’artificialisation') AS source, coalesce(millesime::text, '2009-2024') AS millesime, nom_territoire::text AS nom_territoire FROM {qtable(SCHEMA, CONSUMPTION_VIEW)} WHERE niveau::text = %s AND code_territoire::text = %s AND year IS NOT NULL ORDER BY year::integer """ data = rows(sql, (level, str(territoire))) except Exception: return None if not data: return None series = [ {"year": int(row[0]), "value": round(float(row[1] or 0), 3), "period_label": row[2]} for row in data if row[0] is not None ] if not series: return None total_ha = round(sum(float(item["value"] or 0) for item in series), 3) annual_average_ha = round(total_ha / len(series), 3) if series else 0.0 recent_window = max(1, CONSUMPTION_DEFAULT_RECENT_YEARS) recent_items = series[-recent_window:] recent_ha = round(sum(float(item["value"] or 0) for item in recent_items), 3) recent_period = None if recent_items: recent_period = f"{recent_items[0]['year']}-{recent_items[-1]['year'] + 1}" source = data[-1][3] or "Cerema - Portail national de l’artificialisation" millesime = data[-1][4] or "2009-2024" name = data[-1][5] destinations = cerema_typologies_from_view(territoire, level) return { "available": True, "ready": True, "status": "postgis_cache", "mode": "postgis_cache", "official": True, "source_id": "cerema_portail_artificialisation", "source": source, "producer": "Cerema", "provider": "Cerema", "period": f"{series[0]['year']}-{series[-1]['year'] + 1}", "millesime": millesime, "unit": "ha", "total_ha": total_ha, "annual_average_ha": annual_average_ha, "recent_ha": recent_ha, "recent_period": recent_period, "series": series, "destinations": destinations, "by_sector": destinations, "territory": {"code": str(territoire), "name": name, "level": level}, "methodology": { "producer": "Cerema", "portal": "Portail national de l'artificialisation des sols", "base": "Fichiers Fonciers retraités pour le suivi de la consommation d'espaces naturels, agricoles et forestiers.", "definition": "Consommation d'espaces naturels, agricoles et forestiers transformés vers des espaces urbanisés/artificialises, exprimée en hectares.", "warning": "La consommation ENAF est différente de l'emprise bâtie et du nombre de bâtiments construits.", }, "sources": [ {"label": "Consommation ENAF", "source": source, "producer": "Cerema", "millesime": millesime, "status": "postgis_cache"} ], } def cerema_typologies_from_view(territoire: str, niveau: str | None = None) -> list[dict[str, Any]]: """Agrège les champs Cerema par typologie d'usage de la consommation ENAF.""" if not CONSUMPTION_USE_CEREMA or not CONSUMPTION_TYPOLOGY_VIEW or not pg_configured(): return [] level = normalize_level(niveau) try: sql = f""" SELECT label::text AS label, coalesce(sum(value_ha), 0)::double precision AS value_ha FROM {qtable(SCHEMA, CONSUMPTION_TYPOLOGY_VIEW)} WHERE niveau::text = %s AND code_territoire::text = %s GROUP BY label ORDER BY value_ha DESC """ data = rows(sql, (level, str(territoire))) except Exception: return [] return [ {"label": row[0], "value_ha": round(float(row[1] or 0), 3)} for row in data if row[0] is not None and float(row[1] or 0) > 0 ] def cerema_typology_comparison(territoire: str, niveau: str | None, consumption: dict[str, Any] | None = None) -> dict[str, Any] | None: """Remplace temporairement l'écran OCS2D par une lecture Cerema par destination.""" sectors = [] if consumption: sectors = consumption.get("destinations") or consumption.get("by_sector") or [] if not sectors: sectors = cerema_typologies_from_view(territoire, niveau) if not sectors: return None total = sum(float(item.get("value_ha") or item.get("value") or 0) for item in sectors) def sector_value(*needles: str) -> float: value = 0.0 for item in sectors: label = str(item.get("label") or "").lower() if any(n in label for n in needles): value += float(item.get("value_ha") or item.get("value") or 0) return round(value, 3) habitat = sector_value("habitat") activites = sector_value("activ") infra = sector_value("rout", "ferr", "infrastructure") mixte = sector_value("mix", "inconnu", "non class") categories = [ {"label": item.get("label"), "value_ha": round(float(item.get("value_ha") or item.get("value") or 0), 3)} for item in sectors ] categories = [item for item in categories if item["value_ha"] > 0] mutations = [ {"label": item["label"], "value_ha": item["value_ha"], "message": "Destination Cerema de la consommation ENAF."} for item in categories ] return { "ready": True, "status": "postgis_cache", "source_type": "cerema_typologies", "source": "Cerema - Portail national de l'artificialisation", "producer": "Cerema", "years": {"old": 2009, "new": 2024}, "categories": categories, "mutations": mutations, "urbanized_ha": habitat, "agricultural_ha": activites, "natural_ha": infra, "naf_to_urban_ha": round(total, 3), "mixte_ha": mixte, "methodology": { "principle": "Lecture par destination de la consommation ENAF issue des champs typologiques Cerema.", "warning": "Cette lecture ne remplace pas un vrai millésime OCS2D : elle indique à quoi a été destinée la consommation d'espaces, pas l'ensemble des changements d'occupation du sol.", }, } def demo_consumption(timeline: list[dict[str, Any]], external_status: dict[str, Any] | None = None) -> dict[str, Any]: # Série fictive inspirée du rythme de construction démo. À remplacer par ENAF Cerema. values: list[dict[str, Any]] = [] for year in range(2009, 2025): stat = next((item for item in timeline if item["year"] == year), {}) built_ha = float(stat.get("surface_nouvelle_m2") or 0) / 10000 baseline = max(0.9, 4.2 - (year - 2009) * 0.16) value = round(baseline + built_ha * 2.6, 1) values.append({"year": year, "value": value}) total = round(sum(item["value"] for item in values), 1) recent = round(sum(item["value"] for item in values[-5:]), 1) avg = round(total / len(values), 1) if values else 0 status = external_status or {} warnings = ["Valeurs fictives de démonstration : ne pas utiliser comme chiffres réels."] if status.get("status"): warnings.append(f"Cache Cerema non utilisé : {status.get('status')}.") return { "ready": False, "status": "demo_fallback", "official": False, "source_id": "demo", "source": "Démo locale - cache Cerema à rafraîchir", "producer": "AULA / démonstration", "period": "2009-2024", "unit": "ha", "total_ha": total, "annual_average_ha": avg, "recent_ha": recent, "recent_period": "2020-2024", "series": values, "destinations": [], "external_status": status, "methodology": status.get("methodology") or { "producer": "Cerema", "portal": "Portail national de l'artificialisation des sols", "base": "Fichiers Fonciers retraités pour le Portail national de l'artificialisation", "limits": ["La consommation d'ENAF est distincte de l'emprise bâtie."], }, "warnings": warnings, } def official_or_demo_consumption( territoire: str, niveau: str, timeline: list[dict[str, Any]], *, live: bool = False, ) -> dict[str, Any]: """Renvoie la consommation Cerema depuis PostGIS si disponible, sinon conserve la démo.""" if CONSUMPTION_SOURCE_MODE == "demo": return demo_consumption(timeline) cached_pg = consumption_from_view(territoire, niveau) if cached_pg: return cached_pg if artificialisation is None: return demo_consumption(timeline, { "status": "conso_postgis_absente", "methodology": { "warning": "La vue PostGIS v_conso_enaf_cerema n'a pas renvoyé de données pour ce territoire." }, }) mode = os.environ.get("FONCIER_CONSUMPTION_SOURCE", "cache").lower() if mode == "demo": return demo_consumption(timeline) allow_live = ( live or mode == "cerema" or os.environ.get("FONCIER_CEREMA_AUTO_LIVE", "false").lower() in {"1", "true", "yes"} ) try: official = artificialisation.consumption_for_territory( territoire, normalize_level(niveau), live=allow_live, force=False, ) if official.get("available") or official.get("series"): official["official"] = True official["ready"] = True official["available"] = True official["source_id"] = "cerema_portail_artificialisation" official.setdefault("source_mode", official.get("mode") or "cerema_cache_or_live") return official return demo_consumption(timeline, { "status": official.get("status") or official.get("source_mode") or "cache_absent", "methodology": { **(official.get("methodology") or {}), "refresh_command": official.get("refresh_command") or f"python scripts/refresh_external_cache.py --source artificialisation --level {normalize_level(niveau)} --force", "live_endpoint": f"/api/external/artificialisation/consumption?territoire={territoire}&niveau={normalize_level(niveau)}&live=true", "warning": (official.get("methodology") or {}).get("warning") or "Le bloc conserve des valeurs fictives tant que le cache Cerema n'est pas rafraîchi.", }, }) except Exception as exc: # noqa: BLE001 - l'application doit rester utilisable hors ligne return demo_consumption(timeline, { "status": "erreur_cerema", "methodology": { "definition": "Consommation d'espaces naturels, agricoles et forestiers.", "message": f"Le connecteur Cerema n'a pas répondu : {exc}", "warning": "Le bloc reste affiché avec des valeurs fictives de démonstration.", }, }) def narrative_chapters() -> list[dict[str, Any]]: return [ { "id": "observer", "label": "Observer", "target": "mapSection", "question": "Comment le territoire s'est-il bâti depuis 1970 ?", "summary": "Carte temporelle du bâti, rythme annuel, usages et décennies d'accélération.", }, { "id": "comprendre", "label": "Comprendre", "target": "socioSection", "question": "Quels moteurs socio-démographiques expliquent cette évolution ?", "summary": "Population, ménages, logements et densité issus des recensements INSEE.", }, { "id": "mesurer", "label": "Mesurer", "target": "consumptionSection", "question": "Combien d'espace naturel, agricole ou forestier a été consommé ?", "summary": "Consommation ENAF issue du Cerema / Portail national de l'artificialisation.", }, { "id": "comparer", "label": "Comparer", "target": "ocs2dSection", "question": "Comment l'occupation du sol a-t-elle changé ?", "summary": "Comparaison visuelle OCS2D entre deux millésimes et lecture des mutations.", }, { "id": "anticiper", "label": "Anticiper", "target": "anticipateSection", "question": "Que disent les trajectoires démographiques et résidentielles ?", "summary": "Volet en cours de développement : Omphale EPCI, besoins logements et pression foncière tendancielle.", }, { "id": "agir", "label": "Agir", "target": "actionSection", "question": "Où peut-on réduire la pression foncière ?", "summary": "Volet en cours de développement : densification, renouvellement urbain, friches et potentiels de recyclage.", }, ] def demo_ocs2d(territoire: str, niveau: str | None = None) -> dict[str, Any]: level = normalize_level(niveau) modifier = 0.58 if level == "commune" else 1.0 code_shift = sum(ord(c) for c in str(territoire)) % 7 base = 1 + code_shift / 30 years = {"old": 2005, "new": 2021} categories = [ {"label": "Espaces urbanisés", "old_ha": round(820 * modifier * base, 1), "new_ha": round(1015 * modifier * base, 1)}, {"label": "Espaces agricoles", "old_ha": round(1380 * modifier * base, 1), "new_ha": round(1198 * modifier * base, 1)}, {"label": "Espaces naturels", "old_ha": round(310 * modifier * base, 1), "new_ha": round(286 * modifier * base, 1)}, {"label": "Espaces en eau / autres", "old_ha": round(74 * modifier * base, 1), "new_ha": round(85 * modifier * base, 1)}, ] mutations = [ {"label": "Agricole → urbain", "value_ha": round(122 * modifier * base, 1), "message": "Extension urbaine à objectiver par les millésimes OCS2D."}, {"label": "Naturel → urbain", "value_ha": round(26 * modifier * base, 1), "message": "Mutation sensible à vérifier avec les couches locales."}, {"label": "Urbain → renaturé / autre", "value_ha": round(9 * modifier * base, 1), "message": "Signal faible, utile pour repérer les recompositions."}, ] return { "ready": False, "status": "demo_fallback", "source": "OCS2D Hauts-de-France / Geo2France - à brancher sur les millésimes AULA", "producer": "Région Hauts-de-France / partenaires OCS2D", "years": years, "categories": categories, "mutations": mutations, "urbanized_ha": categories[0]["new_ha"], "agricultural_ha": categories[1]["new_ha"], "natural_ha": categories[2]["new_ha"], "naf_to_urban_ha": round(mutations[0]["value_ha"] + mutations[1]["value_ha"], 1), "methodology": { "principle": "Comparer deux millésimes OCS2D pour montrer les changements d'occupation du sol.", "warning": "Valeurs fictives dans le prototype. En production, brancher les millésimes OCS2D, harmoniser la nomenclature et documenter les règles de mutation.", }, } def demo_omphale(territoire: str, niveau: str | None, socio: dict[str, Any], consumption: dict[str, Any]) -> dict[str, Any]: level = normalize_level(niveau) current = int(socio.get("population_current") or 0) projected = int(socio.get("population_projection") or current) delta = projected - current household_size = 2.15 if level == "epci" else 2.08 housing_need = max(0, round(abs(delta) / household_size + (socio.get("households") or 0) * 0.015)) recent_ha = float(consumption.get("recent_ha") or 0) pressure = "modérée" if recent_ha > 35 or housing_need > 1800: pressure = "élevée" elif recent_ha < 12 and housing_need < 600: pressure = "faible" series = socio.get("series") or [] return { "source": "INSEE OMPHALE - données EPCI à intégrer", "scale": "EPCI" if level == "epci" else "EPCI de référence, commune replacée dans la trajectoire intercommunale", "scenario": "central", "projection_year": socio.get("projection_year") or 2050, "population_current": current, "population_projection": projected, "population_delta": delta, "housing_need_estimate": int(housing_need), "land_pressure": pressure, "series": series, "hypotheses": [ "Projection démographique de référence à l'échelle EPCI.", "Besoin logements calculé ici comme ordre de grandeur de démonstration.", "La trajectoire foncière doit croiser population, ménages, logements autorisés et densité des opérations.", ], "warning": "Les projections Omphale ne sont pas des certitudes. Elles doivent être affichées avec scénario, millésime, horizon et échelle. À la commune, on n'affiche pas une projection Omphale directe.", } def demo_action_levers(territoire: str, niveau: str | None, timeline: list[dict[str, Any]], consumption: dict[str, Any]) -> dict[str, Any]: last10 = [item for item in timeline if int(item.get("year") or 0) >= END_YEAR - 9] recent_buildings = sum(int(item.get("batiments_nouveaux") or 0) for item in last10) recent_surface_ha = sum(float(item.get("surface_nouvelle_m2") or 0) for item in last10) / 10000 density_recent = round(recent_buildings / recent_surface_ha, 1) if recent_surface_ha else 0 density_target = max(18, round(density_recent * 1.35, 1)) if density_recent else 24 modifier = 0.62 if normalize_level(niveau) == "commune" else 1.0 brownfields = [ {"name": "Ancien site d'activité", "type": "activité", "surface_ha": round(3.4 * modifier, 1), "mobilisability": "à étudier", "message": "Potentiel de renouvellement sous réserve de contraintes techniques."}, {"name": "Friche commerciale", "type": "commerce", "surface_ha": round(1.8 * modifier, 1), "mobilisability": "mobilisable", "message": "Site intéressant pour limiter l'extension en périphérie."}, {"name": "Emprise ferroviaire / technique", "type": "infrastructure", "surface_ha": round(2.6 * modifier, 1), "mobilisability": "complexe", "message": "Potentiel long terme, à croiser avec accès et pollution."}, ] total_brownfield = round(sum(item["surface_ha"] for item in brownfields), 1) potential_units = round(total_brownfield * density_target) levers = [ {"label": "Densifier les opérations récentes", "priority": "fort", "message": f"Passer d'environ {density_recent or 'n.c.'} à {density_target} bâtiments/ha sur les secteurs adaptés."}, {"label": "Recycler les friches recensées", "priority": "fort", "message": f"{len(brownfields)} sites de démonstration représentent {total_brownfield} ha de potentiel à qualifier."}, {"label": "Prioriser le déjà urbanisé", "priority": "moyen", "message": "Croiser friches, centralités, vacance, réseaux et contraintes pour limiter la consommation ENAF."}, ] return { "density_recent_buildings_per_ha": density_recent, "density_target_buildings_per_ha": density_target, "brownfields_count": len(brownfields), "brownfields_area_ha": total_brownfield, "brownfields": brownfields, "potential_units": int(potential_units), "levers": levers, "source": "Recensements friches AULA - à brancher sur la base métier", "methodology": "Prototype : les friches sont fictives. En production, brancher les recensements AULA avec statut, contraintes, surface mobilisable et date de mise à jour.", } def demo_priorities(territoire: str, niveau: str | None, consumption: dict[str, Any], action_levers: dict[str, Any], ocs2d: dict[str, Any]) -> dict[str, Any]: conso = float(consumption.get("recent_ha") or 0) friches = float(action_levers.get("brownfields_area_ha") or 0) mutation = float(ocs2d.get("naf_to_urban_ha") or 0) density = float(action_levers.get("density_recent_buildings_per_ha") or 0) score = min(100, max(0, round(46 + mutation * 0.12 + friches * 2.6 - conso * 0.35 + max(0, 25 - density) * 0.9))) profiles = [ {"label": "Forte pression / marges internes", "level": "priorité 1", "message": "À cibler pour renouvellement urbain, friches et formes plus compactes."}, {"label": "Consommation élevée / densité faible", "level": "vigilance", "message": "À regarder pour faire évoluer les formes urbaines et limiter l'étalement."}, {"label": "Potentiel friches / centralités", "level": "opportunité", "message": "À mobiliser en priorité avant extension."}, ] return { "score": score, "label": "potentiel élevé" if score >= 70 else ("potentiel à confirmer" if score >= 45 else "potentiel limité"), "message": "Score de démonstration combinant consommation récente, mutations OCS2D, densité et friches recensées.", "profiles": profiles, "warning": "Indice expérimental : il doit rester explicable, paramétrable et validé métier avant usage décisionnel.", } def build_insights( timeline: list[dict[str, Any]], overview: dict[str, Any], consumption: dict[str, Any] | None = None, ocs2d: dict[str, Any] | None = None, action_levers: dict[str, Any] | None = None, ) -> list[str]: peak_year = overview.get("pic_annee") peak_value = overview.get("pic_batiments") total = overview.get("batiments_cumules") or 0 surface_ha = (overview.get("surface_batie_cumulee_m2") or 0) / 10000 conso_recent = (consumption or {}).get("recent_ha") mutation = (ocs2d or {}).get("naf_to_urban_ha") friches = (action_levers or {}).get("brownfields_area_ha") return [ f"Le récit commence par l'apparition du bâti depuis {START_YEAR} pour repérer les périodes et secteurs d'accélération.", f"Dans le jeu courant, le pic observé est {peak_year} avec {peak_value} nouveaux bâtiments ; l'année finale compte {total:,} bâtiments et {surface_ha:.1f} ha d'emprise bâtie.".replace(',', ' '), f"La consommation ENAF récente affichée est de {conso_recent} ha et les mutations OCS2D NAF → urbain représentent {mutation} ha dans le jeu de démonstration.", f"Les leviers d'action portent sur la densité, le renouvellement urbain et les friches recensées, avec {friches} ha de friches fictives à remplacer par les recensements AULA.", "Chaque chiffre doit rester rattaché à sa source, son millésime et son échelle : bâtiments, ENAF Cerema, OCS2D, Omphale et friches ne mesurent pas la même chose.", ] @app.get("/") def root(): return FileResponse(STATIC_DIR / "index.html") @app.get("/3d") def map_3d(): return FileResponse(STATIC_DIR / "3d.html") @app.get("/api/health") def health(): return { "ok": True, "mode": "postgis" if pg_configured() else "demo", "start_year": START_YEAR, "end_year": END_YEAR, "buildings_view": qtable(SCHEMA, BUILDINGS_VIEW), "bdnb_view": qtable(SCHEMA, BDNB_VIEW), "bdnb_height_col": BDNB_HEIGHT_COL, "static": str(STATIC_DIR), } def slugify_for_url(value: str | None) -> str: text = unicodedata.normalize("NFD", str(value or "")) text = "".join(ch for ch in text if unicodedata.category(ch) != "Mn") text = text.replace("œ", "oe").replace("Œ", "oe").replace("æ", "ae").replace("Æ", "ae") text = re.sub(r"[^A-Za-z0-9]+", "-", text).strip("-").lower() text = re.sub(r"-+", "-", text) return text def banatic_commune_url(code: str | None, name: str | None) -> str | None: code_text = str(code or "").strip() if not re.fullmatch(r"\d{5}", code_text): return None cleaned_name = re.sub(r"^\d{5}\s*[-–—:]?\s*", "", str(name or "")).strip() cleaned_name = re.sub(r"\s*\(\s*\d{5}\s*\)\s*$", "", cleaned_name).strip() slug = slugify_for_url(cleaned_name) if not slug or slug.isdigit(): return None return f"https://www.banatic.interieur.gouv.fr/commune/{code_text}-{slug}#niveau_vie" @app.get("/api/story") def story(): return { "title": "Foncier narratif", "subtitle": "Comprendre l'évolution du foncier et identifier les leviers d'action", "question": "Comment le territoire s'est développé, pourquoi il a consommé de l'espace, et quels leviers seront étudiés ensuite ?", "source": "Démo locale - à remplacer par Fichiers fonciers, OCS2D, Omphale EPCI, Cerema/artificialisation et recensements friches AULA", "start_year": START_YEAR, "end_year": END_YEAR, "default_level": DEFAULT_LEVEL, "chapters": narrative_chapters(), "method_url": "/api/narrative/methodology", } @app.get("/api/territoires") def territoires(niveau: str = Query(DEFAULT_LEVEL, pattern="^(commune|epci)$")): level = normalize_level(niveau) if pg_configured(): if level == "commune": sql = f""" SELECT DISTINCT code_insee::text, commune::text FROM {qtable(SCHEMA, BUILDINGS_VIEW)} WHERE code_insee IS NOT NULL AND commune IS NOT NULL ORDER BY commune """ else: sql = f""" SELECT code_territoire::text, nom_territoire::text FROM {qtable(SCHEMA, TERRITORIES_VIEW)} ORDER BY nom_territoire """ return [ { "code": c, "nom": n, "niveau": level, "banatic_url": banatic_commune_url(c, n) if level == "commune" else None, } for c, n in rows(sql) ] return demo_territories(level) @app.get("/api/bounds") def territory_bounds( territoire: str = Query("62001"), niveau: str = Query(DEFAULT_LEVEL, pattern="^(commune|epci)$"), ): """Renvoie l'emprise du territoire sélectionné en WGS84 pour caler la carte. On utilise toutes les géométries disponibles du territoire, indépendamment de l'année affichée. Cela évite une carte vide au démarrage lorsqu'aucun bâtiment n'apparaît exactement à l'année de départ. """ if pg_configured(): sql = f""" SELECT ST_XMin(e)::double precision AS minx, ST_YMin(e)::double precision AS miny, ST_XMax(e)::double precision AS maxx, ST_YMax(e)::double precision AS maxy, nb::bigint AS nb_geometries FROM ( SELECT ST_Extent(ST_Transform({qident(GEOM_COL)}, 4326)) AS e, count(*) AS nb FROM {qtable(SCHEMA, BUILDINGS_VIEW)} WHERE {qident(scope_column(niveau))} = %s AND {qident(GEOM_COL)} IS NOT NULL ) src WHERE e IS NOT NULL """ try: result = rows(sql, (territoire,)) if not result or result[0][0] is None: return {"available": False, "territoire": territoire, "niveau": normalize_level(niveau), "bounds": None, "nb_geometries": 0} minx, miny, maxx, maxy, nb = result[0] return { "available": True, "territoire": territoire, "niveau": normalize_level(niveau), "bounds": [[float(miny), float(minx)], [float(maxy), float(maxx)]], "nb_geometries": int(nb or 0), } except Exception as exc: raise HTTPException(500, detail=str(exc)) from exc features = filter_demo_features(territoire, niveau) coords = [] def walk(value): if isinstance(value, list) and value and isinstance(value[0], (int, float)): coords.append(value) elif isinstance(value, list): for item in value: walk(item) for feature in features: walk(feature.get("geometry", {}).get("coordinates", [])) if not coords: return {"available": False, "territoire": territoire, "niveau": normalize_level(niveau), "bounds": None, "nb_geometries": 0} xs = [c[0] for c in coords] ys = [c[1] for c in coords] return {"available": True, "territoire": territoire, "niveau": normalize_level(niveau), "bounds": [[min(ys), min(xs)], [max(ys), max(xs)]], "nb_geometries": len(features)} @app.get("/api/timeline/stats") def timeline_stats(territoire: str = Query("62001"), niveau: str = Query(DEFAULT_LEVEL, pattern="^(commune|epci)$")): if pg_configured(): sql = f""" WITH years AS ( SELECT generate_series(%s::integer, %s::integer) AS year ), annual AS ( SELECT annee_apparition::int AS year, {count_sql_expr()} AS nouveaux, coalesce(sum(surface_emprise_m2), 0)::double precision AS surface_nouvelle_m2 FROM {qtable(SCHEMA, BUILDINGS_VIEW)} WHERE {qident(scope_column(niveau))} = %s AND annee_apparition BETWEEN %s::integer AND %s::integer GROUP BY annee_apparition::int ) SELECT y.year, coalesce(a.nouveaux, 0) AS nouveaux, sum(coalesce(a.nouveaux, 0)) OVER (ORDER BY y.year) AS cumules, coalesce(a.surface_nouvelle_m2, 0) AS surface_nouvelle_m2, sum(coalesce(a.surface_nouvelle_m2, 0)) OVER (ORDER BY y.year) AS surface_cumulee_m2 FROM years y LEFT JOIN annual a ON a.year = y.year ORDER BY y.year """ data = rows(sql, (START_YEAR, END_YEAR, territoire, START_YEAR, END_YEAR)) return [ { "year": int(y), "batiments_nouveaux": int(n), "batiments_cumules": int(c), "surface_nouvelle_m2": float(sn), "surface_cumulee_m2": float(sc), } for y, n, c, sn, sc in data ] return timeline_for_features(filter_demo_features(territoire, niveau)) @app.get("/api/batiments") def batiments( territoire: str = Query("62001"), niveau: str = Query(DEFAULT_LEVEL, pattern="^(commune|epci)$"), year: int = Query(1970, ge=1800, le=2100), mode: str = Query("cumul", pattern="^(cumul|nouveaux)$"), ): if pg_configured(): year_filter = "(annee_apparition IS NULL OR annee_apparition <= %s::integer)" if mode == "cumul" else "annee_apparition = %s::integer" sql = f""" SELECT json_build_object( 'type', 'FeatureCollection', 'features', coalesce(json_agg(json_build_object( 'type', 'Feature', 'properties', json_build_object( 'id_batiment', id_batiment, 'annee_apparition', annee_apparition, 'decennie', CASE WHEN annee_apparition IS NULL THEN NULL WHEN annee_apparition <= 1970 THEN 1970 ELSE (floor(annee_apparition::numeric / 10)::int * 10) END, 'decennie_label', CASE WHEN annee_apparition IS NULL THEN 'Non daté' WHEN annee_apparition <= 1970 THEN 'Avant 1970' ELSE ((floor(annee_apparition::numeric / 10)::int * 10)::text || 's') END, 'surface_emprise_m2', surface_emprise_m2, 'usage', usage, 'hauteur_m', hauteur_m, 'altitude_sol_m', altitude_sol_m, 'source_geom', source_geom, 'fictive_geom_cstr', fictive_geom_cstr, 'match_ff_quality', match_ff_quality {feature_weight_property()}, 'source_annee', source_annee, 'qualite_annee', qualite_annee ), 'geometry', ST_AsGeoJSON(ST_Transform({qident(GEOM_COL)}, 4326))::json )), '[]'::json) )::text FROM {qtable(SCHEMA, BUILDINGS_VIEW)} WHERE {qident(scope_column(niveau))} = %s AND {year_filter} AND {qident(GEOM_COL)} IS NOT NULL """ try: value = rows(sql, (territoire, year))[0][0] return json.loads(value) except Exception as exc: raise HTTPException(500, detail=str(exc)) from exc feats = [] for feature in filter_demo_features(territoire, niveau): y = feature_year(feature) if y is None: continue if (mode == "cumul" and y <= year) or (mode == "nouveaux" and y == year): feats.append(feature) return {"type": "FeatureCollection", "features": feats} @app.get("/api/bdnb/buildings3d") def bdnb_buildings3d( territoire: str = Query("62001"), niveau: str = Query(DEFAULT_LEVEL, pattern="^(commune|epci)$"), year: int = Query(2025, ge=1800, le=2100), limit: int = Query(8000, ge=1, le=50000), ): """GeoJSON simplifié pour une extrusion 3D MapLibre. En production EPCI/Agence, privilégier des tuiles vectorielles ou PMTiles. Ce endpoint sert à valider rapidement le potentiel 3D à partir du champ hauteur. """ if not pg_configured(): feats = [] for feature in filter_demo_features(territoire, niveau): y = feature_year(feature) if y is not None and y <= year: demo = json.loads(json.dumps(feature)) demo.setdefault("properties", {})["hauteur_m"] = demo["properties"].get("hauteur_m") or 8 feats.append(demo) return {"type": "FeatureCollection", "features": feats, "mode": "demo"} if not BDNB_HEIGHT_COL: raise HTTPException(500, detail="FONCIER_BDNB_HEIGHT_COL n'est pas configuré") # La 3D s'appuie sur la même vue que l'application principale. # Cela évite les erreurs lorsque FONCIER_GEOM_COL=geom_web n'existe que # dans v_batiments_historique_app, et pas dans la vue BDNB brute. source_view = BUILDINGS_VIEW height_col = BDNB_HEIGHT_COL or "hauteur_m" year_col = BDNB_YEAR_COL or "annee_apparition" where_year = f"AND ({qident(year_col)} IS NULL OR {qident(year_col)} <= %s::integer)" params: list[Any] = [territoire, year, limit] sql = f""" SELECT json_build_object( 'type', 'FeatureCollection', 'features', coalesce(json_agg(json_build_object( 'type', 'Feature', 'properties', json_build_object( 'id_batiment', id_batiment, 'annee_apparition', {qident(year_col)}, 'decennie', CASE WHEN {qident(year_col)} IS NULL THEN NULL WHEN {qident(year_col)} <= 1970 THEN 1970 ELSE (floor({qident(year_col)}::numeric / 10)::int * 10) END, 'decennie_label', CASE WHEN {qident(year_col)} IS NULL THEN 'Non daté' WHEN {qident(year_col)} <= 1970 THEN 'Avant 1970' ELSE ((floor({qident(year_col)}::numeric / 10)::int * 10)::text || 's') END, 'hauteur_m', greatest(coalesce({qident(height_col)}, 3), 1), 'surface_emprise_m2', surface_emprise_m2, 'usage', usage, 'source', 'BDNB + Fichiers fonciers' ), 'geometry', ST_AsGeoJSON(ST_Transform(ST_SimplifyPreserveTopology({qident(GEOM_COL)}, 0.00001), 4326))::json )), '[]'::json) )::text FROM ( SELECT * FROM {qtable(SCHEMA, source_view)} WHERE {qident(scope_column(niveau))} = %s {where_year} AND {qident(GEOM_COL)} IS NOT NULL ORDER BY coalesce({qident(height_col)}, 0) DESC NULLS LAST LIMIT %s ) src """ try: value = rows(sql, tuple(params))[0][0] data = json.loads(value) data["mode"] = "postgis" data["view"] = qtable(SCHEMA, source_view) data["height_col"] = height_col data["warning"] = "GeoJSON de démonstration. Pour un EPCI complet, passer en tuiles vectorielles ou PMTiles." return data except Exception as exc: raise HTTPException(500, detail=str(exc)) from exc @app.get("/api/indicators") def indicators( territoire: str = Query("62001"), niveau: str = Query(DEFAULT_LEVEL, pattern="^(commune|epci)$"), conso_live: bool = Query(False, description="Interroger la couche Cerema/ArcGIS si le cache local est absent."), ): timeline = timeline_stats(territoire, niveau) last = timeline[-1] if timeline else {} peak = max(timeline, key=lambda item: item.get("batiments_nouveaux", 0), default={}) overview = { "batiments_cumules": int(last.get("batiments_cumules") or 0), "surface_batie_cumulee_m2": float(last.get("surface_cumulee_m2") or 0), "pic_annee": peak.get("year"), "pic_batiments": int(peak.get("batiments_nouveaux") or 0), } try: usages = usage_stats_for_pg(territoire, niveau) if pg_configured() else usage_stats_for_demo(territoire, niveau) except Exception: usages = usage_stats_for_demo("demo") consumption = official_or_demo_consumption(territoire, niveau, timeline, live=conso_live) sociodemo = official_or_demo_sociodemo(territoire, niveau) ocs2d = cerema_typology_comparison(territoire, niveau, consumption) or demo_ocs2d(territoire, niveau) omphale = demo_omphale(territoire, niveau, sociodemo, consumption) action_levers = demo_action_levers(territoire, niveau, timeline, consumption) priorities = demo_priorities(territoire, niveau, consumption, action_levers, ocs2d) return { "territoire": territoire, "niveau": normalize_level(niveau), "chapters": narrative_chapters(), "overview": overview, "timeline": timeline, "decades": decade_stats(timeline), "usages": usages, "sociodemo": sociodemo, "consumption": consumption, "ocs2d": ocs2d, "omphale": omphale, "action_levers": action_levers, "priorities": priorities, "insights": build_insights(timeline, overview, consumption, ocs2d, action_levers), "api_sources": source_catalog(), "api_cache_status_url": "/api/external/status", "agency_scope_url": "/api/agency/scope", "external_blocks_ready": [ "socio-démo INSEE Melodi", "projection OMPHALE EPCI importée / fichier interne", "consommation ENAF Cerema / portail artificialisation", "comparaisons OCS2D par millésime", "friches issues des recensements AULA", "Sitadel logements et locaux", "Géorisques", "référentiel communes/EPCI API Géo", ], } @app.get("/api/narrative/methodology") def narrative_methodology(): return { "goal": "Expliquer l'évolution du foncier à partir du bâti, de la socio-démographie, de la consommation d'espace, de l'OCS2D et des leviers d'action.", "chapters": narrative_chapters(), "source_rules": [ {"block": "Bâti depuis 1970", "source": "Fichiers fonciers / traitements AULA", "limit": "Année de construction à qualifier."}, {"block": "Socio-démo", "source": "INSEE RP", "limit": "Données observées, millésime à afficher."}, {"block": "Omphale", "source": "INSEE OMPHALE / fichier AULA à l'EPCI", "limit": "Projection EPCI, pas projection communale directe."}, {"block": "Consommation ENAF", "source": "Cerema / Portail national de l'artificialisation", "limit": "Mesure différente de l'emprise bâtie et de l'OCS2D."}, {"block": "OCS2D", "source": "Millésimes OCS2D Hauts-de-France", "limit": "Comparer des nomenclatures harmonisées couvert / usage."}, {"block": "Friches", "source": "Recensements AULA", "limit": "Donnée métier à dater, qualifier et mettre à jour."}, ], } @app.get("/api/agency/scope") def agency_scope_endpoint( live: bool = Query(False, description="Appeler l'API Geo si le cache n'existe pas ou doit être rafraîchi."), force: bool = Query(False, description="Forcer le rafraîchissement du cache."), ): if geo_api is None: return load_scope() try: return geo_api.agency_scope(live=live, force=force) except ExternalApiError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc @app.get("/api/external/sources") def external_sources(): return source_catalog() @app.get("/api/external/status") def external_status(): names = ["agency_communes"] names += [f"geo_epci_{code}" for code in agency_epci_codes()] names += [f"geo_epci_{code}_communes" for code in agency_epci_codes()] names += [ "datagouv_dataset_consommation-despaces-naturels-agricoles-et-forestiers", "cerema_artificialisation_layer_info", "datagouv_dataset_liste-des-permis-de-construire-et-autres-autorisations-durbanisme", "insee_melodi_catalog_all", ] result = { "scope": load_scope(), "caches": cache_status(names), } if artificialisation is not None: try: result["artificialisation"] = artificialisation.local_status() except Exception as exc: # noqa: BLE001 result["artificialisation"] = {"status": "error", "error": str(exc)} return result @app.get("/api/external/artificialisation/metadata") def artificialisation_metadata(force: bool = Query(False)): if artificialisation is None: raise HTTPException(status_code=503, detail="Connecteur artificialisation non disponible") try: return artificialisation.metadata(force=force) except ExternalApiError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc @app.get("/api/consumption") def consumption_endpoint( territoire: str = Query("62001"), niveau: str = Query(DEFAULT_LEVEL, pattern="^(commune|epci)$"), live: bool = Query(False, description="Conservé pour compatibilité. La V1 lit d'abord PostGIS."), ): timeline = timeline_stats(territoire, niveau) return official_or_demo_consumption(territoire, niveau, timeline, live=live) @app.get("/api/external/artificialisation/status") def artificialisation_status(): if artificialisation is None: raise HTTPException(status_code=503, detail="Connecteur artificialisation non disponible") return artificialisation.local_status() @app.get("/api/external/artificialisation/cerema/layer") def artificialisation_cerema_layer(force: bool = Query(False)): if artificialisation is None: raise HTTPException(status_code=503, detail="Connecteur artificialisation non disponible") try: return artificialisation.cerema_layer_info(force=force) except ExternalApiError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc @app.get("/api/external/artificialisation/cerema") def artificialisation_cerema_summary( code: str = Query(..., description="Code INSEE commune ou SIREN EPCI"), niveau: str = Query(DEFAULT_LEVEL, pattern="^(commune|epci)$"), force: bool = Query(False), enforce_scope: bool = Query(True, description="Limiter au périmètre AULA configuré."), ): if artificialisation is None: raise HTTPException(status_code=503, detail="Connecteur artificialisation non disponible") try: return artificialisation.consumption_summary_for_territory( code, normalize_level(niveau), force=force, enforce_agency_scope=enforce_scope, ) except ExternalApiError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc @app.get("/api/external/artificialisation/cerema/rows") def artificialisation_cerema_rows( code: str = Query(..., description="Code INSEE commune ou SIREN EPCI"), niveau: str = Query(DEFAULT_LEVEL, pattern="^(commune|epci)$"), force: bool = Query(False), enforce_scope: bool = Query(True, description="Limiter au périmètre AULA configuré."), ): if artificialisation is None: raise HTTPException(status_code=503, detail="Connecteur artificialisation non disponible") try: return artificialisation.cerema_rows_for_territory( code, normalize_level(niveau), force=force, enforce_agency_scope=enforce_scope, ) except ExternalApiError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc @app.get("/api/external/artificialisation/consumption") def artificialisation_consumption( territoire: str = Query(...), niveau: str = Query(DEFAULT_LEVEL, pattern="^(commune|epci)$"), live: bool = Query(False, description="Déclenche un appel Cerema live si le cache n'existe pas."), force: bool = Query(False, description="Force le rafraîchissement du cache avant lecture."), ): if artificialisation is None: raise HTTPException(status_code=503, detail="Connecteur artificialisation non disponible") try: return artificialisation.consumption_for_territory(territoire, niveau, live=live or force, force=force) except ExternalApiError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc @app.get("/api/external/artificialisation/refresh") def artificialisation_refresh_get( level: str = Query("epci", pattern="^(commune|epci)$"), force: bool = Query(False), ): if artificialisation is None: raise HTTPException(status_code=503, detail="Connecteur artificialisation non disponible") try: return artificialisation.refresh_filtered_cache(level=normalize_level(level), force=force) except ExternalApiError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc @app.post("/api/external/artificialisation/refresh") def artificialisation_refresh_post( level: str = Query("epci", pattern="^(commune|epci)$"), force: bool = Query(False), ): return artificialisation_refresh_get(level=level, force=force) @app.get("/api/external/sitadel/metadata") def sitadel_metadata(force: bool = Query(False)): if sitadel is None: raise HTTPException(status_code=503, detail="Connecteur Sitadel non disponible") try: return sitadel.metadata(force=force) except ExternalApiError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc @app.get("/api/external/georisques/commune/{code_insee}") def georisques_commune(code_insee: str, force: bool = Query(False)): if georisques is None: raise HTTPException(status_code=503, detail="Connecteur Géorisques non disponible") try: risks = georisques.risks_for_communes([code_insee], force=force) pprn = georisques.pprn_for_commune(code_insee, force=force) return {"code_insee": code_insee, "risks": risks, "pprn": pprn} except ExternalApiError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc @app.get("/api/external/insee/catalog") def insee_catalog(q: str | None = Query(None), force: bool = Query(False)): if insee_melodi is None: raise HTTPException(status_code=503, detail="Connecteur INSEE non disponible") try: data = insee_melodi.catalog(force=force) if q: ql = q.lower() data = [item for item in data if ql in json.dumps(item, ensure_ascii=False).lower()] return {"count": len(data), "items": data[:200]} except ExternalApiError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc @app.get("/api/external/insee/population-reference") def insee_population_reference(epci: str = Query(..., description="Code SIREN EPCI"), force: bool = Query(False)): if epci not in agency_epci_codes(): raise HTTPException(status_code=400, detail="EPCI hors périmètre AULA configuré") if insee_melodi is None: raise HTTPException(status_code=503, detail="Connecteur INSEE non disponible") try: return insee_melodi.population_reference_for_epci(epci, force=force) except ExternalApiError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc @app.get("/api/external/insee/sociodemo") def insee_sociodemo_endpoint( territoire: str = Query(...), niveau: str = Query(DEFAULT_LEVEL, pattern="^(commune|epci)$"), ): """Bloc socio-démo utilisé par l'application. Priorité : vue/cache PostGIS -> cache JSON INSEE -> API publique légère -> démo explicite. """ return official_or_demo_sociodemo(territoire, niveau) @app.get("/api/open-data/sources") def open_data_sources(): if open_data_broker is None: return source_catalog() return open_data_broker.sources() @app.get("/api/open-data/aula-scope") def open_data_aula_scope(): if open_data_broker is None: return load_scope() return open_data_broker.scope() @app.get("/api/open-data/context") def open_data_context(territoire: str | None = Query(None)): if open_data_broker is None: return {"territoire": territoire, "sources": source_catalog()} return open_data_broker.context(territoire) @app.post("/api/open-data/refresh/geo") def open_data_refresh_geo(): if open_data_broker is None: raise HTTPException(503, detail="Open data broker non disponible") try: return open_data_broker.refresh_geo_scope() except Exception as exc: raise HTTPException(502, detail=f"API Geo refresh failed: {exc}") from exc