scripts/preview.py

"""Shared production UI with explicitly simulated sessions; no models or physics."""

import json
import os
import sys
import time
from pathlib import Path

os.environ.setdefault("GRADIO_ANALYTICS_ENABLED", "False")
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

import gradio as gr  # noqa: E402

from check_assets import check_assets  # noqa: E402
from ui import (
    FORCE_LIGHT_JS,
    NEXT_SESSION,
    THEME,
    build_demo,
    session_controls,
    status_html,
)  # noqa: E402

# Exact labels from hf-libero 0.1.4, used only by this simulator-free preview.
TASKS = json.loads((ROOT / "scripts/preview_tasks.json").read_text())
SESSIONS = {}


def send(kind, text, request):
    text = text.strip()
    session = SESSIONS.get(request.session_hash)
    if not text or (kind == "stop" and session is None):
        return "", False, gr.skip()
    if kind == "ask" and session is None:
        # As in app.py: a question never starts a session.
        return text, False, gr.skip()
    if kind == "stop":
        session["stop"] = True
        return "", False, gr.skip()
    start = session is None
    if start:
        session = {"stop": False, "chat": [], "qa": []}
        SESSIONS[request.session_hash] = session
    target = session["qa"] if kind == "ask" else session["chat"]
    target.append({"role": "user", "content": text})
    if kind == "ask":
        session["qa"].append(
            {
                "role": "assistant",
                "content": "This is a design preview; scene questions need the connected VLM.",
            }
        )
    else:
        session["task"] = text if kind == "task" else TASKS[7]
        session["goal"] = text if kind == "goal" else None
        session["chat"].append(
            {
                "role": "assistant",
                "content": f"**Preview instruction:** {session['task']}\n\nSimulated selection for interface testing. The robot will not move.",
            }
        )
    return (
        "",
        start,
        status_html("starting", "Preparing the preview") if start else gr.skip(),
    )


def send_goal(text: str, request: gr.Request):
    # As in app.py: a goal that starts a session shows in the task log at once.
    box, start, status = send("goal", text, request)
    return box, start, status, [{"role": "user", "content": text.strip()}] if start else gr.skip()


def send_question(text: str, request: gr.Request):
    return send("ask", text, request)


def send_stop(request: gr.Request):
    return send("stop", "stop", request)


def run(start: bool, request: gr.Request):
    if not start:
        # As in app.py: yielding nothing would blank every output.
        yield gr.skip()
        return
    session = SESSIONS[request.session_hash]
    try:
        yield (
            gr.skip(),
            session["chat"],
            session["qa"],
            *session_controls(TASKS, "starting", "Preparing the preview"),
        )
        time.sleep(2)  # long enough to see the starting popup
        started = time.monotonic()
        while time.monotonic() - started < 12 and not session["stop"]:
            task = session.get("task", TASKS[8])
            yield (
                gr.skip(),
                session["chat"],
                session["qa"],
                *session_controls(
                    TASKS,
                    "steering" if session.get("goal") else "running",
                    task,
                    session.get("goal"),
                    12 - (time.monotonic() - started),
                    note="Simulated UI state · no robot motion",
                ),
            )
            time.sleep(0.25)
        outcome = "stopped" if session["stop"] else "done"
        session["chat"].append(
            {
                "role": "assistant",
                "content": f"Preview {outcome}. No simulation was executed.",
            }
        )
        yield (
            gr.skip(),
            session["chat"],
            session["qa"],
            *session_controls(
                TASKS, outcome, session.get("task", TASKS[8]), note=NEXT_SESSION
            ),
        )
    finally:
        SESSIONS.pop(request.session_hash, None)


def build_preview():
    check_assets()
    web = ROOT / "web"
    player = (
        (ROOT / "player.js")
        .read_text()
        .replace("__SCENE_URL__", json.dumps(f"/gradio_api/file={web / 'scene.glb'}"))
        .replace(
            "__CAMERAS__", json.dumps(json.loads((web / "cameras.json").read_text()))
        )
    )
    gr.set_static_paths(paths=[str(web)])
    return build_demo(
        TASKS,
        player,
        run=run,
        send_goal=send_goal,
        send_question=send_question,
        send_stop=send_stop,
        preview=True,
    )


if __name__ == "__main__":
    build_preview().queue().launch(
        server_name=os.environ.get("GRADIO_SERVER_NAME", "127.0.0.1"),
        server_port=int(os.environ.get("GRADIO_SERVER_PORT", "7860")),
        css=(ROOT / "style.css").read_text(),
        theme=THEME,
        js=FORCE_LIGHT_JS,
        show_error=True,
    )