| from fastapi import FastAPI, File, UploadFile
|
| from fastapi.responses import JSONResponse
|
| import cv2
|
| import numpy as np
|
| from ultralytics import YOLO
|
|
|
| app = FastAPI(title="GiziTrace ML API")
|
|
|
|
|
| model = YOLO("best.onnx", task="detect")
|
|
|
| class_names = {0: "plate", 1: "food"}
|
|
|
| @app.post("/predict")
|
| async def predict_waste(file: UploadFile = File(...)):
|
| try:
|
| contents = await file.read()
|
| nparr = np.frombuffer(contents, np.uint8)
|
| img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
|
|
|
|
|
| results = model(img, conf=0.10)
|
|
|
| detections = []
|
|
|
|
|
| for result in results:
|
| if len(result.boxes) == 0:
|
| continue
|
|
|
| clss = result.boxes.cls.cpu().numpy()
|
| confs = result.boxes.conf.cpu().numpy()
|
|
|
| for idx in range(len(clss)):
|
| class_id = int(clss[idx])
|
| confidence = float(confs[idx])
|
|
|
|
|
| label = class_names.get(class_id, "unknown")
|
|
|
|
|
| detections.append({
|
| "class": label,
|
| "confidence": round(confidence, 2)
|
| })
|
|
|
| return JSONResponse(content={
|
| "status": "success",
|
| "total_objects_detected": len(detections),
|
| "detections": detections
|
| })
|
|
|
| except Exception as e:
|
| return JSONResponse(status_code=500, content={"message": str(e)}) |