Update app.py

app.py

@@ -1,301 +1,66 @@
 import gradio as gr
-import torch
-try:
-    import cv2
-    CV2_AVAILABLE = True
-except ImportError:
-    CV2_AVAILABLE = False
 import numpy as np
 from PIL import Image, ImageDraw, ImageFont
 import io
 import base64
-import os
 from datetime import datetime
 import json
+from fastapi import FastAPI, Request
 
-# Model dan kelas deteksi (dalam implementasi nyata, akan menggunakan model Hugging Face)
-PPE_CLASSES = ["helmet", "vest", "harness", "gloves", "boots", "goggles", "mask"]
-HEAD_PROTECTION = ["helmet"]
-BODY_PROTECTION = ["vest", "harness"]
-HAND_PROTECTION = ["gloves"]
-FOOT_PROTECTION = ["boots", "safety_shoes"]
-EYE_PROTECTION = ["goggles"]
-RESPIRATORY_PROTECTION = ["mask"]
-
-# Tentukan jenis pekerjaan dan APD yang diperlukan
-WORK_TYPES = {
-    "ketinggian": {
-        "required": ["head_protection", "body_protection", "foot_protection"],
-        "recommended": ["hand_protection"]
-    },
-    "konstruksi_umum": {
-        "required": ["head_protection", "foot_protection"],
-        "recommended": ["body_protection", "hand_protection"]
-    },
-    "listrik": {
-        "required": ["head_protection", "hand_protection", "foot_protection"],
-        "recommended": ["eye_protection"]
-    }
-}
-
-def preprocess_image(image):
-    """Preproses gambar untuk analisis"""
-    if isinstance(image, np.ndarray):
-        # Konversi dari numpy array ke PIL Image
-        if CV2_AVAILABLE and image.shape[2] == 3:  # RGB
-            return Image.fromarray(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))
-        else:
-            return Image.fromarray(image)
-    return image
-
-def analyze_image(image):
-    """
-    Fungsi analisis gambar simulasi
-    Dalam implementasi nyata, ini akan menggunakan model Hugging Face
-    """
-    # Dapatkan dimensi gambar
-    if isinstance(image, np.ndarray):
-        height, width = image.shape[:2]
-    else:
-        width, height = image.size
-    
-    # Deteksi orang (simulasi)
-    # Dalam kasus nyata, model deteksi akan mengembalikan kotak untuk semua orang
-    persons = [
-        {"box": [int(width*0.3), int(height*0.1), int(width*0.7), int(height*0.9)], "confidence": 0.95}
-    ]
-    
-    # Deteksi APD (simulasi untuk demo)
-    ppe_detections = []
-    
-    # Simulasikan harness tetapi tidak ada helm, sarung tangan, dan sepatu safety
-    ppe_detections.append({
-        "item": "harness", 
-        "box": [int(width*0.35), int(height*0.3), int(width*0.65), int(height*0.7)],
-        "confidence": 0.88,
-        "ppe_type": "body_protection"
-    })
-    
-    # Mengembalikan orang + APD yang terdeteksi
-    results = []
-    
-    # Tambahkan deteksi orang
-    for person in persons:
-        results.append({
-            "item": "person",
-            "box": person["box"],
-            "confidence": person["confidence"],
-            "ppe_type": None
-        })
-    
-    # Tambahkan deteksi APD
-    results.extend(ppe_detections)
-    
-    return results
-
-def evaluate_compliance(detections, work_type="ketinggian"):
-    """Evaluasi kepatuhan APD berdasarkan deteksi dan jenis pekerjaan"""
-    has_person = any(d["item"] == "person" for d in detections)
-    
-    if not has_person:
-        return "Tidak ada orang terdeteksi", []
-    
-    # Periksa perlindungan yang ada
-    protections = {
-        "head_protection": any(d["ppe_type"] == "head_protection" for d in detections),
-        "body_protection": any(d["ppe_type"] == "body_protection" for d in detections),
-        "hand_protection": any(d["ppe_type"] == "hand_protection" for d in detections),
-        "foot_protection": any(d["ppe_type"] == "foot_protection" for d in detections),
-        "eye_protection": any(d["ppe_type"] == "eye_protection" for d in detections),
-        "respiratory_protection": any(d["ppe_type"] == "respiratory_protection" for d in detections)
-    }
-    
-    # Tentukan persyaratan APD berdasarkan jenis pekerjaan
-    if work_type in WORK_TYPES:
-        required = WORK_TYPES[work_type]["required"]
-        recommended = WORK_TYPES[work_type]["recommended"]
-    else:
-        # Default untuk konstruksi umum jika jenis pekerjaan tidak dikenali
-        required = WORK_TYPES["konstruksi_umum"]["required"]
-        recommended = WORK_TYPES["konstruksi_umum"]["recommended"]
-    
-    # Periksa perlindungan wajib
-    critical_issues = []
-    recommended_issues = []
-    
-    for protection in required:
-        if not protections[protection]:
-            if protection == "head_protection":
-                critical_issues.append("Tidak ada pelindung kepala (helm)")
-            elif protection == "body_protection":
-                critical_issues.append("Tidak ada pelindung tubuh (harness/rompi)")
-            elif protection == "hand_protection":
-                critical_issues.append("Tidak ada pelindung tangan (sarung tangan)")
-            elif protection == "foot_protection":
-                critical_issues.append("Tidak ada pelindung kaki (sepatu safety)")
-            elif protection == "eye_protection":
-                critical_issues.append("Tidak ada pelindung mata (kacamata safety)")
-            elif protection == "respiratory_protection":
-                critical_issues.append("Tidak ada pelindung pernapasan (masker)")
-    
-    # Periksa perlindungan yang direkomendasikan
-    for protection in recommended:
-        if not protections[protection]:
-            if protection == "head_protection":
-                recommended_issues.append("Sebaiknya menggunakan pelindung kepala (helm)")
-            elif protection == "body_protection":
-                recommended_issues.append("Sebaiknya menggunakan pelindung tubuh (harness/rompi)")
-            elif protection == "hand_protection":
-                recommended_issues.append("Sebaiknya menggunakan pelindung tangan (sarung tangan)")
-            elif protection == "foot_protection":
-                recommended_issues.append("Sebaiknya menggunakan pelindung kaki (sepatu safety)")
-            elif protection == "eye_protection":
-                recommended_issues.append("Sebaiknya menggunakan pelindung mata (kacamata safety)")
-            elif protection == "respiratory_protection":
-                recommended_issues.append("Sebaiknya menggunakan pelindung pernapasan (masker)")
-    
-    # Menentukan status kepatuhan
-    all_issues = critical_issues + recommended_issues
-    if len(critical_issues) == 0:
-        if len(recommended_issues) == 0:
-            return "Patuh (Compliant)", []
-        else:
-            return "Patuh dengan Catatan", all_issues
-    else:
-        return "Tidak Patuh (Non-compliant)", all_issues
-
-def draw_detections(image, detections):
-    """Menggambar kotak deteksi pada gambar"""
-    # Buat salinan gambar
-    if isinstance(image, np.ndarray):
-        img_copy = Image.fromarray(image.copy())
-    else:
-        img_copy = image.copy()
-    
-    draw = ImageDraw.Draw(img_copy)
-    
-    # Coba dapatkan font yang tersedia
-    try:
-        font = ImageFont.truetype("arial.ttf", 15)
-    except IOError:
-        try:
-            font = ImageFont.truetype("DejaVuSans.ttf", 15)
-        except IOError:
-            font = ImageFont.load_default()
-    
-    # Gambar kotak deteksi
-    for detection in detections:
-        box = detection["box"]
-        label = detection["item"]
-        confidence = detection["confidence"]
-        ppe_type = detection["ppe_type"]
-        
-        # Pilih warna berdasarkan tipe
-        if label == "person":
-            color = "red"
-        elif ppe_type == "head_protection":
-            color = "blue"
-        elif ppe_type == "body_protection":
-            color = "green"
-        elif ppe_type:
-            color = "purple"
-        else:
-            color = "yellow"
-        
-        # Gambar box and label
-        draw.rectangle([(box[0], box[1]), (box[2], box[3])], outline=color, width=3)
-        text = f"{label}: {confidence:.2f}"
-        
-        # Estimasi ukuran teks jika draw.textsize tidak tersedia
-        text_w, text_h = (100, 15)  # Default
-        if hasattr(draw, 'textsize'):
-            text_w, text_h = draw.textsize(text, font=font)
-        
-        # Background untuk teks
-        draw.rectangle([(box[0], box[1]-text_h-4), (box[0]+text_w, box[1])], fill=color)
-        draw.text((box[0], box[1]-text_h-2), text, fill="white", font=font)
-    
-    return img_copy
+# Setup FastAPI instance
+app = FastAPI()
 
 def detect_ppe(image, work_type="ketinggian"):
