Spaces:
Sleeping
Sleeping
File size: 2,565 Bytes
513c1a9 | 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 | from fastapi import FastAPI, HTTPException, Query
from weather_pipeline import WeatherPipeline
from model import TrafficRiskModel
import os
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
app = FastAPI(title="Weather-Traffic Risk API")
# Setup
WEATHER_API_KEY = os.getenv("WEATHER_API_KEY", "your_api_key_here")
weather_pipeline = WeatherPipeline(WEATHER_API_KEY)
risk_model = TrafficRiskModel()
@app.get("/")
async def root():
return {
"message": "Welcome to the Weather-Traffic Risk Prediction API",
"endpoints": {
"/weather/{city}": "Get weather data and traffic risk prediction for a city"
}
}
@app.get("/weather/{city}")
async def get_weather_risk(
city: str,
lat: float = Query(None, description="Optional latitude for high precision"),
lon: float = Query(None, description="Optional longitude for high precision")
):
# 1. Fetch and process weather data (Lat/Lon takes priority for precision)
data = weather_pipeline.get_weather(city=city, lat=lat, lon=lon)
if "error" in data:
raise HTTPException(status_code=400, detail=data["error"])
# 2. Predict Traffic Risk using ML model
risk_level, confidence = risk_model.predict(
data["temp"],
data["rain"],
data["condition"]
)
# 3. Combine results in the FINAL God-Level format
result = {
"city": city,
"timestamp": data.get("timestamp"),
"source": data.get("source"),
"coordinates": {"lat": lat, "lon": lon} if lat and lon else "Default City Center",
"weather": {
"actual_temp": data["temp"],
"feels_like": data["feels_like"],
"rainfall_mm": data["rain"],
"condition": data["condition"],
"condition_raw": data["condition_raw"]
},
"derived_weather_metrics": data["derived_metrics"],
"ml_prediction": {
"traffic_risk_level": risk_level,
"confidence": round(float(confidence), 2),
"status_color": get_risk_color(risk_level)
}
}
return result
def get_risk_color(risk_level):
mapping = {
"Low": "🟢 Green",
"Medium": "🟡 Yellow",
"High": "🔴 Red"
}
return mapping.get(risk_level, "⚪ Unknown")
if __name__ == "__main__":
import uvicorn
# Train model on startup if needed
if not risk_model.load():
print("Training model...")
risk_model.train()
uvicorn.run(app, host="0.0.0.0", port=8000)
|