change video stream to base64

app.py

@@ -5,7 +5,8 @@ from ultralytics import YOLO
 
 from typing import Dict
 
-from core.defect_detection import *
+# from core.defect_detection import *
+from core.defect_detection_image import *
 from utils import *
 
 import uvicorn, asyncio
@@ -54,7 +55,6 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
-
 # ============================================================
 # ROUTES
 # ============================================================
@@ -70,26 +70,24 @@ async def start_detection(data: Dict):
     webhook_url = data.get("webhook_url")
     cameras     = data.get("cameras", [])
 
+    # -------------------------------
+    # BASIC VALIDATION
+    # -------------------------------
     if not station_id or not parts or not webhook_url or not cameras:
         return JSONResponse(
-                status_code=400,
-                content={
-                    "status": "error", 
-                    "station_id": station_id,
-                    "camera_count": len(cameras),
-                    "message": "Missing required fields"
-                }
-            )
+            status_code=400,
+            content={
+                "status": "error",
+                "station_id": station_id,
+                "camera_count": len(cameras),
+                "message": "Missing required fields"
+            }
+        )
 
     # -------------------------------
     # VALIDATION BEFORE EXECUTION
     # -------------------------------
-    required_parts_fields = [
-        "id", 
-        "pin_api", 
-        "name", 
-        "sku"
-    ]
+    required_parts_fields = ["id", "pin_api", "name", "sku"]
     validation_errors = validate_input(required_parts_fields, station_id, cameras, parts, webhook_url)
 
     if validation_errors:
@@ -97,40 +95,119 @@ async def start_detection(data: Dict):
         for err in validation_errors:
             logger.error(f" - {err}")
         return JSONResponse(
-                status_code=400,
-                content={
-                    "status": "error", 
-                    "station_id": station_id,
-                    "camera_count": len(cameras),
-                    "message": " | ".join(validation_errors)
-                }
-            )
+            status_code=400,
+            content={
+                "status": "error",
+                "station_id": station_id,
+                "camera_count": len(cameras),
+                "message": " | ".join(validation_errors)
+            }
+        )
+
     logger.info(f"[INFO] Get metadata parts")
     model_path = model_by_id_metadata(parts['id'])
 
     logger.info(f"[INFO] Checking model_path")
     if isinstance(model_path, str):
         if not os.path.exists(model_path):
-            logger.info(f"[INFO] Model file not found")
-            return {"status": "error", "message": f"Model file not found: {model_path}"}
+            logger.error(f"[ERROR] Model file not found: {model_path}")
+            return {
+                "status": "error",
+                "message": f"Model file not found: {model_path}"
+            }
         model = YOLO(model_path)
     else:
         model = model_path
 
-    logger.info(f"[START] Station {station_id} → {len(cameras)} kamera diproses")
+    # =====================================================
+    # BASE64 IMAGE DETECTION (NOT VIDEO STREAM)
+    # =====================================================
+    logger.info(f"[START] Station {station_id} → {len(cameras)} camera(s) with base64 images")
 
-    # running background
-    asyncio.create_task(run_detection_group(station_id, cameras, webhook_url, model, parts))
+    # Jalankan detection di background
+    asyncio.create_task(
+        run_detection_group(station_id, cameras, webhook_url, model, parts)
+    )
 
     return JSONResponse(
-            status_code=200,
-            content={
+        status_code=200,
+        content={
             "status": "started",
             "station_id": station_id,
             "camera_count": len(cameras),
-            "message": "Detection is running in background."
-            }
-        )
+            "message": "Base64 image detection is running in background."
+        }
+    )
+
+
+# @app.post("/start-detection") # live stream
+# async def start_detection(data: Dict):
+#     station_id  = data.get("station_id")
+#     parts       = data.get("parts")
+#     webhook_url = data.get("webhook_url")
+#     cameras     = data.get("cameras", [])
+
+#     if not station_id or not parts or not webhook_url or not cameras:
+#         return JSONResponse(
+#                 status_code=400,
+#                 content={
+#                     "status": "error", 
+#                     "station_id": station_id,
+#                     "camera_count": len(cameras),
+#                     "message": "Missing required fields"
+#                 }
+#             )
+
+#     # -------------------------------
+#     # VALIDATION BEFORE EXECUTION
+#     # -------------------------------
+#     required_parts_fields = [
+#         "id", 
+#         "pin_api", 
+#         "name", 
+#         "sku"
+#     ]
+#     validation_errors = validate_input(required_parts_fields, station_id, cameras, parts, webhook_url)
+
+#     if validation_errors:
+#         logger.error("[VALIDATION FAILED] Input data is invalid.")
+#         for err in validation_errors:
+#             logger.error(f" - {err}")
+#         return JSONResponse(
+#                 status_code=400,
+#                 content={
+#                     "status": "error", 
+#                     "station_id": station_id,
+#                     "camera_count": len(cameras),
+#                     "message": " | ".join(validation_errors)
+#                 }
+#             )
+#     logger.info(f"[INFO] Get metadata parts")
+#     model_path = model_by_id_metadata(parts['id'])
+
+#     logger.info(f"[INFO] Checking model_path")
+#     if isinstance(model_path, str):
+#         if not os.path.exists(model_path):
+#             logger.info(f"[INFO] Model file not found")
+#             return {"status": "error", "message": f"Model file not found: {model_path}"}
+#         model = YOLO(model_path)
+#     else:
+#         model = model_path
+
+#     logger.info(f"[START] Station {station_id} → {len(cameras)} kamera diproses")
+
+#     # running background
+#     asyncio.create_task(run_detection_group(station_id, cameras, webhook_url, model, parts))
+
+#     return JSONResponse(
+#             status_code=200,
+#             content={
+#             "status": "started",
+#             "station_id": station_id,
+#             "camera_count": len(cameras),
+#             "message": "Detection is running in background."
+#             }
+#         )
 
 
 # ============================================================

