"""MOSS-VL-Realtime Space — realtime session demo.

Layout: ONE unified media stage (left: upload video/image, or live camera feed,
with a live status badge overlaid at its top-left) · chat log with chat/raw
toggle (right top) · session console (right bottom). A session binds the staged
media, streams it frame by frame (uploads at wall-clock pace, live camera as it
arrives), and the input box sends prompts INTO the running stream (Enter to
send; disabled when no session is live). Chat bubbles are sentence-segmented
(ending punctuation, ~64-token cap); the raw view keeps whole model rounds.
UI language toggles between English and Chinese (header switch button).
"""

import os
import re
import time
import uuid

import gradio as gr

from inference import (
    MOCK,
    Mailbox,
    RoundParser,
    analyze,
    classify_media,
    extract_frames,
    gpu_session,
    load_image_frame,
)

CHAT_TICK_S = 0.1
RAW_TICK_S = 0.5
CAM_STREAM_EVERY_S = 0.5  # live-camera browser capture cadence (~2 fps)

MOCK_BADGE = "\n\n`MOCK mode — scripted model responses`"

# --- Chat bubble segmentation -------------------------------------------------
# One bubble ends at sentence-ending punctuation; a bubble that reaches
# ~BUBBLE_TOKEN_CAP tokens without one is cut at the punctuation nearest the
# cap, or hard-cut at the cap as a last resort.
BUBBLE_TOKEN_CAP = 64
_SENT_END_CJK = "。！？…"
_SENT_END_LAT = ".!?"
_CLOSERS = "”’』」）)】]\"'"
_CUT_PUNCT = "，、；：,;:—–…" + _SENT_END_CJK + _SENT_END_LAT + _CLOSERS
_CJK_RE = re.compile(r"[぀-ヿ㐀-䶿一-鿿가-힯]")


def _cap_pos(text, cap):
    """Index where `text` exceeds ~`cap` tokens (CJK char ≈ 1, latin word ≈ 1), or None."""
    tokens = 0
    in_word = False
    for i, ch in enumerate(text):
        if _CJK_RE.match(ch):
            in_word = False
            tokens += 1
        elif ch.isalnum():
            if not in_word:
                in_word = True
                tokens += 1
        else:
            in_word = False
        if tokens > cap:
            return i
    return None


def _bubble_cut(text):
    """Index to close the open chat bubble at, or None to keep accumulating."""
    for i, ch in enumerate(text):
        if ch in _SENT_END_CJK or ch in _SENT_END_LAT:
            nxt = text[i + 1 : i + 2]
            if ch in _SENT_END_LAT:
                if not nxt:
                    break  # can't confirm yet (decimals, "e.g.") — wait for lookahead
                if nxt.isalnum():
                    continue  # 3.14 / e.g. / mid-word — not a sentence end
            cut = i + 1
            while cut < len(text) and text[cut] in _CLOSERS:
                cut += 1  # keep closing quotes/brackets with their sentence
            return cut
    pos = _cap_pos(text, BUBBLE_TOKEN_CAP)
    if pos is None:
        return None
    window = text[:pos]
    best = max((window.rfind(c) for c in _CUT_PUNCT), default=-1)
    return best + 1 if best > 0 else pos


# UI language packs. Radio choices use (label, value) pairs so switching
# language changes only the display labels — values stay stable and no
# .change events fire.
LANGS = {
    "en": {
        "btn": "中文",
        "title": "# MOSS-VL-Realtime — Live Session\n\n"
                 "Stage a video/image or go live with your camera, **start a session**, "
                 "and chat with the model *while it watches the stream*.",
        "mode_upload": "📁 Upload",
        "mode_camera": "📷 Live camera",
        "file_label": "Video or image",
        "hint": "<small>One media per session — changing it ends the live session.</small>",
        "clear_media": "Clear media",
        "view_chat": "💬 Chat",
        "view_raw": "{ } Raw",
        "idle_ph": "Start a session to ask…",
        "live_ph": "Ask about the stream, Enter to send…",
        "start": "▶ Start session",
        "stop": "⏹ Stop session",
        "clear_msgs": "🗑 Clear messages",
        "adv": "Advanced Settings",
        "max_tokens": "Max New Tokens",
        "temperature": "Temperature",
        "top_p": "Top-p",
        "rep_pen": "Repetition Penalty",
        "fps": "Video FPS",
        "max_frames": "Max Frames",
        "speed": "Playback speed",
        "ff": "Fast-forward",
    },
    "zh": {
        "btn": "English",
        "title": "# MOSS-VL-Realtime — 实时会话\n\n"
                 "上传视频/图片或打开摄像头，**开始会话**，边看边与模型对话。",
        "mode_upload": "📁 上传",
        "mode_camera": "📷 实时摄像头",
        "file_label": "视频或图片",
        "hint": "<small>每次会话绑定一个媒体，更换将结束当前会话。</small>",
        "clear_media": "清除媒体",
        "view_chat": "💬 对话",
        "view_raw": "{ } 原始",
        "idle_ph": "开始会话后可提问…",
        "live_ph": "边看边问，回车发送…",
        "start": "▶ 开始会话",
        "stop": "⏹ 结束会话",
        "clear_msgs": "🗑 清除消息",
        "adv": "高级设置",
        "max_tokens": "最大生成长度",
        "temperature": "温度",
        "top_p": "Top-p",
        "rep_pen": "重复惩罚",
        "fps": "视频采样帧率",
        "max_frames": "最大帧数",
        "speed": "流速",
        "ff": "快进",
    },
}


