Spaces:

Alimiji
/

weather-prediction-api

Sleeping

File size: 9,246 Bytes

e97af1b

"""FastAPI application for weather temperature prediction.

Deployed on Hugging Face Spaces: https://alimiji-weather-prediction-api.hf.space
Last deployment trigger: 2026-01-19
"""

import json
import logging
import os
from contextlib import asynccontextmanager
from pathlib import Path

import joblib
import numpy as np
import requests
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware

from .schemas import (
    BatchPredictionRequest,
    BatchPredictionResponse,
    HealthResponse,
    ModelMetrics,
    PredictionResponse,
    WeatherFeatures,
)

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Global model storage
model = None
model_info = None
metrics = None

# Paths
ROOT = Path(__file__).resolve().parent.parent.parent
MODEL_PATH = ROOT / "models" / "random_forest" / "Production" / "model.pkl"
MODEL_INFO_PATH = ROOT / "models" / "random_forest" / "Production" / "model_info.json"
METRICS_PATH = ROOT / "metrics.json"

# Feature order (must match training)
FEATURE_ORDER = [
    "min_temp",
    "max_temp",
    "global_radiation",
    "sunshine",
    "cloud_cover",
    "precipitation",
    "pressure",
    "snow_depth",
]


def download_model_from_dagshub():
    """Download model from DagsHub via HTTP."""
    dagshub_token = os.getenv("DAGSHUB_TOKEN")
    dagshub_user = os.getenv("DAGSHUB_USER", "Alimiji")

    if not dagshub_token:
        logger.warning("DAGSHUB_TOKEN not set, skipping model download")
        return False

    # DVC file hashes from dvc.lock
    files_to_download = [
        ("models/random_forest/Production/model_info.json", "006851e7426c173879e57b2b91201121"),
        ("models/random_forest/Production/model.pkl", "44bebd223b998cf7e177aed1e73de3a6"),
    ]

    base_url = "https://dagshub.com/Alimiji/mlops_alimiji1.dvc/files/md5"

    logger.info(f"ROOT path: {ROOT}")
    logger.info(f"MODEL_PATH: {MODEL_PATH}")

    try:
        for file_path, md5_hash in files_to_download:
            full_path = ROOT / file_path
            logger.info(f"Target path: {full_path}")

            full_path.parent.mkdir(parents=True, exist_ok=True)

            # DagsHub DVC storage URL format: /files/md5/{first2chars}/{remaining}
            url = f"{base_url}/{md5_hash[:2]}/{md5_hash[2:]}"
            logger.info(f"Downloading {file_path} from {url}...")

            response = requests.get(
                url,
                auth=(dagshub_user, dagshub_token),
                stream=True,
                timeout=(30, 600),  # 30s connect, 600s read timeout for large file
            )
            response.raise_for_status()

            total_size = int(response.headers.get("content-length", 0))
            logger.info(f"File size: {total_size / 1024 / 1024:.1f} MB")

            downloaded = 0
            with open(full_path, "wb") as f:
                for chunk in response.iter_content(chunk_size=65536):
                    f.write(chunk)
                    downloaded += len(chunk)

            logger.info(f"Downloaded {file_path}: {downloaded / 1024 / 1024:.1f} MB")

            # Verify file exists
            if full_path.exists():
                logger.info(f"Verified: {full_path} exists, size: {full_path.stat().st_size}")
            else:
                logger.error(f"File not found after download: {full_path}")
                return False

        return True

    except requests.RequestException as e:
        logger.error(f"Failed to download model: {e}")
        return False
    except Exception as e:
        logger.error(f"Unexpected error downloading model: {e}")
        import traceback

        logger.error(traceback.format_exc())
        return False


