"""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