# --- Small helpers ---


def _status_html(text):
    return f'<div class="moss-status">{text}</div>' if text else ""


def _append_raw(events, event):
    """Append a raw-view event, coalescing consecutive silences."""
    if event.get("event") == "silence" and events and events[-1].get("event") == "silence":
        events[-1]["count"] += 1
        events[-1]["video_ts"] = event["video_ts"]
    else:
        events.append(event)


def _file_path(value):
    """Normalize a gr.File value (str | list | dict | None) to a path or None."""
    if isinstance(value, list):
        value = value[0] if value else None
    if isinstance(value, dict):
        value = value.get("path") or value.get("name")
    return value or None


# --- Session orchestrator (main process; iterates the GPU session) ---


def run_session(
    stage_mode,
    stage_file,
    staged_media,
    initial_text,
    prev_sid,
    lang,
    max_new_tokens,
    temperature,
    top_p,
    repetition_penalty,
    video_fps,
    max_frames,
    playback_speed,
):
    L = LANGS.get(lang) or LANGS["en"]
    skip = (gr.skip(),) * 12
    idle_box = gr.update(interactive=False, placeholder=L["idle_ph"])
    # messages persist after a session -> console shows "Clear messages" first
    btns_ended = (gr.update(visible=False), gr.update(visible=False), gr.update(visible=True))

    live = stage_mode == "camera"
    if live:
        kind, path = "live", None
    else:
        media = staged_media
        if not media:
            p = _file_path(stage_file)
            media = {"kind": classify_media(p), "path": p} if p else None
        if not media:
            gr.Warning("Stage a video or image first · 请先上传视频或图片")
            yield skip
            return
        kind, path = media["kind"], media["path"]

    if prev_sid and Mailbox.is_live(prev_sid):
        gr.Warning("A session is already live · 会话进行中")
        yield skip
        return

    Mailbox.cleanup_stale()
    sid = uuid.uuid4().hex[:12]
    Mailbox.create(sid)
    if live:
        Mailbox.mark_live_camera(sid)  # lets the webcam stream handler start mailing frames

    initial_text = (initial_text or "").strip()
    if live:
        history = [{"role": "user", "content": "📷 Live camera session · 实时摄像头会话"}]
        status = "🔴 Going live · 开始直播…"
    else:
        history = [{"role": "user", "content": {"path": path}}]
        status = "🎞 Preparing the stream · 准备视频流…"
    events = [{"event": "session_start", "media": "live-camera" if live else os.path.basename(path), "kind": kind}]

    # New session owns the display: clear both views, flip the console live,
    # and surface the staged media in the preview.
    if live:
        stage_updates = (gr.skip(), gr.skip(), gr.skip(), gr.skip())
    else:
        stage_updates = (
            gr.update(visible=False),
            gr.update(visible=kind == "video", value=path if kind == "video" else None),
            gr.update(visible=kind == "image", value=path if kind == "image" else None),
            {"kind": kind, "path": path},
        )
    yield (
        history,
        events,
        _status_html(status),
        gr.update(value="", interactive=True, placeholder=L["live_ph"]),
        gr.update(visible=False),
        gr.update(visible=True, interactive=True),
        gr.update(visible=False),
        sid,
        *stage_updates,
    )

    frames = None
    if not live:
        try:
            if kind == "image":
                frames = load_image_frame(path)
            else:
                frames = extract_frames(path, video_fps, max_frames)
        except Exception as exc:
            Mailbox.cleanup(sid)
            history.append({"role": "assistant", "content": f"⚠️ Could not decode the media: {exc}"})
            _append_raw(events, {"event": "error", "message": str(exc)})
            yield (history, events, _status_html(""), idle_box, *btns_ended, None,
                   gr.skip(), gr.skip(), gr.skip(), gr.skip())
            return

    gen_kwargs = {
        "max_new_tokens": int(max_new_tokens),
        "temperature": float(temperature),
        "top_p": float(top_p),
        "repetition_penalty": float(repetition_penalty),
        "do_sample": float(temperature) > 0.0,
    }

    parser = RoundParser()
    bubble_idx = None  # open, sentence-segmented chat bubble
    round_ev = None  # rolling raw event for the open round (streams text live)
    frame_ev = None  # single rolling raw event for frame progress
    end_reason = "session ended"
    last_chat = last_raw = 0.0
    chat_dirty = raw_dirty = status_dirty = False

    def set_status(text):
        nonlocal status, status_dirty
        if text != status:
            status = text
            status_dirty = True

    def _open_bubble():
        history.append({"role": "assistant", "content": ""})
        return len(history) - 1

    def _stamp(idx, ts):
        history[idx]["content"] = history[idx]["content"].rstrip() + f"\n\n⏱ {ts:.1f}s"

    def _feed_bubble(delta, ts):
        """Stream text into the open bubble, splitting at sentence ends / cap.

        Round boundaries do NOT reach this — the real model re-emits its
        response marker every frame mid-utterance, so only the punctuation /
        token-cap rules (and a real round end) may close a bubble. Every
        closed bubble is stamped with the video timestamp of its close.
        """
        nonlocal bubble_idx, chat_dirty
        if not delta:
            return
        if bubble_idx is None:
            bubble_idx = _open_bubble()
        text = history[bubble_idx]["content"] + delta
        while True:
            cut = _bubble_cut(text)
            if cut is None:
                break
            closed, rest = text[:cut], text[cut:].lstrip()
            history[bubble_idx]["content"] = closed
            _stamp(bubble_idx, ts)
            if rest:
                bubble_idx = _open_bubble()
                text = rest
            else:
                bubble_idx = None
                text = ""
                break
        if bubble_idx is not None:
            history[bubble_idx]["content"] = text
        chat_dirty = True

    def handle_ops(ops, ts):
        nonlocal bubble_idx, round_ev, chat_dirty, raw_dirty
        for op, payload in ops:
            if op == "round_open":
                round_ev = {"event": "round", "video_ts": payload, "text": ""}
                events.append(round_ev)
                raw_dirty = True
            elif op == "round_break":
                # raw view keeps one record per model round; the chat bubble
                # flows straight through the boundary
                round_ev = {"event": "round", "video_ts": payload, "text": ""}
                events.append(round_ev)
                raw_dirty = True
            elif op == "text":
                _feed_bubble(payload, ts)
                if round_ev is not None:
                    round_ev["text"] += payload
                    raw_dirty = True
            elif op == "round_close":
                if bubble_idx is not None:
                    _stamp(bubble_idx, payload)
                    bubble_idx = None
                    chat_dirty = True
                if round_ev is not None:
                    round_ev["video_ts"] = payload
                    round_ev = None
                    raw_dirty = True
            elif op == "silence":
                _append_raw(events, {"event": "silence", "video_ts": payload, "count": 1})
                raw_dirty = True
            elif op == "control":
                _append_raw(events, {"event": "control", "token": payload})
                raw_dirty = True

    for ev in gpu_session(sid, frames, initial_text or None, gen_kwargs, playback_speed, live=live):
        etype = ev["type"]
        ts = ev.get("video_ts", 0.0)

        if etype == "session_start":
            _append_raw(
                events,
                {"event": "gpu_session_start", "frames_total": ev["frames_total"], "budget_s": ev["budget_s"], "live": ev.get("live", False)},
            )
            raw_dirty = True
        elif etype == "prompt":
            history.append({"role": "user", "content": ev["text"]})
            _append_raw(events, {"event": "prompt", "video_ts": ts, "text": ev["text"]})
            chat_dirty = raw_dirty = True
        elif etype in ("frame", "chunk_batch"):
            for chunk in ev.get("chunks", []):
                handle_ops(parser.feed(chunk, ts), ts)
            if etype == "frame":
                if ev.get("total"):
                    set_status(f"👀 Observing · t={ts:.1f}s ({ev['frame']}/{ev['total']} frames)")
                else:
                    set_status(f"🔴 Live · t={ts:.1f}s ({ev['frame']} frames)")
                if frame_ev is None:
                    frame_ev = {"event": "frames", "processed": 0, "total": ev.get("total") or "live", "video_ts": 0.0, "dropped": 0}
                    events.append(frame_ev)
                frame_ev.update(processed=ev["frame"], video_ts=ts, dropped=ev["dropped_frames"])
                raw_dirty = True
        elif etype == "tick":
            # idle heartbeat: advance the live clock and let pending updates flush
            if live:
                set_status(f"🔴 Live · t={ts:.1f}s ({ev.get('frames', 0)} frames)")
        elif etype == "postroll":
            set_status("✅ Stream ended — ask follow-ups · 流结束，可继续提问")
            _append_raw(events, {"event": "postroll", "video_ts": ts})
            raw_dirty = True
        elif etype == "session_end":
            end_reason = ev["reason"]
        elif etype == "error":
            history.append({"role": "assistant", "content": f"⚠️ {ev['message']}"})
            _append_raw(events, {"event": "error", "message": ev["message"]})
            end_reason = "error"
            chat_dirty = raw_dirty = True

        now = time.monotonic()
        flush_chat = (chat_dirty or status_dirty) and now - last_chat >= CHAT_TICK_S
        flush_raw = raw_dirty and now - last_raw >= RAW_TICK_S
        if flush_chat or flush_raw:
            if flush_raw:
                last_raw, raw_dirty = now, False
            h_out = s_out = gr.skip()
            if flush_chat:
                last_chat = now
                if chat_dirty:
                    h_out, chat_dirty = history, False
                if status_dirty:
                    s_out, status_dirty = _status_html(status), False
            yield (
                h_out,
                events if flush_raw else gr.skip(),
                s_out,
                gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip(),
                gr.skip(), gr.skip(), gr.skip(), gr.skip(),
            )

    if not any(m["role"] == "assistant" and m["content"] for m in history):
        history.append(
            {"role": "assistant", "content": "🤫 The model stayed silent for this stream. · 模型全程保持沉默。"}
        )
    # session-end marker: a plain message, not a metadata-titled tag bubble
    history.append(
        {"role": "assistant", "content": f"▣ Session ended · {end_reason} · 会话结束"}
    )
    _append_raw(events, {"event": "session_end", "reason": end_reason})

    yield (history, events, _status_html(""), idle_box, *btns_ended, None,
           gr.skip(), gr.skip(), gr.skip(), gr.skip())


