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), }