Buckets:

---

Rabius

/

Navigation_Benchmark-bucket

🏆 Critical Blindspot

Dhaka (Bangladesh), Delhi (India) these are some of the most populated cities in the world. Unlike developed cities like California (USA) their foot pathway is jammed with people. Although it’s chaotic, somehow people manage to have a workable system by human gestures, expressions and common sense (verbal communication is inefficient in chaotic scenarios). As we are developing bipedal robots for general purpose activities, they also should understand these gestures in real time. So I propose a benchmark consisting of ‘human gestures navigation in chaotic narrow foot pathway’ for bipedal robots.

I will create a synthetic small dataset for demonstration and test it on current state of art physical models such as cosmos 3 2B and so forth.

🤖 Dataset Creation

I took a sample image from https://arxiv.org/pdf/2604.00416 and used gemini & seeddance 2.0 to create a small synthetic benchmark for a narrow space. Later I annotated it with possible human actions. Although it doesn’t entirely represent real scenarios in Dhaka or Delhi, it’s a start. I selected cosmos3 to test this dataset.

Prompt “which way should the robot go?”

🔗 Dataset's filestucture

.
├── images/                     
├── videos/
├── .gitattributes
├── README.md
├── cosmos_vs_human_scores.csv
└── poster.jpg

🎯 Human annotated Dataset

📂 Similarity results (Human vs Physical AI model)

🤖 Methodology

With the created synthetic images, I annotated the possible direction based on my human judgements. Later, I passed the dataset to Cosmos 3 2B alongside the prompt “which way should the robot go?”. I stored the output in a csv file, and passed both the human annotated dataset and the output csv to LLM (Sonnet 5.5) to find a similarity score between 0-5.

🤖 Discussion & weaknesses

Cosmos 3 2B scored 14/60 on this synthetic dataset. It failed to show empathy, especially toward children and pregnant woman. The human gesture judgment was also poor compared to human annotation. One of the main weaknesses of this study is human annotation. A single annotator is not enough for drawing conclusions. Also the prompt quality, it was a basic prompt.

🤖Future Direction

Real world dataset is a must to understand the navigation gestures a group of people use. So collecting real data should be the first priority for this benchmark. A 20 to 30 hour video taken on the narrow streets is a good start. Rethinking the methodology is also needed, especially choosing the similarity score.

🔗 Codes

import argparse
import contextlib
import csv
import re
import shutil
import sys
import tempfile
import time
from pathlib import Path

import torch
import transformers
from huggingface_hub import HfApi, HfFileSystem

MODEL_NAME = "nvidia/Cosmos-Reason2-2B"
DATASET_ID = "Rabius/Navigation_Benchmark"
DEFAULT_PROMPT = "Describe what is happening and what the robot should do next."

IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp"}
VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv"}

THINK_SUFFIX = (
    "\n\nAnswer the question using the following format:\n\n"
    "<think>\nYour reasoning.\n</think>\n\n"
    "Write your final answer immediately after the </think> tag."
)


def natural_key(path: Path):
    return [int(t) if t.isdigit() else t.lower() for t in re.split(r"(\d+)", path.name)]


def collect_local(root: Path, include_poster: bool):
    """(kind, Path) for images/ and videos/ in a local copy of the dataset."""
    items = []
    for folder, exts, kind in (("images", IMAGE_EXTS, "image"), ("videos", VIDEO_EXTS, "video")):
        d = root / folder
        if d.exists():
            files = sorted(
                (p for p in d.rglob("*") if p.is_file() and p.suffix.lower() in exts),
                key=natural_key,
            )
            items += [(kind, p) for p in files]
    if include_poster and (root / "poster.jpg").exists():
        items.append(("image", root / "poster.jpg"))
    return items


def collect_remote(include_poster: bool):
    """(kind, repo_path) for images/ and videos/ listed straight from the Hub (no download)."""
    repo_files = HfApi().list_repo_files(DATASET_ID, repo_type="dataset")
    items = []
    for folder, exts, kind in (("images", IMAGE_EXTS, "image"), ("videos", VIDEO_EXTS, "video")):
        paths = sorted(
            (p for p in repo_files if p.startswith(folder + "/") and Path(p).suffix.lower() in exts),
            key=lambda p: natural_key(Path(p)),
        )
        items += [(kind, p) for p in paths]
    if include_poster and "poster.jpg" in repo_files:
        items.append(("image", "poster.jpg"))
    return items


@contextlib.contextmanager
def materialize(source, fs, tmpdir):
    
    if isinstance(source, Path):
        yield source
        return
    dst = Path(tmpdir) / Path(source).name
    with fs.open(f"datasets/{DATASET_ID}/{source}", "rb") as src, open(dst, "wb") as out:
        shutil.copyfileobj(src, out)
    try:
        yield dst
    finally:
        dst.unlink(missing_ok=True)


