Spaces:
Sleeping
Sleeping
File size: 3,366 Bytes
85f8e77 | 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 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | # app.py — Single-file Transit Tracker API for Hugging Face (No extra files required)
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict, Any
import numpy as np
import joblib
import os
import math
import time
app = FastAPI(title="Transit Tracker API - Single File Version")
# -----------------------------
# In-memory Vehicle Locations
# -----------------------------
VEHICLES: Dict[str, Dict[str, Any]] = {}
# -----------------------------
# Train a lightweight ETA model
# -----------------------------
def train_and_save_model():
from sklearn.ensemble import RandomForestRegressor
np.random.seed(42)
N = 3000
dist = np.random.exponential(400, N)
speed = np.random.uniform(5, 15, N)
tod = np.random.uniform(0, 86400, N)
dow = np.random.randint(0, 7, N)
tod_sin = np.sin(2*np.pi * tod/86400)
tod_cos = np.cos(2*np.pi * tod/86400)
eta = dist/(speed*0.277) + np.random.randint(10, 120, N)
X = np.column_stack([dist, speed, tod_sin, tod_cos, dow])
y = eta
model = RandomForestRegressor(n_estimators=30)
model.fit(X, y)
joblib.dump(model, "model.joblib")
print("MODEL TRAINED AND SAVED ✔")
# -----------------------------
# Load or Train Model
# -----------------------------
def load_model():
if not os.path.exists("model.joblib"):
train_and_save_model()
return joblib.load("model.joblib")
MODEL = load_model()
# -----------------------------
# JSON Schemas
# -----------------------------
class VehicleUpdate(BaseModel):
vehicle_id: str
lat: float
lon: float
speed: float = 0.0
class PredictRequest(BaseModel):
distance_to_stop_m: float
speed_mps: float = 4.0
# -----------------------------
# Utils
# -----------------------------
def haversine(lat1, lon1, lat2, lon2):
R = 6371000
p1, p2 = math.radians(lat1), math.radians(lat2)
dphi = math.radians(lat2 - lat1)
dl = math.radians(lon2 - lon1)
a = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
# -----------------------------
# ROOT
# -----------------------------
@app.get("/")
def root():
return {"message": "Transit Tracker API Running", "vehicles": len(VEHICLES)}
# -----------------------------
# LIVE VEHICLE LOCATION APIs
# -----------------------------
@app.post("/vehicle/update")
def update_vehicle(v: VehicleUpdate):
VEHICLES[v.vehicle_id] = {
"lat": v.lat,
"lon": v.lon,
"speed": v.speed,
"timestamp": time.time()
}
return {"success": True, "vehicle": VEHICLES[v.vehicle_id]}
@app.get("/vehicles")
def get_all_vehicles():
return VEHICLES
# -----------------------------
# ETA PREDICTION
# -----------------------------
@app.post("/predict")
def predict_eta(req: PredictRequest):
dist = req.distance_to_stop_m
if dist < 0:
raise HTTPException(400, "distance cannot be negative")
t = time.gmtime()
seconds = t.tm_hour * 3600 + t.tm_min * 60 + t.tm_sec
tod_sin = np.sin(2*np.pi * seconds/86400)
tod_cos = np.cos(2*np.pi * seconds/86400)
dow = t.tm_wday
X = np.array([[dist, req.speed_mps, tod_sin, tod_cos, dow]])
eta_sec = float(MODEL.predict(X)[0])
return {"eta_seconds": eta_sec, "eta_minutes": eta_sec / 60}
|