Spaces:
Sleeping
Sleeping
File size: 2,159 Bytes
725d1d2 6752520 a097d38 725d1d2 6752520 725d1d2 6752520 725d1d2 | 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 | from datetime import datetime
from typing import Any, Dict
from fastapi.responses import JSONResponse
from api.model.model_service import ModelService
from packages import percentage
class ModelController:
def __init__(self, model_service: ModelService):
self.model_service = model_service
async def predict(self, input_data: Dict[str, Any]) -> JSONResponse:
if not self.model_service.is_model_loaded():
raise RuntimeError("Model not loaded on server.")
df = await self.model_service.preprocess_data(input_data)
pred, proba_yes, proba_no, priority = await self.model_service.predict(df)
return JSONResponse(
status_code=200,
content={
"status_code": 200,
"success": True,
"timestamp": datetime.utcnow().isoformat(),
"data": {
"predicted_class": "YES" if pred == 1 else "NO",
"probability_yes": round(proba_yes, 4)
if proba_yes is not None
else None,
"probability_no": round(proba_no, 4)
if proba_no is not None
else None,
"probability_yes_percentage": percentage(proba_yes)
if proba_yes is not None
else None,
"probability_no_percentage": percentage(proba_no)
if proba_no is not None
else None,
"priority": priority,
},
"meta": {
"input_shape": list(df.shape),
},
},
)
async def get_complete_model_info(self) -> JSONResponse:
model_metadata = self.model_service.get_complete_model_info()
return JSONResponse(
status_code=200,
content={
"status_code": 200,
"success": True,
"timestamp": datetime.utcnow().isoformat(),
"data": model_metadata,
},
)
|