import gradio as gr import torch import numpy as np from transformers import OwlViTProcessor, OwlViTForObjectDetection from PIL import Image, ImageDraw import cv2 import torch.nn.functional as F import tempfile import matplotlib.pyplot as plt import matplotlib.cm as cm from io import BytesIO from SuperGluePretrainedNetwork.models.matching import Matching from SuperGluePretrainedNetwork.models.utils import read_image # Set device device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # Load models model = OwlViTForObjectDetection.from_pretrained("google/owlvit-base-patch32").to(device) processor = OwlViTProcessor.from_pretrained("google/owlvit-base-patch32") matching = Matching({ 'superpoint': {'nms_radius': 4, 'keypoint_threshold': 0.005, 'max_keypoints': 1024}, 'superglue': {'weights': 'outdoor', 'sinkhorn_iterations': 20, 'match_threshold': 0.2} }).eval().to(device) # Utility functions def save_array_to_temp_image(arr): rgb_arr = cv2.cvtColor(arr, cv2.COLOR_BGR2RGB) img = Image.fromarray(rgb_arr) temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.png') temp_file_name = temp_file.name temp_file.close() img.save(temp_file_name) return temp_file_name def unified_matching_plot2(image0, image1, kpts0, kpts1, mkpts0, mkpts1, color, text, path=None, show_keypoints=False, fast_viz=False, opencv_display=False, opencv_title='matches', small_text=[]): height = min(image0.shape[0], image1.shape[0]) image0_resized = cv2.resize(image0, (int(image0.shape[1] * height / image0.shape[0]), height)) image1_resized = cv2.resize(image1, (int(image1.shape[1] * height / image1.shape[0]), height)) plt.figure(figsize=(15, 15)) plt.subplot(1, 2, 1) plt.imshow(image0_resized) plt.scatter(kpts0[:, 0], kpts0[:, 1], color='r', s=1) plt.axis('off') plt.subplot(1, 2, 2) plt.imshow(image1_resized) plt.scatter(kpts1[:, 0], kpts1[:, 1], color='r', s=1) plt.axis('off') fig, ax = plt.subplots(figsize=(20, 20)) plt.plot([mkpts0[:, 0], mkpts1[:, 0] + image0_resized.shape[1]], [mkpts0[:, 1], mkpts1[:, 1]], 'r', lw=0.5) plt.scatter(mkpts0[:, 0], mkpts0[:, 1], s=2, marker='o', color='b') plt.scatter(mkpts1[:, 0] + image0_resized.shape[1], mkpts1[:, 1], s=2, marker='o', color='g') plt.imshow(np.hstack([image0_resized, image1_resized]), aspect='auto') plt.suptitle('\n'.join(text), fontsize=20, fontweight='bold') plt.tight_layout() plt.show() buf = BytesIO() plt.savefig(buf, format='png') buf.seek(0) img_arr = np.frombuffer(buf.getvalue(), dtype=np.uint8) buf.close() img = cv2.imdecode(img_arr, 1) img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) plt.close(fig) return img def stitch_images(images): """Stitches a list of images vertically.""" if not images: return Image.new('RGB', (100, 100), color='gray') max_width = max([img.width for img in images]) total_height = sum(img.height for img in images) composite = Image.new('RGB', (max_width, total_height)) y_offset = 0 for img in images: composite.paste(img, (0, y_offset)) y_offset += img.height return composite # Main functions def detect_and_crop(target_image, query_image, threshold=0.5, nms_threshold=0.3): target_sizes = torch.Tensor([target_image.size[::-1]]) inputs = processor(images=target_image, query_images=query_image, return_tensors="pt").to(device) with torch.no_grad(): outputs = model.image_guided_detection(**inputs) img = cv2.cvtColor(np.array(target_image), cv2.COLOR_BGR2RGB) outputs.logits = outputs.logits.cpu() outputs.target_pred_boxes = outputs.target_pred_boxes.cpu() results = processor.post_process_image_guided_detection(outputs=outputs, threshold=threshold, nms_threshold=nms_threshold, target_sizes=target_sizes) boxes, scores = results[0]["boxes"], results[0]["scores"] if len(boxes) == 0: return [], None filtered_boxes = [] for box in boxes: x1, y1, x2, y2 = [int(i) for i in box.tolist()] cropped_img = img[y1:y2, x1:x2] if cropped_img.size != 0: filtered_boxes.append(cropped_img) draw = ImageDraw.Draw(target_image) for box in boxes: draw.rectangle(box.tolist(), outline="red", width=3) return filtered_boxes, target_image def image_matching_no_pyramid(query_img, target_img, visualize=True): temp_query = save_array_to_temp_image(np.array(query_img)) temp_target = save_array_to_temp_image(np.array(target_img)) image1, inp1, scales1 = read_image(temp_target, device, [640*2], 0, True) image0, inp0, scales0 = read_image(temp_query, device, [640*2], 0, True) if image0 is None or image1 is None: return None pred = matching({'image0': inp0, 'image1': inp1}) pred = {k: v[0] for k, v in pred.items()} kpts0, kpts1 = pred['keypoints0'], pred['keypoints1'] matches, conf = pred['matches0'], pred['matching_scores0'] valid = matches > -1 mkpts0 = kpts0[valid] mkpts1 = kpts1[matches[valid]] mconf = conf[valid] color = cm.jet(mconf.detach().cpu().numpy())[:len(mkpts0)] valid_count = np.sum(valid.tolist()) mkpts0_np = mkpts0.cpu().numpy() mkpts1_np = mkpts1.cpu().numpy() try: H, inliers = cv2.findHomography(mkpts0_np, mkpts1_np, cv2.RANSAC, 5.0) except: inliers = 0 num_inliers = np.sum(inliers) if visualize: visualized_img = unified_matching_plot2( image0, image1, kpts0, kpts1, mkpts0, mkpts1, color, ['Matches'], True, False, True, 'Matches', []) else: visualized_img = None return { 'valid': [valid_count], 'inliers': [num_inliers], 'visualized_image': [visualized_img] } def check_object_in_image(query_image, target_image, threshold=50, scale_factor=[0.33, 0.66, 1]): images_to_return = [] cropped_images, bbox_image = detect_and_crop(target_image, query_image) temp_files = [save_array_to_temp_image(i) for i in cropped_images] crop_results = [image_matching_no_pyramid(query_image, Image.open(i), visualize=True) for i in temp_files] cropped_visuals = [] cropped_inliers = [] for result in crop_results: if result: for img in result['visualized_image']: cropped_visuals.append(Image.fromarray(img)) for inliers_ in result['inliers']: cropped_inliers.append(inliers_) images_to_return.append(stitch_images(cropped_visuals)) is_present = any(value >= threshold for value in cropped_inliers) return { 'is_present': is_present, 'image_with_boxes': bbox_image, 'object_detection_inliers': [int(i) for i in cropped_inliers], } def interface(poster_source, media_source, threshold, scale_factor): result1 = check_object_in_image(poster_source, media_source, threshold, scale_factor) if result1['is_present']: return result1['is_present'], result1['image_with_boxes'] result2 = check_object_in_image(poster_source, media_source, threshold, scale_factor) return result2['is_present'], result2['image_with_boxes'] iface = gr.Interface( fn=interface, inputs=[ gr.Image(type="pil", label="Upload a Query Image (Poster)"), gr.Image(type="pil", label="Upload a Target Image (Media)"), gr.Slider(minimum=0, maximum=100, step=1, value=50, label="Threshold"), gr.CheckboxGroup(choices=["0.33", "0.66", "1.0"], value=["0.33", "0.66", "1.0"], label="Scale Factors"), ], outputs=[ gr.Label(label="Object Presence"), gr.Image(type="pil", label="Detected Bounding Boxes"), ], title="Object Detection in Images", description=""" This application allows you to check if an object in a query image (poster) is present in a target image (media). Steps: 1. Upload a Query Image (Poster) 2. Upload a Target Image (Media) 3. Set Threshold 4. Set Scale Factors 5. View Results """ ) if __name__ == "__main__": iface.launch()