-    """
-    Fungsi utama untuk mendeteksi APD dalam gambar
-    """
+    """Fungsi utama untuk deteksi APD (simulasi)"""
     if image is None:
         return None, "Tidak ada gambar", {"error": "Tidak ada gambar yang diberikan"}
     
     try:
-        # Preproses gambar
-        if isinstance(image, np.ndarray) and image.shape[2] == 4 and CV2_AVAILABLE:  # RGBA
-            image = cv2.cvtColor(image, cv2.COLOR_RGBA2RGB)
-        
-        # 1. Analisis gambar
-        detections = analyze_image(image)
+        # Konversi ke PIL Image jika perlu
+        if isinstance(image, np.ndarray):
+            pil_image = Image.fromarray(image)
+        else:
+            pil_image = image
         
-        # 2. Evaluasi kepatuhan berdasarkan jenis pekerjaan
-        compliance_status, compliance_issues = evaluate_compliance(detections, work_type)
+        width, height = pil_image.size
         
-        # 3. Gambar hasil deteksi
-        result_image = draw_detections(image, detections)
+        # Simulasikan hasil deteksi
+        detections = [
+            {
+                "item": "person",
+                "box": [int(width*0.3), int(height*0.1), int(width*0.7), int(height*0.9)],
+                "confidence": 0.95,
+                "ppe_type": None
+            },
+            {
+                "item": "harness",
+                "box": [int(width*0.35), int(height*0.3), int(width*0.65), int(height*0.7)],
+                "confidence": 0.88,
+                "ppe_type": "body_protection"
+            }
+        ]
         