# --- Console + stage handlers ---


def send_prompt(text, sid):
    text = (text or "").strip()
    if not text:
        return gr.skip()
    if not sid or not Mailbox.is_live(sid) or Mailbox.should_stop(sid):
        gr.Warning("No live session · 当前没有进行中的会话")
        return gr.skip()
    Mailbox.write_prompt(sid, text)
    return ""


def signal_stop(sid):
    if sid and Mailbox.is_live(sid):
        Mailbox.signal_stop(sid)
        return gr.update(interactive=False)
    return gr.skip()


def clear_messages():
    # messages wiped -> hand the console back to Start
    return [], [], gr.update(visible=True), gr.update(visible=False)


def sync_console(history):
    """Post-session button reconciliation (start_evt.then).

    The generator's last yield can be dropped client-side when another event
    (a prompt submit) interleaved with the stream — re-assert the console
    state once the session event has fully completed: messages present means
    "Clear messages", an untouched log means "Start session".
    """
    ended_with_msgs = bool(history)
    return (
        gr.update(visible=not ended_with_msgs),
        gr.update(visible=False),
        gr.update(visible=ended_with_msgs),
    )


def on_media_changed(sid):
    # Media is bound for the whole session — changing it ends the session.
    if sid and Mailbox.is_live(sid) and not Mailbox.should_stop(sid):
        Mailbox.signal_stop(sid)
        gr.Info("Session ended — media changed · 媒体已更换，会话结束")


