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`, | |
| `equals` and `contains` filters. | |
| """ | |
| 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"], | |
| } | |
| # cube -> (levels, indicator ids) — from the captured available_cubes | |
| DATA_CUBES = { | |
| "conso_enaf_com": (["commune", "epci", "departement", "region"], [611, 545]), | |
| "stationnement_velo_habitant_com": (["commune", "epci", "departement", "region"], [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]), | |
| } | |
| YEARS = ["2020", "2021", "2022"] | |
| # Synthetic territories (enough to exercise truncation on communes) | |
| TERRITORIES: dict[str, list[tuple[str, str]]] = { | |
| "region": list(REGION_CODES.items()), | |
| "departement": [(f"{i:02d}", f"Département {i:02d}") for i in range(1, 96) if i != 20] | |
| + [("2A", "Corse-du-Sud"), ("2B", "Haute-Corse"), ("13", "Bouches-du-Rhône"), ("69", "Rhône")], | |
| "epci": [ | |
| ("200046977", "Métropole de Lyon"), | |
| ("200054781", "Métropole du Grand Paris"), | |
| ("200054807", "Métropole d'Aix-Marseille-Provence"), | |
| ("243100518", "Toulouse Métropole"), | |
| ] | |
| + [(f"2000{i:05d}", f"Communauté de communes n°{i}") for i in range(1, 60)], | |
| "commune": [("75056", "Paris"), ("13055", "Marseille"), ("69123", "Lyon"), ("31555", "Toulouse")] | |
| + [(f"{i:05d}", f"Commune {i}") for i in range(1000, 1700)], | |
| } | |
| TERRITORIES["departement"] = list(dict.fromkeys(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"]) | |
| 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 = ["annee"] + [d for lv in levels for d in GEO_DIMS[lv]] | |
| cubes.append({ | |
| "name": cube, | |
| "measures": [{"name": f"{cube}.id_{i}"} for i in ids], | |
| "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: | |
| 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["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()] | |
| if dims and not measures: | |
| # Cube.js dedupes dimension-only queries | |
| seen, uniq = set(), [] | |
| for r in rows: | |
| key = tuple(r.get(d) for d in dims) | |
| 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 _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: | |
| return [] | |
| level = levels[0] | |
| real = CAPTURE["query_611_reg_93"]["data"] if (cube == "conso_enaf_com" and level == "region") else None | |
| rows = [] | |
| for code, label in TERRITORIES[level]: | |
| for year in YEARS: | |
| row = { | |
| f"{cube}.{GEO_DIMS[level][0]}": code, | |
| f"{cube}.{GEO_DIMS[level][1]}": label, | |
| f"{cube}.annee": year, | |
| } | |
| for m in measures: | |
| value = f"{(hash((code, year, m)) % 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"] | |
| row[m] = value | |
| rows.append(row) | |
| return rows | |