from fastapi import FastAPI, HTTPException from pydantic import BaseModel from datetime import datetime import pandas as pd from train_model import predict_demand import uvicorn app = FastAPI(title="Dynamic Parking Demand Predictor API") class PredictRequest(BaseModel): h3_cell: str timestamp: str class Config: json_schema_extra = { "example": { "h3_cell": "8c2a100d1a2bfff", "timestamp": "2026-01-19 14:00:00" } } class PredictResponse(BaseModel): h3_cell: str timestamp: str demand_factor: float class HealthResponse(BaseModel): status: str @app.post('/predict', response_model=PredictResponse) def predict(request: PredictRequest): """ Predict demand factor for a given h3_cell and timestamp. Expected JSON payload: { "h3_cell": "string", "timestamp": "YYYY-MM-DD HH:MM:SS" or datetime string } Returns: { "h3_cell": "string", "timestamp": "string", "demand_factor": float } """ try: # Convert timestamp to datetime timestamp = pd.to_datetime(request.timestamp).round('H') # Predict demand demand_factor = predict_demand(request.h3_cell, timestamp) if demand_factor is None: raise HTTPException( status_code=404, detail=f"No data found for h3_cell: {request.h3_cell} at timestamp: {request.timestamp}" ) return PredictResponse( h3_cell=request.h3_cell, timestamp=str(timestamp), demand_factor=demand_factor ) except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get('/health', response_model=HealthResponse) def health(): """Health check endpoint""" return HealthResponse(status="healthy") @app.get('/') def root(): """Root endpoint with API information""" return { "message": "Dynamic Parking Demand Predictor API", "endpoints": { "POST /predict": "Predict demand factor", "GET /health": "Health check", "GET /docs": "Interactive API documentation" } } if __name__ == '__main__': uvicorn.run(app, host='0.0.0.0', port=7860)