Spaces:
Sleeping
Sleeping
File size: 3,584 Bytes
81ba29b 1e48819 81ba29b ad70ed5 81ba29b 1e48819 ad70ed5 1e48819 ad70ed5 81ba29b 1e48819 21d68f6 1e48819 ad70ed5 81ba29b 1e48819 ad70ed5 81ba29b 1e48819 81ba29b ad70ed5 1e48819 ad70ed5 81ba29b 1e48819 81ba29b 1e48819 81ba29b 1e48819 ad70ed5 81ba29b ad70ed5 1e48819 81ba29b ad70ed5 81ba29b ad70ed5 81ba29b ad70ed5 81ba29b ad70ed5 1e48819 81ba29b ad70ed5 81ba29b ad70ed5 1e48819 81ba29b 1e48819 81ba29b 1e48819 ad70ed5 81ba29b 1e48819 c060227 81ba29b ad70ed5 81ba29b ad70ed5 81ba29b | 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 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import numpy as np
import onnxruntime as ort
app = FastAPI(
title="Spandan ONNX API",
version="1.0.0",
description="REST API for Stress, Physical Activity, Focus and Cognitive Engagement prediction using ONNX models."
)
# ==========================================================
# Load Models
# ==========================================================
wearable_model = ort.InferenceSession("full_dataset_model_filewearable_model.onnx")
distraction_model = ort.InferenceSession("distraction_model.onnx")
wearable_input = wearable_model.get_inputs()[0].name
distraction_input = distraction_model.get_inputs()[0].name
CLASS_NAMES = [
"Stress",
"Physical Activity",
"Normal"
]
# ==========================================================
# Request Schemas
# ==========================================================
class WearableInput(BaseModel):
sensor_values: list[float]
class DistractionInput(BaseModel):
sensor_values: list[float]
# ==========================================================
# Health Check
# ==========================================================
@app.get("/")
def root():
return {
"message": "Spandan API Running",
"status": "healthy",
"version": "1.0.0"
}
# ==========================================================
# Model 1 - Stress / Physical Activity / Focus
# ==========================================================
@app.post("/predict/state")
def predict_state(data: WearableInput):
if len(data.sensor_values) != 13:
raise HTTPException(
status_code=400,
detail=f"Expected exactly 13 sensor values, got {len(data.sensor_values)}"
)
x = np.array([data.sensor_values], dtype=np.float32)
try:
logits = wearable_model.run(
None,
{
wearable_input: x
}
)[0][0]
prediction = int(np.argmax(logits))
# Stable Softmax
exp = np.exp(logits - np.max(logits))
probabilities = exp / np.sum(exp)
return {
"prediction": CLASS_NAMES[prediction],
"confidence": round(float(probabilities[prediction]), 4),
"scores": {
CLASS_NAMES[i]: round(float(probabilities[i]), 4)
for i in range(len(CLASS_NAMES))
}
}
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Model inference failed: {str(e)}"
)
# ==========================================================
# Model 2 - Cognitive Engagement
# ==========================================================
@app.post("/predict/engagement")
def predict_engagement(data: DistractionInput):
if len(data.sensor_values) != 7:
raise HTTPException(
status_code=400,
detail=f"Expected exactly 7 sensor values, got {len(data.sensor_values)}"
)
x = np.array([data.sensor_values], dtype=np.float32)
try:
prediction = int(
distraction_model.run(
None,
{
distraction_input: x
}
)[0][0]
)
status = "Neutral" if prediction == 1 else "Distracted"
return {
"prediction": prediction,
"status": status
}
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Model inference failed: {str(e)}"
) |