app.py

import os
import random
import uuid
import json
import time
import asyncio
from threading import Thread
from pathlib import Path
from io import BytesIO
from typing import Optional, Tuple, Dict, Any, Iterable

import gradio as gr
import spaces
import torch
import numpy as np
from PIL import Image
import cv2
import requests
import fitz  # PyMuPDF

from transformers import (
    Qwen3VLMoeForConditionalGeneration,
    AutoProcessor,
    TextIteratorStreamer,
)
from transformers.image_utils import load_image

# --- Theme Definition ---
from gradio.themes import Soft
from gradio.themes.utils import colors, fonts, sizes

# Define a new "Thistle" color palette
colors.thistle = colors.Color(
    name="thistle",
    c50="#F9F5F9",
    c100="#F3ECF4",
    c200="#E8D9EA",
    c300="#DCC5E0",
    c400="#D0B2D6",
    c500="#D8BFD8",  # Thistle
    c600="#B8A2B9",
    c700="#98869A",
    c800="#796A7C",
    c900="#5C505D",
    c950="#423A44",
)

colors.red_gray = colors.Color(
    name="red_gray",
    c50="#f7eded", c100="#f5dcdc", c200="#efb4b4", c300="#e78f8f",
    c400="#d96a6a", c500="#c65353", c600="#b24444", c700="#8f3434",
    c800="#732d2d", c900="#5f2626", c950="#4d2020",
)

class ThistleTheme(Soft):
    def __init__(
        self,
        *,
        primary_hue: colors.Color | str = colors.gray,
        secondary_hue: colors.Color | str = colors.thistle,
        neutral_hue: colors.Color | str = colors.slate,
        text_size: sizes.Size | str = sizes.text_md,
        font: fonts.Font | str | Iterable[fonts.Font | str] = (
            fonts.GoogleFont("Inconsolata"), "Arial", "sans-serif",
        ),
        font_mono: fonts.Font | str | Iterable[fonts.Font | str] = (
            fonts.GoogleFont("IBM Plex Mono"), "ui-monospace", "monospace",
        ),
    ):
        super().__init__(
            primary_hue=primary_hue,
            secondary_hue=secondary_hue,
            neutral_hue=neutral_hue,
            text_size=text_size,
            font=font,
            font_mono=font_mono,
        )
        super().set(
            background_fill_primary="*primary_50",
            background_fill_primary_dark="*primary_900",
            body_background_fill="linear-gradient(135deg, *secondary_200, *secondary_100)",
            body_background_fill_dark="linear-gradient(135deg, *primary_900, *primary_800)",
            button_primary_text_color="*neutral_900",
            button_primary_text_color_hover="white",
            button_primary_background_fill="linear-gradient(90deg, *secondary_400, *secondary_400)",
            button_primary_background_fill_hover="linear-gradient(90deg, *secondary_600, *secondary_600)",
            button_primary_background_fill_dark="linear-gradient(90deg, *secondary_600, *secondary_800)",
            button_primary_background_fill_hover_dark="linear-gradient(90deg, *secondary_500, *secondary_500)",
            button_secondary_text_color="black",
            button_secondary_text_color_hover="white",
            button_secondary_background_fill="linear-gradient(90deg, *primary_300, *primary_300)",
            button_secondary_background_fill_hover="linear-gradient(90deg, *primary_400, *primary_400)",
            button_secondary_background_fill_dark="linear-gradient(90deg, *primary_500, *primary_600)",
            button_secondary_background_fill_hover_dark="linear-gradient(90deg, *primary_500, *primary_500)",
            button_cancel_background_fill=f"linear-gradient(90deg, {colors.red_gray.c400}, {colors.red_gray.c500})",
            button_cancel_background_fill_dark=f"linear-gradient(90deg, {colors.red_gray.c700}, {colors.red_gray.c800})",
            button_cancel_background_fill_hover=f"linear-gradient(90deg, {colors.red_gray.c500}, {colors.red_gray.c600})",
            button_cancel_background_fill_hover_dark=f"linear-gradient(90deg, {colors.red_gray.c800}, {colors.red_gray.c900})",
            button_cancel_text_color="white",
            slider_color="*secondary_400",
            slider_color_dark="*secondary_600",
            block_title_text_weight="600",
            block_border_width="3px",
            block_shadow="*shadow_drop_lg",
            button_primary_shadow="*shadow_drop_lg",
            button_large_padding="11px",
            color_accent_soft="*primary_100",
            block_label_background_fill="*secondary_200",
        )

