fuelboard-relay/app/main.py

# FuelBoard Relay — keyless proxy for the 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.
#
# Flow:
#   FuelBoard app → GET http://<relay>:8788/api/v1/stations
#   relay → OAuth token (cached ~1h) → GET https://api.fuelfinder.service.gov.uk/v1/prices
#   relay caches the snapshot (default 5 min) → returns JSON to the app
#
# 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)

from contextlib import asynccontextmanager
import os
import time
from typing import Optional

import httpx
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import JSONResponse

# ---------------------------------------------------------------------------
# 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/v1/token")
PRICES_URL = os.environ.get("FUEL_API_PRICES_URL", "https://api.fuelfinder.service.gov.uk/v1/prices")
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 default source when OAuth credentials are not configured.
# 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" (CSV if no creds, API if creds), "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,          # parsed JSON payload (list or {"stations": [...]})
    "stations_fetched_at": 0.0,
    "token": None,
    "token_expires_at": 0.0,
    "last_error": None,
    "sync_attempts": 0,
    "sync_failures": 0,
    "source_used": None,
}


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
        stations.append({
            "id": row.get("forecourts.node_id"),
            "name": row.get("forecourts.trading_name"),
            "brand": row.get("forecourts.brand_name"),
            "address": (row.get("forecourts.location.address_line_1") or "") + " " + (row.get("forecourts.location.address_line_2") or ""),
            "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",
        })
    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


def _get_token(client: httpx.Client) -> str:
    """Return a valid bearer token, refreshing via client-credentials grant."""
    if _state["token"] and time.time() < _state["token_expires_at"] - TOKEN_EXPIRY_BUFFER:
        return _state["token"]

    resp = client.post(
        TOKEN_URL,
        data={"grant_type": "client_credentials"},
        auth=(CLIENT_ID, CLIENT_SECRET),
    )
    resp.raise_for_status()
    payload = resp.json()
    token = payload["access_token"]
    # Prefer expires_in (seconds); default to 1h if absent.
    expires_in = int(payload.get("expires_in", 3600))
    _state["token"] = token
    _state["token_expires_at"] = time.time() + expires_in
    return token


def _fetch_prices() -> dict:
    """Fetch the full prices snapshot, returning the JSON payload."""
    with httpx.Client(timeout=30) as client:
        token = _get_token(client)
        resp = client.get(
            PRICES_URL,
            headers={"Authorization": f"Bearer {token}"},
        )
        resp.raise_for_status()
        return resp.json()


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_prices()
            _state["source_used"] = "api"
        else:  # auto: API if creds present, otherwise public CSV
            if not DEMO_MODE:
                _state["stations"] = _fetch_prices()
                _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", 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 station dict.

    The Fuel Finder payload nests prices under keys like 'prices' -> 'E10'.
    Tolerates both flat and nested shapes.
    """
    prices = station.get("prices") or station.get("price") or {}
    if isinstance(prices, dict):
        for key in (fuel.upper(), fuel.upper() + "_PRICE", "price_" + fuel.upper()):
            if key in prices:
                val = prices[key]
                if isinstance(val, (int, float)):
                    return float(val)
    # flat fallback
    for key in (fuel.upper(), "PRICE_" + fuel.upper()):
        if key in station:
            val = station[key]
            if isinstance(val, (int, float)):
                return float(val)
    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,
        "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: float = Query(20.0, description="Search radius in km"),
    limit: int = Query(10, ge=1, le=1000),
):
    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():
        price = _fuel_price(station, fuel)
        if price is None:
            continue
        s_lat = station.get("latitude", station.get("lat"))
        s_lng = station.get("longitude", station.get("lng"))
        # All known fuel grades for this station (E5/E10/DIESEL) so clients can
        # switch fuel tabs without another request.
        prices = {}
        for grade in ("E5", "E10", "DIESEL"):
            p = _fuel_price(station, grade)
            if p is not None:
                prices[grade] = p
        item = {
            "id": station.get("id") or station.get("station_id"),
            "name": station.get("name") or station.get("station_name"),
            "brand": station.get("brand") or station.get("operator") or "",
            "address": station.get("address") or "",
            "postcode": station.get("postcode") or "",
            "lat": s_lat,
            "lng": s_lng,
            "price": price,
            "prices": prices,
            "price_updated": station.get("price_updated") or station.get("updated_at"),
        }
        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:
        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["distance_km"])
    else:
        rows.sort(key=lambda r: r["price"])

    return {"fuel": fuel, "count": len(rows[:limit]), "stations": rows[:limit]}


if __name__ == "__main__":
    import uvicorn

    port = int(os.environ.get("FUEL_RELAY_PORT", "8788"))
    uvicorn.run(app, host="0.0.0.0", port=port)