def on_file_staged(value, sid):
    on_media_changed(sid)
    path = _file_path(value)
    if not path:
        return None, gr.update(visible=True), gr.update(visible=False, value=None), gr.update(visible=False, value=None)
    kind = classify_media(path)
    return (
        {"kind": kind, "path": path},
        gr.update(visible=False),
        gr.update(visible=kind == "video", value=path if kind == "video" else None),
        gr.update(visible=kind == "image", value=path if kind == "image" else None),
    )


def on_file_cleared(sid):
    on_media_changed(sid)
    return None, gr.update(visible=True, value=None), gr.update(visible=False, value=None), gr.update(visible=False, value=None)


def on_mode_change(mode, sid):
    on_media_changed(sid)
    live = mode == "camera"
    return (
        None,  # staged media reset on mode switch
        gr.update(visible=not live, value=None),
        gr.update(visible=False, value=None),
        gr.update(visible=False, value=None),
        gr.update(visible=live),
    )


def on_cam_frame(frame, sid):
    """Browser webcam tick: mail the frame to the GPU worker when live."""
    if frame is None or not sid:
        return
    if not Mailbox.is_live_camera(sid) or Mailbox.should_stop(sid):
        return
    from PIL import Image

    Mailbox.write_frame(sid, Image.fromarray(frame))


def clear_stages(sid):
    on_media_changed(sid)
    return (
        "upload",
        None,
        gr.update(visible=True, value=None),
        gr.update(visible=False, value=None),
        gr.update(visible=False, value=None),
        gr.update(visible=False),
    )


def toggle_view(choice):
    chat = choice == "chat"
    return gr.update(visible=chat), gr.update(visible=not chat)


def on_chat_clear():
    return []


