app.py

"""DuplexJev-32B-Vision on ZeroGPU: judge what it hears against what it sees, in one forward pass."""
import os
import time

import gradio as gr
import numpy as np
import spaces

MODEL_ID = os.environ.get("DUPLEXJEV_VISION_MODEL", "adventists-ai/DuplexJev-32B-Vision")
N_PERM = int(os.environ.get("DUPLEXJEV_NPERM", "2"))  # answers averaged over both option orders, as on the model card
MOCK = os.environ.get("DUPLEXJEV_MOCK") == "1"
SR = 16000

if MOCK:
    class _Mock:
        def decide(self, audio=None, image=None, questions=None, n_perm=2):
            rng = np.random.default_rng(len(questions))
            out = {}
            for q in questions:
                p = rng.dirichlet(np.ones(len(q["options"])))
                out[q["id"]] = {"answer": q["options"][int(p.argmax())], "confidence": float(p.max()),
                                "probs": dict(zip(q["options"], map(float, p)))}
            return out
    VD = _Mock()
else:
    from duplexjev.vision import VisionDecider
    VD = VisionDecider.from_pretrained(MODEL_ID, device="cuda")  # on cuda at import time, as ZeroGPU expects

PRESETS = {
    "Face ↔ voice · 表情和语气": [
        ("face matches voice", "Does the face in the picture show the same emotion as the speaker's voice?", ["Yes", "No"]),
        ("face emotion", "What emotion does the face in the picture show?", ["happy", "angry", "sad"]),
        ("voice emotion", "What is the speaker's emotional state?", ["neutral", "happy", "angry", "sad"]),
        ("speaker gender", "What is the perceived gender of the speaker?", ["female", "male"]),
    ],
    "Red word ↔ stress · 标红词和重读": [
        ("red word stressed", "Is the word in red in the picture the word the speaker stresses?", ["Yes", "No"]),
    ],
    "Caption ↔ sound · 文字说明和声音": [
        ("caption matches", "Does the caption in the picture describe the sound you hear?", ["Yes", "No"]),
        ("sound", "Besides speech, which sound can be heard in this audio?",
         ["laughter", "breathing", "coughing", "a sigh", "None of these"]),
    ],
}


def preset_text(name):
    return "\n".join(f"{i} | {q} | " + "; ".join(o) for i, q, o in PRESETS[name])


def parse(text):
    qs = []
    for n, line in enumerate(l for l in text.splitlines() if l.strip()):
        parts = [p.strip() for p in line.split("|")]
        if len(parts) < 3:
            raise gr.Error(f"Line {n + 1}: use  name | question | option 1; option 2; …")
        opts = [o.strip() for o in parts[2].split(";") if o.strip()]
        if len(opts) < 2:
            raise gr.Error(f"Line {n + 1}: give at least two options, separated by ';'")
        qs.append({"id": parts[0] or f"q{n + 1}", "text": parts[1], "options": opts})
    if not qs:
        raise gr.Error("Add at least one question.")
    if len(qs) > 8:
        raise gr.Error("At most 8 questions per request in this demo.")
    return qs


