app.py

from fastapi import FastAPI
import joblib
import pandas as pd
from datetime import datetime
from typing import Literal, Annotated
from pydantic import BaseModel, Field

import warnings
warnings.filterwarnings("ignore", category=UserWarning)



import os
import requests

HF_REPO = "samithcs/heart-rate-models"
HEART_MODEL_FILENAME = "Heart_Rate_Predictor_model.joblib"
ANOMALY_MODEL_FILENAME = "Anomaly_Detector_model.joblib"


MODEL_DIR = os.path.join("artifacts", "model_trainer")
os.makedirs(MODEL_DIR, exist_ok=True)

def download_from_hf(filename):
    local_path = os.path.join(MODEL_DIR, filename)

  
    if os.path.exists(local_path):
        print(f"✅ {filename} already exists at {local_path}")
        return local_path

 
    url = f"https://huggingface.co/{HF_REPO}/resolve/main/{filename}"
    print(f"⬇️ Downloading {filename} from {url} ...")
    with requests.get(url, stream=True) as r:
        r.raise_for_status()
        with open(local_path, "wb") as f:
            for chunk in r.iter_content(chunk_size=8192):
                f.write(chunk)
    print(f"✅ Downloaded {filename} to {local_path}")
    return local_path



download_from_hf(HEART_MODEL_FILENAME)
download_from_hf(ANOMALY_MODEL_FILENAME)





# ===============================
# Define request schemas
# ===============================

class HeartRateInput(BaseModel):
    age: Annotated[int, Field(..., gt=0, lt=120, description="The age of the user")]
    gender: Annotated[Literal['M', 'F'], Field(..., description="Gender of the user")]
    weight_kg: Annotated[float, Field(..., gt=0, description='Weight of the user')]
    height_cm: Annotated[float, Field(..., gt=0, lt=250, description='Height of the user')]
    bmi: Annotated[float, Field(..., gt=0, lt=100, description='BMI of the user')]
    fitness_level: Annotated[Literal['lightly_active', 'fairly_active', 'sedentary', 'very_active'], Field(..., description="Fitness level")]
    performance_level: Annotated[Literal['low', 'moderate', 'high'], Field(..., description="Performance level")]
    resting_hr: Annotated[int, Field(..., gt=0, lt=120, description="Resting HR")]
    max_hr: Annotated[int, Field(..., gt=0, lt=220, description="Max HR")]
    activity_type: Annotated[Literal['sleeping', 'walking', 'resting', 'light', 'commuting', 'exercise'], Field(..., description="Activity type")]
    activity_intensity: Annotated[float, Field(..., gt=0.0, description="Activity intensity")]
    steps_5min: Annotated[int, Field(..., gt=0, description="Steps in 5 min")]
    calories_5min: Annotated[float, Field(..., gt=0, description="Calories in 5 min")]
    hrv_rmssd: Annotated[float, Field(..., gt=0, description="Heart rate variability RMSSD")]
    stress_score: Annotated[int, Field(..., gt=0, lt=100, description="Stress score")]
    signal_quality: Annotated[float, Field(..., gt=0, description="Signal quality")]
    skin_temperature: Annotated[float, Field(..., gt=0, description="Skin temperature")]
    device_battery: Annotated[int, Field(..., gt=0, description="Device battery")]
    elevation_gain: Annotated[int, Field(..., ge=0, description="Elevation gain")]
    sleep_stage: Annotated[Literal['light_sleep', 'deep_sleep', 'rem_sleep'], Field(..., description="Sleep stage")]
    date: Annotated[datetime, Field(..., description="Timestamp")]