def switch_lang(lang, sid):
    lang = "zh" if lang == "en" else "en"
    L = LANGS[lang]
    in_session = bool(sid and Mailbox.is_live(sid))
    ph = L["live_ph"] if in_session else L["idle_ph"]
    return (
        lang,
        gr.update(value=L["btn"]),
        L["title"] + (MOCK_BADGE if MOCK else ""),
        gr.update(choices=[(L["mode_upload"], "upload"), (L["mode_camera"], "camera")]),
        gr.update(label=L["file_label"]),
        L["hint"],
        gr.update(value=L["clear_media"]),
        gr.update(choices=[(L["view_chat"], "chat"), (L["view_raw"], "raw")]),
        gr.update(placeholder=ph),
        gr.update(value=L["start"]),
        gr.update(value=L["stop"]),
        gr.update(value=L["clear_msgs"]),
        gr.update(label=L["adv"]),
        gr.update(label=L["max_tokens"]),
        gr.update(label=L["temperature"]),
        gr.update(label=L["top_p"]),
        gr.update(label=L["rep_pen"]),
        gr.update(label=L["fps"]),
        gr.update(label=L["max_frames"]),
        gr.update(label=L["speed"], choices=[("1×", "1×"), ("2×", "2×"), (L["ff"], "Fast-forward")]),
    )


# --- UI ---

CSS = """
#col-container { max-width: 1250px; margin: 0 auto; --console-h: 46px; --stage-h: 620px; }
#links-row { align-items: center; }
#links-md { flex-grow: 1; }
#lang-btn { flex-grow: 0 !important; min-width: 84px; }
/* the two columns mirror each other: [group: toggle bar + stage] + bottom row.
   The message stage is pinned to --stage-h (chat scrolls inside — it must not
   grow with content); the media stage stretches to match so the toggle bars
   and the bottom rows stay strictly aligned across columns */
#message-stage {
  flex-grow: 0 !important; height: var(--stage-h); min-height: var(--stage-h); max-height: var(--stage-h);
  display: flex; flex-direction: column;
}
#media-stage { flex-grow: 1 !important; display: flex; flex-direction: column; min-height: 0; position: relative; }
/* the group nests gr-group > gr-group > .styler — every layer must stretch */
#media-stage > *, #media-stage .styler,
#message-stage > *, #message-stage .styler {
  flex-grow: 1 !important; display: flex; flex-direction: column; min-height: 0;
}
#stage-file, #preview-video, #preview-image, #live-cam { flex-grow: 1 !important; min-height: 0; }
/* chat/raw fill the fixed stage and scroll internally — never grow with content */
#chatbot, #raw-json {
  flex-grow: 1 !important; min-height: 0 !important; height: auto !important; max-height: none !important;
}
#stage-file .boundedheight, #preview-video .boundedheight, #preview-image .boundedheight, #live-cam .boundedheight,
#stage-file [data-testid="file"], #preview-video video, #preview-image img, #live-cam img {
  height: 100% !important; max-height: none !important;
}
#stage-hint { flex-grow: 0 !important; }
/* session status badge overlaid at the media stage's top-left (below the
   segmented toggle bar); empty when no session is live */
#status-line { position: absolute; top: 48px; left: 12px; z-index: 30; flex: none !important; }
#status-line .moss-status {
  display: inline-block; background: rgba(17, 17, 17, 0.72); color: #ffd34d;
  font-size: 0.82rem; line-height: 1.5; padding: 3px 12px; border-radius: 999px;
  pointer-events: none; backdrop-filter: blur(2px); white-space: nowrap;
}
/* segmented toggle bars (media source / chat-raw view): two half-width buttons,
   centered text, no radio dot — black = off, accent yellow = on */
#stage-mode, #view-toggle { flex-grow: 0 !important; flex-shrink: 0 !important; min-height: fit-content; }
#stage-mode .wrap, #view-toggle .wrap { display: flex; flex-direction: row; gap: 0; width: 100%; }
#stage-mode label, #view-toggle label {
  flex: 1 1 50%; justify-content: center; text-align: center; margin: 0;
  padding: var(--spacing-sm) 0; cursor: pointer; border-radius: 0;
  background: #141414; color: #9aa0a6; border: 1px solid var(--border-color-primary);
  transition: background 0.15s, color 0.15s;
}
#stage-mode label:first-child, #view-toggle label:first-child { border-radius: var(--radius-md) 0 0 var(--radius-md); }
#stage-mode label:last-child, #view-toggle label:last-child { border-radius: 0 var(--radius-md) var(--radius-md) 0; }
#stage-mode label.selected, #view-toggle label.selected {
  background: var(--color-accent); color: #111; font-weight: 600;
}
#stage-mode input[type="radio"], #view-toggle input[type="radio"] { display: none; }
/* bottom rows sit on the same baseline with the same height: the clear-media
   button (left) and the prompt console (right) share --console-h */
#clear-media, #session-console { flex-grow: 0 !important; margin-top: auto; }
#clear-media { height: var(--console-h); min-height: var(--console-h); }
#session-console { align-items: stretch; }
#session-console button { white-space: nowrap; }
#session-console .form {
  flex: 1 1 auto; border: none !important; background: transparent !important; box-shadow: none !important;
}
/* prompt box: bare input with the send button INSIDE its right edge — no grey
   padded wrapper around the input */
#prompt-box { height: var(--console-h); }
#prompt-box .input-container { position: relative; height: var(--console-h); }
#prompt-box textarea {
  height: var(--console-h) !important; min-height: var(--console-h) !important;
  max-height: var(--console-h) !important; resize: none;
  /* single 20px line centered inside the 46px bar (13 + 20 + 13);
     long input scrolls with no visible scrollbar */
  padding: 13px 2.8rem 13px 14px; line-height: 20px;
  overflow-y: auto; scrollbar-width: none;
}
#prompt-box textarea::-webkit-scrollbar { display: none; width: 0; }
#prompt-box .submit-button { position: absolute; right: 8px; top: 50%; transform: translateY(-50%); }
#start-btn, #stop-btn, #clear-btn { height: var(--console-h) !important; min-height: var(--console-h) !important; }
/* live camera: no record/stop controls — the feed auto-starts (JS below) and
   frames only reach the model while a session is live (backend-gated) */
#live-cam .button-wrap { display: none !important; }
/* scrollbars: no rail is drawn; the thumb is invisible until the pointer
   enters the rail strip at the scrollable edge (.sb-hot, set by the page
   script) or grabs the thumb itself. scrollbar-color drives modern
   Chrome/Firefox; the ::-webkit rules cover older WebKit. */
* { scrollbar-width: thin; scrollbar-color: transparent transparent; }
.sb-hot { scrollbar-color: rgba(128, 128, 128, 0.55) transparent !important; }
::-webkit-scrollbar { width: 10px; height: 10px; background: transparent !important; }
::-webkit-scrollbar-track, ::-webkit-scrollbar-corner { background: transparent !important; }
::-webkit-scrollbar-button { display: none !important; width: 0 !important; height: 0 !important; }
::-webkit-scrollbar-thumb { background: transparent; border-radius: 5px; }
.sb-hot::-webkit-scrollbar-thumb,
::-webkit-scrollbar-thumb:hover, ::-webkit-scrollbar-thumb:active { background: rgba(128, 128, 128, 0.6) !important; }
"""