-        # 4. Identifikasi APD yang hilang
-        missing_ppe = []
-        for issue in compliance_issues:
-            if "Tidak ada" in issue:
-                missing_ppe.append(issue.replace("Tidak ada ", ""))
+        # Simulasikan hasil kepatuhan
+        compliance_status = "Tidak Patuh (Non-compliant)"
+        compliance_issues = ["Tidak ada pelindung kepala (helm)", "Tidak ada pelindung kaki (sepatu safety)"]
         
-        # 5. Siapkan respons - format data array dengan indeks dimulai dari 1 bukan 0
-        # Proses detections agar terformat dengan baik
-        formatted_detections = {}
-        for idx, detection in enumerate(detections, 1):
-            # Format box array agar lebih rapi
-            box_values = detection["box"]
-            formatted_box = {}
-            for i, coord in enumerate(box_values):
-                formatted_box[str(i+1)] = coord
-                
-            # Buat entry detection dengan format yang rapi
-            formatted_detections[str(idx)] = {
-                "item": detection["item"],
-                "box": formatted_box,
-                "confidence": detection["confidence"],
-                "ppe_type": detection["ppe_type"]
-            }
-            
-        # Format issues dengan indeks dimulai dari 1
-        formatted_issues = {}
-        for idx, issue in enumerate(compliance_issues, 1):
-            formatted_issues[str(idx)] = issue
-            
-        # Format missing_ppe dengan indeks dimulai dari 1
-        formatted_missing = {}
-        for idx, item in enumerate(missing_ppe, 1):
-            formatted_missing[str(idx)] = item
-            
-        # Format recommendations dengan indeks dimulai dari 1
-        recommendations = []
-        for issue in compliance_issues:
-            if "Tidak ada" in issue:
-                recommendations.append(f"Pastikan pekerja menggunakan {issue.replace('Tidak ada ', '')}")
-            elif "Sebaiknya menggunakan" in issue:
-                recommendations.append(issue)
-                
-        formatted_recommendations = {}
-        for idx, rec in enumerate(recommendations, 1):
-            formatted_recommendations[str(idx)] = rec
+        # Gambar hasil deteksi
+        result_image = draw_detections(pil_image, detections)
         
-        # Buat hasil analisis yang terformat dengan baik
+        # Format JSON untuk respons API
         result = {
-            "detections": formatted_detections,
+            "detections": {str(i+1): d for i, d in enumerate(detections)},
             "compliance": {
                 "status": compliance_status,
-                "issues": formatted_issues
+                "issues": {str(i+1): issue for i, issue in enumerate(compliance_issues)}
             },
-            "missing_ppe": formatted_missing,
             "work_type": work_type,
             "summary": f"Deteksi: {len(detections)} objek. Status: {compliance_status}",
             "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
-            "recommendations": formatted_recommendations
+            "recommendations": {
+                "1": "Pastikan pekerja menggunakan pelindung kepala (helm)",
+                "2": "Pastikan pekerja menggunakan pelindung kaki (sepatu safety)"
+            }
         }
         
         return result_image, compliance_status, result
