Initial commit

This commit is contained in:
2026-07-11 19:30:52 +02:00
commit 7183ae31bc
8 changed files with 2831 additions and 0 deletions
+611
View File
@@ -0,0 +1,611 @@
import asyncio
import json
import os
import time
from pathlib import Path
from typing import Optional
import httpx
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address
load_dotenv()
# Rate limiter — uses client IP. When behind a reverse proxy, ensure the proxy
# sets X-Forwarded-For so get_remote_address resolves correctly.
limiter = Limiter(key_func=get_remote_address)
app = FastAPI(title="Travel Monitor")
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
BASE_DIR = Path(__file__).parent
LOCATIONS_FILE = Path(os.getenv("DATA_DIR", "/data")) / "locations.json"
STATIC_DIR = BASE_DIR / "static"
# ── Config ────────────────────────────────────────────────────────────────────
APP_TITLE = os.getenv("APP_TITLE", "Travel Monitor")
HQ_NAME = os.getenv("HQ_NAME", "")
HQ_ADDRESS = os.getenv("HQ_ADDRESS", "")
HQ_LAT = os.getenv("HQ_LAT", "")
HQ_LNG = os.getenv("HQ_LNG", "")
REFRESH_INTERVAL = int(os.getenv("REFRESH_INTERVAL", "300"))
THRESHOLD_YELLOW = float(os.getenv("THRESHOLD_YELLOW", "1.20"))
THRESHOLD_RED = float(os.getenv("THRESHOLD_RED", "1.50"))
THRESHOLD_DARK_RED = float(os.getenv("THRESHOLD_DARK_RED", "1.70"))
ROUTING_PROVIDER = os.getenv("ROUTING_PROVIDER", "openrouteservice").lower()
# Provider API keys — only the one matching ROUTING_PROVIDER is required
ORS_API_KEY = os.getenv("ORS_API_KEY", "") # openrouteservice
GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY", "") # google
MAPBOX_API_KEY = os.getenv("MAPBOX_API_KEY", "") # mapbox
HERE_API_KEY = os.getenv("HERE_API_KEY", "") # here
# ── Shared state ──────────────────────────────────────────────────────────────
_cache: dict = {}
_hq_coords: Optional[tuple] = None
# ── Persistence ───────────────────────────────────────────────────────────────
def load_locations() -> list:
if not LOCATIONS_FILE.exists():
return []
with open(LOCATIONS_FILE) as f:
return json.load(f)
def save_locations(locations: list):
with open(LOCATIONS_FILE, "w") as f:
json.dump(locations, f, indent=2)
def get_status(ratio: float) -> str:
if ratio < THRESHOLD_YELLOW:
return "green"
elif ratio < THRESHOLD_RED:
return "yellow"
elif ratio < THRESHOLD_DARK_RED:
return "red"
else:
return "dark_red"
# ── Provider: OpenRouteService ────────────────────────────────────────────────
async def ors_geocode(client: httpx.AsyncClient, address: str) -> tuple:
r = await client.get(
"https://api.openrouteservice.org/geocode/search",
params={"api_key": ORS_API_KEY, "text": address, "size": 1},
timeout=10,
)
r.raise_for_status()
data = r.json()
if not data.get("features"):
raise ValueError(f"No result for: {address}")
coords = data["features"][0]["geometry"]["coordinates"]
return (coords[1], coords[0]) # lat, lng
async def ors_directions(client: httpx.AsyncClient, hq: tuple, dest: tuple) -> dict:
# NOTE: ORS's free driving-car profile does NOT factor in live traffic at all —
# it always returns the same historical/typical duration regardless of time of day.
# If you need real-time traffic awareness, use google, mapbox, or here instead.
body = {
"coordinates": [[hq[1], hq[0]], [dest[1], dest[0]]],
"preference": "fastest",
}
r = await client.post(
"https://api.openrouteservice.org/v2/directions/driving-car",
json=body,
headers={"Authorization": ORS_API_KEY, "Content-Type": "application/json"},
timeout=15,
)
r.raise_for_status()
route = r.json()["routes"][0]
return {
"duration": route["summary"]["duration"],
"distance_km": round(route["summary"]["distance"] / 1000, 1),
"geometry": route.get("geometry"),
"geometry_type": "polyline",
}
# ── Provider: Google Maps ─────────────────────────────────────────────────────
async def google_geocode(client: httpx.AsyncClient, address: str) -> tuple:
r = await client.get(
"https://maps.googleapis.com/maps/api/geocode/json",
params={"address": address, "key": GOOGLE_API_KEY},
timeout=10,
)
r.raise_for_status()
data = r.json()
if data.get("status") != "OK" or not data.get("results"):
raise ValueError(f"No result for: {address} ({data.get('status')})")
loc = data["results"][0]["geometry"]["location"]
return (loc["lat"], loc["lng"])
async def google_directions(client: httpx.AsyncClient, hq: tuple, dest: tuple) -> dict:
r = await client.get(
"https://maps.googleapis.com/maps/api/directions/json",
params={
"origin": f"{hq[0]},{hq[1]}",
"destination": f"{dest[0]},{dest[1]}",
"mode": "driving",
"departure_time": "now", # required to get live-traffic duration
"traffic_model": "best_guess",
"key": GOOGLE_API_KEY,
},
timeout=15,
)
r.raise_for_status()
data = r.json()
if data.get("status") != "OK" or not data.get("routes"):
raise ValueError(f"No route found ({data.get('status')})")
leg = data["routes"][0]["legs"][0]
poly = data["routes"][0]["overview_polyline"]["points"]
# duration_in_traffic only appears when departure_time is set; fall back to duration if missing
duration = leg.get("duration_in_traffic", leg["duration"])["value"]
return {
"duration": duration,
"distance_km": round(leg["distance"]["value"] / 1000, 1),
"geometry": poly,
"geometry_type": "polyline",
}
# ── Provider: Mapbox ──────────────────────────────────────────────────────────
async def mapbox_geocode(client: httpx.AsyncClient, address: str) -> tuple:
import urllib.parse
encoded = urllib.parse.quote(address)
r = await client.get(
f"https://api.mapbox.com/geocoding/v5/mapbox.places/{encoded}.json",
params={"access_token": MAPBOX_API_KEY, "limit": 1},
timeout=10,
)
r.raise_for_status()
data = r.json()
if not data.get("features"):
raise ValueError(f"No result for: {address}")
coords = data["features"][0]["geometry"]["coordinates"]
return (coords[1], coords[0]) # lat, lng
async def mapbox_directions(client: httpx.AsyncClient, hq: tuple, dest: tuple) -> dict:
coords = f"{hq[1]},{hq[0]};{dest[1]},{dest[0]}"
# driving-traffic profile factors in live + historical traffic conditions
# (plain "driving" profile ignores traffic entirely)
r = await client.get(
f"https://api.mapbox.com/directions/v5/mapbox/driving-traffic/{coords}",
params={
"access_token": MAPBOX_API_KEY,
"geometries": "polyline",
"overview": "full",
},
timeout=15,
)
r.raise_for_status()
data = r.json()
if not data.get("routes"):
raise ValueError("No route found")
route = data["routes"][0]
return {
"duration": route["duration"],
"distance_km": round(route["distance"] / 1000, 1),
"geometry": route["geometry"],
"geometry_type": "polyline",
}
# ── Provider: HERE ────────────────────────────────────────────────────────────
async def here_geocode(client: httpx.AsyncClient, address: str) -> tuple:
r = await client.get(
"https://geocode.search.hereapi.com/v1/geocode",
params={"q": address, "apiKey": HERE_API_KEY, "limit": 1},
timeout=10,
)
r.raise_for_status()
data = r.json()
if not data.get("items"):
raise ValueError(f"No result for: {address}")
pos = data["items"][0]["position"]
return (pos["lat"], pos["lng"])
async def here_directions(client: httpx.AsyncClient, hq: tuple, dest: tuple) -> dict:
# routingMode=fast + return=summary already factors in live traffic by default on HERE.
# departureTime=now makes this explicit so cached/historical routes aren't served.
r = await client.get(
"https://router.hereapi.com/v8/routes",
params={
"apiKey": HERE_API_KEY,
"transportMode": "car",
"origin": f"{hq[0]},{hq[1]}",
"destination": f"{dest[0]},{dest[1]}",
"return": "summary,polyline",
"routingMode": "fast",
"departureTime": "now",
},
timeout=15,
)
r.raise_for_status()
data = r.json()
if not data.get("routes"):
raise ValueError("No route found")
section = data["routes"][0]["sections"][0]
duration = section["summary"]["duration"] # seconds
# HERE returns a Flexible Polyline encoded string
polyline_str = section.get("polyline", None)
return {
"duration": duration,
"distance_km": round(section["summary"]["length"] / 1000, 1),
"geometry": polyline_str,
"geometry_type": "here_polyline",
}
# ── Provider dispatch ─────────────────────────────────────────────────────────
PROVIDERS = {
"openrouteservice": (ors_geocode, ors_directions),
"google": (google_geocode, google_directions),
"mapbox": (mapbox_geocode, mapbox_directions),
"here": (here_geocode, here_directions),
}
async def geocode_address(address: str) -> tuple:
if ROUTING_PROVIDER not in PROVIDERS:
raise ValueError(f"Unknown ROUTING_PROVIDER: '{ROUTING_PROVIDER}'. "
f"Valid options: {', '.join(PROVIDERS)}")
geocode_fn, _ = PROVIDERS[ROUTING_PROVIDER]
async with httpx.AsyncClient() as client:
return await geocode_fn(client, address)
async def fetch_travel_time(hq: tuple, dest: tuple) -> dict:
if ROUTING_PROVIDER not in PROVIDERS:
raise ValueError(f"Unknown ROUTING_PROVIDER: '{ROUTING_PROVIDER}'")
_, directions_fn = PROVIDERS[ROUTING_PROVIDER]
try:
async with httpx.AsyncClient() as client:
result = await directions_fn(client, hq, dest)
print(f"[{ROUTING_PROVIDER}] duration={result['duration']}s "
f"({result['duration']/60:.1f}min)")
return result
except httpx.HTTPStatusError as e:
# Capture the response body so we see the actual provider error message
try:
body = e.response.json()
except Exception:
body = e.response.text
raise ValueError(f"HTTP {e.response.status_code} from {ROUTING_PROVIDER}: {body}") from e
# ── HQ resolution ─────────────────────────────────────────────────────────────
async def get_hq_coords() -> tuple:
global _hq_coords
if _hq_coords:
return _hq_coords
if HQ_LAT and HQ_LNG:
_hq_coords = (float(HQ_LAT), float(HQ_LNG))
elif HQ_ADDRESS:
_hq_coords = await geocode_address(HQ_ADDRESS)
else:
raise ValueError("No HQ location configured in .env")
return _hq_coords
# ── Refresh all locations ─────────────────────────────────────────────────────
_provider_errors: dict = {} # loc_id -> {"message": str, "at": int}
_last_refresh_ok: Optional[int] = None
_last_refresh_attempt: Optional[int] = None
_hq_error: Optional[str] = None
# Semaphore limits concurrent routing API calls — prevents rate limit errors
# when you have many locations. 5 at a time is safe for all providers.
_fetch_semaphore = asyncio.Semaphore(5)
async def fetch_with_semaphore(hq: tuple, dest: tuple) -> dict:
async with _fetch_semaphore:
return await fetch_travel_time(hq, dest)
async def refresh_all_locations():
global _last_refresh_ok, _last_refresh_attempt, _hq_error
locations = load_locations()
_last_refresh_attempt = int(time.time())
if not locations:
return
try:
hq = await get_hq_coords()
_hq_error = None
except Exception as e:
_hq_error = str(e)
print(f"[ERROR] Could not resolve HQ: {e}")
return
tasks = [fetch_with_semaphore(hq, (loc["lat"], loc["lng"])) for loc in locations]
results = await asyncio.gather(*tasks, return_exceptions=True)
updated = False
any_success = False
for loc, result in zip(locations, results):
lid = loc["id"]
if isinstance(result, Exception):
print(f"[WARN] Failed for {loc['name']}: {result!r}")
_provider_errors[lid] = {"message": str(result), "at": int(time.time())}
if lid in _cache:
_cache[lid]["error"] = True
continue
any_success = True
_provider_errors.pop(lid, None)
one_way_seconds = int(result["duration"])
geometry = result.get("geometry")
geometry_type = result.get("geometry_type", "polyline")
if "baseline_seconds" not in loc or loc["baseline_seconds"] is None:
loc["baseline_seconds"] = one_way_seconds
updated = True
baseline = loc["baseline_seconds"]
ratio = one_way_seconds / baseline if baseline > 0 else 1.0
_cache[lid] = {
"current_seconds": one_way_seconds,
"baseline_seconds": baseline,
"ratio": round(ratio, 3),
"status": get_status(ratio),
"one_way_min": round(one_way_seconds / 60),
"round_trip_min": round(one_way_seconds / 60 * 2),
"distance_km": result.get("distance_km"),
"geometry": geometry,
"geometry_type": geometry_type,
"last_updated": int(time.time()),
"error": False,
}
if any_success:
_last_refresh_ok = int(time.time())
if updated:
save_locations(locations)
# ── Background refresh loop ───────────────────────────────────────────────────
@app.on_event("startup")
async def startup():
asyncio.create_task(background_refresh())
async def background_refresh():
while True:
try:
await refresh_all_locations()
except Exception as e:
print(f"[ERROR] Background refresh: {e}")
await asyncio.sleep(REFRESH_INTERVAL)
# ── API routes ────────────────────────────────────────────────────────────────
@app.get("/api/config")
@limiter.limit("60/minute")
async def get_config(request: Request):
if HQ_NAME:
hq_label = HQ_NAME
elif HQ_ADDRESS:
hq_label = HQ_ADDRESS
else:
hq_label = f"{HQ_LAT}, {HQ_LNG}"
try:
hq = await get_hq_coords()
return {
"app_title": APP_TITLE,
"hq_label": hq_label,
"hq_lat": hq[0],
"hq_lng": hq[1],
"refresh_interval": REFRESH_INTERVAL,
"routing_provider": ROUTING_PROVIDER,
"thresholds": {
"yellow": THRESHOLD_YELLOW,
"red": THRESHOLD_RED,
"dark_red": THRESHOLD_DARK_RED,
},
}
except Exception as e:
raise HTTPException(500, str(e))
@app.get("/api/locations")
@limiter.limit("60/minute")
async def get_locations(request: Request):
locations = load_locations()
result = []
for loc in locations:
entry = {
"id": loc["id"],
"name": loc["name"],
"lat": loc["lat"],
"lng": loc["lng"],
}
if loc["id"] in _cache:
entry.update(_cache[loc["id"]])
elif loc["id"] in _provider_errors:
entry["error"] = True
result.append(entry)
return result
@app.get("/api/status")
@limiter.limit("30/minute")
async def get_status_summary(request: Request):
locations = load_locations()
failing = [
{
"id": loc["id"],
"name": loc["name"],
"message": _provider_errors[loc["id"]]["message"],
"at": _provider_errors[loc["id"]]["at"],
}
for loc in locations
if loc["id"] in _provider_errors
]
return {
"routing_provider": ROUTING_PROVIDER,
"hq_error": _hq_error,
"last_refresh_attempt": _last_refresh_attempt,
"last_refresh_ok": _last_refresh_ok,
"refresh_interval": REFRESH_INTERVAL,
"total_locations": len(locations),
"failing_locations": failing,
"healthy": _hq_error is None and len(failing) == 0,
}
@app.post("/api/refresh")
@limiter.limit("10/minute")
async def manual_refresh(request: Request):
await refresh_all_locations()
return {"ok": True}
class AddByAddress(BaseModel):
name: str
address: str
class AddByCoords(BaseModel):
name: str
lat: float
lng: float
class RenameLocation(BaseModel):
name: str
@app.post("/api/locations/add/address")
@limiter.limit("20/minute")
async def add_by_address(request: Request, body: AddByAddress):
try:
lat, lng = await geocode_address(body.address)
except Exception as e:
raise HTTPException(400, f"Geocoding failed: {e}")
locations = load_locations()
new_id = str(int(time.time() * 1000))
locations.append({"id": new_id, "name": body.name, "lat": lat, "lng": lng, "baseline_seconds": None})
save_locations(locations)
asyncio.create_task(refresh_all_locations())
return {"ok": True, "id": new_id, "lat": lat, "lng": lng}
@app.post("/api/locations/add/coords")
@limiter.limit("20/minute")
async def add_by_coords(request: Request, body: AddByCoords):
locations = load_locations()
new_id = str(int(time.time() * 1000))
locations.append({"id": new_id, "name": body.name, "lat": body.lat, "lng": body.lng, "baseline_seconds": None})
save_locations(locations)
asyncio.create_task(refresh_all_locations())
return {"ok": True, "id": new_id}
@app.patch("/api/locations/{loc_id}/rename")
@limiter.limit("20/minute")
async def rename_location(request: Request, loc_id: str, body: RenameLocation):
locations = load_locations()
for loc in locations:
if loc["id"] == loc_id:
loc["name"] = body.name.strip()
save_locations(locations)
return {"ok": True}
raise HTTPException(404, "Location not found")
@app.delete("/api/locations")
@limiter.limit("20/minute")
async def remove_locations(request: Request, ids: list[str]):
locations = load_locations()
locations = [l for l in locations if l["id"] not in ids]
save_locations(locations)
for lid in ids:
_cache.pop(lid, None)
return {"ok": True}
class SimulateDelay(BaseModel):
ratio: float # e.g. 1.35 = simulate 35% worse than baseline. Use 1.0 to clear.
@app.post("/api/locations/{loc_id}/simulate-delay")
@limiter.limit("20/minute")
async def simulate_delay(request: Request, loc_id: str, body: SimulateDelay):
"""Debug/testing endpoint. Call with body {"ratio": 1.6} to fake a 60% delay
on this location, to verify status colors and thresholds update correctly
without waiting for real traffic changes. POST {"ratio": 1.0} to clear it.
Overwritten by the next real refresh unless REFRESH_INTERVAL is paused."""
if loc_id not in _cache:
raise HTTPException(404, "No cached data for this location yet — refresh first")
baseline = _cache[loc_id]["baseline_seconds"]
fake_seconds = int(baseline * body.ratio)
_cache[loc_id]["current_seconds"] = fake_seconds
_cache[loc_id]["ratio"] = round(body.ratio, 3)
_cache[loc_id]["status"] = get_status(body.ratio)
_cache[loc_id]["one_way_min"] = round(fake_seconds / 60)
_cache[loc_id]["round_trip_min"] = round(fake_seconds / 60 * 2)
_cache[loc_id]["simulated"] = body.ratio != 1.0
_cache[loc_id]["last_updated"] = int(time.time())
return {"ok": True, "simulated_ratio": body.ratio, "status": _cache[loc_id]["status"]}
@app.post("/api/locations/{loc_id}/reset-baseline")
@limiter.limit("20/minute")
async def reset_baseline(request: Request, loc_id: str):
locations = load_locations()
for loc in locations:
if loc["id"] == loc_id:
loc["baseline_seconds"] = None
save_locations(locations)
if loc_id in _cache:
_cache[loc_id]["baseline_seconds"] = None
_cache[loc_id]["ratio"] = 1.0
_cache[loc_id]["status"] = "green"
asyncio.create_task(refresh_all_locations())
return {"ok": True}
raise HTTPException(404, "Location not found")
class ReorderBody(BaseModel):
ids: list[str]
@app.post("/api/locations/reorder")
@limiter.limit("30/minute")
async def reorder_locations(request: Request, body: ReorderBody):
locations = load_locations()
loc_map = {l["id"]: l for l in locations}
reordered = [loc_map[i] for i in body.ids if i in loc_map]
# Append any locations not in the reorder list (safety net)
seen = set(body.ids)
reordered += [l for l in locations if l["id"] not in seen]
save_locations(reordered)
return {"ok": True}
# ── Serve frontend ────────────────────────────────────────────────────────────
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
@app.get("/")
async def root():
return FileResponse(STATIC_DIR / "index.html")