210 lines
8.6 KiB
Python
210 lines
8.6 KiB
Python
"""Resolve driving distances between sites.
|
|
|
|
The :class:`DistanceResolver` class provides a simple mechanism to
|
|
determine the distance in miles between two points. It is designed to
|
|
prefer a local route catalogue (CSV) if available, fall back to
|
|
external API calls when API keys are configured and, as a last
|
|
resort, compute a straight-line distance using the haversine
|
|
formula.
|
|
|
|
Caching is performed to avoid repeated API calls or calculations. A
|
|
time-to-live (TTL) can be specified when constructing the resolver
|
|
although it is currently not enforced in the simple in-memory
|
|
implementation. Distances are rounded to one decimal place as
|
|
required by HR mileage claim forms.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import math
|
|
import os
|
|
import time
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from typing import Dict, Optional, Tuple
|
|
|
|
try:
|
|
import httpx # type: ignore
|
|
except ImportError: # Optional dependency. If unavailable, API calls will be skipped.
|
|
httpx = None # type: ignore
|
|
|
|
from ..logic.detect_itinerary import haversine_distance
|
|
|
|
|
|
@dataclass
|
|
class _CacheEntry:
|
|
distance: float
|
|
timestamp: float
|
|
|
|
|
|
class DistanceResolver:
|
|
"""Resolve driving distances between two locations.
|
|
|
|
The resolver consults an in-memory cache, a local route catalogue,
|
|
an optional external API and finally falls back to a straight-line
|
|
calculation using the haversine formula. Distances are cached for
|
|
the lifetime of the object. Rounding to one decimal mile is
|
|
applied uniformly.
|
|
"""
|
|
|
|
def __init__(self, route_csv_path: Optional[str] = None, api_key: Optional[str] = None,
|
|
http_client: Optional[object] = None, ttl_seconds: float = 365 * 24 * 3600,
|
|
vehicle_label: str = "SH11 DRV (Own 1.6CC Diesel Car/Van)", job_role: str = "ICT Technician"):
|
|
"""Initialise the distance resolver.
|
|
|
|
Parameters
|
|
----------
|
|
route_csv_path : str, optional
|
|
Path to a CSV file containing pre-approved route distances.
|
|
The file should have at least three columns: origin,
|
|
destination and miles. The entries are assumed to be
|
|
directional; if symmetric distances are desired both
|
|
directions must be provided.
|
|
api_key : str, optional
|
|
API key for the Google Routes API. If omitted, API calls
|
|
will be skipped.
|
|
http_client : :class:`httpx.Client`, optional
|
|
HTTP client instance to use for API requests. A new client
|
|
will be created if not provided.
|
|
ttl_seconds : float, optional
|
|
Time-to-live for cache entries in seconds. Expired
|
|
entries are recomputed on demand. The default is one year.
|
|
"""
|
|
|
|
self.api_key = api_key
|
|
# Only store an HTTP client if provided and httpx is available.
|
|
# When httpx is unavailable the client will be ignored and API
|
|
# calls will be skipped.
|
|
self.http_client = http_client if httpx is not None else None
|
|
self.ttl_seconds = ttl_seconds
|
|
self.vehicle_label = vehicle_label
|
|
self.job_role = job_role
|
|
self.cache: Dict[Tuple[str, str], _CacheEntry] = {}
|
|
# Load route catalogue
|
|
self.route_catalog: Dict[Tuple[str, str], float] = {}
|
|
if route_csv_path and os.path.exists(route_csv_path):
|
|
with open(route_csv_path, "r", encoding="utf-8") as f:
|
|
reader = csv.reader(f)
|
|
for row in reader:
|
|
if not row or row[0].startswith("#"):
|
|
continue
|
|
try:
|
|
origin, destination, miles_str = row[:3]
|
|
miles = float(miles_str)
|
|
self.route_catalog[(origin.strip(), destination.strip())] = miles
|
|
except Exception:
|
|
# Skip malformed entries silently
|
|
continue
|
|
|
|
def _get_from_cache(self, origin: str, dest: str) -> Optional[float]:
|
|
"""Retrieve a cached distance if present and unexpired."""
|
|
entry = self.cache.get((origin, dest))
|
|
if entry is None:
|
|
return None
|
|
if (time.time() - entry.timestamp) > self.ttl_seconds:
|
|
# Expired
|
|
return None
|
|
return entry.distance
|
|
|
|
def _set_cache(self, origin: str, dest: str, distance: float) -> None:
|
|
"""Cache the given distance for the origin/destination pair."""
|
|
self.cache[(origin, dest)] = _CacheEntry(distance=distance, timestamp=time.time())
|
|
|
|
def resolve(self, origin_name: str, dest_name: str, origin_coords: Tuple[float, float], dest_coords: Tuple[float, float]) -> float:
|
|
"""Resolve the distance between two sites in miles.
|
|
|
|
This method will consult the cache, route catalogue, external API
|
|
and finally compute a haversine distance. Once resolved, the
|
|
distance is cached and rounded to one decimal place.
|
|
|
|
Parameters
|
|
----------
|
|
origin_name : str
|
|
Canonical name of the origin site. Used for cache and
|
|
catalogue lookups.
|
|
dest_name : str
|
|
Canonical name of the destination site.
|
|
origin_coords : tuple(float, float)
|
|
Latitude and longitude of the origin in decimal degrees.
|
|
dest_coords : tuple(float, float)
|
|
Latitude and longitude of the destination in decimal degrees.
|
|
|
|
Returns
|
|
-------
|
|
float
|
|
The resolved driving distance in miles, rounded to one
|
|
decimal place.
|
|
"""
|
|
|
|
# First check the cache
|
|
cached = self._get_from_cache(origin_name, dest_name)
|
|
if cached is not None:
|
|
return cached
|
|
# Second consult the route catalogue
|
|
catalogue_key = (origin_name, dest_name)
|
|
if catalogue_key in self.route_catalog:
|
|
dist = self.route_catalog[catalogue_key]
|
|
rounded = round(dist, 1)
|
|
self._set_cache(origin_name, dest_name, rounded)
|
|
return rounded
|
|
# Attempt to call external API if configured
|
|
if self.api_key:
|
|
try:
|
|
dist = self._call_google_routes_api(origin_coords, dest_coords)
|
|
if dist is not None:
|
|
rounded = round(dist, 1)
|
|
self._set_cache(origin_name, dest_name, rounded)
|
|
return rounded
|
|
except Exception:
|
|
# Swallow API errors and fall back
|
|
pass
|
|
# Fall back to haversine distance
|
|
dist = haversine_distance(origin_coords[0], origin_coords[1], dest_coords[0], dest_coords[1])
|
|
rounded = round(dist, 1)
|
|
self._set_cache(origin_name, dest_name, rounded)
|
|
return rounded
|
|
|
|
def _call_google_routes_api(self, origin_coords: Tuple[float, float], dest_coords: Tuple[float, float]) -> Optional[float]:
|
|
"""Call the Google Maps Routes API to compute driving distance.
|
|
|
|
Note that this is a blocking call. The caller should ensure that
|
|
network access is permitted and that a valid API key has been
|
|
configured. If the request fails or the response cannot be
|
|
parsed, ``None`` is returned.
|
|
"""
|
|
|
|
# Construct the API request
|
|
# See https://developers.google.com/maps/documentation/routes for details
|
|
base_url = "https://routes.googleapis.com/directions/v2:computeRoutes"
|
|
# Compose JSON payload
|
|
payload = {
|
|
"origin": {"location": {"latLng": {"latitude": origin_coords[0], "longitude": origin_coords[1]}}},
|
|
"destination": {"location": {"latLng": {"latitude": dest_coords[0], "longitude": dest_coords[1]}}},
|
|
"travelMode": "DRIVE",
|
|
"routingPreference": "TRAFFIC_AWARE",
|
|
"computeAlternativeRoutes": False,
|
|
"units": "IMPERIAL",
|
|
}
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"X-Goog-Api-Key": self.api_key,
|
|
"X-Goog-FieldMask": "routes.duration,routes.distanceMeters",
|
|
}
|
|
# If httpx is unavailable, or no API key is configured, skip API call
|
|
if httpx is None or self.http_client is None:
|
|
return None
|
|
resp = self.http_client.post(base_url, json=payload, headers=headers)
|
|
if resp.status_code != 200:
|
|
return None
|
|
try:
|
|
data = resp.json()
|
|
routes = data.get("routes") or []
|
|
if not routes:
|
|
return None
|
|
# Distance is returned in meters; convert to miles
|
|
meters = routes[0]["distanceMeters"]
|
|
miles = meters / 1609.34
|
|
return float(miles)
|
|
except Exception:
|
|
return None |