README.md

---
license: mit
---

## setup uv env 

```
#!/bin/bash
# Set up RAM+ in an isolated venv with a compatible transformers version.
# recognize-anything uses APIs removed in transformers>=4.46, so we pin to 4.45.2.
set -e

VENV=".venv-ram"
WEIGHTS_DIR="models/ram_plus"
WEIGHTS_FILE="$WEIGHTS_DIR/ram_plus_swin_large_14m.pth"

# ---------------------------------------------------------------------------
# 1. Create venv
# ---------------------------------------------------------------------------
if [ ! -d "$VENV" ]; then
    echo "Creating $VENV …"
    uv venv "$VENV" --python 3.11
else
    echo "$VENV already exists"
fi

PIP="$VENV/bin/pip"

# ---------------------------------------------------------------------------
# 2. Install PyTorch (reuse system CUDA version if available)
# ---------------------------------------------------------------------------
if ! "$VENV/bin/python" -c "import torch" 2>/dev/null; then
    echo "Installing PyTorch …"
    CUDA_VER=$(python3 -c "import torch; print('cu' + torch.version.cuda.replace('.',''))" 2>/dev/null || echo "cpu")
    if [ "$CUDA_VER" = "cpu" ]; then
        uv pip install --python "$VENV/bin/python" torch torchvision
    else
        uv pip install --python "$VENV/bin/python" torch torchvision \
            --index-url "https://download.pytorch.org/whl/$CUDA_VER"
    fi
fi

# ---------------------------------------------------------------------------
# 3. Install recognize-anything + pinned transformers
# ---------------------------------------------------------------------------
if ! "$VENV/bin/python" -c "import ram" 2>/dev/null; then
    echo "Installing recognize-anything with transformers==4.45.2 …"
    uv pip install --python "$VENV/bin/python" \
        "transformers==4.45.2" \
        timm pillow scipy fairscale \
        git+https://github.com/xinyu1205/recognize-anything.git
fi

# ---------------------------------------------------------------------------
# 4. Install Streamlit in the RAM venv
# ---------------------------------------------------------------------------
if ! "$VENV/bin/python" -c "import streamlit" 2>/dev/null; then
    echo "Installing streamlit …"
    uv pip install --python "$VENV/bin/python" streamlit
fi

# ---------------------------------------------------------------------------
# 5. Download weights
# ---------------------------------------------------------------------------
mkdir -p "$WEIGHTS_DIR"
if [ ! -f "$WEIGHTS_FILE" ]; then
    echo "Downloading RAM+ weights (~3 GB) …"
    "$VENV/bin/python" -c "
from huggingface_hub import hf_hub_download
path = hf_hub_download(
    repo_id='ma7583/ramplus',
    filename='ram_plus_swin_large_14m.pth',
    local_dir='$WEIGHTS_DIR',
)
print('Downloaded to:', path)
"
else
    echo "Weights already present at $WEIGHTS_FILE"
fi

echo ""
echo "Setup complete. Run the app with:"
echo "  $VENV/bin/streamlit run tests/explore_ram.py"


```


## run streamlit for demo