thistle_theme = ThistleTheme()

# --- Custom CSS ---
css = """
:root {
    --color-grey-50: #f9fafb;
}
"""

# --- App Constants & Setup ---
MAX_MAX_NEW_TOKENS = 4096
DEFAULT_MAX_NEW_TOKENS = 2048
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

print("CUDA_VISIBLE_DEVICES=", os.environ.get("CUDA_VISIBLE_DEVICES"))
print("torch.__version__ =", torch.__version__)
print("torch.version.cuda =", torch.version.cuda)
print("cuda available:", torch.cuda.is_available())
print("cuda device count:", torch.cuda.device_count())
if torch.cuda.is_available():
    print("current device:", torch.cuda.current_device())
    print("device name:", torch.cuda.get_device_name(torch.cuda.current_device()))
print("Using device:", device)

# --- Model Loading ---
MODEL_ID_Q3VL = "Qwen/Qwen3-VL-30B-A3B-Instruct"
processor_q3vl = AutoProcessor.from_pretrained(MODEL_ID_Q3VL, trust_remote_code=True, use_fast=False)
model_q3vl = Qwen3VLMoeForConditionalGeneration.from_pretrained(
    MODEL_ID_Q3VL,
    trust_remote_code=True,
    dtype=torch.float16
).to(device).eval()

# --- Backend Functions ---
def downsample_video(video_path):
    vidcap = cv2.VideoCapture(video_path)
    total_frames = int(vidcap.get(cv2.CAP_PROP_FRAME_COUNT))
    frames = []
    frame_indices = np.linspace(0, total_frames - 1, min(total_frames, 10), dtype=int)
    for i in frame_indices:
        vidcap.set(cv2.CAP_PROP_POS_FRAMES, i)
        success, image = vidcap.read()
        if success:
            image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
            frames.append(Image.fromarray(image))
    vidcap.release()
    return frames

def convert_pdf_to_images(file_path: str, dpi: int = 200):
    if not file_path:
        return []
    images = []
    pdf_document = fitz.open(file_path)
    zoom = dpi / 72.0
    mat = fitz.Matrix(zoom, zoom)
    for page_num in range(len(pdf_document)):
        page = pdf_document.load_page(page_num)
        pix = page.get_pixmap(matrix=mat)
        img_data = pix.tobytes("png")
        images.append(Image.open(BytesIO(img_data)))
    pdf_document.close()
    return images

def get_initial_pdf_state() -> Dict[str, Any]:
    return {"pages": [], "total_pages": 0, "current_page_index": 0}

def load_and_preview_pdf(file_path: Optional[str]) -> Tuple[Optional[Image.Image], Dict[str, Any], str]:
    state = get_initial_pdf_state()
    if not file_path:
        return None, state, '<div style="text-align:center;">No file loaded</div>'
    try:
        pages = convert_pdf_to_images(file_path)
        if not pages:
            return None, state, '<div style="text-align:center;">Could not load file</div>'
        state["pages"], state["total_pages"] = pages, len(pages)
        return pages[0], state, f'<div style="text-align:center;">Page 1 / {state["total_pages"]}</div>'
    except Exception as e:
        return None, state, f'<div style="text-align:center;">Failed to load preview: {e}</div>'

def navigate_pdf_page(direction: str, state: Dict[str, Any]):
    if not state or not state["pages"]:
        return None, state, '<div style="text-align:center;">No file loaded</div>'
    idx, total = state["current_page_index"], state["total_pages"]
    new_idx = max(0, idx - 1) if direction == "prev" else min(total - 1, idx + 1)
    state["current_page_index"] = new_idx
    return state["pages"][new_idx], state, f'<div style="text-align:center;">Page {new_idx + 1} / {total}</div>'

