app application

.gitignore

@@ -0,0 +1,303 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be added to the global gitignore or merged into this project gitignore.  For a PyCharm
+#  project, it is recommended to ignore these files.
+.idea/
+
+# VS Code
+.vscode/
+
+# macOS
+.DS_Store
+.AppleDouble
+.LSOverride
+
+# Icon must end with two \r
+Icon
+
+# Thumbnails
+._*
+
+# Files that might appear in the root of a volume
+.DocumentRevisions-V100
+.fseventsd
+.Spotlight-V100
+.TemporaryItems
+.Trashes
+.VolumeIcon.icns
+.com.apple.timemachine.donotpresent
+
+# Directories potentially created on remote AFP share
+.AppleDB
+.AppleDesktop
+Network Trash Folder
+Temporary Items
+.apdisk
+
+# Windows
+Thumbs.db
+Thumbs.db:encryptable
+ehthumbs.db
+ehthumbs_vista.db
+
+# Dump file
+*.stackdump
+
+# Folder config file
+[Dd]esktop.ini
+
+# Recycle Bin used on file shares
+$RECYCLE.BIN/
+
+# Windows Installer files
+*.cab
+*.msi
+*.msix
+*.msm
+*.msp
+
+# Windows shortcuts
+*.lnk
+
+# Linux
+*~
+
+# temporary files which can be created if a process still has a handle open of a deleted file
+.fuse_hidden*
+
+# KDE directory preferences
+.directory
+
+# Linux trash folder which might appear on any partition or disk
+.Trash-*
+
+# .nfs files are created when an open file is removed but is still being accessed
+.nfs*
+
+# Project-specific ignores
+# Model files and caches
+models/
+*.bin
+*.safetensors
+*.onnx
+*.pb
+*.tflite
+*.pth
+*.pt
+*.h5
+*.pkl
+*.pickle
+
+# Hugging Face cache
+.cache/
+transformers_cache/
+huggingface_hub/
+
+# Gradio specific
+gradio_cached_examples/
+flagged/
+
+# Temporary files
+temp/
+tmp/
+*.tmp
+*.temp
+
+# Log files
+*.log
+logs/
+
+# Data files (if sensitive)
+data/
+datasets/
+*.csv
+*.json
+*.xml
+*.yaml
+*.yml
+
+# Media files (examples might be large)
+examples/
+assets/
+media/
+*.mp4
+*.avi
+*.mov
+*.jpg
+*.jpeg
+*.png
+*.gif
+*.bmp
+*.tiff
+*.webp
+
+# Configuration files with secrets
+config.ini
+secrets.json
+.secrets
+credentials.json
+
+# Backup files
+*.bak
+*.backup
+*.old
+
+# Node.js (if using any frontend tools)
+node_modules/
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log* 

app.py