```
"""RAM+ Video Object Explorer — Streamlit app.

Usage:
    streamlit run tests/explore_ram.py
"""

import json
import subprocess
import tempfile
from pathlib import Path
from PIL import Image
import streamlit as st
import torch

VIDEO_DIR  = Path(".")
WEIGHTS    = Path("models/ram_plus/ram_plus_swin_large_14m.pth")
CACHE_DIR  = Path("tests/mbs/ram_tags")
IMAGE_SIZE = 384

st.set_page_config(page_title="RAM+ Explorer", layout="wide")

# ---------------------------------------------------------------------------
# Load model
# ---------------------------------------------------------------------------

@st.cache_resource(show_spinner="Loading RAM+ model…")
def load_model(weights: str):
    from ram.models import ram_plus
    import torchvision.transforms as T

    device = "cuda" if torch.cuda.is_available() else "cpu"
    model = ram_plus(pretrained=weights, image_size=IMAGE_SIZE, vit="swin_l")
    model.eval().to(device)

    transform = T.Compose([
        T.Resize((IMAGE_SIZE, IMAGE_SIZE)),
        T.ToTensor(),
        T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
    ])
    return model, transform, device


if not WEIGHTS.exists():
    st.error(f"Weights not found at `{WEIGHTS}`. Run `bash run_ram_setup.sh` first.")
    st.stop()

model, transform, device = load_model(str(WEIGHTS))

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def extract_frames(video_path: Path, fps: float) -> list[tuple[float, Path]]:
    tmp = Path(tempfile.mkdtemp(prefix="ram-frames-"))
    subprocess.run(
        ["ffmpeg", "-y", "-i", str(video_path), "-vf", f"fps={fps}", "-q:v", "2",
         str(tmp / "frame_%05d.jpg")],
        capture_output=True, check=True, timeout=300,
    )
    frames = sorted(tmp.glob("frame_*.jpg"))
    return [(i / fps, p) for i, p in enumerate(frames)]


@torch.inference_mode()
def tag_frame(img_path: Path, threshold: float) -> list[str]:
    img = Image.open(img_path).convert("RGB")
    tensor = transform(img).unsqueeze(0).to(device)
    model.threshold = threshold
    tags, _ = model.generate_tag(tensor)
    return [t.strip() for t in tags[0].split("|") if t.strip()]


CACHE_DIR.mkdir(parents=True, exist_ok=True)

def cache_path(video_name: str, fps: float, threshold: float) -> Path:
    return CACHE_DIR / f"{Path(video_name).stem}_{fps}fps_t{threshold:.2f}.json"


@st.cache_data(show_spinner=False)
def get_tags(video_name: str, fps: float, threshold: float, video_dir: str) -> list[dict]:
    cp = cache_path(video_name, fps, threshold)
    if cp.exists():
        return json.loads(cp.read_text())

    frames = extract_frames(Path(video_dir) / video_name, fps)
    results = []
    bar = st.progress(0, text=f"Tagging {len(frames)} frames…")
    for i, (ts, fp) in enumerate(frames):
        results.append({"timestamp": round(ts, 2), "tags": tag_frame(fp, threshold), "frame": str(fp)})
        bar.progress((i + 1) / len(frames), text=f"Frame {i+1}/{len(frames)}")
    bar.empty()

    cp.write_text(json.dumps(results, ensure_ascii=False))
    return results

# ---------------------------------------------------------------------------
# UI
# ---------------------------------------------------------------------------

with st.sidebar:
    st.header("Settings")
    video_dir_input = st.text_input("Video directory", value=str(VIDEO_DIR))
    video_dir = Path(video_dir_input)

videos = sorted(p.name for p in video_dir.glob("*.mp4")) if video_dir.exists() else []
if not videos:
    st.error(f"No .mp4 files found in `{video_dir}`")
    st.stop()

with st.sidebar:
    video_name = st.selectbox(f"Video ({len(videos)} total)", videos)

col_search, col_fps, col_thresh = st.columns([4, 1, 1])
with col_search:
    tag_filter = st.text_input("Search tags", placeholder="e.g. person, shelf, phone")
with col_fps:
    fps = st.select_slider("Sampling fps", [0.25, 0.5, 1.0, 2.0], value=0.5)
with col_thresh:
    threshold = st.slider("RAM Threshold", 0.1, 0.95, 0.68, step=0.05)

cached = cache_path(video_name, fps, threshold).exists()
st.caption(f"{'✓ cached' if cached else '⚡ will run inference'}")

with st.spinner(f"Processing {video_name}…"):
    frames = get_tags(video_name, fps, threshold, str(video_dir))

# Filter by tag
if tag_filter.strip():
    needles = [n.strip().lower() for n in tag_filter.split(",") if n.strip()]
    frames = [f for f in frames if any(
        any(n in t.lower() for t in f["tags"]) for n in needles
    )]
    st.info(f"{len(frames)} frames match **{tag_filter}**")
else:
    # Show tag cloud when no filter active
    all_tags: dict[str, int] = {}
    for f in frames:
        for t in f["tags"]:
            all_tags[t] = all_tags.get(t, 0) + 1
    top = sorted(all_tags.items(), key=lambda x: -x[1])[:60]
    st.write("**All tags:** " + "  ·  ".join(f"`{t}` ({n})" for t, n in top))
    st.divider()

# Frame grid
if frames:
    cols = st.columns(4)
    for i, frame in enumerate(frames):
        with cols[i % 4]:
            st.image(frame["frame"], use_container_width=True)
            ts = frame["timestamp"]
            m, s = divmod(int(ts), 60)
            st.caption(f"**{m}:{s:02d}**  {' · '.join(frame['tags'][:5])}")
else:
    st.warning("No frames match the search.")

```