@spaces.GPU
def model_stream_response(prompt_text: str, images: list, max_new_tokens: int, temperature: float, top_p: float, top_k: int, repetition_penalty: float):
    messages = [{"role": "user", "content": []}]
    for img in images:
        messages[0]["content"].append({"type": "image"})
    messages[0]["content"].append({"type": "text", "text": prompt_text})
    
    prompt_full = processor_q3vl.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
    inputs = processor_q3vl(text=[prompt_full], images=images, return_tensors="pt", padding=True).to(device)
    
    streamer = TextIteratorStreamer(processor_q3vl, skip_prompt=True, skip_special_tokens=True)
    
    generation_kwargs = {
        **inputs,
        "streamer": streamer,
        "max_new_tokens": max_new_tokens,
        "do_sample": True,
        "temperature": temperature,
        "top_p": top_p,
        "top_k": top_k,
        "repetition_penalty": repetition_penalty,
    }
    
    thread = Thread(target=model_q3vl.generate, kwargs=generation_kwargs)
    thread.start()
    
    buffer = ""
    for new_text in streamer:
        buffer += new_text
        yield buffer.replace("<|im_end|>", ""), buffer.replace("<|im_end|>", "")
        time.sleep(0.01)

def generate_image(text: str, image: Image.Image, *args):
    if image is None:
        yield "Please upload an image.", "Please upload an image."
        return
    yield from model_stream_response(text, [image], *args)

def generate_video(text: str, video_path: str, *args):
    if video_path is None:
        yield "Please upload a video.", "Please upload a video."
        return
    frames = downsample_video(video_path)
    if not frames:
        yield "Could not process video.", "Could not process video."
        return
    yield from model_stream_response(text, frames, *args)

def generate_pdf(text: str, state: Dict[str, Any], *args):
    if not state or not state["pages"]:
        yield "Please upload a PDF file first.", "Please upload a PDF file first."
        return
    
    full_response = ""
    for i, image in enumerate(state["pages"]):
        page_header = f"--- Page {i+1}/{len(state['pages'])} ---\n"
        yield full_response + page_header, full_response + page_header
        
        # This is a simplified approach. For true streaming of the whole PDF, a more complex logic would be needed.
        # Here we just get the full response for the page and then append it.
        final_page_text = ""
        for page_text, _ in model_stream_response(text, [image], *args):
             yield full_response + page_header + page_text, full_response + page_header + page_text
             final_page_text = page_text
        
        full_response += page_header + final_page_text + "\n\n"

def generate_caption(image: Image.Image, *args):
    if image is None:
        yield "Please upload an image for captioning.", "Please upload an image for captioning."
        return
    
    system_prompt = (
        "You are an AI assistant that rigorously follows this response protocol: For every input image, "
        "your primary task is to write a precise caption that captures the essence of the image in clear, "
        "concise, and contextually accurate language. Along with the caption, provide a structured set of "
        "attributes describing the visual elements, including details such as objects, people, actions, "
        "colors, environment, mood, and other notable characteristics. Ensure captions are precise, neutral, "
        "and descriptive, avoiding unnecessary elaboration or subjective interpretation unless explicitly required. "
        "Do not reference the rules or instructions in the output; only return the formatted caption, attributes, and class_name."
    )
    yield from model_stream_response(system_prompt, [image], *args)

# --- Gradio Interface ---
image_examples = [["Describe the safety measures in the image. Conclude (Safe / Unsafe)..", "images/5.jpg"], ["Convert this page to doc [markdown] precisely.", "images/3.png"]]
video_examples = [["Explain the video in detail.", "videos/2.mp4"]]
caption_examples = [["images/3.png"], ["images/5.jpg"]]

