File size: 2,380 Bytes
e118065
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
76270bc
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
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)