app.py

import os
import shutil
import sys
import warnings
import random
import time
import logging
import dotenv
import fal_client
import requests
import base64
from io import BytesIO
from typing import Dict, List, Tuple, Union, Optional

# dotenv.load_dotenv()

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# Download model weights only if they don't exist
if not os.path.exists("groundingdino_swint_ogc.pth"):
    os.system("wget https://huggingface.co/ShilongLiu/GroundingDINO/resolve/main/groundingdino_swint_ogc.pth")

if not os.path.exists("sam_hq_vit_l.pth"):
    os.system("wget https://huggingface.co/lkeab/hq-sam/resolve/main/sam_hq_vit_l.pth")

# Add paths
sys.path.append(os.path.join(os.getcwd(), "GroundingDINO"))
sys.path.append(os.path.join(os.getcwd(), "sam-hq"))
warnings.filterwarnings("ignore")

import numpy as np
import torch
import torchvision
import gradio as gr
import argparse
from PIL import Image, ImageDraw, ImageFont


# Grounding DINO
import GroundingDINO.groundingdino.datasets.transforms as T
from GroundingDINO.groundingdino.models import build_model
from GroundingDINO.groundingdino.util.slconfig import SLConfig
from GroundingDINO.groundingdino.util.utils import clean_state_dict, get_phrases_from_posmap

# segment anything
from segment_anything import build_sam_vit_l, SamPredictor 

# Constants
CONFIG_FILE = 'GroundingDINO/groundingdino/config/GroundingDINO_SwinT_OGC.py'
GROUNDINGDINO_CHECKPOINT = "groundingdino_swint_ogc.pth"
SAM_CHECKPOINT = 'sam_hq_vit_l.pth'
OUTPUT_DIR = "outputs"
FAL_KEY = os.getenv("FAL_KEY")
UPLOAD_DIR = "./tmp/images"

os.makedirs(UPLOAD_DIR, exist_ok=True)

# Global variables for model caching
_models = {
    'groundingdino': None,
    'sam_predictor': None
}

# Enable GPU if available with proper error handling
try:
    device = 'cuda' if torch.cuda.is_available() else 'cpu'
    logger.info(f"Using device: {device}")
except Exception as e:
    logger.warning(f"Error detecting GPU, falling back to CPU: {e}")
    device = 'cpu'


class ModelManager:
    """Manages model loading, unloading, and provides error handling"""
    
    @staticmethod
    def load_model(model_name: str) -> None:
        """Load a model if not already loaded"""
        try:
            if model_name == 'groundingdino' and _models['groundingdino'] is None:
                logger.info("Loading GroundingDINO model...")
                start_time = time.time()
                
                if not os.path.exists(GROUNDINGDINO_CHECKPOINT):
                    raise FileNotFoundError(f"GroundingDINO checkpoint not found at {GROUNDINGDINO_CHECKPOINT}")
                
                args = SLConfig.fromfile(CONFIG_FILE)
                args.device = device
                model = build_model(args)
                checkpoint = torch.load(GROUNDINGDINO_CHECKPOINT, map_location="cpu")
                load_res = model.load_state_dict(clean_state_dict(checkpoint["model"]), strict=False)
                logger.info(f"GroundingDINO load result: {load_res}")
                _ = model.eval()
                _models['groundingdino'] = model
                
                logger.info(f"GroundingDINO model loaded in {time.time() - start_time:.2f} seconds")
                
            elif model_name == 'sam' and _models['sam_predictor'] is None:
                logger.info("Loading SAM-HQ model...")
                start_time = time.time()
                
                if not os.path.exists(SAM_CHECKPOINT):
                    raise FileNotFoundError(f"SAM checkpoint not found at {SAM_CHECKPOINT}")
                
                sam = build_sam_vit_l(checkpoint=SAM_CHECKPOINT)
                sam.to(device=device)
                _models['sam_predictor'] = SamPredictor(sam)
                
                logger.info(f"SAM-HQ model loaded in {time.time() - start_time:.2f} seconds")

                
        except Exception as e:
            logger.error(f"Error loading {model_name} model: {e}")
            raise RuntimeError(f"Failed to load {model_name} model: {e}")
    
    @staticmethod
    def get_model(model_name: str):
        """Get a model, loading it if necessary"""
        if model_name not in _models or _models[model_name] is None:
            ModelManager.load_model(model_name)
        return _models[model_name]
    
    @staticmethod
    def unload_model(model_name: str) -> None:
        """Unload a model to free memory"""
        if model_name in _models and _models[model_name] is not None:
            logger.info(f"Unloading {model_name} model")
            _models[model_name] = None
            if device == 'cuda':
                torch.cuda.empty_cache()


