Spaces:
Runtime error
Runtime error
File size: 2,665 Bytes
8b19fd0 | 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 | from __future__ import annotations
import os
from contextlib import asynccontextmanager
import uvicorn
from fastapi import FastAPI, File, HTTPException, UploadFile
from fastapi.concurrency import run_in_threadpool
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from inference import get_model_path, load_model, predict_image_bytes
class ProbabilityResponse(BaseModel):
negative: float
neutral: float
positive: float
class PredictionResponse(BaseModel):
label: str
fis_score: float
cnn_prediction_name: str
probabilities: ProbabilityResponse
def get_allowed_origins() -> list[str]:
configured_origins = os.getenv("AI_ENGINE_ALLOWED_ORIGINS", "")
if configured_origins.strip():
return [origin.strip() for origin in configured_origins.split(",") if origin.strip()]
return [
"http://localhost:3000",
"http://127.0.0.1:3000",
]
@asynccontextmanager
async def lifespan(_: FastAPI):
yield
app = FastAPI(
title="Soft Computing AI Engine",
description="FastAPI service for CNN + FIS image sentiment inference.",
version="0.1.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=get_allowed_origins(),
allow_credentials=False,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/health")
def health():
model_path = get_model_path()
return {
"status": "ok",
"model_exists": model_path.exists(),
"model_loaded": load_model.cache_info().currsize > 0,
"model_path": str(model_path),
}
@app.post("/predict", response_model=PredictionResponse)
async def predict(file: UploadFile = File(...)):
if not file.content_type or not file.content_type.startswith("image/"):
raise HTTPException(status_code=400, detail="Upload harus berupa file gambar.")
image_bytes = await file.read()
await file.close()
if not image_bytes:
raise HTTPException(status_code=400, detail="File gambar kosong.")
try:
prediction = await run_in_threadpool(predict_image_bytes, image_bytes)
except FileNotFoundError as error:
raise HTTPException(status_code=500, detail=str(error)) from error
except ValueError as error:
raise HTTPException(status_code=400, detail=str(error)) from error
except Exception as error:
raise HTTPException(
status_code=500,
detail=f"Prediksi gagal dijalankan: {error}",
) from error
return PredictionResponse.model_validate(prediction)
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
|