Update app.py

app.py

@@ -1,212 +1,103 @@
-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"
+import numpy as np
+from vision_agent.tools import *
+from pillow_heif import register_heif_opener
+from typing import Dict
+
+# Register HEIF opener
+register_heif_opener()
+
+import vision_agent as va
+
+def analyze_mixed_boxes(image) -> Dict:
+    """
+    Analyzes an image containing mixed types of beverages, specifically water bottles and beverage cans.
+    1) Loads the image from the provided path.
+    2) Uses the 'countgd_object_detection' tool with the prompt 'water bottle, beverage can' to detect items.
+    3) Splits detections into a top shelf and bottom shelf by comparing detection center to the image's vertical midpoint.
+    4) Calculates how many water bottles and beverage cans are on each shelf and overall, along with average confidence scores.
+    5) Overlays bounding boxes on the image to visualize detections, then saves the annotated image.
+    6) Returns a dictionary summarizing the distribution of water bottles and beverage cans.
+
+    Parameters:
+        image (PIL.Image): The uploaded image.
+
+    Returns:
+        dict: Summary of the analysis with keys:
+            - total_items (int): total number of detected items
+            - total_water_bottles (int): total count of detected water bottles
+            - total_beverage_cans (int): total count of detected beverage cans
+            - top_shelf (dict): counts of bottles and cans on top shelf
+            - bottom_shelf (dict): counts of bottles and cans on bottom shelf
+            - confidence (dict): average confidence scores for bottles and cans
+    """
+    # Convert the uploaded image to a numpy array
+    image = np.array(image)
+    height, width = image.shape[:2]
+
+    # Detect water bottles and beverage cans
+    detections = countgd_object_detection("water bottle, beverage can", image)
+
+    # Separate detections into top shelf and bottom shelf
+    mid_height = height / 2
+    top_shelf_dets = []
+    bottom_shelf_dets = []
+    for det in detections:
+        cy = ((det["bbox"][1] + det["bbox"][3]) / 2) * height
+        if cy < mid_height:
+            top_shelf_dets.append(det)
         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)
+            bottom_shelf_dets.append(det)
 
-        # 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)
+    # Count items by label and calculate average confidence
+    water_bottles = [det for det in detections if det["label"] == "water bottle"]
+    beverage_cans = [det for det in detections if det["label"] == "beverage can"]
 
-        return output_path, result_text
+    avg_bottle_conf = (sum(det["score"] for det in water_bottles) / len(water_bottles)
+                       if water_bottles else 0)
+    avg_can_conf = (sum(det["score"] for det in beverage_cans) / len(beverage_cans)
+                    if beverage_cans else 0)
 
-    except Exception as e:
-        return temp_path, f"Error: {str(e)}"
-    finally:
-        os.remove(temp_path)
+    top_water_bottles = [det for det in top_shelf_dets if det["label"] == "water bottle"]
+    top_beverage_cans = [det for det in top_shelf_dets if det["label"] == "beverage can"]
+    bottom_water_bottles = [det for det in bottom_shelf_dets if det["label"] == "water bottle"]
+    bottom_beverage_cans = [det for det in bottom_shelf_dets if det["label"] == "beverage can"]
 
-# ========== 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>""")
+    # Overlay bounding boxes and save the annotated image
+    annotated_image = overlay_bounding_boxes(image, detections)
     
-    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])
+    # Convert annotated image back to PIL format for Gradio output
+    annotated_image_pil = Image.fromarray(annotated_image)
+    
+    # Return the result
+    result = {
+        "total_items": len(detections),
+        "total_water_bottles": len(water_bottles),
+        "total_beverage_cans": len(beverage_cans),
+        "top_shelf": {
+            "water_bottles": len(top_water_bottles),
+            "beverage_cans": len(top_beverage_cans),
+        },
+        "bottom_shelf": {
+            "water_bottles": len(bottom_water_bottles),
+            "beverage_cans": len(bottom_beverage_cans),
+        },
+        "confidence": {
+            "water_bottles": round(avg_bottle_conf, 2),
+            "beverage_cans": round(avg_can_conf, 2),
+        },
+        "annotated_image": annotated_image_pil  # return annotated image for display
+    }
+    
+    return result
+
+# Gradio Interface
+iface = gr.Interface(
+    fn=analyze_mixed_boxes,
+    inputs=gr.Image(type="pil"),  # allows image upload
+    outputs=[gr.JSON(), gr.Image(type="pil")],  # display result and annotated image
+    title="Beverage Detection Analysis",
+    description="Upload an image containing water bottles and beverage cans, and the tool will analyze the distribution on shelves and display an annotated image.",
+)
 
 iface.launch()