app.py

import os
import numpy as np
import tempfile
import requests
import cv2
import gradio as gr
from dotenv import load_dotenv
from roboflow import Roboflow
import subprocess

# ========== Konfigurasi ========== 
load_dotenv()

# Roboflow Config
rf_api_key = os.getenv("ROBOFLOW_API_KEY")
workspace = os.getenv("ROBOFLOW_WORKSPACE")
project_name = os.getenv("ROBOFLOW_PROJECT")
model_version = int(os.getenv("ROBOFLOW_MODEL_VERSION"))

# countgd Model Configuration
COUNTGD_API_KEY = os.getenv("COUNTGD_API_KEY")
COUNTGD_MODEL_URL = "https://api.landing.ai/v1/tools/countgd-object-detection"  # Replace with the correct API endpoint

# Inisialisasi Model
rf = Roboflow(api_key=rf_api_key)
project = rf.workspace(workspace).project(project_name)
yolo_model = project.version(model_version).model

# ========== Fungsi Deteksi Kombinasi ========== 
def detect_combined(image):
    with tempfile.NamedTemporaryFile(delete=False, suffix=".jpg") as temp_file:
        image.save(temp_file, format="JPEG")
        temp_path = temp_file.name

    try:
        # ========== [1] YOLO: Deteksi Produk Nestlé (Per Class) ========== 
        yolo_pred = yolo_model.predict(temp_path, confidence=50, overlap=80).json()

        # Hitung per class Nestlé
        nestle_class_count = {}
        nestle_boxes = []
        for pred in yolo_pred['predictions']:
            class_name = pred['class']
            nestle_class_count[class_name] = nestle_class_count.get(class_name, 0) + 1
            nestle_boxes.append((pred['x'], pred['y'], pred['width'], pred['height']))

        total_nestle = sum(nestle_class_count.values())

        # ========== [2] countgd: Deteksi Produk dengan countgd Model ========== 
        # Make a request to the countgd model API (adjust parameters accordingly)
        with open(temp_path, 'rb') as img_file:
            response = requests.post(
                COUNTGD_MODEL_URL,
                headers={"Authorization": f"Bearer {COUNTGD_API_KEY}"},
                files={"image": img_file},
                data={"prompts": ["water bottle", "beverage can"]}
            )

        # Handle the response from the countgd model
        if response.status_code == 200:
            countgd_pred = response.json()['detections']
        else:
            return temp_path, f"Error calling countgd API: {response.text}"

        # Filter & Hitung Kompetitor
        competitor_class_count = {}
        competitor_boxes = []
        for obj in countgd_pred:
            # Filter and process the detections
            class_name = obj['label']
            if class_name.lower() in ['water bottle', 'beverage can']:  # Modify this as needed
                competitor_class_count[class_name] = competitor_class_count.get(class_name, 0) + 1
                competitor_boxes.append({
                    "class": class_name,
                    "box": obj['bbox'],
                    "confidence": obj['score']
                })

        total_competitor = sum(competitor_class_count.values())

        # ========== [3] Format Output ========== 
        result_text = "Product Nestle\n\n"
        for class_name, count in nestle_class_count.items():
            result_text += f"{class_name}: {count}\n"
        result_text += f"\nTotal Products Nestle: {total_nestle}\n\n"

        # Unclassified Products (from countgd model)
        if competitor_class_count:
            result_text += f"Total Unclassified Products: {total_competitor}\n"
        else:
            result_text += "No Unclassified Products detected\n"

        # ========== [4] Visualisasi ========== 
        img = cv2.imread(temp_path)

        # Nestlé (Hijau)
        for pred in yolo_pred['predictions']:
            x, y, w, h = pred['x'], pred['y'], pred['width'], pred['height']
            cv2.rectangle(img, (int(x-w/2), int(y-h/2)), (int(x+w/2), int(y+h/2)), (0,255,0), 2)
            cv2.putText(img, pred['class'], (int(x-w/2), int(y-h/2-10)), 
                       cv2.FONT_HERSHEY_SIMPLEX, 0.55, (0,255,0), 2)

        # Kompetitor (Merah) with countgd detections
        for comp in competitor_boxes:
            x1, y1, x2, y2 = comp['box']
            cv2.rectangle(img, (int(x1), int(y1)), (int(x2), int(y2)), (0, 0, 255), 2)
            cv2.putText(img, f"{comp['class']} {comp['confidence']:.2f}",
                        (int(x1), int(y1-10)), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (0, 0, 255), 2)
            
        output_path = "/tmp/combined_output.jpg"
        cv2.imwrite(output_path, img)

        return output_path, result_text

    except Exception as e:
        return temp_path, f"Error: {str(e)}"
    finally:
        os.remove(temp_path)

