import argparse import math from io import StringIO from typing import Any, Dict, List, Optional, Tuple import numpy as np import pandas as pd import requests from geopy.geocoders import Nominatim from geopy.location import Location from tabulate import tabulate from constants import ENDPOINT, HEADERS, SORT_KV def get_location(address: str) -> tuple[float, float]: geolocator = Nominatim(user_agent="FuelNearMe") result = geolocator.geocode(address) if not isinstance(result, Location): raise ValueError(f"Failed to get location from address: '{address}'") return (result.latitude, result.longitude) def get_latest_data() -> tuple[pd.DataFrame, Optional[str]]: response = requests.get(ENDPOINT, headers=HEADERS, timeout=10) response.raise_for_status() return pd.read_csv(StringIO(response.text)), response.headers.get("Last-Modified") def filter_df( dframe: pd.DataFrame, arguments: argparse.Namespace, loc: Tuple[float, float] ) -> List[Dict[str, Any]]: def bounding_box() -> pd.DataFrame: lat, lon = loc deg_lat = arguments.radius / 69.0 deg_lon = arguments.radius / (69.0 * math.cos(math.radians(lat))) return dframe[ dframe["forecourts.location.latitude"].between(lat - deg_lat, lat + deg_lat) & dframe["forecourts.location.longitude"].between( lon - deg_lon, lon + deg_lon ) ] def haversine_miles(lat2: np.ndarray, lon2: np.ndarray) -> np.ndarray: R = 3958.8 lat1, lon1 = np.radians(loc[0]), np.radians(loc[1]) lat2, lon2 = np.radians(lat2), np.radians(lon2) dlat = lat2 - lat1 dlon = lon2 - lon1 a = np.sin(dlat / 2) ** 2 + np.cos(lat1) * np.cos(lat2) * np.sin(dlon / 2) ** 2 return R * 2 * np.arcsin(np.sqrt(a)) def pence_to_pounds(col: pd.Series) -> pd.Series: return (col / 100).round(2).where(col.notna(), other="N/A") df = bounding_box().copy() df["distance"] = haversine_miles( df["forecourts.location.latitude"].to_numpy(), df["forecourts.location.longitude"].to_numpy(), ).round(1) df = df[df["distance"] < arguments.radius] df = df.assign( e5_price=pence_to_pounds(df["forecourts.fuel_price.E5"]), e10_price=pence_to_pounds(df["forecourts.fuel_price.E10"]), diesel_price=pence_to_pounds(df["forecourts.fuel_price.B7S"]), ) return df.rename(columns={"forecourts.trading_name": "station_name"})[ ["station_name", "distance", "e5_price", "e10_price", "diesel_price"] ].to_dict(orient="records") def sort_stations(stations: list[dict], sort: str) -> list[dict]: sort_key = SORT_KV[sort] return sorted(stations, key=lambda d: d[sort_key] if d[sort_key] != "N/A" else 999) def output_stations(stations: List[Dict[str, Any]]) -> None: if not stations: print("[*] No stations found.") return print( tabulate( stations, headers={ "station_name": "Station Name", "distance": "Distance (miles)", "e5_price": "E5 (£/L)", "e10_price": "E10 (£/L)", "diesel_price": "B7S (£/L)", }, floatfmt=".2f", ) )