def load_model():
    """Load the trained model and metadata."""
    global model, model_info, metrics

    # Try to download model if not present
    if not MODEL_PATH.exists():
        logger.info("Model not found locally, downloading from DagsHub...")
        download_model_from_dagshub()

    if not MODEL_PATH.exists():
        logger.error(f"Model not found at {MODEL_PATH}")
        raise FileNotFoundError(f"Model not found at {MODEL_PATH}")

    logger.info(f"Loading model from {MODEL_PATH}")
    model = joblib.load(MODEL_PATH)

    if MODEL_INFO_PATH.exists():
        model_info = json.loads(MODEL_INFO_PATH.read_text(encoding="utf-8"))
        logger.info(f"Model info loaded: run_id={model_info.get('run_id', 'unknown')}")
    else:
        model_info = {"run_id": "unknown"}

    if METRICS_PATH.exists():
        metrics = json.loads(METRICS_PATH.read_text(encoding="utf-8"))
        logger.info("Metrics loaded successfully")

    logger.info("Model loaded successfully")


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan manager."""
    # Startup
    try:
        load_model()
    except FileNotFoundError as e:
        logger.warning(f"Model not loaded at startup: {e}")
    yield
    # Shutdown
    logger.info("Shutting down API")


# Create FastAPI app
app = FastAPI(
    title="Weather Temperature Prediction API",
    description="API for predicting mean temperature based on weather features using Random Forest model",
    version="1.0.0",
    lifespan=lifespan,
)

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/", tags=["Root"])
async def root():
    """Root endpoint with API information."""
    return {
        "message": "Weather Temperature Prediction API",
        "docs": "/docs",
        "health": "/health",
    }


@app.get("/health", response_model=HealthResponse, tags=["Health"])
async def health_check():
    """Check API health and model status."""
    return HealthResponse(
        status="healthy" if model is not None else "degraded",
        model_loaded=model is not None,
        model_path=str(MODEL_PATH) if model is not None else None,
    )


@app.post("/predict", response_model=PredictionResponse, tags=["Predictions"])
async def predict(features: WeatherFeatures):
    """
    Predict mean temperature from weather features.

    Returns the predicted mean temperature in Celsius.
    """
    if model is None:
        raise HTTPException(status_code=503, detail="Model not loaded")

    # Convert features to numpy array in correct order
    feature_values = [getattr(features, name) for name in FEATURE_ORDER]
    X = np.array([feature_values])

    # Make prediction
    prediction = model.predict(X)[0]

    return PredictionResponse(
        predicted_mean_temp=round(float(prediction), 2),
        model_version=model_info.get("run_id", "unknown") if model_info else "unknown",
    )


@app.post("/predict/batch", response_model=BatchPredictionResponse, tags=["Predictions"])
async def predict_batch(request: BatchPredictionRequest):
    """
    Batch prediction for multiple weather instances.

    Accepts up to 1000 instances per request.
    """
    if model is None:
        raise HTTPException(status_code=503, detail="Model not loaded")

    # Convert all instances to numpy array
    X = np.array([[getattr(instance, name) for name in FEATURE_ORDER] for instance in request.instances])

    # Make predictions
    predictions = model.predict(X)

    return BatchPredictionResponse(
        predictions=[round(float(p), 2) for p in predictions],
        model_version=model_info.get("run_id", "unknown") if model_info else "unknown",
        count=len(predictions),
    )


@app.get("/metrics", response_model=ModelMetrics, tags=["Model Info"])
async def get_metrics():
    """Get model performance metrics from the last evaluation."""
    if metrics is None:
        raise HTTPException(status_code=404, detail="Metrics not available")

    return ModelMetrics(
        train_rmse=metrics["train"]["rmse"],
        train_mae=metrics["train"]["mae"],
        train_r2=metrics["train"]["r2"],
        valid_rmse=metrics["valid"]["rmse"],
        valid_mae=metrics["valid"]["mae"],
        valid_r2=metrics["valid"]["r2"],
        test_rmse=metrics["test"]["rmse"],
        test_mae=metrics["test"]["mae"],
        test_r2=metrics["test"]["r2"],
    )


@app.get("/model/info", tags=["Model Info"])
async def get_model_info():
    """Get information about the loaded model."""
    if model_info is None:
        raise HTTPException(status_code=404, detail="Model info not available")

    return {
        "run_id": model_info.get("run_id"),
        "experiment_name": model_info.get("experiment_name"),
        "model_type": model_info.get("model_type"),
        "params": model_info.get("params"),
        "features": FEATURE_ORDER,
    }


@app.post("/model/reload", tags=["Model Info"])
async def reload_model():
    """Reload the model from disk."""
    try:
        load_model()
        return {"status": "success", "message": "Model reloaded successfully"}
    except FileNotFoundError as e:
        raise HTTPException(status_code=404, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to reload model: {str(e)}")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)