def transform_image(image_pil: Image.Image) -> torch.Tensor:
    """Transform PIL image for GroundingDINO"""
    transform = T.Compose([
        T.RandomResize([800], max_size=1333),
        T.ToTensor(),
        T.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]),
    ])
    image, _ = transform(image_pil, None)  # 3, h, w
    return image


def get_grounding_output(
    image: torch.Tensor, 
    caption: str, 
    box_threshold: float, 
    text_threshold: float, 
    with_logits: bool = True
) -> Tuple[torch.Tensor, torch.Tensor, List[str]]:
    """Run GroundingDINO to get bounding boxes from text prompt"""
    try:
        model = ModelManager.get_model('groundingdino')
        
        # Format caption
        caption = caption.lower().strip()
        if not caption.endswith("."):
            caption = caption + "."

        with torch.no_grad():
            outputs = model(image[None], captions=[caption])
            
        logits = outputs["pred_logits"].cpu().sigmoid()[0]  # (nq, 256)
        boxes = outputs["pred_boxes"].cpu()[0]  # (nq, 4)

        # Filter output
        logits_filt = logits.clone()
        boxes_filt = boxes.clone()
        filt_mask = logits_filt.max(dim=1)[0] > box_threshold
        logits_filt = logits_filt[filt_mask]  # num_filt, 256
        boxes_filt = boxes_filt[filt_mask]  # num_filt, 4

        # Get phrases
        tokenizer = model.tokenizer
        tokenized = tokenizer(caption)
        pred_phrases = []
        scores = []
        
        for logit, box in zip(logits_filt, boxes_filt):
            pred_phrase = get_phrases_from_posmap(
                logit > text_threshold, tokenized, tokenizer)
            if with_logits:
                pred_phrases.append(pred_phrase + f"({str(logit.max().item())[:4]})")
            else:
                pred_phrases.append(pred_phrase)
            scores.append(logit.max().item())

        return boxes_filt, torch.Tensor(scores), pred_phrases
    
    except Exception as e:
        logger.error(f"Error in grounding output: {e}")
        # Return empty results instead of crashing
        return torch.Tensor([]), torch.Tensor([]), []


def draw_mask(mask: np.ndarray, draw: ImageDraw.Draw) -> None:
    """Draw mask on image"""

    color = (255, 255, 255, 255)

    nonzero_coords = np.transpose(np.nonzero(mask))
    for coord in nonzero_coords:
        draw.point(coord[::-1], fill=color)


def draw_box(box: torch.Tensor, draw: ImageDraw.Draw, label: Optional[str]) -> None:
    """Draw bounding box on image"""
    color = tuple(np.random.randint(0, 255, size=3).tolist())
    draw.rectangle(((box[0], box[1]), (box[2], box[3])), outline=color, width=2)

    if label:
        font = ImageFont.load_default()
        if hasattr(font, "getbbox"):
            bbox = draw.textbbox((box[0], box[1]), str(label), font)
        else:
            w, h = draw.textsize(str(label), font)
            bbox = (box[0], box[1], w + box[0], box[1] + h)
        draw.rectangle(bbox, fill=color)
        draw.text((box[0], box[1]), str(label), fill="white")


