Delete EurybiaMini3.0Gradio.app

EurybiaMini3.0Gradio.app

@@ -1,536 +0,0 @@
-import os
-import glob
-import time
-import threading
-import requests
-import wikipedia
-import torch
-import cv2
-import numpy as np
-from io import BytesIO
-from PIL import Image
-import base64  # Added import
-
-import gradio as gr
-from ultralytics import YOLO
-from transformers import pipeline, AutoTokenizer, AutoModelForCausalLM
-from diffusers import MarigoldDepthPipeline  # Updated import for depth model
-from realesrgan import RealESRGANer
-from basicsr.archs.rrdbnet_arch import RRDBNet
-
-# Set environment variable for PyTorch MPS fallback before importing torch
-os.environ['PYTORCH_ENABLE_MPS_FALLBACK'] = '1'
-
-# Initialize Models
-def initialize_models():
-    models = {}
-    
-    # Device detection
-    if torch.cuda.is_available():
-        device = 'cuda'
-    elif torch.backends.mps.is_available():
-        device = 'mps'
-    else:
-        device = 'cpu'
-    models['device'] = device
-
-    print(f"Using device: {device}")
-
-    # Initialize the RoBERTa model for question answering
-    try:
-        models['qa_pipeline'] = pipeline(
-            "question-answering", model="deepset/roberta-base-squad2", device=0 if device == 'cuda' else -1)
-        print("RoBERTa QA pipeline initialized.")
-    except Exception as e:
-        print(f"Error initializing the RoBERTa model: {e}")
-        models['qa_pipeline'] = None
-
-    # Initialize the Gemma model
-    try:
-        models['gemma_tokenizer'] = AutoTokenizer.from_pretrained("google/gemma-2-2b-it")
-        models['gemma_model'] = AutoModelForCausalLM.from_pretrained(
-            "google/gemma-2-2b-it",
-            device_map="auto",
-            torch_dtype=torch.bfloat16 if device == 'cuda' else torch.float32
-        )
-        print("Gemma model initialized.")
-    except Exception as e:
-        print(f"Error initializing the Gemma model: {e}")
-        models['gemma_model'] = None
-
-    # Initialize the depth estimation model using MarigoldDepthPipeline exactly as per your sample
-    try:
-        if device == 'cuda':
-            models['depth_pipe'] = MarigoldDepthPipeline.from_pretrained(
-                "prs-eth/marigold-depth-lcm-v1-0",
-                variant="fp16",
-                torch_dtype=torch.float16
-            ).to('cuda')
-        else:
-            # For CPU or MPS devices, keep on 'cpu' to avoid unsupported operators
-            models['depth_pipe'] = MarigoldDepthPipeline.from_pretrained(
-                "prs-eth/marigold-depth-lcm-v1-0",
-                torch_dtype=torch.float32
-            ).to('cpu')
-        print("Depth estimation model initialized.")
-    except Exception as e:
-        error_message = f"Error initializing the depth estimation model: {e}"
-        print(error_message)
-        models['depth_pipe'] = None
-        models['depth_init_error'] = error_message  # Store the error message
-
-    # Initialize the upscaling model
-    try:
-        upscaler_model_path = 'weights/RealESRGAN_x4plus.pth'  # Ensure this path is correct
-        if not os.path.exists(upscaler_model_path):
-            print(f"Upscaling model weights not found at {upscaler_model_path}. Please download them.")
-            models['upscaler'] = None
-        else:
-            # Define the model architecture
-            model = RRDBNet(num_in_ch=3, num_out_ch=3, num_feat=64,
-                            num_block=23, num_grow_ch=32, scale=4)
-
-            # Initialize RealESRGANer
-            models['upscaler'] = RealESRGANer(
-                scale=4,
-                model_path=upscaler_model_path,
-                model=model,
-                pre_pad=0,
-                half=(device == 'cuda'),
-                device=device
-            )
-            print("Real-ESRGAN upscaler initialized.")
-    except Exception as e:
-        print(f"Error initializing the upscaling model: {e}")
-        models['upscaler'] = None
-
-    # Initialize YOLO model
-    try:
-        source_weights_path = "/Users/David/Downloads/WheelOfFortuneLab-DavidDriscoll/Eurybia1.3/mbari_315k_yolov8.pt"
-        if not os.path.exists(source_weights_path):
-            print(f"YOLO weights not found at {source_weights_path}. Please download them.")
-            models['yolo_model'] = None
-        else:
-            models['yolo_model'] = YOLO(source_weights_path)
-            print("YOLO model initialized.")
-    except Exception as e:
-        print(f"Error initializing YOLO model: {e}")
-        models['yolo_model'] = None
-
-    return models
-
-models = initialize_models()
-
-# Utility Functions
-def search_class_description(class_name):
-    wikipedia.set_lang("en")
-    wikipedia.set_rate_limiting(True)
-    description = ""
-
-    try:
-        page = wikipedia.page(class_name)
-        if page:
-            description = page.content[:5000]  # Get more content
-    except Exception as e:
-        print(f"Error fetching description for {class_name}: {e}")
-
-    return description
-
-def search_class_image(class_name):
-    wikipedia.set_lang("en")
-    wikipedia.set_rate_limiting(True)
-    img_url = ""
-
-    try:
-        page = wikipedia.page(class_name)
-        if page:
-            for img in page.images:
-                if img.lower().endswith(('.jpg', '.jpeg', '.png', '.gif')):
-                    img_url = img
-                    break
-    except Exception as e:
-        print(f"Error fetching image for {class_name}: {e}")
-
-    return img_url
-
-def process_image(image):
-    if models['yolo_model'] is None:
-        return None, "YOLO model is not initialized.", "YOLO model is not initialized.", [], None
-    
-    try:
-        if image is None:
-            return None, "No image uploaded.", "No image uploaded.", [], None
-
-        # Convert Gradio Image to OpenCV format
-        image_np = np.array(image)
-        if image_np.dtype != np.uint8:
-            image_np = image_np.astype(np.uint8)
-
-        if len(image_np.shape) != 3 or image_np.shape[2] != 3:
-            return None, "Invalid image format. Please upload a RGB image.", "Invalid image format. Please upload a RGB image.", [], None
-
-        image_cv = cv2.cvtColor(image_np, cv2.COLOR_RGB2BGR)
-
-        # Store the original image before drawing bounding boxes
-        original_image_cv = image_cv.copy()
-        original_image_pil = Image.fromarray(cv2.cvtColor(original_image_cv, cv2.COLOR_BGR2RGB))
-
-        # Perform YOLO prediction
-        results = models['yolo_model'].predict(
-            source=image_cv, conf=0.075)[0]  # Lowered the threshold
-
-        bounding_boxes = []
-        image_processed = image_cv.copy()
-
-        if results.boxes is not None:
-            for box in results.boxes:
-                x1, y1, x2, y2 = map(int, box.xyxy[0].tolist())
-                class_name = models['yolo_model'].names[int(box.cls)]
-                confidence = box.conf.item() * 100  # Convert to percentage
-
-                bounding_boxes.append({
-                    "coords": (x1, y1, x2, y2),
-                    "class_name": class_name,
-                    "confidence": confidence
-                })
-
-                cv2.rectangle(image_processed, (x1, y1), (x2, y2), (0, 0, 255), 2)
-                cv2.putText(image_processed, f'{class_name} {confidence:.2f}%',
-                            (x1, y1 - 10), cv2.FONT_HERSHEY_SIMPLEX,
-                            0.9, (0, 0, 255), 2)
-
-        # Convert back to PIL Image
-        processed_image = Image.fromarray(cv2.cvtColor(image_processed, cv2.COLOR_BGR2RGB))
-
-        # Prepare detection info
-        if bounding_boxes:
-            detection_info = "\n".join(
-                [f'{box["class_name"]}: {box["confidence"]:.2f}%' for box in bounding_boxes]
-            )
-        else:
-            detection_info = "No detections found."
-
-        # Prepare detection details as Markdown
-        if bounding_boxes:
-            details = ""
-            for idx, box in enumerate(bounding_boxes):
-                class_name = box['class_name']
-                confidence = box['confidence']
-                description = search_class_description(class_name)
-                img_url = search_class_image(class_name)
-                img_md = ""
-                if img_url:
-                    try:
-                        headers = {
-                            'User-Agent': 'MyApp/1.0 (https://example.com/contact; [email protected])'
-                        }
-                        response = requests.get(img_url, headers=headers, timeout=10)
-                        img_data = response.content
-                        img = Image.open(BytesIO(img_data)).convert("RGB")
-                        img.thumbnail((400, 400))  # Resize for faster loading
-                        buffered = BytesIO()
-                        img.save(buffered, format="PNG")
-                        img_str = base64.b64encode(buffered.getvalue()).decode()
-                        img_md = f"![{class_name}](data:image/png;base64,{img_str})\n\n"
-                    except Exception as e:
-                        print(f"Error fetching image for {class_name}: {e}")
-                details += f"### {idx+1}. {class_name} ({confidence:.2f}%)\n\n"
-                if description:
-                    details += f"{description}\n\n"
-                if img_md:
-                    details += f"{img_md}\n\n"
-            detection_details_md = details
-        else:
-            detection_details_md = "No detections to show."
-
-        return processed_image, detection_info, detection_details_md, bounding_boxes, original_image_pil
-    except Exception as e:
-        print(f"Error processing image: {e}")
-        return None, f"Error processing image: {e}", f"Error processing image: {e}", [], None
-
-def ask_eurybia(question, state):
-    if not question.strip():
-        return "Please enter a valid question.", state
-
-    if not state['bounding_boxes']:
-        return "No detected objects to ask about.", state
-
-    # Combine descriptions of all detected objects as context
-    context = ""
-    for box in state['bounding_boxes']:
-        description = search_class_description(box['class_name'])
-        if description:
-            context += description + "\n"
-
-    if not context.strip():
-        return "No sufficient context available to answer the question.", state
-
-    try:
-        if models['qa_pipeline'] is None:
-            return "QA pipeline is not initialized.", state
-
-        answer = models['qa_pipeline'](question=question, context=context)
-        answer_text = answer['answer'].strip()
-        if not answer_text:
-            return "I couldn't find an answer to that question based on the detected objects.", state
-        return answer_text, state
-    except Exception as e:
-        print(f"Error during question answering: {e}")
-        return f"Error during question answering: {e}", state
-
-def enhance_image(cropped_image_pil):
-    if models['upscaler'] is None:
-        return None, "Upscaling model is not initialized."
-    
-    try:
-        input_image = cropped_image_pil.convert("RGB")
-        img = np.array(input_image)
-
-        # Run the model to enhance the image
-        output, _ = models['upscaler'].enhance(img, outscale=4)
-
-        enhanced_image = Image.fromarray(output)
-
-        return enhanced_image, "Image enhanced successfully."
-    except Exception as e:
-        print(f"Error during image enhancement: {e}")
-        return None, f"Error during image enhancement: {e}"
-
-def run_depth_prediction(original_image):
-    if models['depth_pipe'] is None:
-        error_msg = models.get('depth_init_error', "Depth estimation model is not initialized.")
-        return None, error_msg
-
-    try:
-        if original_image is None:
-            return None, "No image uploaded for depth prediction."
-
-        # Prepare the image
-        input_image = original_image.convert("RGB")
-
-        # Run the depth pipeline
-        result = models['depth_pipe'](input_image)
-
-        # Access the depth prediction
-        depth_prediction = result.prediction  # Adjust based on sample code
-
-        # Visualize the depth map
-        vis_depth = models['depth_pipe'].image_processor.visualize_depth(depth_prediction)
-        
-        # Ensure vis_depth is a list and extract the first image
-        if isinstance(vis_depth, list) and len(vis_depth) > 0:
-            vis_depth_image = vis_depth[0]
-        else:
-            vis_depth_image = vis_depth  # Fallback if not a list
-
-        return vis_depth_image, "Depth prediction completed."
-    except Exception as e:
-        print(f"Error during depth prediction: {e}")
-        return None, f"Error during depth prediction: {e}"
-
-# Gradio Interface Components
-with gr.Blocks() as demo:
-    gr.Markdown("# Eurybia Mini - Object Detection and Analysis Tool")
-
-    with gr.Tab("Upload & Process"):
-        with gr.Row():
-            with gr.Column():
-                image_input = gr.Image(type="pil", label="Upload Image")
-                process_button = gr.Button("Process Image")
-                clear_button = gr.Button("Clear")
-            with gr.Column():
-                processed_image = gr.Image(type="pil", label="Processed Image")
-                detection_info = gr.Textbox(label="Detection Information", lines=10)
-
-    with gr.Tab("Detection Details"):
-        with gr.Accordion("Click to see detection details", open=False):
-            detection_details_md = gr.Markdown("No detections to show.")
-
-    with gr.Tab("Ask Eurybia"):
-        with gr.Row():
-            with gr.Column():
-                question_input = gr.Textbox(label="Ask a question about the detected objects")
-                ask_button = gr.Button("Ask Eurybia")
-            with gr.Column():
-                answer_output = gr.Markdown(label="Eurybia's Answer")
-
-    with gr.Tab("Depth Estimation"):
-        with gr.Row():
-            with gr.Column():
-                depth_button = gr.Button("Run Depth Prediction")
-            with gr.Column():
-                depth_output = gr.Image(type="pil", label="Depth Map")
-                depth_status = gr.Textbox(label="Status", lines=2)
-        
-        # Display error message if depth estimation model failed to initialize
-        if models.get('depth_init_error'):
-            gr.Markdown(f"**Depth Estimation Initialization Error:** {models['depth_init_error']}")
-
-    with gr.Tab("Enhance Detected Objects"):
-        if models['yolo_model'] is not None and models['upscaler'] is not None:
-            with gr.Row():
-                detected_objects = gr.Dropdown(choices=[], label="Select Detected Object", interactive=True)
-                enhance_btn = gr.Button("Enhance Image")
-            with gr.Column():
-                enhanced_image = gr.Image(type="pil", label="Enhanced Image")
-                enhance_status = gr.Textbox(label="Status", lines=2)
-        else:
-            gr.Markdown("**Warning:** YOLO model or Upscaling model is not initialized. Image enhancement functionality will be unavailable.")
-
-    with gr.Tab("Credits"):
-        gr.Markdown("""
-# Credits and Licensing Information
-
-This project utilizes various open-source libraries, tools, pretrained models, and datasets. Below is the list of components used and their respective credits/licenses:
-
-## Libraries
-- **Python** - Python Software Foundation License (PSF License)
-- **Gradio** - Licensed under the Apache License 2.0
-- **Torch (PyTorch)** - Licensed under the BSD 3-Clause License
-- **OpenCV (cv2)** - Licensed under the Apache License 2.0
-- **NumPy** - Licensed under the BSD License
-- **Pillow (PIL)** - Licensed under the HPND License
-- **Requests** - Licensed under the Apache License 2.0
-- **Wikipedia API** - Licensed under the MIT License
-- **Transformers** - Licensed under the Apache License 2.0
-- **Diffusers** - Licensed under the Apache License 2.0
-- **Real-ESRGAN** - Licensed under the MIT License
-- **BasicSR** - Licensed under the Apache License 2.0
-- **Ultralytics YOLO** - Licensed under the GPL-3.0 License
-
-## Pretrained Models
-- **deepset/roberta-base-squad2 (RoBERTa)** - Model provided by Hugging Face under the Apache License 2.0.
-- **google/gemma-2-2b-it** - Model provided by Hugging Face under the Apache License 2.0.
-- **prs-eth/marigold-depth-lcm-v1-0** - Licensed under the Apache License 2.0.
-- **Real-ESRGAN model weights (RealESRGAN_x4plus.pth)** - Distributed under the MIT License.
-- **FathomNet MBARI 315K YOLOv8 Model**:
-  - **Dataset**: Sourced from [FathomNet](https://fathomnet.org).
-  - **Model**: Derived from MBARI’s curated dataset of 315,000 marine annotations.
-  - **License**: Dataset and models adhere to MBARI’s Creative Commons Attribution-NonCommercial 4.0 International License (CC BY-NC 4.0).
-
-## Datasets
-- **FathomNet MBARI Dataset**:
-  - A large-scale dataset for marine biodiversity image annotations.
-  - All content adheres to the [CC BY-NC 4.0 License](https://creativecommons.org/licenses/by-nc/4.0/).
-
-## Acknowledgments
-- **Ultralytics YOLO**: For the YOLOv8 architecture used for object detection.
-- **FathomNet and MBARI**: For providing the marine dataset and annotations that support object detection in underwater imagery.
-- **Gradio**: For providing an intuitive interface for machine learning applications.
-- **Hugging Face**: For pretrained models and pipelines (e.g., Transformers, Diffusers).
-- **Real-ESRGAN**: For image enhancement and upscaling models.
-- **Wikipedia API**: For fetching object descriptions and images.
-""")
-
-    # Hidden state to store bounding boxes, original and processed images
-    state = gr.State({"bounding_boxes": [], "last_image": None, "original_image": None})
-
-    # Event Handlers
-    def on_process_image(image, state):
-        processed_img, info, details, bounding_boxes, original_image_pil = process_image(image)
-        if processed_img is not None:
-            # Update the state with new bounding boxes and images
-            state['bounding_boxes'] = bounding_boxes
-            state['last_image'] = processed_img
-            state['original_image'] = original_image_pil
-            # Update the dropdown choices for detected objects
-            choices = [f"{idx+1}. {box['class_name']} ({box['confidence']:.2f}%)" for idx, box in enumerate(bounding_boxes)]
-        else:
-            choices = []
-        return processed_img, info, details, gr.update(choices=choices), state
-
-    process_button.click(
-        on_process_image,
-        inputs=[image_input, state],
-        outputs=[processed_image, detection_info, detection_details_md, detected_objects, state]
-    )
-
-    def on_clear(state):
-        state = {"bounding_boxes": [], "last_image": None, "original_image": None}
-        return None, "No detections found.", "No detections to show.", gr.update(choices=[]), state
-
-    clear_button.click(
-        on_clear,
-        inputs=state,
-        outputs=[processed_image, detection_info, detection_details_md, detected_objects, state]
-    )
-
-    def on_ask_eurybia(question, state):
-        answer, state = ask_eurybia(question, state)
-        return answer, state
-
-    ask_button.click(
-        on_ask_eurybia,
-        inputs=[question_input, state],
-        outputs=[answer_output, state]
-    )
-
-    def on_depth_prediction(state):
-        original_image = state.get('original_image')
-        depth_img, status = run_depth_prediction(original_image)
-        return depth_img, status
-
-    depth_button.click(
-        on_depth_prediction,
-        inputs=state,
-        outputs=[depth_output, depth_status]
-    )
-
-    def on_enhance_image(selected_object, state):
-        if not selected_object:
-            return None, "No object selected.", state
-
-        try:
-            idx = int(selected_object.split('.')[0]) - 1
-            box = state['bounding_boxes'][idx]
-            class_name = box['class_name']
-            x1, y1, x2, y2 = box['coords']
-
-            if not state.get('last_image'):
-                return None, "Processed image is not available.", state
-
-            # Ensure processed_image is stored in state
-            processed_img_pil = state['last_image']
-            if not isinstance(processed_img_pil, Image.Image):
-                return None, "Processed image is in an unsupported format.", state
-
-            # Convert processed_image to OpenCV format with checks
-            processed_img_cv = np.array(processed_img_pil)
-            if processed_img_cv.dtype != np.uint8:
-                processed_img_cv = processed_img_cv.astype(np.uint8)
-
-            if len(processed_img_cv.shape) != 3 or processed_img_cv.shape[2] != 3:
-                return None, "Invalid processed image format.", state
-
-            processed_img_cv = cv2.cvtColor(processed_img_cv, cv2.COLOR_RGB2BGR)
-
-            # Crop the detected object from the processed image
-            cropped_img_cv = processed_img_cv[y1:y2, x1:x2]
-            if cropped_img_cv.size == 0:
-                return None, "Cropped image is empty.", state
-
-            cropped_img_pil = Image.fromarray(cv2.cvtColor(cropped_img_cv, cv2.COLOR_BGR2RGB))
-
-            # Enhance the cropped image
-            enhanced_img, status = enhance_image(cropped_img_pil)
-            return enhanced_img, status, state
-        except Exception as e:
-            return None, f"Error: {e}", state
-
-    if models['yolo_model'] is not None and models['upscaler'] is not None:
-        enhance_btn.click(
-            on_enhance_image,
-            inputs=[detected_objects, state],
-            outputs=[enhanced_image, enhance_status, state]
-        )
-
-    # Optional: Add a note if the depth model isn't initialized
-    if models['depth_pipe'] is None and not models.get('depth_init_error'):
-        gr.Markdown("**Warning:** Depth estimation model is not initialized. Depth prediction functionality will be unavailable.")
-
-    # Optional: Add a note if the upscaler isn't initialized
-    if models['upscaler'] is None:
-        gr.Markdown("**Warning:** Upscaling model is not initialized. Image enhancement functionality will be unavailable.")
-
-# Launch the Gradio app
-demo.launch()