import argparse import base64 import logging import os import sys import traceback import threading from collections import Counter from io import BytesIO from typing import Dict, List, Optional, Tuple import gradio as gr import pandas as pd import requests import torch import uvicorn from fastapi import FastAPI, File, Form, HTTPException, UploadFile from fastapi.responses import JSONResponse from PIL import Image, ImageDraw, ImageStat from transformers import ( DetrForObjectDetection, DetrForSegmentation, DetrImageProcessor, YolosForObjectDetection, YolosImageProcessor, ) import nest_asyncio # ------------------------------ # Configuration # ------------------------------ # Logging configuration logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", ) logger = logging.getLogger(__name__) # Model and processing constants CONFIDENCE_THRESHOLD: float = 0.5 VALID_MODELS: List[str] = [ "facebook/detr-resnet-50", "facebook/detr-resnet-101", "facebook/detr-resnet-50-panoptic", "facebook/detr-resnet-101-panoptic", "hustvl/yolos-tiny", "hustvl/yolos-base", ] MODEL_DESCRIPTIONS: Dict[str, str] = { "facebook/detr-resnet-50": ( "DETR with ResNet-50 backbone for object detection. Fast and accurate for general use." ), "facebook/detr-resnet-101": ( "DETR with ResNet-101 backbone for object detection. More accurate but slower than ResNet-50." ), "facebook/detr-resnet-50-panoptic": ( "DETR with ResNet-50 for panoptic segmentation. Detects objects and segments scenes." ), "facebook/detr-resnet-101-panoptic": ( "DETR with ResNet-101 for panoptic segmentation. High accuracy for complex scenes." ), "hustvl/yolos-tiny": ( "YOLOS Tiny model. Lightweight and fast, ideal for resource-constrained environments." ), "hustvl/yolos-base": ( "YOLOS Base model. Balances speed and accuracy for object detection." ), } # Port configuration DEFAULT_GRADIO_PORT: int = 7860 DEFAULT_FASTAPI_PORT: int = 8000 PORT_RANGE: range = range(7860, 7870) # Try ports 7860-7869 MAX_PORT_ATTEMPTS: int = 10 # Thread-safe storage for lazy-loaded models and processors models: Dict[str, any] = {} processors: Dict[str, any] = {} model_lock = threading.Lock() # ------------------------------ # Model Loading # ------------------------------ def load_model_and_processor(model_name: str) -> Tuple[any, any]: """ Load and cache the specified model and processor thread-safely. Args: model_name: Name of the model to load (must be in VALID_MODELS). Returns: Tuple containing the loaded model and processor. Raises: ValueError: If the model_name is invalid or loading fails. """ with model_lock: if model_name not in models: logger.info(f"Loading model: {model_name}") try: if "yolos" in model_name: models[model_name] = YolosForObjectDetection.from_pretrained(model_name) processors[model_name] = YolosImageProcessor.from_pretrained(model_name) elif "panoptic" in model_name: models[model_name] = DetrForSegmentation.from_pretrained(model_name) processors[model_name] = DetrImageProcessor.from_pretrained(model_name) else: models[model_name] = DetrForObjectDetection.from_pretrained(model_name) processors[model_name] = DetrImageProcessor.from_pretrained(model_name) logger.debug(f"Model {model_name} loaded successfully") except Exception as e: logger.error(f"Failed to load model {model_name}: {str(e)}") raise ValueError(f"Failed to load model: {str(e)}") return models[model_name], processors[model_name] # ------------------------------ # Image Processing # ------------------------------ def process(image: Image.Image, model_name: str) -> Tuple[Image.Image, List[str], List[float], List[str], List[float], Dict[str, str]]: """ Process an image for object detection or panoptic segmentation. Args: image: PIL Image to process. model_name: Name of the model to use (must be in VALID_MODELS). Returns: Tuple containing: - Annotated image (PIL Image). - List of detected object names. - List of confidence scores for detected objects. - List of unique object names. - List of confidence scores for unique objects. - Dictionary of image properties (format, size, etc.). Raises: ValueError: If the model_name is invalid. RuntimeError: If processing fails due to model or image issues. """ if model_name not in VALID_MODELS: raise ValueError(f"Invalid model: {model_name}. Choose from: {VALID_MODELS}") try: # Load model and processor model, processor = load_model_and_processor(model_name) logger.debug(f"Processing image with model: {model_name}") # Prepare image for processing inputs = processor(images=image, return_tensors="pt") with torch.no_grad(): outputs = model(**inputs) # Initialize drawing context draw = ImageDraw.Draw(image) object_names: List[str] = [] confidence_scores: List[float] = [] object_counter = Counter() target_sizes = torch.tensor([image.size[::-1]]) # Process panoptic segmentation or object detection if "panoptic" in model_name: processed_sizes = torch.tensor([[inputs["pixel_values"].shape[2], inputs["pixel_values"].shape[3]]]) results = processor.post_process_panoptic(outputs, target_sizes=target_sizes, processed_sizes=processed_sizes)[0] for segment in results["segments_info"]: label = segment["label_id"] label_name = model.config.id2label.get(label, "Unknown") score = segment.get("score", 1.0) # Apply segmentation mask if available if "masks" in results and segment["id"] < len(results["masks"]): mask = results["masks"][segment["id"]].cpu().numpy() if mask.shape[0] > 0 and mask.shape[1] > 0: mask_image = Image.fromarray((mask * 255).astype("uint8")) colored_mask = Image.new("RGBA", image.size, (0, 0, 0, 0)) mask_draw = ImageDraw.Draw(colored_mask) r, g, b = (segment["id"] * 50) % 255, (segment["id"] * 100) % 255, (segment["id"] * 150) % 255 mask_draw.bitmap((0, 0), mask_image, fill=(r, g, b, 128)) image = Image.alpha_composite(image.convert("RGBA"), colored_mask).convert("RGB") draw = ImageDraw.Draw(image) if score > CONFIDENCE_THRESHOLD: object_names.append(label_name) confidence_scores.append(float(score)) object_counter[label_name] = float(score) else: results = processor.post_process_object_detection(outputs, target_sizes=target_sizes)[0] for score, label, box in zip(results["scores"], results["labels"], results["boxes"]): if score > CONFIDENCE_THRESHOLD: x, y, x2, y2 = box.tolist() draw.rectangle([x, y, x2, y2], outline="#32CD32", width=2) label_name = model.config.id2label.get(label.item(), "Unknown") text = f"{label_name}: {score:.2f}" text_bbox = draw.textbbox((0, 0), text) text_width, text_height = text_bbox[2] - text_bbox[0], text_bbox[3] - text_bbox[1] draw.text((x2 - text_width - 2, y - text_height - 2), text, fill="#32CD32") object_names.append(label_name) confidence_scores.append(float(score)) object_counter[label_name] = float(score) # Compile unique objects and confidences unique_objects = list(object_counter.keys()) unique_confidences = [object_counter[obj] for obj in unique_objects] # Calculate image properties properties: Dict[str, str] = { "Format": image.format if hasattr(image, "format") and image.format else "Unknown", "Size": f"{image.width}x{image.height}", "Width": f"{image.width} px", "Height": f"{image.height} px", "Mode": image.mode, "Aspect Ratio": ( f"{round(image.width / image.height, 2)}" if image.height != 0 else "Undefined" ), "File Size": "Unknown", "Mean (R,G,B)": "Unknown", "StdDev (R,G,B)": "Unknown", } # Compute file size try: buffered = BytesIO() image.save(buffered, format="PNG") properties["File Size"] = f"{len(buffered.getvalue()) / 1024:.2f} KB" except Exception as e: logger.error(f"Error calculating file size: {str(e)}") # Compute color statistics try: stat = ImageStat.Stat(image) properties["Mean (R,G,B)"] = ", ".join(f"{m:.2f}" for m in stat.mean) properties["StdDev (R,G,B)"] = ", ".join(f"{s:.2f}" for s in stat.stddev) except Exception as e: logger.error(f"Error calculating color statistics: {str(e)}") return image, object_names, confidence_scores, unique_objects, unique_confidences, properties except Exception as e: logger.error(f"Error in process: {str(e)}\n{traceback.format_exc()}") raise RuntimeError(f"Failed to process image: {str(e)}") # ------------------------------ # FastAPI Setup # ------------------------------ app = FastAPI(title="Object Detection API") @app.post("/detect") async def detect_objects_endpoint( file: Optional[UploadFile] = File(None), image_url: Optional[str] = Form(None), model_name: str = Form(VALID_MODELS[0]), ) -> JSONResponse: """ FastAPI endpoint to detect objects in an image from file upload or URL. Args: file: Uploaded image file (optional). image_url: URL of the image (optional). model_name: Model to use for detection (default: first VALID_MODELS entry). Returns: JSONResponse containing the processed image (base64), detected objects, and confidences. Raises: HTTPException: If input validation fails or processing errors occur. """ try: # Validate input if (file is None and not image_url) or (file is not None and image_url): raise HTTPException( status_code=400, detail="Provide either an image file or an image URL, not both.", ) # Load image if file: if not file.content_type.startswith("image/"): raise HTTPException(status_code=400, detail="File must be an image") contents = await file.read() image = Image.open(BytesIO(contents)).convert("RGB") else: response = requests.get(image_url, timeout=10) response.raise_for_status() image = Image.open(BytesIO(response.content)).convert("RGB") if model_name not in VALID_MODELS: raise HTTPException( status_code=400, detail=f"Invalid model. Choose from: {VALID_MODELS}", ) # Process image detected_image, detected_objects, detected_confidences, unique_objects, unique_confidences, _ = process( image, model_name ) # Encode image as base64 buffered = BytesIO() detected_image.save(buffered, format="PNG") img_base64 = base64.b64encode(buffered.getvalue()).decode("utf-8") img_url = f"data:image/png;base64,{img_base64}" return JSONResponse( content={ "image_url": img_url, "detected_objects": detected_objects, "confidence_scores": detected_confidences, "unique_objects": unique_objects, "unique_confidence_scores": unique_confidences, } ) except requests.RequestException as e: logger.error(f"Error fetching image from URL: {str(e)}") raise HTTPException(status_code=400, detail=f"Failed to fetch image: {str(e)}") except Exception as e: logger.error(f"Error in FastAPI endpoint: {str(e)}\n{traceback.format_exc()}") raise HTTPException(status_code=500, detail=f"Error processing image: {str(e)}") # ------------------------------ # Gradio UI Setup # ------------------------------ def create_gradio_ui() -> gr.Blocks: """ Create and configure the Gradio UI for object detection. Returns: Gradio Blocks object representing the UI. Raises: RuntimeError: If UI creation fails. """ try: with gr.Blocks(theme=gr.themes.Default(primary_hue="blue", secondary_hue="gray")) as app: gr.Markdown( f""" # 🚀 Object Detection App Upload an image or provide a URL to detect objects using state-of-the-art transformer models (DETR, YOLOS). Running on port: {os.getenv('GRADIO_SERVER_PORT', 'auto-selected')} """ ) with gr.Tabs(): with gr.Tab("📷 Image Upload"): with gr.Row(): with gr.Column(scale=1): gr.Markdown("### Input") model_choice = gr.Dropdown( choices=VALID_MODELS, value=VALID_MODELS[0], label="🔎 Select Model", info="Choose a model for object detection or panoptic segmentation.", ) model_info = gr.Markdown( f"**Model Info**: {MODEL_DESCRIPTIONS[VALID_MODELS[0]]}", visible=True, ) image_input = gr.Image(type="pil", label="📷 Upload Image") image_url_input = gr.Textbox( label="🔗 Image URL", placeholder="https://example.com/image.jpg", ) with gr.Row(): submit_btn = gr.Button("✨ Detect", variant="primary") clear_btn = gr.Button("🗑ī¸ Clear", variant="secondary") model_choice.change( fn=lambda model_name: ( f"**Model Info**: {MODEL_DESCRIPTIONS.get(model_name, 'No description available.')}" ), inputs=model_choice, outputs=model_info, ) with gr.Column(scale=2): gr.Markdown("### Results") error_output = gr.Textbox( label="⚠ī¸ Errors", visible=False, lines=3, max_lines=5, ) output_image = gr.Image( type="pil", label="đŸŽ¯ Detected Image", interactive=False, ) with gr.Row(): objects_output = gr.DataFrame( label="📋 Detected Objects", interactive=False, value=None, ) unique_objects_output = gr.DataFrame( label="🔍 Unique Objects", interactive=False, value=None, ) properties_output = gr.DataFrame( label="📄 Image Properties", interactive=False, value=None, ) def process_for_gradio(image: Optional[Image.Image], url: Optional[str], model_name: str) -> Tuple[ Optional[Image.Image], Optional[pd.DataFrame], Optional[pd.DataFrame], Optional[pd.DataFrame], str ]: """ Process image for Gradio UI and return results. Args: image: Uploaded PIL Image (optional). url: Image URL (optional). model_name: Model to use for detection. Returns: Tuple of detected image, objects DataFrame, unique objects DataFrame, properties DataFrame, and error message. """ try: if image is None and not url: return None, None, None, None, "Please provide an image or URL" if image and url: return None, None, None, None, "Please provide either an image or URL, not both" if url: response = requests.get(url, timeout=10) response.raise_for_status() image = Image.open(BytesIO(response.content)).convert("RGB") detected_image, objects, scores, unique_objects, unique_scores, properties = process( image, model_name ) objects_df = ( pd.DataFrame( { "Object": objects, "Confidence Score": [f"{score:.2f}" for score in scores], } ) if objects else pd.DataFrame(columns=["Object", "Confidence Score"]) ) unique_objects_df = ( pd.DataFrame( { "Unique Object": unique_objects, "Confidence Score": [f"{score:.2f}" for score in unique_scores], } ) if unique_objects else pd.DataFrame(columns=["Unique Object", "Confidence Score"]) ) properties_df = ( pd.DataFrame([properties]) if properties else pd.DataFrame(columns=properties.keys()) ) return detected_image, objects_df, unique_objects_df, properties_df, "" except requests.RequestException as e: error_msg = f"Error fetching image from URL: {str(e)}" logger.error(f"{error_msg}\n{traceback.format_exc()}") return None, None, None, None, error_msg except Exception as e: error_msg = f"Error processing image: {str(e)}" logger.error(f"{error_msg}\n{traceback.format_exc()}") return None, None, None, None, error_msg submit_btn.click( fn=process_for_gradio, inputs=[image_input, image_url_input, model_choice], outputs=[output_image, objects_output, unique_objects_output, properties_output, error_output], ) clear_btn.click( fn=lambda: [None, "", None, None, None, None], inputs=None, outputs=[ image_input, image_url_input, output_image, objects_output, unique_objects_output, properties_output, error_output, ], ) with gr.Tab("🔗 JSON Output"): gr.Markdown("### Process Image for JSON Output") image_input_json = gr.Image(type="pil", label="📷 Upload Image") image_url_input_json = gr.Textbox( label="🔗 Image URL", placeholder="https://example.com/image.jpg", ) url_model_choice = gr.Dropdown( choices=VALID_MODELS, value=VALID_MODELS[0], label="🔎 Select Model", ) url_model_info = gr.Markdown( f"**Model Info**: {MODEL_DESCRIPTIONS[VALID_MODELS[0]]}", visible=True, ) url_submit_btn = gr.Button("🔄 Process", variant="primary") url_output = gr.JSON(label="API Response") url_model_choice.change( fn=lambda model_name: ( f"**Model Info**: {MODEL_DESCRIPTIONS.get(model_name, 'No description available.')}" ), inputs=url_model_choice, outputs=url_model_info, ) def process_url_for_gradio(image: Optional[Image.Image], url: Optional[str], model_name: str) -> Dict: """ Process image from file or URL for Gradio UI and return JSON response. Args: image: Uploaded PIL Image (optional). url: Image URL (optional). model_name: Model to use for detection. Returns: Dictionary with processed image (base64), detected objects, and confidences. """ try: if image is None and not url: return {"error": "Please provide an image or URL"} if image and url: return {"error": "Please provide either an image or URL, not both"} if url: response = requests.get(url, timeout=10) response.raise_for_status() image = Image.open(BytesIO(response.content)).convert("RGB") detected_image, objects, scores, unique_objects, unique_scores, _ = process( image, model_name ) buffered = BytesIO() detected_image.save(buffered, format="PNG") img_base64 = base64.b64encode(buffered.getvalue()).decode("utf-8") return { "image_url": f"data:image/png;base64,{img_base64}", "detected_objects": objects, "confidence_scores": scores, "unique_objects": unique_objects, "unique_confidence_scores": unique_scores, } except requests.RequestException as e: error_msg = f"Error fetching image from URL: {str(e)}" logger.error(f"{error_msg}\n{traceback.format_exc()}") return {"error": error_msg} except Exception as e: error_msg = f"Error processing image: {str(e)}" logger.error(f"{error_msg}\n{traceback.format_exc()}") return {"error": error_msg} url_submit_btn.click( fn=process_url_for_gradio, inputs=[image_input_json, image_url_input_json, url_model_choice], outputs=[url_output], ) with gr.Tab("ℹī¸ Help"): gr.Markdown( """ ## How to Use - **Image Upload**: Select a model, upload an image or provide a URL, and click "Detect" to see detected objects and image properties. - **JSON Output**: Upload an image or enter a URL, select a model, and click "Process" to get results in JSON format. - **Models**: Choose from DETR (object detection or panoptic segmentation) or YOLOS (lightweight detection). - **Clear**: Reset all inputs and outputs using the "Clear" button in the Image Upload tab. - **Errors**: Check the error box (Image Upload) or JSON response (JSON Output) for issues. ## Tips - Use high-quality images for better detection results. - Panoptic models (e.g., DETR-ResNet-50-panoptic) provide segmentation masks for complex scenes. - For faster processing, try YOLOS-Tiny on resource-constrained devices. """ ) return app except Exception as e: logger.error(f"Error creating Gradio UI: {str(e)}\n{traceback.format_exc()}") raise RuntimeError(f"Failed to create Gradio UI: {str(e)}") # ------------------------------ # Launcher # ------------------------------ def parse_args() -> argparse.Namespace: """ Parse command-line arguments with defaults and ignore unrecognized arguments. Returns: Parsed arguments as a Namespace object. Raises: SystemExit: If argument parsing fails (handled by argparse). """ parser = argparse.ArgumentParser( description="Launcher for Object Detection App with Gradio UI and optional FastAPI server." ) parser.add_argument( "--gradio-port", type=int, default=DEFAULT_GRADIO_PORT, help=f"Port for the Gradio UI (default: {DEFAULT_GRADIO_PORT}).", ) parser.add_argument( "--enable-fastapi", action="store_true", default=False, help="Enable the FastAPI server (disabled by default).", ) parser.add_argument( "--fastapi-port", type=int, default=DEFAULT_FASTAPI_PORT, help=f"Port for the FastAPI server if enabled (default: {DEFAULT_FASTAPI_PORT}).", ) # Parse known arguments and ignore unrecognized ones (e.g., Jupyter kernel args) args, _ = parser.parse_known_args() return args def find_available_port(start_port: int, port_range: range, max_attempts: int) -> Optional[int]: """ Find an available port within the specified range. Args: start_port: Initial port to try (e.g., from args or environment). port_range: Range of ports to attempt. max_attempts: Maximum number of ports to try. Returns: Available port number, or None if no port is found. Raises: OSError: If port binding fails for reasons other than port in use. """ import socket port = start_port attempts = 0 # Check environment variable GRADIO_SERVER_PORT env_port = os.getenv("GRADIO_SERVER_PORT") if env_port and env_port.isdigit(): port = int(env_port) logger.info(f"Using GRADIO_SERVER_PORT from environment: {port}") while attempts < max_attempts: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: try: s.bind(("0.0.0.0", port)) logger.debug(f"Port {port} is available") return port except OSError as e: if e.errno == 98: # Port in use logger.debug(f"Port {port} is in use") port = port + 1 if port < max(port_range) else min(port_range) attempts += 1 else: raise except Exception as e: logger.error(f"Error checking port {port}: {str(e)}") raise logger.error(f"No available port found in range {min(port_range)}-{max(port_range)} after {max_attempts} attempts") return None def run_fastapi_server(host: str, port: int) -> None: """ Run the FastAPI server using Uvicorn. Args: host: Host address for the FastAPI server. port: Port for the FastAPI server. """ try: uvicorn.run(app, host=host, port=port) except Exception as e: logger.error(f"Error running FastAPI server: {str(e)}\n{traceback.format_exc()}") sys.exit(1) def main() -> None: """ Main function to launch Gradio UI and optional FastAPI server. Raises: SystemExit: If the application is interrupted or encounters an error. """ try: # Apply nest_asyncio to allow nested event loops in Jupyter/Colab nest_asyncio.apply() # Parse command-line arguments args = parse_args() logger.info(f"Parsed arguments: {args}") # Find available port for Gradio gradio_port = find_available_port(args.gradio_port, PORT_RANGE, MAX_PORT_ATTEMPTS) if gradio_port is None: logger.error("Failed to find an available port for Gradio UI") sys.exit(1) # Launch FastAPI server in a separate thread if enabled if args.enable_fastapi: logger.info(f"Starting FastAPI server on port {args.fastapi_port}") fastapi_thread = threading.Thread( target=run_fastapi_server, args=("0.0.0.0", args.fastapi_port), daemon=True ) fastapi_thread.start() # Launch Gradio UI logger.info(f"Starting Gradio UI on port {gradio_port}") app = create_gradio_ui() app.launch(server_port=gradio_port, server_name="0.0.0.0") except KeyboardInterrupt: logger.info("Application terminated by user.") sys.exit(0) except OSError as e: logger.error(f"Port binding error: {str(e)}") sys.exit(1) except Exception as e: logger.error(f"Error running application: {str(e)}\n{traceback.format_exc()}") sys.exit(1) if __name__ == "__main__": main()