def run_grounded_sam(input_image):
    """Main function to run GroundingDINO and SAM-HQ"""
    try:
        # Create output directory
        os.makedirs(OUTPUT_DIR, exist_ok=True)
        text_prompt = 'car'
        task_type = 'text'
        box_threshold = 0.3
        text_threshold = 0.25
        iou_threshold = 0.8
        hq_token_only = True
        
        # Process input image
        if isinstance(input_image, dict):
            # Input from gradio sketch component
            scribble = np.array(input_image["mask"])
            image_pil = input_image["image"].convert("RGB")
        else:
            # Direct image input
            image_pil = input_image.convert("RGB") if input_image else None
            scribble = None
            
        if image_pil is None:
            logger.error("No input image provided")
            return [Image.new('RGB', (400, 300), color='gray')]

        # Transform image for GroundingDINO
        transformed_image = transform_image(image_pil)
        
        # Load models as needed
        ModelManager.load_model('groundingdino')
        size = image_pil.size
        H, W = size[1], size[0]

        # Run GroundingDINO with provided text
        boxes_filt, scores, pred_phrases = get_grounding_output(
            transformed_image, text_prompt, box_threshold, text_threshold
        )

        if boxes_filt is not None:
            # Scale boxes to image dimensions
            for i in range(boxes_filt.size(0)):
                boxes_filt[i] = boxes_filt[i] * torch.Tensor([W, H, W, H])
                boxes_filt[i][:2] -= boxes_filt[i][2:] / 2
                boxes_filt[i][2:] += boxes_filt[i][:2]
            
            # Apply non-maximum suppression if we have multiple boxes
            if boxes_filt.size(0) > 1:
                logger.info(f"Before NMS: {boxes_filt.shape[0]} boxes")
                nms_idx = torchvision.ops.nms(boxes_filt, scores, iou_threshold).numpy().tolist()
                boxes_filt = boxes_filt[nms_idx]
                pred_phrases = [pred_phrases[idx] for idx in nms_idx]
                logger.info(f"After NMS: {boxes_filt.shape[0]} boxes")
        
        # Load SAM model
        ModelManager.load_model('sam')
        sam_predictor = ModelManager.get_model('sam_predictor')
        
        # Set image for SAM
        image = np.array(image_pil)
        sam_predictor.set_image(image)
        
        # Run SAM 
        # Use boxes for these task types
        if boxes_filt.size(0) == 0:
            logger.warning("No boxes detected")
            return [image_pil, Image.new('RGBA', size, color=(0, 0, 0, 0))]
        
        transformed_boxes = sam_predictor.transform.apply_boxes_torch(boxes_filt, image.shape[:2]).to(device)
        
        masks, _, _ = sam_predictor.predict_torch(
            point_coords=None,
            point_labels=None,
            boxes=transformed_boxes,
            multimask_output=False,
            hq_token_only=hq_token_only,
        )
        
        # Create mask image
        mask_image = Image.new('RGBA', size, color=(0, 0, 0, 0))
        mask_draw = ImageDraw.Draw(mask_image)
        
        # Draw masks
        for mask in masks:
            draw_mask(mask[0].cpu().numpy(), mask_draw)
        
        # Draw boxes and points on original image
        image_draw = ImageDraw.Draw(image_pil)

        for box, label in zip(boxes_filt, pred_phrases):
            draw_box(box, image_draw, label)

        return mask_image
    
    except Exception as e:
        logger.error(f"Error in run_grounded_sam: {e}")
        # Return original image on error
        if isinstance(input_image, dict) and "image" in input_image:
            return [input_image["image"], Image.new('RGBA', input_image["image"].size, color=(0, 0, 0, 0))]
        elif isinstance(input_image, Image.Image):
            return [input_image, Image.new('RGBA', input_image.size, color=(0, 0, 0, 0))]
        else:
            return [Image.new('RGB', (400, 300), color='gray'), Image.new('RGBA', (400, 300), color=(0, 0, 0, 0))]

def image_gaussian_blur(image: torch.Tensor, radius: float) -> torch.Tensor:
    if image.ndim == 4:  # Remove batch dimension if present
        image = image.squeeze(0)
    pil_image = tensor2pil(image)
    blurred_pil_image = pil_image.filter(ImageFilter.GaussianBlur(radius))
    return pil2tensor(blurred_pil_image).squeeze(0)

def load_image(image_path: str) -> torch.Tensor:
    image = Image.open(image_path).convert("RGBA")
    image_tensor = torch.from_numpy(np.array(image)).permute(2, 0, 1).float() / 255.0
    return image_tensor

