Spaces:
Running
Running
| """Fake Cube.js client for offline tests. | |
| Reproduces the behaviour of the real Indicateurs API observed on the live Space | |
| (tests/live_capture.json): metadata cube, source cube whose dimensions do NOT | |
| include `libelle`, data cubes named {theme}_{maille} with measures `id_{n}` and | |
| geographic dimensions geocode_*/libelle_*, string-typed numeric values, `limit`, | |
| `order`, `equals` / `contains` / `set` filters. As on the real API, breakdown | |
| columns (secteur, mode_transport…) and `date_mesure` are returned in every row | |
| even when not requested, and a dimension-only query returns distinct rows. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| from pathlib import Path | |
| from typing import Any | |
| from src.api_client import BadRequestError, CubeJsClient | |
| from src.models import REGION_CODES | |
| CAPTURE = json.loads((Path(__file__).parent / "live_capture.json").read_text(encoding="utf-8")) | |
| META_DIMS = [ | |
| "id", "libelle", "unite", "description", "methode_calcul", "annees_disponibles", | |
| "mailles_disponibles", "maille_mini_disponible", "couverture_geographique", | |
| "completion_region", "completion_departement", "completion_epci", "completion_commune", | |
| "thematique_fnv", | |
| ] | |
| # As on the real API: no `libelle` dimension on the source cube. | |
| SOURCE_DIMS = [ | |
| "id_indicateur", "nom_source", "description", "producteur_source", | |
| "distributeur_source", "license_source", "lien_page", "date_derniere_extraction", | |
| ] | |
| GEO_DIMS = { | |
| "commune": ["geocode_commune", "libelle_commune"], | |
| "epci": ["geocode_epci", "libelle_epci"], | |
| "departement": ["geocode_departement", "libelle_departement"], | |
| "region": ["geocode_region", "libelle_region"], | |
| } | |
| ALL_LEVELS = ["commune", "epci", "departement", "region"] | |
| # cube -> (levels, indicator ids) — from the captured available_cubes | |
| DATA_CUBES = { | |
| "conso_enaf_com": (ALL_LEVELS, [611, 545]), | |
| "stationnement_velo_habitant_com": (ALL_LEVELS, [94]), | |
| "surfaces_agricoles_com": (["commune"], [23]), | |
| "surface_bio_epci": (["epci", "region"], [23]), | |
| "surface_bio_dpt": (["departement", "region"], [23, 42]), | |
| "dechets_population_dpt": (["departement", "region"], [42]), | |
| "emission_ges_secteur_com": (ALL_LEVELS, [1]), | |
| "parts_modales_actifs_com": (ALL_LEVELS, [1053, 955]), | |
| # commune/EPCI indicator whose commune cube also carries département/région columns | |
| "climadiag_vague_chaleur_com": (ALL_LEVELS, [1298]), | |
| "climadiag_vague_chaleur_epci": (["epci"], [1298]), | |
| } | |
| YEARS = ["2020", "2021", "2022"] | |
| CUBE_YEARS = {"climadiag_vague_chaleur_com": ["2005", "2030", "2050", "2100"], | |
| "climadiag_vague_chaleur_epci": ["2005", "2030", "2050", "2100"]} | |
| # Breakdown columns, always returned by the API | |
| CUBE_AXES = { | |
| "emission_ges_secteur_com": {"secteur": ["Agriculture", "Déchets", "Energie", "Industrie hors Energie", | |
| "Résidentiel", "Tertiaire", "Transports"]}, | |
| "parts_modales_actifs_com": {"mode_transport": ["Marche", "Transports en commun", "Vélo (depuis 2017)", "Voiture"]}, | |
| } | |
| EXTRA_MEASURES = {"climadiag_vague_chaleur_com": ["scenario_bas", "scenario_haut"], | |
| "climadiag_vague_chaleur_epci": ["scenario_bas", "scenario_haut"]} | |
| # Metadata not in the capture (axes, summable…), as returned by the real API | |
| META_OVERRIDES: dict[int, dict[str, Any]] = { | |
| 1: {"noms_axes": "secteur", "summable": True, | |
| "valeur_axes": CUBE_AXES["emission_ges_secteur_com"], "annees_disponibles": "2020, 2021, 2022"}, | |
| 1053: {"libelle": "Part d'actifs selon le mode de transport principalement utilisé pour aller travailler", | |
| "unite": "%", "noms_axes": "mode_transport", "summable": False, | |
| "valeur_axes": CUBE_AXES["parts_modales_actifs_com"], "mailles_disponibles": "commune, epci, departement, region", | |
| "annees_disponibles": "2020, 2021, 2022"}, | |
| 955: {"noms_axes": "", "summable": False, "unite": "%", "annees_disponibles": "2020, 2021, 2022"}, | |
| 1298: {"libelle": "Nombre annuel de jours en vague de chaleur", "unite": "jours", "summable": False, | |
| "mailles_disponibles": "commune, epci", "annees_disponibles": "2005, 2030, 2050, 2100", | |
| "completion_epci": 0.82}, | |
| } | |
| # Synthetic territories (enough to exercise truncation on communes) | |
| METRO_REGIONS = [code for code in REGION_CODES if not code.startswith("0")] | |
| DEPARTEMENTS = dict.fromkeys( | |
| [(f"{i:02d}", f"Département {i:02d}") for i in range(1, 96) if i not in (13, 20, 22, 69)] | |
| + [("2A", "Corse-du-Sud"), ("2B", "Haute-Corse"), ("13", "Bouches-du-Rhône"), ("69", "Rhône"), | |
| ("22", "Côtes-d'Armor"), ("974", "La Réunion")] | |
| ) | |
| EPCIS = { | |
| "200046977": "Métropole de Lyon", | |
| "200054781": "Métropole du Grand Paris", | |
| "200054807": "Métropole d'Aix-Marseille-Provence", | |
| "243100518": "Toulouse Métropole", | |
| "243400017": "Montpellier Méditerranée Métropole", | |
| "200069409": "Communauté d'agglomération Saint-Brieuc Armor Agglomération", | |
| "249740119": "Communauté intercommunale du Nord de La Réunion", | |
| **{f"2000{i:05d}": f"Communauté de communes n°{i}" for i in range(1, 60)}, | |
| } | |
| DPT_REGION = {"75": "11", "93": "11", "13": "93", "69": "84", "31": "76", "34": "76", | |
| "22": "53", "35": "53", "56": "53", "974": "04"} | |
| # commune: (label, epci, département) | |
| COMMUNES: dict[str, tuple[str, str, str]] = { | |
| "75056": ("Paris", "200054781", "75"), | |
| "13055": ("Marseille", "200054807", "13"), | |
| "69123": ("Lyon", "200046977", "69"), | |
| "31555": ("Toulouse", "243100518", "31"), | |
| "34172": ("Montpellier", "243400017", "34"), | |
| "34057": ("Castelnau-le-Lez", "243400017", "34"), | |
| "22278": ("Saint-Brieuc", "200069409", "22"), | |
| "35258": ("Saint-Brieuc-des-Iffs", "200000001", "35"), | |
| "56208": ("Saint-Brieuc-de-Mauron", "200000002", "56"), | |
| "93066": ("Saint-Denis", "200054781", "93"), | |
| "97411": ("Saint-Denis", "249740119", "974"), | |
| **{f"{i:05d}": (f"Commune {i}", f"2000{i % 59 + 1:05d}", f"{i % 90 + 1:02d}") for i in range(1000, 1700)}, | |
| } | |
| def _region_of(dpt: str) -> str: | |
| return DPT_REGION.get(dpt) or METRO_REGIONS[sum(map(ord, dpt)) % len(METRO_REGIONS)] | |
| TERRITORIES: dict[str, list[tuple[str, str]]] = { | |
| "region": [(code, REGION_CODES[code]) for code in METRO_REGIONS], | |
| "departement": list(DEPARTEMENTS), | |
| "epci": list(EPCIS.items()), | |
| "commune": [(code, c[0]) for code, c in COMMUNES.items()], | |
| } | |
| DEPARTEMENT_LABELS = dict(TERRITORIES["departement"]) | |
| def _indicator_rows() -> list[dict[str, Any]]: | |
| """All indicators known from the capture, richest metadata when available.""" | |
| rows: dict[int, dict[str, Any]] = {} | |
| for item in CAPTURE["list_all"]["indicators"]: | |
| rows[item["id"]] = dict(item) | |
| for key, det in CAPTURE["details"].items(): | |
| rows[int(key)].update(det["metadata"]) | |
| for ind_id, override in META_OVERRIDES.items(): | |
| rows.setdefault(ind_id, {"id": ind_id}).update(override) | |
| return [rows[k] for k in sorted(rows)] | |
| class FakeCubeJsClient(CubeJsClient): | |
| def __init__(self) -> None: | |
| super().__init__(token="fake-token") | |
| self.queries: list[dict[str, Any]] = [] | |
| self.fail_meta = False | |
| async def get_meta(self) -> dict[str, Any]: | |
| if self.fail_meta: | |
| raise BadRequestError("meta unavailable") | |
| cubes = [ | |
| {"name": "indicateur_metadata", "measures": [], "dimensions": [ | |
| {"name": f"indicateur_metadata.{d}"} for d in META_DIMS]}, | |
| {"name": "indicateur_x_source_metadata", "measures": [], "dimensions": [ | |
| {"name": f"indicateur_x_source_metadata.{d}"} for d in SOURCE_DIMS]}, | |
| ] | |
| for cube, (levels, ids) in DATA_CUBES.items(): | |
| dims = ["date_mesure", "annee"] + [d for lv in levels for d in GEO_DIMS[lv]] + list(CUBE_AXES.get(cube, {})) | |
| measures = [f"id_{i}" for i in ids] + EXTRA_MEASURES.get(cube, []) | |
| cubes.append({ | |
| "name": cube, | |
| "measures": [{"name": f"{cube}.{m}"} for m in measures], | |
| "dimensions": [{"name": f"{cube}.{d}"} for d in dims], | |
| }) | |
| return {"cubes": cubes} | |
| async def load(self, query: dict[str, Any]) -> dict[str, Any]: | |
| self.queries.append(query) | |
| dims: list[str] = query.get("dimensions", []) | |
| measures: list[str] = query.get("measures", []) | |
| limit = int(query.get("limit", 10000)) | |
| filters = query.get("filters", []) | |
| members = dims + measures | |
| cube = members[0].split(".")[0] if members else "" | |
| if cube == "indicateur_metadata": | |
| rows = [{f"indicateur_metadata.{k}": v for k, v in r.items()} for r in _indicator_rows()] | |
| elif cube == "indicateur_x_source_metadata": | |
| for d in dims: | |
| if d.split(".", 1)[1] not in SOURCE_DIMS: | |
| raise BadRequestError( | |
| f"Bad request to API. Details: {{'type': 'UserError', 'error': \"'{d.split('.', 1)[1]}' " | |
| f"not found for path '{d}'\"}}" | |
| ) | |
| rows = [{ | |
| "indicateur_x_source_metadata.id_indicateur": 611, | |
| "indicateur_x_source_metadata.nom_source": "fichiers_fonciers", | |
| "indicateur_x_source_metadata.description": "Fichiers fonciers (Cerema)", | |
| "indicateur_x_source_metadata.producteur_source": "Cerema", | |
| "indicateur_x_source_metadata.distributeur_source": "Cerema", | |
| "indicateur_x_source_metadata.license_source": "Licence ouverte", | |
| "indicateur_x_source_metadata.lien_page": "https://artificialisation.developpement-durable.gouv.fr", | |
| "indicateur_x_source_metadata.date_derniere_extraction": "2024-06-01", | |
| }] | |
| elif cube in DATA_CUBES and not measures and sum(d.split(".", 1)[1].startswith("geocode_") for d in dims) > 1: | |
| rows = self._hierarchy_rows(cube) | |
| elif cube in DATA_CUBES: | |
| rows = self._data_rows(cube, dims, measures) | |
| else: | |
| raise BadRequestError(f"Unknown cube {cube}") | |
| for f in filters: | |
| member, op, values = f["member"], f["operator"], [str(v) for v in f.get("values", [])] | |
| if op == "equals": | |
| rows = [r for r in rows if str(r.get(member)) in values] | |
| elif op == "contains": | |
| needle = values[0].lower() | |
| rows = [r for r in rows if needle in str(r.get(member, "")).lower()] | |
| elif op == "set": | |
| rows = [r for r in rows if r.get(member) is not None] | |
| for member, direction in reversed(list((query.get("order") or {}).items())): | |
| rows.sort(key=lambda r: str(r.get(member) or ""), reverse=direction == "desc") | |
| if dims and not measures: | |
| # Cube.js dedupes dimension-only queries | |
| seen, uniq = set(), [] | |
| for r in rows: | |
| key = json.dumps([r.get(d) for d in dims], sort_keys=True, default=str) | |
| if key not in seen: | |
| seen.add(key) | |
| uniq.append({d: r.get(d) for d in dims}) | |
| rows = uniq | |
| return {"data": rows[:limit]} | |
| def _hierarchy_rows(self, cube: str) -> list[dict[str, Any]]: | |
| """One row per commune with its EPCI, département and région (territory index).""" | |
| rows = [] | |
| for code, (label, epci, dpt) in COMMUNES.items(): | |
| rows.append({ | |
| f"{cube}.geocode_commune": code, f"{cube}.libelle_commune": label, | |
| f"{cube}.geocode_epci": epci, f"{cube}.libelle_epci": EPCIS[epci], | |
| f"{cube}.geocode_departement": dpt, | |
| f"{cube}.libelle_departement": DEPARTEMENT_LABELS.get(dpt, f"Département {dpt}"), | |
| f"{cube}.geocode_region": _region_of(dpt), | |
| }) | |
| return rows | |
| def _data_rows(self, cube: str, dims: list[str], measures: list[str]) -> list[dict[str, Any]]: | |
| levels = [lv for lv in GEO_DIMS if f"{cube}.{GEO_DIMS[lv][0]}" in dims] | |
| if not levels: | |
| # dimension `annee` only (latest-year lookup): one territory stands for all | |
| levels = [DATA_CUBES[cube][0][-1]] | |
| level = levels[0] | |
| real = CAPTURE["query_611_reg_93"]["data"] if (cube == "conso_enaf_com" and level == "region") else None | |
| axes = CUBE_AXES.get(cube, {}) | |
| modalities: list[dict[str, str]] = [{}] | |
| for axis, values in axes.items(): | |
| modalities = [{**m, axis: v} for m in modalities for v in values] | |
| rows = [] | |
| for code, label in TERRITORIES[level]: | |
| for year in CUBE_YEARS.get(cube, YEARS): | |
| for modality in modalities: | |
| row = { | |
| f"{cube}.{GEO_DIMS[level][0]}": code, | |
| f"{cube}.{GEO_DIMS[level][1]}": label, | |
| f"{cube}.annee": year, | |
| f"{cube}.date_mesure": f"{year}-01-01T00:00:00.000", | |
| f"{cube}.date_mesure.year": f"{year}-01-01T00:00:00.000", | |
| **{f"{cube}.{a}": v for a, v in modality.items()}, | |
| } | |
| for m in measures: | |
| value: Any = f"{(hash((code, year, m, tuple(modality.values()))) % 100000) / 10:.4f}" | |
| if real and code == "93": | |
| match = next((r for r in real if r["annee"] == year), None) | |
| if match: | |
| value = match["valeur"] | |
| if m.endswith(".id_955") and (modality.get("mode_transport") != "Vélo (depuis 2017)" or year != "2022"): | |
| value = None # « - vélo » indicator: only the vélo modality, since 2017 | |
| if m.endswith(".id_611") and code == "13" and level == "departement": | |
| value = None # a territory without data at this level | |
| row[m] = value | |
| rows.append(row) | |
| return rows | |