core/defect_detection_image.py

@@ -0,0 +1,228 @@
+import os, cv2, time, base64, asyncio, httpx
+import numpy as np
+
+from datetime import datetime
+from dotenv import load_dotenv
+from typing import Dict, List
+from utils import *
+
+load_dotenv()
+
+MODEL_VERSION   = os.getenv("MODEL_VERSION","v1.0.0")
+WEBHOOK_URL     = os.getenv("WEBHOOK_URL")
+
+WEBHOOK_TIMEOUT = float(os.getenv("WEBHOOK_TIMEOUT", "10.0"))
+
+# ============================================================
+# DEFECT DETECTION FROM BASE64 IMAGE
+# ============================================================
+def detect_defect_from_base64(station_id: str, camera_id: str, image_base64: str, model=None):
+    """
+    Detect defect from a single Base64 image.
+    - Decode Base64 → OpenCV image
+    - Run YOLO model once
+    - Return result (OK / NG) + annotated image base64
+    """
+
+    try:
+        img_data = base64.b64decode(image_base64)
+        np_arr = np.frombuffer(img_data, np.uint8)
+        frame = cv2.imdecode(np_arr, cv2.IMREAD_COLOR)
+        if frame is None:
+            raise ValueError("Decoded image is None")
+    except Exception as e:
+        logger.error(f"[ERROR] Cannot decode base64 image for camera {camera_id}: {e}")
+        return {
+            "station_id": station_id,
+            "camera_id": camera_id,
+            "status": "error",
+            "status_defect": "",
+            "image_base64": "",
+            "detections": [],
+            "message": f"Invalid base64 image"
+        }
+        
+        
+    # ========== YOLO DETECTION ==========
+    if model:
+        # Predict
+        results = model.predict(source=frame, conf=0.4, imgsz=640, verbose=False)
+        boxes = results[0].boxes
+
+        if len(boxes) > 0:
+            for box in boxes:
+                cls = int(box.cls[0])
+                conf = float(box.conf[0])
+                xyxy = [int(x) for x in box.xyxy[0].tolist()]
+                defect_name = model.names.get(cls, f"class_{cls}").lower()
+
+                x1, y1, x2, y2 = xyxy
+
+                # Ambil warna berdasarkan defect
+                try:
+                    color = color_defect(defect_name)
+                except Exception:
+                    color = color_defect('other')
+
+                # Draw bounding box di frame
+                cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2)
+
+                # Label
+                label = f"{defect_name.upper()} {conf:.2f}"
+                (w, h), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.6, 2)
+                cv2.rectangle(frame, (x1, y1 - 20), (x1 + w, y1), color, -1)
+                cv2.putText(frame, label, (x1, y1 - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
+
+
+                # Convert annotated image ke Base64
+                _, buffer = cv2.imencode(".jpg", frame)
+                frame_base64 = base64.b64encode(buffer).decode("utf-8")
+
+                # Save annotated image
+                # output_dir = "outputs/images"
+                # os.makedirs(output_dir, exist_ok=True)
+                # filename = f"{station_id}_{camera_id}_NG_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jpg"
+                # filepath = os.path.join(output_dir, filename)
+                # cv2.imwrite(filepath, frame)
+
+                # logger.info(f"[SAVED] NG image saved to {filepath}")
+
+                logger.info(f"[DETECTED] Camera {camera_id} → {defect_name} ({conf:.2f})")
+
+                return {
+                    "station_id": station_id,
+                    "camera_id": camera_id,
+                    "status": "success",
+                    "status_defect": "NG",
+                    "image_base64": frame_base64,
+                    "detections": [{
+                        "class": defect_name,
+                        "confidence": conf,
+                        "bbox": xyxy
+                    }],
+                    "message": f"Detected as defect"
+                }
+
+    # ========== NO DEFECT ==========
+    _, buffer = cv2.imencode(".jpg", frame)
+    frame_base64 = base64.b64encode(buffer).decode("utf-8")
+
+    # Save OK image (no bbox)
+    # output_dir = "outputs/images"
+    # os.makedirs(output_dir, exist_ok=True)
+    # filename = f"{station_id}_{camera_id}_OK_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jpg"
+    # filepath = os.path.join(output_dir, filename)
+    # cv2.imwrite(filepath, frame)
+    # logger.info(f"[SAVED] OK image saved to {filepath}")
+
+    logger.info(f"[OK] Camera {camera_id} → No defect detected.")
+    return {
+        "station_id": station_id,
+        "camera_id": camera_id,
+        "status": "success",
+        "status_defect": "OK",
+        "image_base64": frame_base64,
+        "detections": [],
+        "message": f"Detected as normal (no defect)"
+    }
+
+# ============================================================
+# ASYNC WRAPPERS
+# ============================================================
+async def _detect_camera_image(station_id: str, camera: Dict, model=None):
+    """Run detect_defect_from_base64 in thread for async parallel."""
+    return await asyncio.to_thread(
+        detect_defect_from_base64,
+        station_id,
+        camera["camera_id"],
+        camera["image_base64"],
+        model
+    )
+
+async def run_detection_group(
+    station_id: str,
+    cameras: List[Dict],
+    webhook_url: str,
+    model=None,
+    parts: Dict = None
+):
+    """
+    Run defect detection for multiple cameras in parallel and send webhook.
+    All results are serialized safely for JSON.
+    """
+    parts = parts or {}
+
+    logger.info(f"[START] Station {station_id} → {len(cameras)} camera(s)")
+
+    # Run detection async parallel
+    results = await asyncio.gather(
+        *[_detect_camera_image(station_id, cam, model) for cam in cameras],
+        return_exceptions=True
+    )
+
+    # Convert exceptions to dict
+    clean_results = []
+    for r in results:
+        if isinstance(r, Exception):
+            clean_results.append({
+                "status": "error",
+                "message": str(r)
+            })
+        else:
+            clean_results.append(r)
+
+    # Determine overall status
+    has_error = any(r.get("status") == "error" for r in clean_results)
+    all_error = all(r.get("status") == "error" for r in clean_results)
+
+    if all_error:
+        status = "error"
+        message = "All cameras failed during detection"
+    elif has_error:
+        status = "partial_error"
+        message = "Some cameras failed during detection"
+    else:
+        status = "success"
+        message = "Success detecting defects"
+
+    # Build payload
+    payload = {
+        "status": status,
+        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime()),
+        "model_version": MODEL_VERSION,
+        "message": message,
+        "parts": parts,
+        # "data": make_serializable(clean_results),
+    }
+
+    # Send webhook
+    try:
+        async with httpx.AsyncClient(timeout=WEBHOOK_TIMEOUT) as client:
+            await client.post(webhook_url, json=payload)
+            logger.info(f"[DONE] Station {station_id}")
+    except Exception as e:
+        logger.exception(f"[ERROR] Webhook failed for Station {station_id}: {e}")
+
+    return payload
+
+# ============================================================
+# JSON SERIALIZABLE HELPER
+# ============================================================
+def make_serializable(obj):
+    """Convert object to JSON-serializable format."""
+    if isinstance(obj, (int, float, str, bool)) or obj is None:
+        return obj
+    elif isinstance(obj, (list, tuple)):
+        return [make_serializable(i) for i in obj]
+    elif isinstance(obj, dict):
+        return {k: make_serializable(v) for k, v in obj.items()}
+    elif isinstance(obj, datetime):
+        return obj.isoformat()
+    elif isinstance(obj, np.integer):
+        return int(obj)
+    elif isinstance(obj, np.floating):
+        return float(obj)
+    elif isinstance(obj, np.ndarray):
+        return obj.tolist()
+    else:
+        return str(obj)

utils.py

@@ -46,8 +46,10 @@ def validate_input(required_parts_fields, station_id, cameras, parts, webhook_ur
         for index, cam in enumerate(cameras):
             if "camera_id" not in cam or not str(cam["camera_id"]).strip():
                 errors.append(f"camera[{index}].camera_id missing or empty")
-            if "rtsp_url" not in cam or not str(cam["rtsp_url"]).strip():
-                errors.append(f"camera[{index}].rtsp_url missing or empty")
+            if "image_base64" not in cam or not str(cam["image_base64"]).strip():
+                errors.append(f"camera[{index}].image_base64 missing or empty")
+            # if "rtsp_url" not in cam or not str(cam["rtsp_url"]).strip():
+            #     errors.append(f"camera[{index}].rtsp_url missing or empty")
 
     # Validate webhook
     if not webhook_url or not webhook_url.startswith("http"):