Spaces:
Sleeping
Sleeping
File size: 4,108 Bytes
284e818 f2552b2 284e818 f2552b2 284e818 eea5f65 e6a157d 284e818 e6a157d 284e818 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | """Fetch and normalise French fuel-price data from the open-data portal.
Data source: https://public.opendatasoft.com/explore/dataset/prix_des_carburants_j_7
The dataset holds the prices reported by every petrol station in France over the
last 7 days, refreshed continuously upstream.
"""
from __future__ import annotations
import json
import os
from datetime import datetime, timezone
import requests
DATASET = "prix_des_carburants_j_7"
EXPORT_URL = (
"https://public.opendatasoft.com/api/explore/v2.1/"
f"catalog/datasets/{DATASET}/exports/json"
)
# Human label -> price field in the dataset (ordered: most common fuel first).
FUELS: dict[str, str] = {
"Gazole": "price_gazole",
"SP95": "price_sp95",
"SP98": "price_sp98",
"E10": "price_e10",
"E85": "price_e85",
"GPLc": "price_gplc",
}
# Plausible price band (€/L). Real road fuels in France sit well inside this;
# anything outside is a data-entry error and is dropped.
PRICE_MIN, PRICE_MAX = 0.4, 4.0
def fetch_raw(timeout: int = 180) -> list[dict]:
"""Download every record from the dataset's bulk export endpoint."""
resp = requests.get(EXPORT_URL, params={"timezone": "Europe/Paris"}, timeout=timeout)
resp.raise_for_status()
return resp.json()
def normalise(raw: list[dict]) -> list[dict]:
"""Keep geolocated stations that report at least one price; tidy the fields."""
stations = []
for r in raw:
geo = r.get("geo_point") or {}
lat, lon = geo.get("lat"), geo.get("lon")
if lat is None or lon is None:
continue
# Some stations report nonsense prices (e.g. 0.002 €/L); keep only
# plausible values so they don't skew the cheapest/cost figures.
prices = {
label: (v if isinstance(v, (int, float)) and PRICE_MIN <= v <= PRICE_MAX else None)
for label, field in FUELS.items()
for v in (r.get(field),)
}
if not any(v is not None for v in prices.values()):
continue
stations.append(
{
"id": r.get("id"),
"name": (r.get("name") or "").strip() or "Station",
"brand": (r.get("brand") or "").strip(),
"address": (r.get("address") or "").strip(),
"city": (r.get("city") or "").strip(),
"cp": (r.get("cp") or "").strip(),
"lat": lat,
"lon": lon,
"prices": prices,
"update": r.get("update"),
"highway": r.get("pop") == "A", # A = autoroute, R = route
"open24": is_24h(r.get("automate_24_24")),
"services": clean_services(r.get("services")),
}
)
return stations
def is_24h(value) -> bool:
"""Whether the station has 24/7 automated payment (field is bool or Oui/Non)."""
if value is True:
return True
return isinstance(value, str) and value.strip().lower() in ("oui", "yes", "true", "1")
def clean_services(value) -> list[str]:
"""Normalise the `services` field (list, or {'service': [...]}) to a string list."""
if isinstance(value, dict):
value = value.get("service")
if not isinstance(value, list):
return []
return [str(s).strip() for s in value if str(s).strip()]
def fetch_stations(cache_path: str | None = None) -> list[dict]:
"""Fetch + normalise the live dataset, optionally caching the result to disk."""
stations = normalise(fetch_raw())
if cache_path:
os.makedirs(os.path.dirname(cache_path) or ".", exist_ok=True)
payload = {
"fetched_at": datetime.now(timezone.utc).isoformat(),
"count": len(stations),
"stations": stations,
}
with open(cache_path, "w", encoding="utf-8") as fh:
json.dump(payload, fh, ensure_ascii=False)
return stations
if __name__ == "__main__":
data = fetch_stations()
print(f"Fetched {len(data)} geolocated stations.")
sample = data[0]
print("Example:", json.dumps(sample, ensure_ascii=False, indent=2))
|