apt/fuelboard-relay-api
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

FuelBoard Relay API: official Fuel Finder API adapter (sibling of CSV relay)

Browse Source 
- Talks to the OFFICIAL GOV.UK Fuel Finder API instead of the public CSV
  mirror: OAuth token (/api/v1/oauth/generate_access_token), batched
  paging (500/page via batch-number) over /api/v1/pfs and
  /api/v1/pfs/fuel-prices, joined on node_id.
- Price strings parsed to pence with pounds->pence correction (< 2.0);
  50-500p sane band keeps junk out of the cheapest reference.
- Fuel mapping: E5/E10 as-is; B7/B7_STANDARD/B7S/B7P/B10 -> DIESEL;
  HVO/SDV dropped (app tracks E5/E10/DIESEL only).
- lat/lng parsed from nested location strings.
- Output contract IDENTICAL to fuelboard-relay (/health + /api/v1/stations),
  so the app/widget/alerts work unchanged; switch = point app at :8789.
- CSV mirror kept as automatic fallback (FUEL_SOURCE=auto) so the full
  path is testable before GOV.UK One Login creds exist.
- 12 unit tests against documented API shapes (no network).
This commit is contained in:
FuelBoard Contributor
2026-08-12 15:34:57 +01:00
commit b315f4dc47
8 changed files with 817 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/main.py
+524
View File
@@ -0,0 +1,524 @@
# FuelBoard Relay API — keyless proxy for the official UK Fuel Finder API
#
# Holds the GOV.UK Fuel Finder OAuth 2.0 client credentials server-side so the
# iOS app / widget never embeds a secret. The app calls this relay with no auth
# and gets stations + prices as simple JSON.
#
# This is the API-source relay (sibling of fuelboard-relay, which reads the
# public CSV mirror). It talks to the OFFICIAL Fuel Finder API:
#
# FuelBoard app → GET http://<relay>:8789/api/v1/stations
# relay → OAuth token (cached ~1h) → GET /api/v1/pfs (station info)
# GET /api/v1/pfs/fuel-prices (prices)
# relay joins on node_id, normalises to the SAME output shape as the CSV
# relay, caches the snapshot (default 5 min) → returns JSON to the app
#
# The /api/v1/stations contract is IDENTICAL to fuelboard-relay, so the app,
# widget and alerts work unchanged against either relay — switching sources is
# just pointing the app at a different host:port.
#
# Endpoints:
# GET /health — cache status + last sync timestamp
# GET /api/v1/stations — full station list with current prices (keyless)
# GET /api/v1/stations?fuel=e10&lat=53.72&lng=-1.85&radius=10
# — optional server-side filter/sort (cheapest first)
#
# Data shape notes (from developer.fuel-finder.service.gov.uk):
# - Token: POST /api/v1/oauth/generate_access_token with client_id/client_secret
# → { "data": { "access_token": ..., "expires_in": 3600 } }
# - Stations: GET /api/v1/pfs → array, batched 500/page via ?batch-number=N
# - Prices: GET /api/v1/pfs/fuel-prices → array, same batching
# - Price values are DECIMAL STRINGS in pence ("0120.0000" = 120.0p); some
# stations report pounds (< 2.0) — multiply by 100.
# - lat/lng are STRINGS inside location: { latitude, longitude }.
# - Fuel types: E10, E5, B7_STANDARD (diesel), B10, HVO, SDV, ...
from contextlib import asynccontextmanager
import os
import time
from typing import Optional
import httpx
from fastapi import FastAPI, HTTPException, Query
# ---------------------------------------------------------------------------
# Config (all via environment; never hardcode credentials)
# ---------------------------------------------------------------------------
CLIENT_ID = os.environ.get("FUEL_API_CLIENT_ID", "")
CLIENT_SECRET = os.environ.get("FUEL_API_CLIENT_SECRET", "")
TOKEN_URL = os.environ.get(
"FUEL_API_TOKEN_URL",
"https://api.fuelfinder.service.gov.uk/api/v1/oauth/generate_access_token",
)
# Official Fuel Finder endpoints (batched, 500 records/page).
PFS_URL = os.environ.get("FUEL_API_PFS_URL", "https://api.fuelfinder.service.gov.uk/api/v1/pfs")
PRICES_URL = os.environ.get(
"FUEL_API_PRICES_URL",
"https://api.fuelfinder.service.gov.uk/api/v1/pfs/fuel-prices",
)
PAGE_SIZE = int(os.environ.get("FUEL_API_PAGE_SIZE", "500"))
CACHE_TTL = int(os.environ.get("FUEL_CACHE_TTL", "300")) # seconds (5 min)
TOKEN_EXPIRY_BUFFER = int(os.environ.get("FUEL_TOKEN_BUFFER", "60")) # seconds
# Public full-UK mirror of the Fuel Finder dataset (GitHub Actions, hourly),
# used as the fallback source when OAuth credentials are not configured so the
# relay is testable end-to-end before GOV.UK One Login creds exist.
# Licence: Open Government Licence v3.0. No auth required.
CSV_URL = os.environ.get(
"FUEL_CSV_URL",
"https://raw.githubusercontent.com/matthewgall/fuelfinder-archive/main/data.csv",
)
CSV_REFRESH_TTL = int(os.environ.get("FUEL_CSV_TTL", "3600")) # seconds (1h)
# Source selection: "auto" (API if creds, CSV fallback), "csv", "api"
SOURCE = os.environ.get("FUEL_SOURCE", "auto").lower()
DEMO_MODE = not (CLIENT_ID and CLIENT_SECRET) and SOURCE != "csv"
# ---------------------------------------------------------------------------
# In-memory cache state
# ---------------------------------------------------------------------------
_state = {
"stations": None, # normalised station list [{...}, ...]
"stations_fetched_at": 0.0,
"token": None,
"token_expires_at": 0.0,
"last_error": None,
"sync_attempts": 0,
"sync_failures": 0,
"source_used": None,
}
# ---------------------------------------------------------------------------
# Normalisation helpers (shared by CSV fallback and API path)
# ---------------------------------------------------------------------------
# Map official Fuel Finder fuel types → the app's grades (E5/E10/DIESEL).
# The app tracks exactly these three; everything else (HVO, SDV, B10, ...) is
# dropped so it can't poison the cheapest-price reference.
FUEL_MAP = {
"E5": "E5",
"E10": "E10",
"B7": "DIESEL",
"B7_STANDARD": "DIESEL",
"B7S": "DIESEL",
"B7P": "DIESEL",
"B10": "DIESEL",
}
def _parse_price(value) -> Optional[float]:
"""Parse a Fuel Finder price value into pence/litre, or None.
The API sends DECIMAL STRINGS in pence ("0120.0000" = 120.0p). Some
stations report pounds (values below 2.0) — detect and multiply by 100.
Returns None for junk (non-numeric, or outside the sane 50–500p band so a
pounds-denominated column can't poison the cheapest reference).
"""
if value is None:
return None
try:
raw = float(value)
except (TypeError, ValueError):
return None
if raw < 2.0:
# Pounds-denominated ("1.299" → 129.9p). Guard against a near-zero
# empty/placeholder value being inflated to something plausible.
raw *= 100
if raw < 50 or raw > 500:
return None
return raw
def _normalise_station(station_id, name, brand, address, postcode, lat, lng,
prices, price_updated=None, is_motorway=False) -> Optional[dict]:
"""Build a station dict in the relay's canonical output shape.
Returns None if the station has no tracked fuel prices (sells nothing the
app monitors) or its location is unusable.
"""
if not prices:
return None
if lat is None or lng is None:
return None
return {
"id": station_id,
"name": name,
"brand": brand or "",
"address": address or "",
"postcode": postcode or "",
"lat": lat,
"lng": lng,
"prices": prices, # {"E5": 144.9, "E10": 137.9, "DIESEL": ...}
"price_updated": price_updated,
"is_motorway": bool(is_motorway),
}
def _normalise_api_payload(stations_info: list, fuel_prices: list) -> list:
"""Join the official API's two endpoints on node_id → canonical stations.
stations_info: records from GET /api/v1/pfs (site details + location).
fuel_prices: records from GET /api/v1/pfs/fuel-prices (prices array).
"""
info_by_id = {}
for rec in stations_info:
node_id = rec.get("node_id")
if not node_id:
continue
loc = rec.get("location") or {}
lat = _parse_lat_lng(loc.get("latitude"))
lng = _parse_lat_lng(loc.get("longitude"))
info_by_id[node_id] = {
"name": rec.get("trading_name"),
"brand": rec.get("brand_name"),
"address": _join_address(loc),
"postcode": loc.get("postcode"),
"lat": lat,
"lng": lng,
"is_motorway": rec.get("is_motorway_service_station", False),
}
stations = []
for rec in fuel_prices:
node_id = rec.get("node_id")
info = info_by_id.get(node_id)
if info is None:
continue
prices = {}
latest_updated = None
for entry in rec.get("fuel_prices") or []:
fuel_type = (entry.get("fuel_type") or "").upper()
grade = FUEL_MAP.get(fuel_type)
if grade is None:
continue
price = _parse_price(entry.get("price"))
if price is None:
continue
prices[grade] = price
updated = entry.get("price_last_updated")
if updated and (latest_updated is None or updated > latest_updated):
latest_updated = updated
station = _normalise_station(
station_id=node_id,
name=info["name"],
brand=info["brand"],
address=info["address"],
postcode=info["postcode"],
lat=info["lat"],
lng=info["lng"],
prices=prices,
price_updated=latest_updated,
is_motorway=info["is_motorway"],
)
if station is not None:
stations.append(station)
return stations
def _parse_lat_lng(value) -> Optional[float]:
"""API sends lat/lng as STRINGS — parse to float, tolerate garbage."""
if value is None or value == "":
return None
try:
return float(value)
except (TypeError, ValueError):
return None
def _join_address(loc: dict) -> str:
parts = [loc.get("address_line_1"), loc.get("address_line_2")]
return " ".join(p for p in parts if p) or ""
# ---------------------------------------------------------------------------
# CSV fallback (public mirror — same parser as fuelboard-relay)
# ---------------------------------------------------------------------------
def _parse_csv_to_stations(text: str) -> list:
"""Parse the Fuel Finder archive CSV into a station list.
Header uses dotted keys: forecourts.trading_name, forecourts.location.latitude,
forecourts.fuel_price.E10, forecourts.price_change_effective_timestamp.E10, etc.
"""
import csv
import io
reader = csv.DictReader(io.StringIO(text))
stations = []
for row in reader:
def num(key):
try:
val = row.get(key)
return float(val) if val not in (None, "") else None
except (TypeError, ValueError):
return None
lat = num("forecourts.location.latitude")
lng = num("forecourts.location.longitude")
prices = {}
for grade, label in (("E5", "E5"), ("E10", "E10"), ("B7S", "DIESEL"), ("B7P", "DIESEL")):
p = num(f"forecourts.fuel_price.{grade}")
if p is not None and grade not in prices:
prices[label] = p
station = _normalise_station(
station_id=row.get("forecourts.node_id"),
name=row.get("forecourts.trading_name"),
brand=row.get("forecourts.brand_name"),
address=_join_address({
"address_line_1": row.get("forecourts.location.address_line_1"),
"address_line_2": row.get("forecourts.location.address_line_2"),
}),
postcode=row.get("forecourts.location.postcode"),
lat=lat,
lng=lng,
prices=prices,
price_updated=row.get("forecourts.price_change_effective_timestamp.E10")
or row.get("forecourts.price_change_effective_timestamp.E5"),
is_motorway=row.get("forecourts.is_motorway_service_station") == "true",
)
if station is not None:
stations.append(station)
return stations
def _fetch_csv() -> list:
"""Download the hourly full-UK CSV mirror."""
with httpx.Client(timeout=120) as client:
resp = client.get(CSV_URL)
resp.raise_for_status()
stations = _parse_csv_to_stations(resp.text)
if not stations:
raise RuntimeError("CSV parse returned zero stations")
return stations
# ---------------------------------------------------------------------------
# Official API fetch (OAuth + batched paging + node_id join)
# ---------------------------------------------------------------------------
def _get_token(client: httpx.Client) -> str:
"""Return a valid bearer token, refreshing via client-credentials grant.
The Fuel Finder token endpoint returns the token under "data":
{ "data": { "access_token": "...", "expires_in": 3600 }, ... }
Tolerates a flat { "access_token": ... } shape too (older docs).
"""
if _state["token"] and time.time() < _state["token_expires_at"] - TOKEN_EXPIRY_BUFFER:
return _state["token"]
resp = client.post(
TOKEN_URL,
json={"client_id": CLIENT_ID, "client_secret": CLIENT_SECRET},
)
resp.raise_for_status()
payload = resp.json()
data = payload.get("data") if isinstance(payload.get("data"), dict) else payload
token = data.get("access_token") or payload.get("access_token")
if not token:
raise RuntimeError(f"Token response missing access_token: {list(payload.keys())}")
expires_in = int(data.get("expires_in", payload.get("expires_in", 3600)))
_state["token"] = token
_state["token_expires_at"] = time.time() + expires_in
return token
def _fetch_all_pages(client: httpx.Client, url: str) -> list:
"""Fetch every page of a batched Fuel Finder endpoint (500/page)."""
results = []
batch = 1
while True:
resp = client.get(
url,
params={"batch-number": batch},
headers={"Authorization": f"Bearer {_get_token(client)}"},
)
resp.raise_for_status()
page = resp.json()
if not isinstance(page, list) or not page:
break
results.extend(page)
if len(page) < PAGE_SIZE:
break
batch += 1
if batch > 200: # safety cap (~100k records)
break
return results
def _fetch_api() -> list:
"""Fetch the full dataset from the official API and normalise it.
Two batched endpoints joined on node_id:
1. GET /api/v1/pfs → site details (trading_name, location, ...)
2. GET /api/v1/pfs/fuel-prices → per-station fuel_prices array
"""
with httpx.Client(timeout=60) as client:
# Warm the token once so both loops reuse it (token endpoint is the
# most rate-sensitive; docs say reuse until near expiry).
_get_token(client)
stations_info = _fetch_all_pages(client, PFS_URL)
fuel_prices = _fetch_all_pages(client, PRICES_URL)
stations = _normalise_api_payload(stations_info, fuel_prices)
if not stations:
raise RuntimeError("API fetch returned zero stations")
return stations
# ---------------------------------------------------------------------------
# Cache refresh
# ---------------------------------------------------------------------------
def _refresh_cache() -> None:
"""Refresh the station cache if stale (or on first run)."""
_state["sync_attempts"] += 1
try:
if SOURCE == "csv":
_state["stations"] = _fetch_csv()
_state["source_used"] = "csv"
elif SOURCE == "api":
_state["stations"] = _fetch_api()
_state["source_used"] = "api"
else: # auto: official API if creds present, otherwise public CSV
if not DEMO_MODE:
_state["stations"] = _fetch_api()
_state["source_used"] = "api"
else:
_state["stations"] = _fetch_csv()
_state["source_used"] = "csv"
_state["stations_fetched_at"] = time.time()
_state["last_error"] = None
except Exception as exc: # keep serving stale cache on failure
_state["sync_failures"] += 1
_state["last_error"] = f"{type(exc).__name__}: {exc}"
@asynccontextmanager
async def lifespan(_: FastAPI):
# Always try to warm the cache. In auto mode with no credentials this
# fetches the public full-UK CSV; with credentials it uses the OAuth API.
_refresh_cache()
yield
app = FastAPI(title="FuelBoard Relay API", version="1.0.0", lifespan=lifespan)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _stations_list() -> list:
if _state["stations"] is None:
return []
data = _state["stations"]
if isinstance(data, list):
return data
if isinstance(data, dict):
for key in ("stations", "prices", "data"):
if isinstance(data.get(key), list):
return data[key]
return []
def _distance_km(lat1: float, lng1: float, lat2: float, lng2: float) -> float:
import math
r = 6371.0
dlat = math.radians(lat2 - lat1)
dlng = math.radians(lng2 - lng1)
a = math.sin(dlat / 2) ** 2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlng / 2) ** 2
return r * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
def _fuel_price(station: dict, fuel: str) -> Optional[float]:
"""Extract a price for the fuel grade from a normalised station dict."""
prices = station.get("prices") or {}
raw = prices.get(fuel.upper())
if isinstance(raw, (int, float)) and 50 <= raw <= 500:
return float(raw)
return None
# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------
@app.get("/health")
def health():
return {
"status": "ok" if not DEMO_MODE and _state["stations"] is not None else "degraded",
"demo_mode": DEMO_MODE,
"source": _state["source_used"],
"stations_cached": _state["stations"] is not None,
"stations_count": len(_stations_list()),
"last_sync": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(_state["stations_fetched_at"])) if _state["stations_fetched_at"] else None,
"sync_attempts": _state["sync_attempts"],
"sync_failures": _state["sync_failures"],
"last_error": _state["last_error"],
}
@app.get("/api/v1/stations")
def stations(
fuel: str = Query("e10", description="Fuel grade: e10, e5, diesel"),
lat: Optional[float] = Query(None, description="User latitude"),
lng: Optional[float] = Query(None, description="User longitude"),
radius: Optional[float] = Query(None, description="Search radius in km (optional; omit for the full dataset)"),
limit: int = Query(10000, ge=1, le=10000),
):
if _state["stations"] is None:
raise HTTPException(status_code=502, detail="No data cached yet and upstream fetch failed")
# Trigger a refresh if stale; keep serving stale cache if it fails.
if time.time() - _state["stations_fetched_at"] > CACHE_TTL:
_refresh_cache()
rows = []
for station in _stations_list():
s_lat = station.get("lat")
s_lng = station.get("lng")
# All known fuel grades for this station (E5/E10/DIESEL) so clients can
# switch fuel tabs without another request. Every station is returned
# regardless of the requested fuel — clients filter on-device.
prices = {}
for grade in ("E5", "E10", "DIESEL"):
p = _fuel_price(station, grade)
if p is not None:
prices[grade] = p
if not prices:
continue # station sells nothing we track — skip
item = {
"id": station.get("id"),
"name": station.get("name"),
"brand": station.get("brand") or "",
"address": station.get("address") or "",
"postcode": station.get("postcode") or "",
"lat": s_lat,
"lng": s_lng,
"price": prices.get(fuel.upper()),
"prices": prices,
"price_updated": station.get("price_updated"),
}
if lat is not None and lng is not None and isinstance(s_lat, (int, float)) and isinstance(s_lng, (int, float)):
item["distance_km"] = round(_distance_km(lat, lng, s_lat, s_lng), 2)
rows.append(item)
if lat is not None and lng is not None:
# Optional server-side radius filter — used by the alert path for a
# focused fresh fetch. The main app fetch omits radius and gets all.
if radius is not None:
rows = [r for r in rows if r.get("distance_km") is not None and r["distance_km"] <= radius]
rows.sort(key=lambda r: r.get("distance_km") if r.get("distance_km") is not None else float("inf"))
else:
rows.sort(key=lambda r: r["price"] if r["price"] is not None else float("inf"))
return {"fuel": fuel, "count": len(rows[:limit]), "stations": rows[:limit]}
if __name__ == "__main__":
import uvicorn
port = int(os.environ.get("FUEL_RELAY_PORT", "8789"))
uvicorn.run(app, host="0.0.0.0", port=port)