vision.py

"""Photograph your ingredients; a small VLM lists what it sees.

Uses MiniCPM-V 4.6 (openbmb) — an open-vocabulary vision-language model that can
name produce, pantry items and even read spice-jar labels. We ask it for a plain
ingredient list, then hand that straight to the seasoning planner. (LocateAnything
is for bounding-box grounding of known targets; here we want open discovery, so a
describe-and-list VLM is the right tool.)

~2B params, so it sits comfortably alongside Mellum 2's 12B under the 32B cap.

Real inference runs only on a GPU (the Space). With MOCK_VISION=1 — or whenever
torch/the model can't load — `detect_ingredients` returns a clearly-labelled
sample list so the photo→pantry→plan flow is demonstrable offline.
"""

import os
import re

VISION_MODEL_ID = os.environ.get("VISION_MODEL_ID", "openbmb/MiniCPM-V-4.6")
# Vision has its own backend because, unlike the reasoning model, MiniCPM-V has a
# free hosted API — so you can run it real with no GPU at all.
#   openbmb : OpenBMB/ModelBest free hosted API (no GPU, recommended real path)
#   modal   : your Modal endpoint   |   zerogpu : in-Space GPU
VISION_BACKEND = os.environ.get("VISION_BACKEND", "zerogpu")
# Mock only when asked, OR in MOCK_LLM dev runs that DIDN'T pick a real vision
# backend — so `MOCK_LLM=1 VISION_BACKEND=openbmb` gives real vision + scripted reasoning.
MOCK_VISION = os.environ.get("MOCK_VISION") == "1" or (
    os.environ.get("MOCK_LLM") == "1" and "VISION_BACKEND" not in os.environ)
MODAL_VISION_URL = os.environ.get("MODAL_VISION_URL", "")
# OpenBMB hosted API (OpenAI-compatible). Public free key ships as the default but
# is shared/rate-limited — override with your own from platform.modelbest.cn.
OPENBMB_API_URL = os.environ.get("MINICPM_API_URL", "https://api.modelbest.cn/v1/chat/completions")
OPENBMB_API_KEY = os.environ.get("MINICPM_API_KEY", "sk-pQ8L2zF3XmR5kY9wV4jB7hN1tC6vM0xG3aD5sH2bJ9lK4cZ8")
OPENBMB_API_MODEL = os.environ.get("MINICPM_API_MODEL", "MiniCPM-V-4.6-Instruct")

DETECT_PROMPT = (
    "You are looking at a photo of someone's kitchen ingredients. List ONLY the "
    "specific food ingredients you can clearly identify. Use canonical SINGULAR "
    "names a recipe would use (e.g. 'cumin', 'garlic', 'lentil', 'tomato', "
    "'papaya'). Do NOT use vague category words like 'produce', 'vegetables', "
    "'spices', or 'herbs' — name the actual item. Reply with a single "
    "comma-separated list and nothing else."
)

# Shown offline so the demo flow works without a GPU — clearly not a real read.
_MOCK_DETECTION = [
    "lentil", "cumin", "coriander seed", "turmeric", "garlic",
    "onion", "ginger", "tomato", "lemon",
]

_model = None
_tokenizer = None


def _load():
    """Lazy-load MiniCPM-V. Module-level load is fine on ZeroGPU, but lazy keeps
    startup cheap when vision isn't used."""
    global _model, _tokenizer
    if _model is not None:
        return
    import torch
    from transformers import AutoModel, AutoTokenizer

    _tokenizer = AutoTokenizer.from_pretrained(VISION_MODEL_ID, trust_remote_code=True)
    _model = AutoModel.from_pretrained(
        VISION_MODEL_ID, trust_remote_code=True, dtype=torch.bfloat16
    ).eval().to("cuda")


def _to_image(image):
    """Accept a filepath or a PIL image; return RGB PIL."""
    from PIL import Image

    if isinstance(image, str):
        return Image.open(image).convert("RGB")
    return image.convert("RGB")


def _parse_list(text: str) -> list[str]:
    """Turn the model's reply into clean, de-duplicated ingredient names."""
    text = re.sub(r"^[^:]*:", "", text.strip())  # drop any "I can see:" preamble
    items, seen = [], set()
    for part in re.split(r"[,\n;]+", text):
        name = re.sub(r"^[\s\-\*\d\.\)]+", "", part).strip().lower()  # strip bullets/numbering
        name = re.sub(r"\s+", " ", name).strip(" .")
        if 1 < len(name) <= 30 and name not in seen:
            seen.add(name)
            items.append(name)
    return items[:20]


def _detect_zerogpu(pil) -> str:
    _load()
    msgs = [{"role": "user", "content": [pil, DETECT_PROMPT]}]
    reply = _model.chat(image=None, msgs=msgs, tokenizer=_tokenizer,
                        sampling=False, max_new_tokens=200)
    return reply if isinstance(reply, str) else str(reply)


def _detect_modal(pil) -> str:
    import base64
    import io

    import httpx

    buf = io.BytesIO()
    pil.save(buf, format="JPEG")
    payload = {"image_b64": base64.b64encode(buf.getvalue()).decode(), "prompt": DETECT_PROMPT}
    resp = httpx.post(MODAL_VISION_URL, json=payload, timeout=180)
    resp.raise_for_status()
    return resp.json()["text"]


def _detect_openbmb(pil) -> str:
    """Call OpenBMB's free hosted MiniCPM-V API (OpenAI-compatible). No GPU."""
    import base64
    import io

    import httpx

    buf = io.BytesIO()
    pil.save(buf, format="JPEG")
    data_uri = "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode()
    payload = {
        "model": OPENBMB_API_MODEL,
        "messages": [{"role": "user", "content": [
            {"type": "text", "text": DETECT_PROMPT},
            {"type": "image_url", "image_url": {"url": data_uri}},
        ]}],
    }
    resp = httpx.post(OPENBMB_API_URL, json=payload, timeout=120,
                      headers={"Authorization": f"Bearer {OPENBMB_API_KEY}"})
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]


def detect_ingredients(image) -> tuple[list[str], str]:
    """Return (ingredient_names, source_note). Never raises — vision failures
    degrade to an empty list with an explanatory note so the UI stays usable."""
    if image is None:
        return [], "No photo provided."
    if MOCK_VISION:
        return list(_MOCK_DETECTION), "🔬 mock vision (deploy on GPU for a real read)"
    try:
        pil = _to_image(image)
        if VISION_BACKEND == "openbmb":
            raw = _detect_openbmb(pil)
        elif VISION_BACKEND == "modal":
            raw = _detect_modal(pil)
        else:
            raw = _detect_zerogpu(pil)
        names = _parse_list(raw)
        return names, f"👁️ detected by MiniCPM-V ({VISION_BACKEND})"
    except Exception as exc:
        return [], f"Vision unavailable ({type(exc).__name__}: {exc}). Type your pantry instead."