def _to16k(audio):
    from math import gcd
    from scipy.signal import resample_poly
    sr, x = audio
    x = np.asarray(x)
    if x.dtype.kind == "i":
        x = x.astype(np.float32) / float(np.iinfo(x.dtype).max + 1)
    x = x.astype(np.float32)
    if x.ndim == 2:
        x = x.mean(axis=1)
    if sr != SR:
        g = gcd(int(sr), SR)
        x = resample_poly(x, SR // g, int(sr) // g).astype(np.float32)
    return x[: SR * 15]


@spaces.GPU(duration=10, size="xlarge")  # 33 B in bf16 needs the full 96 GB card; a short reservation keeps free runs
def _run(x, img, qs):
    t = time.perf_counter()
    res = VD.decide(audio=x, image=img, questions=qs) if MOCK else VD.decide(audio=x, image=img, questions=qs, n_perm=N_PERM)
    return res, (time.perf_counter() - t) * 1000


def decide(image, audio, text):
    if image is None and audio is None:
        raise gr.Error("Add a picture and/or a voice clip (or pick an example).")
    x = _to16k(audio) if audio is not None else None
    qs = parse(text)
    res, ms = _run(x, image, qs)
    rows = []
    for q in qs:
        r = res[q["id"]]
        others = " · ".join(f"{o} {p:.0%}" for o, p in sorted(r["probs"].items(), key=lambda kv: -kv[1]) if o != r["answer"])
        rows.append([q["id"], r["answer"], f"{r['confidence']:.0%}", others])
    head = f"### {len(qs)} decisions about the voice + picture · 0 decode steps · **{ms:.0f} ms** on the GPU"
    return head, rows


R = "examples/"
EXAMPLES = [
    [R + "rp_1072_TSI_HAP_XX.jpg", R + "rp_1072_TSI_HAP_XX.wav", "Face ↔ voice · 表情和语气"],
    [R + "rp_1072_TSI_SAD_XX.jpg", R + "rp_1072_TSI_HAP_XX.wav", "Face ↔ voice · 表情和语气"],
    [R + "rp_1072_TSI_ANG_XX.jpg", R + "rp_1072_TSI_ANG_XX.wav", "Face ↔ voice · 表情和语气"],
    [R + "rp_1085_WSI_HAP_XX.jpg", R + "rp_1085_WSI_ANG_XX.wav", "Face ↔ voice · 表情和语气"],
    [R + "rp_1074_MTI_SAD_XX.jpg", R + "rp_1074_MTI_SAD_XX.wav", "Face ↔ voice · 表情和语气"],
    [R + "rp_1078_ITS_HAP_XX.jpg", R + "rp_1078_ITS_SAD_XX.wav", "Face ↔ voice · 表情和语气"],
    [R + "v11_st_red_he.png", R + "v11_st_he.wav", "Red word ↔ stress · 标红词和重读"],
    [R + "v11_st_red_money.png", R + "v11_st_he.wav", "Red word ↔ stress · 标红词和重读"],
    [R + "v11_cap_laugh.png", R + "v11_laugh.wav", "Caption ↔ sound · 文字说明和声音"],
    [R + "v11_cap_laugh.png", R + "v11_cough.wav", "Caption ↔ sound · 文字说明和声音"],
]
LABELS = ["real photo · happy face + happy voice", "same person, sad face + happy voice", "real photo · angry face + angry voice",
          "happy face + angry voice", "real photo · sad face + sad voice", "happy face + sad voice",
          "“I didn't say HE…” · red word: he", "“I didn't say HE…” · red word: money", "caption “laughter” + laughter",
          "caption “laughter” + a cough"]

INTRO = """
# 👁️🎧 DuplexJev-32B-Vision — it hears, and now it sees
**One forward pass over a voice and a picture.** Ask multiple-choice questions that need both: *does the face match the
voice? is the red word the one they stress? does the caption describe the sound?* Each answer is one constrained token —
nothing is decoded. Research preview · [model card](https://huggingface.co/adventists-ai/DuplexJev-32B-Vision) ·
⭐ [GitHub](https://github.com/adventists-ai/duplexjev) · speech-only demo: [DuplexJev](https://huggingface.co/spaces/BitKnight/DuplexJev) ·
🏢 commercial licence & hosted API: **jiejin@adventists.ai**

<sub>⏱️ The first run after the Space has been idle takes a few seconds while Hugging Face attaches a GPU. Free GPU time per visitor
is limited by Hugging Face — [log in](https://huggingface.co/login) for more runs. 首次运行需要几秒分配 GPU；登录 Hugging Face 可获得更多免费次数。</sub>
"""

FOOT = """
<sub>Model: DuplexJev-32B-Vision (CC BY-NC 4.0, research preview); answers averaged over both option orders. Questions work best in English;
speech can be English or Chinese. One picture per request. Real-photo clips: CREMA-D (ODbL / DbCL), actors not seen in training.
Pictures and audio are processed for this request only and not stored. Do not use the outputs to make decisions about individuals.</sub>
"""

with gr.Blocks(title="DuplexJev-32B-Vision — voice + picture decisions", theme=gr.themes.Soft(primary_hue="emerald")) as demo:
    gr.Markdown(INTRO)
    with gr.Row():
        with gr.Column(scale=5):
            with gr.Row():
                image = gr.Image(type="pil", sources=["upload", "webcam", "clipboard"], label="Picture", height=260)
                audio = gr.Audio(sources=["microphone", "upload"], type="numpy", label="Voice (≤ 15 s)")
            preset = gr.Radio(list(PRESETS), value="Face ↔ voice · 表情和语气", label="Question set")
            with gr.Accordion("Questions (editable): name | question | option 1; option 2; …", open=False):
                table = gr.Textbox(value=preset_text("Face ↔ voice · 表情和语气"), lines=6, show_label=False)
            btn = gr.Button("Decide ⚡", variant="primary")
            gr.Examples(EXAMPLES, inputs=[image, audio, preset], example_labels=LABELS, label="Examples")
        with gr.Column(scale=5):
            head = gr.Markdown("### Results appear here")
            out = gr.Dataframe(headers=["decision", "answer", "confidence", "other options"], wrap=True,
                               interactive=False, column_widths=["24%", "26%", "14%", "36%"])
    gr.Markdown(FOOT)
    preset.change(preset_text, preset, table)
    btn.click(decide, [image, audio, table], [head, out], api_name="decide")

if __name__ == "__main__":
    demo.queue(max_size=40).launch()