car_data.py

"""Fetch real French car models with their official fuel consumption.

Source: ADEME "Car Labelling" open dataset (combined-cycle WLTP consumption in
L/100 km for every model commercialised in France). We map each model's energy
type to the matching pump fuel and dedupe to one entry per commercial model.
"""
from __future__ import annotations

import json
import urllib.parse
import urllib.request

ADEME_URL = "https://data.ademe.fr/data-fair/api/v1/datasets/ademe-car-labelling/lines"
SELECT = "Marque,Description_Commerciale,Energie,Conso_vitesse_mixte_Min,Conso_vitesse_mixte_Max"

# ADEME "Energie" -> our pump-fuel label (None = no liquid fuel, e.g. pure electric).
ENERGIE_FUEL = {
    "ESSENCE": "E10",
    "GAZOLE": "Gazole",
    "GAZ+ELEC HNR": "Gazole",      # diesel hybrid
    "ESS+ELEC HNR": "E10",         # petrol hybrid
    "ELEC+ESSENC HR": "E10",       # plug-in petrol hybrid
    "ELEC+GAZOLE HR": "Gazole",    # plug-in diesel hybrid
    "SUPERETHANOL": "E85",
    "ESS+G.P.L.": "GPLc",
    "ELECTRIC": None,
}

# Suffix appended to the model name so hybrids are explicit in the picker.
HYBRID_TAG = {
    "ESS+ELEC HNR": " — hybride",
    "GAZ+ELEC HNR": " — hybride",
    "ELEC+ESSENC HR": " — hybride rechargeable",
    "ELEC+GAZOLE HR": " — hybride rechargeable",
}


def _fetch_all_rows(timeout: int, page_size: int = 10000) -> list[dict]:
    """Page through every line of the dataset via data-fair's `after` cursor.

    The API caps a single response, so we follow the server-provided `next` link
    (which carries the `after=<cursor>`) until the whole dataset is collected.
    """
    url = ADEME_URL + f"?size={page_size}&select=" + urllib.parse.quote(SELECT)
    rows: list[dict] = []
    for _ in range(200):  # hard safety cap on the number of pages
        req = urllib.request.Request(url, headers={"User-Agent": "petrol-map/1.0"})
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            data = json.load(resp)
        page = data.get("results", [])
        rows.extend(page)
        total = data.get("total")
        nxt = data.get("next")
        if not page or not nxt or (isinstance(total, int) and len(rows) >= total):
            break
        url = nxt
    return rows


def fetch_cars(timeout: int = 90) -> list[list]:
    """Return [[label, fuel, consumption_L_per_100km], ...], sorted by label."""
    rows = _fetch_all_rows(timeout)

    agg: dict[tuple, list[float]] = {}  # (label, fuel) -> [sum_cons, n]
    for r in rows:
        energie = (r.get("Energie") or "").strip()
        fuel = ENERGIE_FUEL.get(energie)
        if not fuel:
            continue
        marque = (r.get("Marque") or "").strip()
        desc = (r.get("Description_Commerciale") or "").strip()
        if not marque or not desc:
            continue
        vals = [v for v in (r.get("Conso_vitesse_mixte_Min"), r.get("Conso_vitesse_mixte_Max"))
                if isinstance(v, (int, float)) and v > 0]
        if not vals:
            continue
        cons = sum(vals) / len(vals)
        label = f"{marque} {desc}{HYBRID_TAG.get(energie, '')}"
        key = (label, fuel)
        if key in agg:
            agg[key][0] += cons
            agg[key][1] += 1
        else:
            agg[key] = [cons, 1]

    cars = [[label, fuel, round(s / n, 1)] for (label, fuel), (s, n) in agg.items()]
    cars.sort(key=lambda c: c[0])
    return cars


if __name__ == "__main__":
    cars = fetch_cars()
    blob = json.dumps(cars, ensure_ascii=False, separators=(",", ":"))
    print(f"{len(cars)} models, embedded size ~{len(blob) / 1024:.0f} KB")
    for c in cars[:5]:
        print("  ", c)