def split_image_with_alpha(image: torch.Tensor):
    out_images = image[:3, :, :]
    out_alphas = image[3, :, :] if image.shape[0] > 3 else torch.ones_like(image[0, :, :])
    result = (out_images.unsqueeze(0), 1.0 - out_alphas.unsqueeze(0))
    return result

def pil2numpy(image: Image.Image):
    return np.array(image).astype(np.float32) / 255.0

def numpy2pil(image: np.ndarray, mode=None):
    return Image.fromarray(np.clip(255.0 * image, 0, 255).astype(np.uint8), mode)

def pil2tensor(image: Image.Image):
    return torch.from_numpy(pil2numpy(image)).unsqueeze(0)

def invert(image):
    s = 1.0 - image
    return s

def tensor2pil(image: torch.Tensor, mode=None):
    if image.ndim == 2:  # Grayscale image
        image = image.unsqueeze(0)  # Add channel dimension

    if image.ndim != 3 or image.shape[1:] == (0, 0):
        raise ValueError(f"Invalid tensor dimensions: {image.shape}")

    if image.shape[0] == 1:  # Single channel, replicate to 3 channels
        image = image.repeat(3, 1, 1)
    elif image.shape[0] != 3:
        raise ValueError("Unexpected number of channels in the image tensor")
    
    return numpy2pil(image.cpu().numpy().transpose(1, 2, 0), mode=mode)

def extract_high_frequency(image: torch.Tensor, blur_radius: float = 5.0) -> torch.Tensor:
    """Extract high-frequency details by subtracting the blurred image from the original."""
    if image.ndim == 4:
        image = image.squeeze(0)
    
    blurred = image_gaussian_blur(image, blur_radius)
    
    if blurred.ndim == 4:
        blurred = blurred.squeeze(0)
    elif blurred.ndim == 3 and blurred.shape[0] != 3:
        blurred = blurred.permute(2, 0, 1)
    
    high_freq = image - blurred
    return high_freq

def image_blend_mask(image_a, image_b, mask, blend_percentage):

    # Convert images to PIL
    img_a = tensor2pil(image_a)
    img_b = tensor2pil(image_b)
    mask = ImageOps.invert(tensor2pil(mask).convert('L'))

    # Mask image
    masked_img = Image.composite(img_a, img_b, mask.resize(img_a.size))

    # Blend image
    blend_mask = Image.new(mode="L", size=img_a.size,
                           color=(round(blend_percentage * 255)))
    blend_mask = ImageOps.invert(blend_mask)
    img_result = Image.composite(img_a, masked_img, blend_mask)

    del img_a, img_b, blend_mask, mask

    return (pil2tensor(img_result), )

def encode_image(image):
    buffer = BytesIO()
    image.save(buffer, format="PNG")
    encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
    return f"data:image/png;base64,{encoded_image}"

def generate_ai_bg(input_img, prompt):
    hf_input_img = encode_image(input_img)

    handler = fal_client.submit(
        "fal-ai/iclight-v2",
        arguments={
            "prompt": prompt,
            "image_url": hf_input_img
        },
        webhook_url="https://optional.webhook.url/for/results",
    )

    request_id = handler.request_id

    status = fal_client.status("fal-ai/iclight-v2", request_id, with_logs=True)

    result = fal_client.result("fal-ai/iclight-v2", request_id)

    ic_light_img = result['images'][0]['url']

    return ic_light_img