# Page-load JS: (1) auto-start the live camera preview whenever the camera
# stage is visible — grant-access overlay first, then the (hidden) Record
# toggle; clicking only the exact "Record" state can never stop a running
# stream. __mossCamHold (a timestamp) pauses the watcher from the moment a
# clear/mode-switch-away fires until the camera stage is actually hidden —
# otherwise the watcher could re-acquire the camera in the gap before the
# server hides the component. The hold expires after 5s as a backstop.
# (2) keep the raw event view pinned to the bottom while it streams, unless
# the user has scrolled up to read.
PAGE_JS = """
() => {
  if (!window.__mossSbHot) {
    window.__mossSbHot = true;
    let hot = null;
    document.addEventListener('mousemove', (e) => {
      let el = e.target instanceof Element ? e.target : null;
      let found = null;
      while (el && el !== document.documentElement) {
        const vs = el.scrollHeight > el.clientHeight + 4;
        const hs = el.scrollWidth > el.clientWidth + 4;
        const cs = vs || hs ? getComputedStyle(el) : null;
        if (cs && /(auto|scroll)/.test(cs.overflowY + cs.overflowX)) {
          const r = el.getBoundingClientRect();
          if ((vs && r.right - e.clientX <= 18) || (hs && r.bottom - e.clientY <= 18)) found = el;
          break;
        }
        el = el.parentElement;
      }
      if (found !== hot) {
        if (hot) hot.classList.remove('sb-hot');
        if (found) found.classList.add('sb-hot');
        hot = found;
      }
    }, {passive: true});
  }
  if (window.__mossCamWatch) return;
  window.__mossCamWatch = setInterval(() => {
    if (!window.__mossRawScroll) {
      const rawRoot = document.querySelector('#raw-json');
      if (rawRoot) {
        window.__mossRawScroll = new MutationObserver(() => {
          if (rawRoot.offsetParent === null) return;
          const sc = [...rawRoot.querySelectorAll('*')].find(e => e.scrollHeight > e.clientHeight + 8);
          if (!sc) return;
          if (sc.scrollHeight - sc.scrollTop - sc.clientHeight < 240) sc.scrollTop = sc.scrollHeight;
        });
        window.__mossRawScroll.observe(rawRoot, {childList: true, subtree: true, characterData: true});
      }
    }
    const cam = document.querySelector('#live-cam');
    const visible = cam && cam.offsetParent !== null;
    if (window.__mossCamHold) {
      if (!visible) { window.__mossCamHold = 0; return; }
      if (Date.now() - window.__mossCamHold < 5000) return;
      window.__mossCamHold = 0;
    }
    if (!visible) return;
    const btns = [...cam.querySelectorAll('button')];
    const access = btns.find(b => (b.title + ' ' + b.textContent).includes('Webcam'));
    if (access) { access.click(); return; }
    const rec = btns.find(b => b.textContent.trim() === 'Record');
    if (rec) rec.click();
  }, 700);
}
"""