@@ -0,0 +1,381 @@
+#!/usr/bin/env python
+
+import os
+import re
+import tempfile
+from collections.abc import Iterator
+from threading import Thread
+
+import cv2
+import gradio as gr
+import spaces
+import torch
+from loguru import logger
+from PIL import Image
+from transformers import AutoProcessor, AutoModelForImageTextToText, TextIteratorStreamer
+
+model_id = os.getenv("MODEL_ID", "google/medgemma-4b-it")
+processor = AutoProcessor.from_pretrained(model_id)
+model = AutoModelForImageTextToText.from_pretrained(
+    model_id, device_map="auto", torch_dtype=torch.bfloat16
+)
+
+MAX_NUM_IMAGES = int(os.getenv("MAX_NUM_IMAGES", "5"))
+
+def count_files_in_new_message(paths: list[str]) -> tuple[int, int]:
+    image_count = 0
+    video_count = 0
+    for path in paths:
+        if path.endswith(".mp4"):
+            video_count += 1
+        else:
+            image_count += 1
+    return image_count, video_count
+
+def count_files_in_history(history: list[dict]) -> tuple[int, int]:
+    image_count = 0
+    video_count = 0
+    for item in history:
+        if item["role"] != "user" or isinstance(item["content"], str):
+            continue
+        if item["content"][0].endswith(".mp4"):
+            video_count += 1
+        else:
+            image_count += 1
+    return image_count, video_count
+
+def validate_media_constraints(message: dict, history: list[dict]) -> bool:
+    new_image_count, new_video_count = count_files_in_new_message(message["files"])
+    history_image_count, history_video_count = count_files_in_history(history)
+    image_count = history_image_count + new_image_count
+    video_count = history_video_count + new_video_count
+    if video_count > 1:
+        gr.Warning("Only one video is supported.")
+        return False
+    if video_count == 1:
+        if image_count > 0:
+            gr.Warning("Mixing images and videos is not allowed.")
+            return False
+        if "<image>" in message["text"]:
+            gr.Warning("Using <image> tags with video files is not supported.")
+            return False
+    if video_count == 0 and image_count > MAX_NUM_IMAGES:
+        gr.Warning(f"You can upload up to {MAX_NUM_IMAGES} images.")
+        return False
+    if "<image>" in message["text"] and message["text"].count("<image>") != new_image_count:
+        gr.Warning("The number of <image> tags in the text does not match the number of images.")
+        return False
+    return True
+
+def downsample_video(video_path: str) -> list[tuple[Image.Image, float]]:
+    vidcap = cv2.VideoCapture(video_path)
+    fps = vidcap.get(cv2.CAP_PROP_FPS)
+    total_frames = int(vidcap.get(cv2.CAP_PROP_FRAME_COUNT))
+
+    frame_interval = max(total_frames // MAX_NUM_IMAGES, 1)
+    frames: list[tuple[Image.Image, float]] = []
+
+    for i in range(0, min(total_frames, MAX_NUM_IMAGES * frame_interval), frame_interval):
+        if len(frames) >= MAX_NUM_IMAGES:
+            break
+
+        vidcap.set(cv2.CAP_PROP_POS_FRAMES, i)
+        success, image = vidcap.read()
+        if success:
+            image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
+            pil_image = Image.fromarray(image)
+            timestamp = round(i / fps, 2)
+            frames.append((pil_image, timestamp))
+
+    vidcap.release()
+    return frames
+
+def process_video(video_path: str) -> list[dict]:
+    content = []
+    frames = downsample_video(video_path)
+    for frame in frames:
+        pil_image, timestamp = frame
+        with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as temp_file:
+            pil_image.save(temp_file.name)
+            content.append({"type": "text", "text": f"Frame {timestamp}:"})
+            content.append({"type": "image", "url": temp_file.name})
+    logger.debug(f"{content=}")
+    return content
+
+def process_interleaved_images(message: dict) -> list[dict]:
+    logger.debug(f"{message['files']=}")
+    parts = re.split(r"(<image>)", message["text"])
+    logger.debug(f"{parts=}")
+
+    content = []
+    image_index = 0
+    for part in parts:
+        logger.debug(f"{part=}")
+        if part == "<image>":
+            content.append({"type": "image", "url": message["files"][image_index]})
+            logger.debug(f"file: {message['files'][image_index]}")
+            image_index += 1
+        elif part.strip():
+            content.append({"type": "text", "text": part.strip()})
+        elif isinstance(part, str) and part != "<image>":
+            content.append({"type": "text", "text": part})
+    logger.debug(f"{content=}")
+    return content
+
+def process_new_user_message(message: dict) -> list[dict]:
+    if not message["files"]:
+        return [{"type": "text", "text": message["text"]}]
+
+    if message["files"][0].endswith(".mp4"):
+        return [{"type": "text", "text": message["text"]}, *process_video(message["files"][0])]
+
+    if "<image>" in message["text"]:
+        return process_interleaved_images(message)
+
+    return [
+        {"type": "text", "text": message["text"]},
+        *[{"type": "image", "url": path} for path in message["files"]],
+    ]
+
+def process_history(history: list[dict]) -> list[dict]:
+    messages = []
+    current_user_content: list[dict] = []
+    for item in history:
+        if item["role"] == "assistant":
+            if current_user_content:
+                messages.append({"role": "user", "content": current_user_content})
+                current_user_content = []
+            messages.append({"role": "assistant", "content": [{"type": "text", "text": item["content"]}]})
+        else:
+            content = item["content"]
+            if isinstance(content, str):
+                current_user_content.append({"type": "text", "text": content})
+            else:
+                current_user_content.append({"type": "image", "url": content[0]})
+    return messages
+
+@spaces.GPU(duration=120)
+def run(message: dict, history: list[dict], system_prompt: str = "", max_new_tokens: int = 2048) -> Iterator[str]:
+    if not validate_media_constraints(message, history):
+        yield ""
+        return
+
+    messages = []
+    if system_prompt:
+        messages.append({"role": "system", "content": [{"type": "text", "text": system_prompt}]})
+    messages.extend(process_history(history))
+    messages.append({"role": "user", "content": process_new_user_message(message)})
+
+    inputs = processor.apply_chat_template(
+        messages,
+        add_generation_prompt=True,
+        tokenize=True,
+        return_dict=True,
+        return_tensors="pt",
+    ).to(device=model.device, dtype=torch.bfloat16)
+
+    streamer = TextIteratorStreamer(processor, timeout=30.0, skip_prompt=True, skip_special_tokens=True)
+    generate_kwargs = dict(
+        inputs,
+        max_new_tokens=max_new_tokens,
+        streamer=streamer,
+        temperature=1.0,
+        top_p=0.95,
+        top_k=64,
+        min_p=0.0,
+    )
+    t = Thread(target=model.generate, kwargs=generate_kwargs)
+    t.start()
+
+    output = ""
+    for delta in streamer:
+        output += delta
+        yield output
+
+# Custom CSS for the UI
+custom_css = """
+:root {
+    --primary: #4f46e5;
+    --primary-dark: #4338ca;
+    --text: #1f2937;
+    --background: #f9fafb;
+    --chat-bg: #ffffff;
+    --user-bubble: #e0e7ff;
+    --bot-bubble: #f3f4f6;
+    --border: #e5e7eb;
+}
+
+.dark {
+    --primary: #6366f1;
+    --primary-dark: #4f46e5;
+    --text: #f3f4f6;
+    --background: #111827;
+    --chat-bg: #1f2937;
+    --user-bubble: #4338ca;
+    --bot-bubble: #374151;
+    --border: #4b5563;
+}
+
+body {
+    font-family: 'Inter', sans-serif;
+}
+
+.gr-chatbot {
+    background-color: var(--chat-bg);
+    border-radius: 12px;
+    border: 1px solid var(--border);
+    box-shadow: 0 4px 6px -1px rgba(0, 0, 0, 0.1);
+}
+
+.gr-chat-message {
+    padding: 16px 20px;
+    border-radius: 12px;
+    margin: 8px 0;
+    max-width: 80%;
+}
+
+.gr-chat-message-user {
+    background-color: var(--user-bubble);
+    margin-left: auto;
+    border-bottom-right-radius: 4px;
+}
+
+.gr-chat-message-bot {
+    background-color: var(--bot-bubble);
+    margin-right: auto;
+    border-bottom-left-radius: 4px;
+}
+
+.gr-textbox textarea {
+    min-height: 120px;
+    border-radius: 12px;
+    padding: 16px;
+    background-color: var(--background);
+    color: var(--text);
+    border: 1px solid var(--border);
+}
+
+.gr-button {
+    background-color: var(--primary) !important;
+    color: white !important;
+    border-radius: 8px !important;
+    padding: 10px 20px !important;
+    font-weight: 500 !important;
+    transition: all 0.2s !important;
+}
+
+.gr-button:hover {
+    background-color: var(--primary-dark) !important;
+    transform: translateY(-1px) !important;
+}
+
+.gr-button:active {
+    transform: translateY(0) !important;
+}
+
+.gr-interface {
+    max-width: 900px;
+    margin: 0 auto;
+    padding: 24px;
+}
+
+.gr-header {
+    text-align: center;
+    margin-bottom: 24px;
+}
+
+.gr-header h1 {
+    font-size: 2.5rem;
+    font-weight: 700;
+    color: var(--primary);
+    margin-bottom: 8px;
+}
+
+.gr-header p {
+    color: var(--text);
+    opacity: 0.8;
+    font-size: 1.1rem;
+}
+
+.gr-image-preview {
+    border-radius: 8px;
+    max-width: 100%;
+    max-height: 300px;
+    object-fit: contain;
+}
+
+.gr-video-preview {
+    border-radius: 8px;
+    max-width: 100%;
+    max-height: 300px;
+}
+"""
+
+DESCRIPTION = """\
+## Medical Vision-Language Assistant
+
+This advanced AI assistant can understand and analyze medical images, videos, and text. 
+Upload images or a video along with your questions to get insights.
+
+**Features:**
+- Analyze medical images (X-rays, CT scans, etc.)
+- Process video frames from medical videos
+- Interleave images with text questions
+- Customize system behavior with prompts
+"""
+
+demo = gr.ChatInterface(
+    fn=run,
+    type="messages",
+    chatbot=gr.Chatbot(
+        type="messages", 
+        scale=1, 
+        allow_tags=["image"],
+        bubble_full_width=False,
+        avatar_images=(
+            "assets/user.png",  # User avatar
+            "assets/doctor.png"  # Bot avatar (replace with your own)
+        ),
+        render=False  # We'll handle rendering in CSS
+    ),
+    textbox=gr.MultimodalTextbox(
+        file_types=["image", ".mp4"], 
+        file_count="multiple", 
+        autofocus=True,
+        placeholder="Type your message or upload images/video...",
+    ),
+    multimodal=True,
+    additional_inputs=[
+        gr.Textbox(
+            label="System Prompt", 
+            value="You are a helpful and knowledgeable medical expert. Provide accurate, detailed explanations in clear language.",
+            info="Guide the assistant's behavior and expertise"
+        ),
+        gr.Slider(
+            label="Response Length", 
+            minimum=100, 
+            maximum=4096, 
+            step=10, 
+            value=1024,
+            info="Control how verbose the responses are"
+        ),
+    ],
+    stop_btn=None,
+    title="",
+    description=DESCRIPTION,
+    examples=[
+        ["What abnormalities do you see in this chest X-ray?", "examples/chest_xray.jpg"],
+        ["Explain the key findings in this MRI scan.", "examples/brain_mri.jpg"],
+        ["Describe the progression shown in this video.", "examples/heart_ultrasound.mp4"],
+    ],
+    cache_examples=False,
+    css=custom_css,
+    theme=gr.themes.Default(
+        primary_hue="indigo",
+        secondary_hue="gray",
+        font=["Inter", "sans-serif"]
+    ),
+)
+
+if __name__ == "__main__":
+    demo.launch()

requirements.txt

@@ -0,0 +1,68 @@
+accelerate==1.8.1
+aiofiles==24.1.0
+annotated-types==0.7.0
+anyio==4.9.0
+certifi==2025.6.15
+charset-normalizer==3.4.2
+click==8.2.1
+fastapi==0.115.14
+ffmpy==0.6.0
+filelock==3.18.0
+fsspec==2025.5.1
+gradio==5.35.0
+gradio_client==1.10.4
+groovy==0.1.2
+h11==0.16.0
+hf-xet==1.1.5
+httpcore==1.0.9
+httpx==0.28.1
+huggingface-hub==0.33.1
+idna==3.10
+Jinja2==3.1.6
+loguru==0.7.3
+markdown-it-py==3.0.0
+MarkupSafe==3.0.2
+mdurl==0.1.2
+mpmath==1.3.0
+networkx==3.5
+numpy==2.3.1
+opencv-python==4.11.0.86
+orjson==3.10.18
+packaging==25.0
+pandas==2.3.0
+pillow==11.2.1
+psutil==5.9.8
+pydantic==2.11.7
+pydantic_core==2.33.2
+pydub==0.25.1
+Pygments==2.19.2
+python-dateutil==2.9.0.post0
+python-multipart==0.0.20
+pytz==2025.2
+PyYAML==6.0.2
+regex==2024.11.6
+requests==2.32.4
+rich==14.0.0
+ruff==0.12.1
+safehttpx==0.1.6
+safetensors==0.5.3
+semantic-version==2.10.0
+setuptools==80.9.0
+shellingham==1.5.4
+six==1.17.0
+sniffio==1.3.1
+spaces==0.37.1
+starlette==0.46.2
+sympy==1.14.0
+tokenizers==0.21.2
+tomlkit==0.13.3
+torch==2.7.1
+tqdm==4.67.1
+transformers==4.53.0
+typer==0.16.0
+typing-inspection==0.4.1
+typing_extensions==4.14.0
+tzdata==2025.2
+urllib3==2.5.0
+uvicorn==0.35.0
+websockets==15.0.1