@@ -303,67 +68,78 @@ def detect_ppe(image, work_type="ketinggian"):
     except Exception as e:
         return None, f"Error: {str(e)}", {"error": str(e)}
 
-# Fungsi untuk menerima dan mengolah base64 image
-def process_base64_image(base64_string, work_type="ketinggian"):
+def draw_detections(image, detections):
+    """Menggambar kotak deteksi pada gambar"""
+    img_copy = image.copy()
+    draw = ImageDraw.Draw(img_copy)
+    
     try:
-        # Cek apakah base64 datang dengan header data URL
-        if "data:" in base64_string and ";base64," in base64_string:
-            base64_string = base64_string.split(";base64,")[1]
-        
-        # Decode base64 ke bytes
-        image_bytes = base64.b64decode(base64_string)
+        font = ImageFont.truetype("DejaVuSans.ttf", 15)
+    except IOError:
+        font = ImageFont.load_default()
+    
+    for detection in detections:
+        box = detection["box"]
+        label = detection["item"]
+        confidence = detection["confidence"]
         
-        # Buka gambar dari bytes
-        image = Image.open(io.BytesIO(image_bytes))
+        # Warna berdasarkan tipe
+        color = "red" if label == "person" else "blue"
         
-        # Konversi ke numpy array jika model Anda memerlukan format ini
-        image_np = np.array(image)
+        # Gambar box
+        draw.rectangle([(box[0], box[1]), (box[2], box[3])], outline=color, width=3)
+        text = f"{label}: {confidence:.2f}"
         
-        # Proses gambar dengan fungsi deteksi APD yang ada
-        return detect_ppe(image_np, work_type)
-    except Exception as e:
-        return None, f"Error processing base64 image: {str(e)}", {"error": str(e)}
+        # Label
+        draw.rectangle([(box[0], box[1]-20), (box[0]+100, box[1])], fill=color)
+        draw.text((box[0], box[1]-18), text, fill="white", font=font)
+    
+    return img_copy
 
-# API handler untuk base64 images
-def api_predict(data):
+# API function for FastAPI
+@app.post("/api/predict")
+async def predict_api(request: Request):
+    """API endpoint untuk deteksi APD"""
     try:
+        # Baca JSON dari request
+        data = await request.json()
+        
+        # Validasi data
+        if "data" not in data or not isinstance(data["data"], list) or len(data["data"]) == 0:
+            return {"error": "Invalid data format. Expected: {\"data\": [\"base64_image\", \"work_type\"]}"}
+        
         # Extract data
-        if len(data) == 0:
-            return [None, "Error: No data provided", {"error": "No data provided"}]
+        base64_image = data["data"][0]
+        work_type = data["data"][1] if len(data["data"]) > 1 else "ketinggian"
+        
+        # Hapus header data URL jika ada
+        if "data:" in base64_image and ";base64," in base64_image:
+            base64_image = base64_image.split(";base64,")[1]
         
-        # Data[0] adalah gambar base64, data[1] adalah jenis pekerjaan (opsional)
-        base64_image = data[0]
-        work_type = data[1] if len(data) > 1 else "ketinggian"
+        # Decode base64 ke bytes
+        image_bytes = base64.b64decode(base64_image)
+        
+        # Buka gambar dari bytes
+        image = Image.open(io.BytesIO(image_bytes))
         
-        # Process the image
-        result_image, status, analysis_result = process_base64_image(base64_image, work_type)
+        # Proses gambar
+        result_image, status, analysis_result = detect_ppe(image, work_type)
         
-        # Convert result image to base64 for API response
+        # Konversi hasil gambar ke base64
         if result_image is not None:
             buffered = io.BytesIO()
             result_image.save(buffered, format="JPEG")
             img_str = "data:image/jpeg;base64," + base64.b64encode(buffered.getvalue()).decode()
             return [img_str, status, analysis_result]
-        else:
-            return [None, status, analysis_result]
+        return [None, status, analysis_result]
+    
     except Exception as e:
-        return [None, f"API Error: {str(e)}", {"error": str(e)}]
+        return {"error": str(e)}
 