def blend_details(input_image, relit_image, masked_image):
    with torch.inference_mode():
        # Load and resize images
        # input_image = load_image(input_image_path)
        # relit_image = load_image(relit_image_path)
        # masked_image = load_image(masked_image_path)

        # Resize input image
        input_image = torch.nn.functional.interpolate(
            input_image.unsqueeze(0),
            size=(1024, 1024),
            mode="bicubic",
            align_corners=False
        ).squeeze(0)

        # Resize relit image
        relit_image = torch.nn.functional.interpolate(
            relit_image.unsqueeze(0),
            size=(1024, 1024),
            mode="bicubic",
            align_corners=False
        ).squeeze(0)

        # Resize masked image
        masked_image = torch.nn.functional.interpolate(
            masked_image.unsqueeze(0),
            size=(1024, 1024),
            mode="bicubic",
            align_corners=False
        ).squeeze(0)

        # Split images and get RGB channels
        input_image_rgb = split_image_with_alpha(input_image)[0].squeeze(0)
        relit_image_rgb = split_image_with_alpha(relit_image)[0].squeeze(0)
        
        # Use masked image RGB channels as segmentation mask (average of RGB channels)
        segmentation_mask = masked_image[:3].mean(dim=0)  # Average RGB channels to get grayscale mask
        
        print(f"segmentation_mask shape: {segmentation_mask.shape}")
        
        # Extract high-frequency details from input image
        high_freq_details = extract_high_frequency(input_image_rgb, blur_radius=3.0)

        # Print shapes for debugging
        print(f"high_freq_details shape: {high_freq_details.shape}")
        print(f"segmentation_mask shape: {segmentation_mask.shape}")
        print(f"relit_image_rgb shape: {relit_image_rgb.shape}")

        # Apply high-frequency details only in masked areas
        detail_strength = 0.5
        segmentation_mask = segmentation_mask.unsqueeze(0).repeat(3, 1, 1)  # Expand mask to match RGB channels
        masked_details = high_freq_details * segmentation_mask
        # final_image = relit_image_rgb + (masked_details * detail_strength)
        # final_image = image_blend_mask(relit_image_rgb, masked_details, mask, blend_percentage)
        final_image = relit_image_rgb + masked_details
        print('final_image shape:', final_image.shape)

        # Normalize to [0, 1] range
        final_image = torch.clamp(final_image, 0, 1)

        # Save intermediate results for debugging
        tensor2pil(segmentation_mask).save("output/segmentation_mask.png")
        tensor2pil(high_freq_details).save("output/high_freq_details.png")
        tensor2pil(masked_details).save("output/masked_details.png")
        
        # Save final result
        final_image_pil = tensor2pil(final_image)
        # final_image_pil.save("output/output_image.png")
        return [final_image_pil]

def generate_image(input_img, ai_gen_image, prompt):

    # ai_gen_image = generate_ai_bg(input_img, prompt)

    mask_input_image = run_grounded_sam(input_img)

    final_image = blend_details(input_img, ai_gen_image, mask_input_image)

    return [final_image]

def create_ui():
    """Create Gradio UI for CarViz demo"""
    with gr.Blocks(title="CarViz Demo") as block:
        gr.Markdown("""
        # CarViz
        """)

        with gr.Row():
            with gr.Column():
                input_image = gr.Image(type="pil", label="image")
                ai_image = gr.Image(type="pil", label="image")
                prompt = gr.Textbox(label="Prompt", placeholder="Enter your prompt here...")
                run_button = gr.Button(value='Run')

            with gr.Column():
                gallery = gr.Gallery(
                    label="Generated images", show_label=False, elem_id="gallery"
                )
        
        # Run button
        run_button.click(
            fn=generate_image, 
            inputs=[
                input_image,
                ai_image,
                prompt
            ], 
            outputs=gallery
        )     
 
    return block


if __name__ == "__main__":
    parser = argparse.ArgumentParser("Carviz demo", add_help=True)
    parser.add_argument("--debug", action="store_true", help="using debug mode")
    parser.add_argument("--share", action="store_true", help="share the app")
    parser.add_argument('--no-gradio-queue', action="store_true", help="disable gradio queue")
    parser.add_argument('--port', type=int, default=7860, help="port to run the app")
    parser.add_argument('--host', type=str, default="0.0.0.0", help="host to run the app")
    args = parser.parse_args()

    logger.info(f"Starting CarViz demo with args: {args}")
    
    # Check for model files
    if not os.path.exists(GROUNDINGDINO_CHECKPOINT):
        logger.warning(f"GroundingDINO checkpoint not found at {GROUNDINGDINO_CHECKPOINT}")
    if not os.path.exists(SAM_CHECKPOINT):
        logger.warning(f"SAM-HQ checkpoint not found at {SAM_CHECKPOINT}")
    
    # Create app
    block = create_ui()
    if not args.no_gradio_queue:
        block = block.queue()
    
    # Launch app
    try:
        block.launch(
            debug=args.debug, 
            share=args.share, 
            show_error=True,
            server_name=args.host,
            server_port=args.port
        )
    except Exception as e:
        logger.error(f"Error launching app: {e}")