admin/Mileage-Logger
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit (clean, ignores in place)

Browse Source
This commit is contained in:
Kieran Bolt-Biggs
2025-08-12 01:13:41 +01:00
commit c74790b014
26 changed files with 2331 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
mileage_logger/distance/__init__.py Normal file
View File

@@ -0,0 +1,12 @@
"""Distance resolution utilities.
This subpackage exposes classes and functions that resolve driving
distances between pairs of recognised sites. The primary
implementation is :class:`DistanceResolver`, which first consults a
pre-defined route catalogue before optionally consulting an external
API and finally falling back to a simple geodesic calculation.
"""
from .resolve import DistanceResolver, haversine_distance
__all__ = ["DistanceResolver", "haversine_distance"]

210
mileage_logger/distance/resolve.py Normal file
View File

@@ -0,0 +1,210 @@
"""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