routes/ML_ICA_SensorSignals.py

from fastapi import APIRouter
from pydantic import BaseModel
import joblib
import pandas as pd
from typing import Optional, Any

from .config_huggingface import build_model_url, download_artifact_if_needed

router = APIRouter(tags=["Machine Learning"])


class ICARequest(BaseModel):
    sensor_1: float = 0.5
    sensor_2: float = -0.3
    sensor_3: float = 0.8
    sensor_4: float = -0.1


MODEL_STATE: dict[str, Optional[Any]] = {
    "model": None,
    "error": None,
}

MODEL_URL = build_model_url("ML_ICA_SensorSignals.joblib")


def _ensure_model_loaded() -> None:
    if MODEL_STATE["model"] is not None:
        return
    try:
        model_path = download_artifact_if_needed(MODEL_URL)
        MODEL_STATE["model"] = joblib.load(model_path)
        MODEL_STATE["error"] = None
    except Exception as e:
        MODEL_STATE["error"] = str(e)
        raise


@router.post("/models/ica", summary="Separate mixed sensor signals into independent components with ICA")
def transform_ica(data: ICARequest):
    import traceback
    try:
        _ensure_model_loaded()
    except Exception:
        detail = f"Model not loaded: {MODEL_STATE['error']}" if MODEL_STATE["error"] else "Model not loaded."
        return {"error": detail, "traceback": traceback.format_exc(), "status": 500}

    bundle = MODEL_STATE["model"]
    if bundle is None:
        return {"error": f"Model is None. Error: {MODEL_STATE['error']}", "status": 500}

    scaler        = bundle["scaler"]
    ica           = bundle["ica"]
    feature_names = bundle["feature_names"]

    input_df = pd.DataFrame(
        [[data.sensor_1, data.sensor_2, data.sensor_3, data.sensor_4]],
        columns=feature_names,
    )

    try:
        components = ica.transform(scaler.transform(input_df))[0]
    except Exception as e:
        return {"error": f"Transform failed: {str(e)}", "traceback": traceback.format_exc(), "status": 500}

    return {
        "ic1": round(float(components[0]), 4),
        "ic2": round(float(components[1]), 4),
        "ic3": round(float(components[2]), 4),
        "ic4": round(float(components[3]), 4),
    }