-# Fungsi untuk menguji API secara langsung
-def test_api(test_image, test_work_type):
-    # Konversi gambar ke base64
-    if test_image is not None:
-        buffered = io.BytesIO()
-        test_image.save(buffered, format="JPEG")
-        img_base64 = "data:image/jpeg;base64," + base64.b64encode(buffered.getvalue()).decode()
-        
-        # Panggil API dengan data yang diformat dengan benar
-        result = api_predict([img_base64, test_work_type])
-        return result
-    return [None, "Error: No test image provided", {"error": "No test image provided"}]
+# Buat UI utama
+demo = gr.Blocks(title="K3L APD Detection")
 
-# Buat interface Gradio untuk UI
-with gr.Blocks(title="Sistem Deteksi APD untuk K3L Platform") as demo:
+with demo:
     gr.Markdown("# Sistem Deteksi APD untuk K3L Platform")
     gr.Markdown("Upload gambar untuk analisis kepatuhan APD di lokasi konstruksi")
     
@@ -395,76 +171,24 @@ with gr.Blocks(title="Sistem Deteksi APD untuk K3L Platform") as demo:
     3. Klik tombol "Analisis Gambar"
     4. Lihat hasil analisis dengan kotak deteksi, status kepatuhan, dan rekomendasi
     
-    ### Deteksi APD Wajib untuk Pekerjaan di Ketinggian
-    - Helm (wajib)
-    - Safety harness (wajib)
-    - Sepatu safety (wajib)
-    - Sarung tangan (direkomendasikan)
-    """)
-
-    status_msg = "✅ OpenCV tersedia" if CV2_AVAILABLE else "⚠️ OpenCV tidak tersedia - menggunakan mode fallback"
-    gr.Markdown(f"*Status sistem: {status_msg}*")
-
-# Tambahkan tab untuk testing API
-with gr.Blocks(title="API Testing") as api_test_interface:
-    gr.Markdown("## Test API Endpoint")
-    gr.Markdown("Test the API endpoint with an example image")
-    
-    with gr.Row():
-        with gr.Column():
-            test_input_image = gr.Image(label="Test Image")
-            test_work_type = gr.Radio(
-                label="Work Type", 
-                choices=["ketinggian", "konstruksi_umum", "listrik"],
-                value="ketinggian"
-            )
-            test_btn = gr.Button("Test API", variant="primary")
-        
-        with gr.Column():
-            test_output_image = gr.Image(label="API Result Image")
-            test_status = gr.Textbox(label="API Status")
-            test_json = gr.JSON(label="API Response")
-    
-    test_btn.click(
-        fn=test_api,
-        inputs=[test_input_image, test_work_type],
-        outputs=[test_output_image, test_status, test_json]
-    )
-
-# Buat interface dengan tabs dan definisikan API endpoint di dalam konteks gr.Blocks
-with gr.Blocks() as combined_interface:
-    with gr.Tabs():
-        with gr.TabItem("K3L APD Detection"):
-            demo.render()
-        with gr.TabItem("API Testing"):
-            api_test_interface.render()
+    ### API Endpoint
+    API endpoint tersedia di: `/api/predict`
     
-    # PERBAIKAN: Daftarkan API endpoint DI DALAM konteks gr.Blocks
-    gr.Interface(
-        fn=api_predict,
-        inputs=[gr.JSON(label="Data")],
-        outputs=[
-            gr.Image(label="Result Image"),
-            gr.Textbox(label="Status"),
-            gr.JSON(label="Analysis Result")
-        ],
-        title="K3L PPE Detection API",
-        description="API for detecting PPE in construction sites",
-        flagging_mode="never"
-    ).queue().launch(prevent_thread_lock=True, show_api=True, share=False)
-
-# Simpan fungsi queue untuk berfungsi dengan baik dengan semua interface
-combined_interface.queue()
+    Format request (POST):
+    ```json
+    {
+        "data": [
+            "BASE64_IMAGE",
+            "JENIS_PEKERJAAN"  (opsional)
+        ]
+    }
+    ```
+    """)
 
-# Launch aplikasi dengan semua konfigurasi yang diperlukan
-combined_interface.launch(
-    share=True,      # Mengaktifkan URL publik
-    server_name="0.0.0.0", # Bind ke semua interfaces
-    server_port=7860,      # Port standar Gradio
-    enable_queue=True,     # Mengaktifkan antrian untuk permintaan API
-)
+# Gabungkan Gradio dengan FastAPI
+app = gr.mount_gradio_app(app, demo, path="/")
 
-# CATATAN: Endpoint API yang terdaftar akan tersedia di:
-# POST https://[your-space-url]/api/predict
-# dengan format payload:
-# {"data": ["base64_image_string", "work_type_string"]} 
+# Jalankan aplikasi dengan uvicorn
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)