app.py

import gradio as gr
import numpy as np
from PIL import Image, ImageDraw, ImageFont
import io
import base64
from datetime import datetime
import json
from fastapi import FastAPI, Request

# Setup FastAPI instance
app = FastAPI()

def detect_ppe(image, work_type="ketinggian"):
    """Fungsi utama untuk deteksi APD (simulasi)"""
    if image is None:
        return None, "Tidak ada gambar", {"error": "Tidak ada gambar yang diberikan"}
    
    try:
        # Konversi ke PIL Image jika perlu
        if isinstance(image, np.ndarray):
            pil_image = Image.fromarray(image)
        else:
            pil_image = image
        
        width, height = pil_image.size
        
        # 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"
            }
        ]
        
        # Simulasikan hasil kepatuhan
        compliance_status = "Tidak Patuh (Non-compliant)"
        compliance_issues = ["Tidak ada pelindung kepala (helm)", "Tidak ada pelindung kaki (sepatu safety)"]
        
        # Gambar hasil deteksi
        result_image = draw_detections(pil_image, detections)
        
        # Format JSON untuk respons API
        result = {
            "detections": {str(i+1): d for i, d in enumerate(detections)},
            "compliance": {
                "status": compliance_status,
                "issues": {str(i+1): issue for i, issue in enumerate(compliance_issues)}
            },
            "work_type": work_type,
            "summary": f"Deteksi: {len(detections)} objek. Status: {compliance_status}",
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "recommendations": {
                "1": "Pastikan pekerja menggunakan pelindung kepala (helm)",
                "2": "Pastikan pekerja menggunakan pelindung kaki (sepatu safety)"
            }
        }
        
        return result_image, compliance_status, result
        
    except Exception as e:
        return None, f"Error: {str(e)}", {"error": str(e)}

def draw_detections(image, detections):
    """Menggambar kotak deteksi pada gambar"""
    img_copy = image.copy()
    draw = ImageDraw.Draw(img_copy)
    
    try:
        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"]
        
        # Warna berdasarkan tipe
        color = "red" if label == "person" else "blue"
        
        # Gambar box
        draw.rectangle([(box[0], box[1]), (box[2], box[3])], outline=color, width=3)
        text = f"{label}: {confidence:.2f}"
        
        # 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 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
        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]
        
        # Decode base64 ke bytes
        image_bytes = base64.b64decode(base64_image)
        
        # Buka gambar dari bytes
        image = Image.open(io.BytesIO(image_bytes))
        
        # Proses gambar
        result_image, status, analysis_result = detect_ppe(image, work_type)
        
        # 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]
        return [None, status, analysis_result]
    
    except Exception as e:
        return {"error": str(e)}

# Buat UI utama
demo = gr.Blocks(title="K3L APD Detection")

with demo:
    gr.Markdown("# Sistem Deteksi APD untuk K3L Platform")
    gr.Markdown("Upload gambar untuk analisis kepatuhan APD di lokasi konstruksi")
    
    with gr.Row():
        with gr.Column():
            input_image = gr.Image(label="Upload Gambar")
            work_type = gr.Radio(
                label="Jenis Pekerjaan", 
                choices=["ketinggian", "konstruksi_umum", "listrik"],
                value="ketinggian"
            )
            analyze_btn = gr.Button("Analisis Gambar", variant="primary")
        
        with gr.Column():
            output_image = gr.Image(label="Hasil Deteksi")
            status_output = gr.Textbox(label="Status Kepatuhan")
            output_json = gr.JSON(label="Detail Analisis")
    
    analyze_btn.click(
        fn=detect_ppe,
        inputs=[input_image, work_type],
        outputs=[output_image, status_output, output_json]
    )

    gr.Markdown("""
    ### Panduan Penggunaan
    1. Upload gambar situasi konstruksi 
    2. Pilih jenis pekerjaan (mempengaruhi persyaratan APD)
    3. Klik tombol "Analisis Gambar"
    4. Lihat hasil analisis dengan kotak deteksi, status kepatuhan, dan rekomendasi
    
    ### API Endpoint
    API endpoint tersedia di: `/api/predict`
    
    Format request (POST):
    ```json
    {
        "data": [
            "BASE64_IMAGE",
            "JENIS_PEKERJAAN"  (opsional)
        ]
    }
    ```
    """)

# Gabungkan Gradio dengan FastAPI
app = gr.mount_gradio_app(app, demo, path="/")

# Jalankan aplikasi dengan uvicorn
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=7860)