class AnomalyInput(BaseModel):
    heart_rate: Annotated[float, Field(..., gt=0.0, description="Heart rate")]
    resting_hr_baseline: Annotated[int, Field(..., gt=0, lt=120, description="Resting HR baseline")]
    activity_type: Annotated[Literal['sleeping', 'walking', 'resting', 'light', 'commuting', 'exercise'], Field(..., description="Activity type")]
    activity_intensity: Annotated[float, Field(..., gt=0, description="Activity intensity")]
    steps_5min: Annotated[int, Field(..., gt=0, description="Steps in 5 min")]
    calories_5min: Annotated[float, Field(..., gt=0, description="Calories in 5 min")]
    hrv_rmssd: Annotated[float, Field(..., gt=0, description="Heart rate variability RMSSD")]
    stress_score: Annotated[int, Field(..., gt=0, lt=100, description="Stress score")]
    confidence_score: Annotated[float, Field(..., gt=0.0, description="Confidence score")]
    signal_quality: Annotated[float, Field(..., gt=0, description="Signal quality")]
    skin_temperature: Annotated[float, Field(..., gt=0, description="Skin temperature")]
    device_battery: Annotated[int, Field(..., gt=0, description="Device battery")]
    elevation_gain: Annotated[int, Field(..., ge=0, description="Elevation gain")]
    sleep_stage: Annotated[Literal['light_sleep', 'deep_sleep', 'rem_sleep'], Field(..., description="Sleep stage")]
    date: Annotated[datetime, Field(..., description="Timestamp")]

# ===============================
# Load models
# ===============================

MODEL_DIR = os.path.join("artifacts", "model_trainer")


HEART_MODEL_PATH = os.path.join(MODEL_DIR, "Heart_Rate_Predictor_model.joblib")
ANOMALY_MODEL_PATH = os.path.join(MODEL_DIR, "Anomaly_Detector_model.joblib")


heart_model_artifacts = joblib.load(HEART_MODEL_PATH)
heart_model = heart_model_artifacts['model']
heart_features = heart_model_artifacts['feature_columns']

anomaly_model_artifacts = joblib.load(ANOMALY_MODEL_PATH)
anomaly_model = anomaly_model_artifacts['model']
anomaly_features = anomaly_model_artifacts['feature_columns']

# ===============================
# Create FastAPI app
# ===============================
app = FastAPI(title="Health Monitoring API")

@app.get("/")
def home():
    return {"message": "Health Monitoring API is running!"}

# ===============================
# Utility: preprocess features
# ===============================
def preprocess_heart_features(data_dict: dict) -> pd.DataFrame:
    # Encode datetime
    data_dict['date_encoded'] = data_dict['date'].timestamp()

    # One-hot categorical encodings
    data_dict['gender_M'] = 1 if data_dict['gender'] == 'M' else 0
    data_dict['gender_F'] = 1 if data_dict['gender'] == 'F' else 0

    for act in ['sleeping', 'walking', 'resting', 'light', 'commuting', 'exercise']:
        data_dict[f"activity_type_{act}"] = 1 if data_dict['activity_type'] == act else 0

    for stage in ['light_sleep', 'deep_sleep', 'rem_sleep']:
        data_dict[f"sleep_stage_{stage}"] = 1 if data_dict['sleep_stage'] == stage else 0

    # Restrict to model features only
    return pd.DataFrame([{f: data_dict.get(f, 0) for f in heart_features}])


def preprocess_anomaly_features(data_dict: dict) -> pd.DataFrame:
    data_dict['date_encoded'] = data_dict['date'].timestamp()

    for act in ['sleeping', 'walking', 'resting', 'light', 'commuting', 'exercise']:
        data_dict[f"activity_type_{act}"] = 1 if data_dict['activity_type'] == act else 0

    for stage in ['light_sleep', 'deep_sleep', 'rem_sleep']:
        data_dict[f"sleep_stage_{stage}"] = 1 if data_dict['sleep_stage'] == stage else 0

    return pd.DataFrame([{f: data_dict.get(f, 0) for f in anomaly_features}])

# ===============================
# Endpoints
# ===============================
@app.post("/predict_heart_rate")
def predict_heart_rate(input_data: HeartRateInput):
    try:
        data_dict = input_data.model_dump()
        X = preprocess_heart_features(data_dict)
        prediction = heart_model.predict(X)[0]
        return {"heart_rate_prediction": float(prediction)}
    except Exception as e:
        return {"error": str(e)}


@app.post("/detect_anomaly")
def detect_anomaly(input_data: AnomalyInput):
    try:
        data_dict = input_data.model_dump()
        X = preprocess_anomaly_features(data_dict)
        prediction = anomaly_model.predict(X)[0]
        return {"anomaly_detected": bool(prediction)}
    except Exception as e:
        return {"error": str(e)}



if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=7860)