def build_messages(kind: str, path: Path, prompt: str, fps: int, think: bool):
    text = prompt + (THINK_SUFFIX if think else "")
    if kind == "image":
        media = {"type": "image", "image": str(path.resolve())}
    else:
        media = {"type": "video", "video": str(path.resolve()), "fps": fps}
    return [
        {"role": "system", "content": [{"type": "text", "text": "You are a helpful assistant."}]},
        {"role": "user", "content": [media, {"type": "text", "text": text}]},
    ]


def extract_answer(text: str) -> str:
    """Pull the final answer out of '<think>...</think><answer>...</answer>' style output."""
    m = re.search(r"<answer>(.*?)(?:</answer>|$)", text, flags=re.S)
    if m:
        return m.group(1).strip()
    if "</think>" in text:
        return text.split("</think>")[-1].strip()
    return text.strip()


@torch.inference_mode()
def run_one(model, processor, kind, path, prompt, fps, max_new_tokens, think):
    messages = build_messages(kind, path, prompt, fps, think)
    kwargs = {"fps": fps} if kind == "video" else {}
    inputs = processor.apply_chat_template(
        messages,
        tokenize=True,
        add_generation_prompt=True,
        return_dict=True,
        return_tensors="pt",
        **kwargs,
    ).to(model.device)

    generated = model.generate(**inputs, max_new_tokens=max_new_tokens)
    trimmed = [out[len(inp):] for inp, out in zip(inputs.input_ids, generated)]
    text = processor.batch_decode(
        trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
    )[0]
    return text.strip()


def main():
    ap = argparse.ArgumentParser(description="Cosmos-Reason2-2B batch inference -> CSV")
    ap.add_argument("--prompt", default=DEFAULT_PROMPT)
    ap.add_argument("--output", default="cosmos_reason2_outputs.csv")
    ap.add_argument("--data-dir", default=None,
                    help="Use a local copy of the dataset instead of streaming from the Hub.")
    ap.add_argument("--fps", type=int, default=4, help="Video sampling FPS (model card: 4). Lower it if you run out of VRAM.")
    ap.add_argument("--max-new-tokens", type=int, default=1024,
                    help="Default 1024 is plenty for short answers; use 4096+ with --think.")
    ap.add_argument("--think", action="store_true",
                    help="Ask for <think> reasoning and also save it in a full_output column (slower, longer).")
    ap.add_argument("--include-poster", action="store_true", help="Also run poster.jpg from the dataset root.")
    args = ap.parse_args()

    if not torch.cuda.is_available():
        sys.exit("No CUDA GPU detected. This script needs a GPU.")

    # ---- Data ----
    fs = HfFileSystem()
    if args.data_dir:
        media = collect_local(Path(args.data_dir), args.include_poster)
    else:
        print(f"Listing {DATASET_ID} on the Hub (streaming, no full download) ...")
        media = collect_remote(args.include_poster)
    if not media:
        sys.exit("No images/videos found.")
    print(f"Found {sum(k == 'image' for k, _ in media)} images and "
          f"{sum(k == 'video' for k, _ in media)} videos.")

    # ---- Model ----
    dtype = torch.bfloat16 if torch.cuda.is_bf16_supported() else torch.float16
    print(f"Loading {MODEL_NAME} ({dtype}) on {torch.cuda.get_device_name(0)} ...")
    model = transformers.Qwen3VLForConditionalGeneration.from_pretrained(
        MODEL_NAME, dtype=dtype, device_map="auto", attn_implementation="sdpa"
    )
    model.eval()
    processor = transformers.AutoProcessor.from_pretrained(MODEL_NAME)

    # ---- Inference ----
    think = args.think
    with tempfile.TemporaryDirectory() as tmpdir, \
            open(args.output, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.writer(f)
        writer.writerow(["name", "type", "prompt", "answer"] + (["full_output"] if think else []))

        for i, (kind, source) in enumerate(media, 1):
            name = Path(source).name
            print(f"[{i}/{len(media)}] {kind}: {name} ...", end=" ", flush=True)
            t0 = time.time()
            try:
                with materialize(source, fs, tmpdir) as local_path:
                    full = run_one(model, processor, kind, local_path, args.prompt,
                                   args.fps, args.max_new_tokens, think)
                answer = extract_answer(full)
                print(f"done ({time.time() - t0:.1f}s)")
            except torch.cuda.OutOfMemoryError:
                full = answer = "ERROR: CUDA out of memory (try a lower --fps or --max-new-tokens)"
                print("OOM")
            except Exception as e:  # keep going so one bad file doesn't kill the run
                full = answer = f"ERROR: {type(e).__name__}: {e}"
                print("failed")
            finally:
                torch.cuda.empty_cache()

            writer.writerow([name, kind, args.prompt, answer] + ([full] if think else []))
            f.flush()  # results are saved as we go

    print(f"\nSaved {len(media)} rows to {Path(args.output).resolve()}")


if __name__ == "__main__":
    main()