# ========== Fungsi untuk Deteksi Video ========== 

def convert_video_to_mp4(input_path, output_path):
    try:
        subprocess.run(['ffmpeg', '-i', input_path, '-vcodec', 'libx264', '-acodec', 'aac', output_path], check=True)
        return output_path
    except subprocess.CalledProcessError as e:
        return None, f"Error converting video: {e}"

def detect_objects_in_video(video_path):
    temp_output_path = "/tmp/output_video.mp4"
    temp_frames_dir = tempfile.mkdtemp()
    frame_count = 0

    try:
        # Convert video to MP4 if necessary
        if not video_path.endswith(".mp4"):
            video_path, err = convert_video_to_mp4(video_path, temp_output_path)
            if not video_path:
                return None, f"Video conversion error: {err}"

        # Read video and process frames
        video = cv2.VideoCapture(video_path)
        frame_rate = int(video.get(cv2.CAP_PROP_FPS))
        frame_width = int(video.get(cv2.CAP_PROP_FRAME_WIDTH))
        frame_height = int(video.get(cv2.CAP_PROP_FRAME_HEIGHT))
        frame_size = (frame_width, frame_height)

        # VideoWriter for output video
        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
        output_video = cv2.VideoWriter(temp_output_path, fourcc, frame_rate, frame_size)

        while True:
            ret, frame = video.read()
            if not ret:
                break

            # Process predictions for the current frame using countgd model (same as in image detection)
            frame_path = os.path.join(temp_frames_dir, f"frame_{frame_count}.jpg")
            cv2.imwrite(frame_path, frame)

            # Get predictions from countgd (adjust accordingly for video frames)
            response = requests.post(
                COUNTGD_MODEL_URL,
                headers={"Authorization": f"Bearer {COUNTGD_API_KEY}"},
                files={"image": open(frame_path, 'rb')},
                data={"prompts": ["water bottle", "beverage can"]}
            )

            # Process the response (similarly to what was done for image detection)
            if response.status_code == 200:
                countgd_pred = response.json()['detections']
            else:
                continue

            # Drawing detections on frames
            for obj in countgd_pred:
                x1, y1, x2, y2 = obj['bbox']
                cv2.rectangle(frame, (int(x1), int(y1)), (int(x2), int(y2)), (0, 0, 255), 2)
                cv2.putText(frame, f"{obj['label']} {obj['score']:.2f}", 
                            (int(x1), int(y1-10)), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (0, 0, 255), 2)

            # Write processed frame to output video
            output_video.write(frame)
            frame_count += 1

        video.release()
        output_video.release()

        return temp_output_path

    except Exception as e:
        return None, f"An error occurred: {e}"

# ========== Gradio Interface ========== 
with gr.Blocks(theme=gr.themes.Base(primary_hue="teal", secondary_hue="teal", neutral_hue="slate")) as iface:
    gr.Markdown("""<div style="text-align: center;"><h1>NESTLE - STOCK COUNTING</h1></div>""")
    
    with gr.Row():
        with gr.Column():
            input_image = gr.Image(type="pil", label="Input Image")
            detect_image_button = gr.Button("Detect Image")
            output_image = gr.Image(label="Detect Object")
            output_text = gr.Textbox(label="Counting Object")
            detect_image_button.click(fn=detect_combined, inputs=input_image, outputs=[output_image, output_text])

        with gr.Column():
            input_video = gr.Video(label="Input Video")
            detect_video_button = gr.Button("Detect Video")
            output_video = gr.Video(label="Output Video")
            detect_video_button.click(fn=detect_objects_in_video, inputs=input_video, outputs=[output_video])

iface.launch()