Spaces:
Sleeping
Sleeping
File size: 2,962 Bytes
6d9b502 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | # app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict, Any
import joblib, os, time, math
import numpy as np
APP_PORT = int(os.environ.get("PORT", 7860)) # HF maps to 7860
app = FastAPI(title="Transit Tracker API")
# In-memory store for vehicles
VEHICLES: Dict[str, Dict[str, Any]] = {}
MODEL = None
MODEL_PATH = "models/model.joblib"
# -- Utilities
def haversine(lat1, lon1, lat2, lon2):
# returns meters
R = 6371000
phi1, phi2 = math.radians(lat1), math.radians(lat2)
dphi = math.radians(lat2-lat1)
dlambda = math.radians(lon2-lon1)
a = math.sin(dphi/2)**2 + math.cos(phi1)*math.cos(phi2)*math.sin(dlambda/2)**2
return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))
def load_model():
global MODEL
if MODEL is None:
if not os.path.exists(MODEL_PATH):
raise RuntimeError("Model file not found: " + MODEL_PATH)
MODEL = joblib.load(MODEL_PATH)
return MODEL
# -- Request models
class VehicleUpdate(BaseModel):
vehicle_id: str
lat: float
lon: float
speed: float = 0.0 # meters/sec
bearing: float = None
timestamp: str = None # ISO
class PredictRequest(BaseModel):
distance_to_stop_m: float
speed_mps: float = 3.0
timestamp_iso: str = None # optional
# -- Endpoints
@app.get("/health")
def health():
ok = os.path.exists(MODEL_PATH)
return {"status": "ok" if ok else "model-missing", "model_path": MODEL_PATH, "time": time.time()}
@app.post("/vehicle/update")
def vehicle_update(v: VehicleUpdate):
VEHICLES[v.vehicle_id] = {
"lat": v.lat,
"lon": v.lon,
"speed": v.speed,
"bearing": v.bearing,
"timestamp": v.timestamp or time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
}
return {"ok": True, "vehicle_id": v.vehicle_id}
@app.get("/vehicles")
def get_vehicles():
return VEHICLES
@app.post("/predict")
def predict(req: PredictRequest):
# safety checks
if req.distance_to_stop_m < 0:
raise HTTPException(status_code=400, detail="distance must be >= 0")
model = load_model()
# compute TOD features (simple UTC seconds)
if req.timestamp_iso:
try:
import datetime
dt = datetime.datetime.fromisoformat(req.timestamp_iso.replace("Z",""))
sec = dt.hour*3600 + dt.minute*60 + dt.second
dow = dt.weekday()
except:
sec = time.gmtime().tm_hour*3600
dow = time.gmtime().tm_wday
else:
sec = time.gmtime().tm_hour*3600 + time.gmtime().tm_min*60 + time.gmtime().tm_sec
dow = time.gmtime().tm_wday
tod_sin = np.sin(2*np.pi*sec/86400)
tod_cos = np.cos(2*np.pi*sec/86400)
X = [[float(req.distance_to_stop_m), float(req.speed_mps), float(tod_sin), float(tod_cos), int(dow)]]
eta_seconds = float(model.predict(X)[0])
return {"eta_seconds": eta_seconds, "eta_minutes": eta_seconds/60.0}
|