# Close the live feed (stop streaming + release the camera) before the server
# hides the component. Passed as the js half of clear/mode-switch events; must
# return the inputs unchanged for the python fn. mode_arg names the positional
# input carrying the stage-mode value: entering camera mode must NOT hold the
# watcher (the stage is about to become visible and should auto-start).
def _cam_stop_js(n_args, mode_arg=None):
    args = ", ".join(f"a{i}" for i in range(n_args))
    hold = (
        "window.__mossCamHold = Date.now();"
        if mode_arg is None
        else f"if (a{mode_arg} !== 'camera') window.__mossCamHold = Date.now();"
    )
    return f"""
({args}) => {{
  {hold}
  const cam = document.querySelector('#live-cam');
  if (cam) {{
    const stop = [...cam.querySelectorAll('button')].find(b => b.textContent.trim() === 'Stop');
    if (stop) stop.click();
    cam.querySelectorAll('video').forEach(v => {{
      if (v.srcObject) {{ v.srcObject.getTracks().forEach(t => t.stop()); v.srcObject = null; }}
    }});
  }}
  return [{args}];
}}
"""

with gr.Blocks(title="MOSS-VL-Realtime Demo") as demo:
    EN = LANGS["en"]
    with gr.Column(elem_id="col-container"):
        title_md = gr.Markdown(EN["title"] + (MOCK_BADGE if MOCK else ""))
        with gr.Row(elem_id="links-row"):
            gr.Markdown(
                "[Model Card](https://huggingface.co/OpenMOSS-Team/MOSS-VL-Realtime) | "
                "[GitHub](https://github.com/OpenMOSS/MOSS-VL)",
                elem_id="links-md",
            )
            lang_btn = gr.Button(EN["btn"], size="sm", scale=0, elem_id="lang-btn")

        with gr.Row(equal_height=True):
            with gr.Column(scale=5):
                with gr.Group(elem_id="media-stage"):
                    stage_mode = gr.Radio(
                        [(EN["mode_upload"], "upload"), (EN["mode_camera"], "camera")],
                        value="upload",
                        show_label=False,
                        container=False,
                        elem_id="stage-mode",
                    )
                    status_line = gr.HTML("", elem_id="status-line")
                    stage_file = gr.File(
                        file_types=["image", "video"],
                        label=EN["file_label"],
                        height=470,
                        elem_id="stage-file",
                    )
                    preview_video = gr.Video(
                        visible=False, interactive=False, height=470, show_label=False,
                        autoplay=False, elem_id="preview-video",
                    )
                    preview_image = gr.Image(
                        visible=False, interactive=False, height=470, show_label=False,
                        elem_id="preview-image",
                    )
                    live_cam = gr.Image(
                        sources=["webcam"],
                        streaming=True,
                        visible=False,
                        height=470,
                        show_label=False,
                        elem_id="live-cam",
                    )
                    stage_hint = gr.Markdown(EN["hint"], elem_id="stage-hint")
                clear_media_btn = gr.Button(
                    EN["clear_media"], size="sm", variant="secondary", elem_id="clear-media"
                )

            with gr.Column(scale=6):
                with gr.Group(elem_id="message-stage"):
                    view_toggle = gr.Radio(
                        [(EN["view_chat"], "chat"), (EN["view_raw"], "raw")],
                        value="chat",
                        show_label=False,
                        container=False,
                        elem_id="view-toggle",
                    )
                    chatbot = gr.Chatbot(
                        height=550,
                        show_label=False,
                        buttons=["copy"],
                        autoscroll=True,
                        group_consecutive_messages=False,  # one bubble per segmented sentence
                        elem_id="chatbot",
                    )
                    raw_json = gr.JSON(
                        value=[], show_label=False, visible=False, height=550, elem_id="raw-json"
                    )
                with gr.Row(elem_id="session-console"):
                    prompt_box = gr.Textbox(
                        scale=1,
                        lines=1,
                        interactive=False,
                        show_label=False,
                        container=False,
                        submit_btn=True,
                        placeholder=EN["idle_ph"],
                        elem_id="prompt-box",
                    )
                    start_btn = gr.Button(EN["start"], variant="primary", scale=0, elem_id="start-btn")
                    stop_btn = gr.Button(EN["stop"], variant="stop", visible=False, scale=0, elem_id="stop-btn")
                    clear_msgs_btn = gr.Button(
                        EN["clear_msgs"], variant="secondary", visible=False, scale=0, elem_id="clear-btn"
                    )

        with gr.Accordion(EN["adv"], open=False) as adv_acc:
            with gr.Row():
                max_new_tokens = gr.Slider(64, 4096, value=512, step=64, label=EN["max_tokens"])
                temperature = gr.Slider(0.0, 1.5, value=0.0, step=0.05, label=EN["temperature"])
            with gr.Row():
                top_p = gr.Slider(0.1, 1.0, value=1.0, step=0.05, label=EN["top_p"])
                repetition_penalty = gr.Slider(1.0, 2.0, value=1.0, step=0.05, label=EN["rep_pen"])
            with gr.Row():
                video_fps = gr.Slider(0.1, 4.0, value=1.0, step=0.1, label=EN["fps"])
                max_frames = gr.Slider(8, 256, value=64, step=8, label=EN["max_frames"])
                playback_speed = gr.Radio(
                    [("1×", "1×"), ("2×", "2×"), (EN["ff"], "Fast-forward")],
                    value="1×",
                    label=EN["speed"],
                )

    session_sid = gr.State(None)
    staged_media = gr.State(None)
    ui_lang = gr.State("en")

    start_evt = start_btn.click(
        run_session,
        inputs=[
            stage_mode, stage_file, staged_media, prompt_box, session_sid, ui_lang,
            max_new_tokens, temperature, top_p, repetition_penalty,
            video_fps, max_frames, playback_speed,
        ],
        outputs=[chatbot, raw_json, status_line, prompt_box, start_btn, stop_btn, clear_msgs_btn,
                 session_sid, stage_file, preview_video, preview_image, staged_media],
        api_visibility="private",
        concurrency_limit=1,
        show_progress="hidden",
    )
    start_evt.then(
        sync_console, [chatbot], [start_btn, stop_btn, clear_msgs_btn], api_visibility="private"
    )
    # cancels=[start_evt] is deliberately NOT wired to Stop: it would skip
    # session.close() and the final UI-restore yield. Stop is a mailbox flag.
    prompt_box.submit(send_prompt, [prompt_box, session_sid], [prompt_box], api_visibility="private")
    stop_btn.click(signal_stop, [session_sid], [stop_btn], api_visibility="private")
    clear_msgs_btn.click(
        clear_messages, None, [chatbot, raw_json, start_btn, clear_msgs_btn], api_visibility="private"
    )
    view_toggle.change(toggle_view, [view_toggle], [chatbot, raw_json], api_visibility="private")
    chatbot.clear(on_chat_clear, None, [raw_json], api_visibility="private")

    lang_btn.click(
        switch_lang,
        [ui_lang, session_sid],
        [ui_lang, lang_btn, title_md, stage_mode, stage_file, stage_hint, clear_media_btn,
         view_toggle, prompt_box, start_btn, stop_btn, clear_msgs_btn, adv_acc,
         max_new_tokens, temperature, top_p, repetition_penalty,
         video_fps, max_frames, playback_speed],
        api_visibility="private",
    )

    stage_file.upload(
        on_file_staged, [stage_file, session_sid],
        [staged_media, stage_file, preview_video, preview_image], api_visibility="private",
    )
    for clear_event in (stage_file.clear, stage_file.delete):
        clear_event(
            on_file_cleared, [session_sid],
            [staged_media, stage_file, preview_video, preview_image], api_visibility="private",
        )
    stage_mode.change(
        on_mode_change, [stage_mode, session_sid],
        [staged_media, stage_file, preview_video, preview_image, live_cam], api_visibility="private",
        js=_cam_stop_js(2, mode_arg=0),
    )
    live_cam.stream(
        on_cam_frame, [live_cam, session_sid], None,
        stream_every=CAM_STREAM_EVERY_S, show_progress="hidden", api_visibility="private",
    )
    clear_media_btn.click(
        clear_stages, [session_sid],
        [stage_mode, staged_media, stage_file, preview_video, preview_image, live_cam],
        api_visibility="private",
        js=_cam_stop_js(1),
    )

    gr.api(analyze, api_name="analyze")
    demo.load(None, None, None, js=PAGE_JS, api_visibility="private")


if __name__ == "__main__":
    # In gradio 6.x, theme/css/mcp_server are launch() parameters (the 6.0
    # migration moved them off the Blocks constructor).
    demo.queue(max_size=64)
    demo.launch(theme=gr.themes.Citrus(), css=CSS, mcp_server=True)