with gr.Blocks(theme=thistle_theme, css=css) as demo:
    pdf_state = gr.State(value=get_initial_pdf_state())
    gr.Markdown("# **Qwen3-VL-Demo**")
    
    with gr.Row():
        with gr.Column(scale=2):
            with gr.Tabs():
                # Image Tab
                with gr.TabItem("Image Inference"):
                    image_query = gr.Textbox(label="Query Input", placeholder="Enter your query here...")
                    image_upload = gr.Image(type="pil", label="Image", height=290)
                    image_submit = gr.Button("Submit", variant="primary")
                    gr.Examples(examples=image_examples, inputs=[image_query, image_upload])

                # Video Tab
                with gr.TabItem("Video Inference"):
                    video_query = gr.Textbox(label="Query Input", placeholder="Enter your query here...")
                    video_upload = gr.Video(label="Video", height=290)
                    video_submit = gr.Button("Submit", variant="primary")
                    gr.Examples(examples=video_examples, inputs=[video_query, video_upload])

                # PDF Tab
                with gr.TabItem("PDF Inference"):
                    with gr.Row():
                        with gr.Column(scale=1):
                            pdf_query = gr.Textbox(label="Query Input", placeholder="e.g., 'Summarize this document'")
                            pdf_upload = gr.File(label="Upload PDF", file_types=[".pdf"])
                            pdf_submit = gr.Button("Submit", variant="primary")
                        with gr.Column(scale=1):
                            pdf_preview_img = gr.Image(label="PDF Preview", height=290, interactive=False)
                            with gr.Row():
                                prev_page_btn = gr.Button("◀ Previous")
                                page_info = gr.HTML('<div style="text-align:center;">No file loaded</div>')
                                next_page_btn = gr.Button("Next ▶")
                
                # Caption Tab
                with gr.TabItem("Caption"):
                    caption_image_upload = gr.Image(type="pil", label="Image to Caption", height=290)
                    caption_submit = gr.Button("Generate Caption", variant="primary")
                    gr.Examples(examples=caption_examples, inputs=[caption_image_upload])

            # Advanced Options
            with gr.Accordion("Advanced options", open=False):
                adv_opts = [
                    gr.Slider(label="Max new tokens", minimum=1, maximum=MAX_MAX_NEW_TOKENS, step=1, value=DEFAULT_MAX_NEW_TOKENS),
                    gr.Slider(label="Temperature", minimum=0.1, maximum=4.0, step=0.1, value=0.6),
                    gr.Slider(label="Top-p (nucleus sampling)", minimum=0.05, maximum=1.0, step=0.05, value=0.9),
                    gr.Slider(label="Top-k", minimum=1, maximum=1000, step=1, value=50),
                    gr.Slider(label="Repetition penalty", minimum=1.0, maximum=2.0, step=0.05, value=1.2)
                ]

        # Output Column
        with gr.Column(scale=3):
            gr.Markdown("## Output")
            output = gr.Textbox(label="Raw Output Stream", interactive=False, lines=14, show_copy_button=True)
            with gr.Accordion("(Result.md)", open=False):
                markdown_output = gr.Markdown(label="(Result.Md)")
                
    # Event handlers
    image_submit.click(
        fn=generate_image,
        inputs=[image_query, image_upload, max_new_tokens, temperature, top_p, top_k, repetition_penalty],
        outputs=[output, markdown_output]
    )
    video_submit.click(
        fn=generate_video,
        inputs=[video_query, video_upload, max_new_tokens, temperature, top_p, top_k, repetition_penalty],
        outputs=[output, markdown_output]
    )
    pdf_submit.click(
        fn=generate_pdf,
        inputs=[pdf_query, pdf_state, max_new_tokens, temperature, top_p, top_k, repetition_penalty],
        outputs=[output, markdown_output]
    )
    pdf_upload.change(
        fn=load_and_preview_pdf,
        inputs=[pdf_upload],
        outputs=[pdf_preview_img, pdf_state, page_info]
    )
    prev_page_btn.click(
        fn=lambda s: navigate_pdf_page("prev", s),
        inputs=[pdf_state],
        outputs=[pdf_preview_img, pdf_state, page_info]
    )
    next_page_btn.click(
        fn=lambda s: navigate_pdf_page("next", s),
        inputs=[pdf_state],
        outputs=[pdf_preview_img, pdf_state, page_info]
    )
    caption_submit.click(
        fn=generate_caption,
        inputs=[caption_image_upload] + adv_opts, 
        outputs=[output, markdown_output])
    
    pdf_upload.change(fn=load_and_preview_pdf, inputs=[pdf_upload], outputs=[pdf_preview_img, pdf_state, page_info])
    prev_page_btn.click(fn=lambda s: navigate_pdf_page("prev", s), inputs=[pdf_state], outputs=[pdf_preview_img, pdf_state, page_info])
    next_page_btn.click(fn=lambda s: navigate_pdf_page("next", s), inputs=[pdf_state], outputs=[pdf_preview_img, pdf_state, page_info])

if __name__ == "__main__":
    demo.queue(max_size=50).launch(mcp_server=True, ssr_mode=False, show_error=True)