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("""

NESTLE - STOCK COUNTING

""") 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()