| |
| |
| |
| |
| |
|
|
| """ |
| Gradio-based web UI for OpenEnv environments. |
| |
| Replaces the legacy HTML/JavaScript interface when ENABLE_WEB_INTERFACE is set. |
| Mount at /web via gr.mount_gradio_app() from create_web_interface_app(). |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import re |
| from typing import Any, Dict, List, Optional |
|
|
| import gradio as gr |
|
|
| from .types import EnvironmentMetadata |
|
|
|
|
| def _escape_md(text: str) -> str: |
| """Escape Markdown special characters in user-controlled content.""" |
| return re.sub(r"([\\`*_\{\}\[\]()#+\-.!|~>])", r"\\\1", str(text)) |
|
|
|
|
| def _format_observation(data: Dict[str, Any]) -> str: |
| """Format reset/step response for Markdown display.""" |
| lines: List[str] = [] |
| obs = data.get("observation", {}) |
| if isinstance(obs, dict): |
| if obs.get("prompt"): |
| lines.append(f"**Prompt:**\n\n{_escape_md(obs['prompt'])}\n") |
| messages = obs.get("messages", []) |
| if messages: |
| lines.append("**Messages:**\n") |
| for msg in messages: |
| sender = _escape_md(str(msg.get("sender_id", "?"))) |
| content = _escape_md(str(msg.get("content", ""))) |
| cat = _escape_md(str(msg.get("category", ""))) |
| lines.append(f"- `[{cat}]` Player {sender}: {content}") |
| lines.append("") |
| reward = data.get("reward") |
| done = data.get("done") |
| if reward is not None: |
| lines.append(f"**Reward:** `{reward}`") |
| if done is not None: |
| lines.append(f"**Done:** `{done}`") |
| return "\n".join(lines) if lines else "*No observation data*" |
|
|
|
|
| def _readme_section(metadata: Optional[EnvironmentMetadata]) -> str: |
| """README content for the left panel.""" |
| if not metadata or not metadata.readme_content: |
| return "*No README available.*" |
| return metadata.readme_content |
|
|
|
|
| def get_gradio_display_title( |
| metadata: Optional[EnvironmentMetadata], |
| fallback: str = "OpenEnv Environment", |
| ) -> str: |
| """Return the title used for the Gradio app (browser tab and Blocks).""" |
| name = metadata.name if metadata else fallback |
| return f"OpenEnv Agentic Environment: {name}" |
|
|
|
|
| def build_gradio_app( |
| web_manager: Any, |
| action_fields: List[Dict[str, Any]], |
| metadata: Optional[EnvironmentMetadata], |
| is_chat_env: bool, |
| title: str = "OpenEnv Environment", |
| quick_start_md: Optional[str] = None, |
| ) -> gr.Blocks: |
| """ |
| Build a Gradio Blocks app for the OpenEnv web interface. |
| |
| Args: |
| web_manager: WebInterfaceManager (reset/step_environment, get_state). |
| action_fields: Field dicts from _extract_action_fields(action_cls). |
| metadata: Environment metadata for README/name. |
| is_chat_env: If True, single message textbox; else form from action_fields. |
| title: App title (overridden by metadata.name when present; see get_gradio_display_title). |
| quick_start_md: Optional Quick Start markdown (class names already replaced). |
| |
| Returns: |
| gr.Blocks to mount with gr.mount_gradio_app(app, blocks, path="/web"). |
| """ |
| readme_content = _readme_section(metadata) |
| display_title = get_gradio_display_title(metadata, fallback=title) |
|
|
| async def reset_env(): |
| try: |
| data = await web_manager.reset_environment() |
| obs_md = _format_observation(data) |
| return ( |
| obs_md, |
| json.dumps(data, indent=2), |
| "Environment reset successfully.", |
| ) |
| except Exception as e: |
| return ("", "", f"Error: {e}") |
|
|
| def _step_with_action(action_data: Dict[str, Any]): |
| async def _run(): |
| try: |
| data = await web_manager.step_environment(action_data) |
| obs_md = _format_observation(data) |
| return ( |
| obs_md, |
| json.dumps(data, indent=2), |
| "Step complete.", |
| ) |
| except Exception as e: |
| return ("", "", f"Error: {e}") |
|
|
| return _run |
|
|
| async def step_chat(message: str): |
| if not (message or str(message).strip()): |
| return ("", "", "Please enter an action message.") |
| action = {"message": str(message).strip()} |
| return await _step_with_action(action)() |
|
|
| def get_state_sync(): |
| try: |
| data = web_manager.get_state() |
| return json.dumps(data, indent=2) |
| except Exception as e: |
| return f"Error: {e}" |
|
|
| with gr.Blocks(title=display_title) as demo: |
| with gr.Row(): |
| with gr.Column(scale=1, elem_classes="col-left"): |
| if quick_start_md: |
| with gr.Accordion("Quick Start", open=True): |
| gr.Markdown(quick_start_md) |
| with gr.Accordion("README", open=False): |
| gr.Markdown(readme_content) |
|
|
| with gr.Column(scale=2, elem_classes="col-right"): |
| obs_display = gr.Markdown( |
| value=("# Playground\n\nClick **Reset** to start a new episode."), |
| ) |
| with gr.Group(): |
| if is_chat_env: |
| action_input = gr.Textbox( |
| label="Action message", |
| placeholder="e.g. Enter your message...", |
| ) |
| step_inputs = [action_input] |
| step_fn = step_chat |
| else: |
| step_inputs = [] |
| for field in action_fields: |
| name = field["name"] |
| field_type = field.get("type", "text") |
| label = name.replace("_", " ").title() |
| placeholder = field.get("placeholder", "") |
| if field_type == "checkbox": |
| inp = gr.Checkbox(label=label) |
| elif field_type == "number": |
| inp = gr.Number(label=label) |
| elif field_type == "select": |
| choices = field.get("choices") or [] |
| inp = gr.Dropdown( |
| choices=choices, |
| label=label, |
| allow_custom_value=False, |
| ) |
| elif field_type in ("textarea", "tensor"): |
| inp = gr.Textbox( |
| label=label, |
| placeholder=placeholder, |
| lines=3, |
| ) |
| else: |
| inp = gr.Textbox( |
| label=label, |
| placeholder=placeholder, |
| ) |
| step_inputs.append(inp) |
|
|
| async def step_form(*values): |
| if not action_fields: |
| return await _step_with_action({})() |
| action_data = {} |
| for i, field in enumerate(action_fields): |
| if i >= len(values): |
| break |
| name = field["name"] |
| val = values[i] |
| if field.get("type") == "checkbox": |
| action_data[name] = bool(val) |
| elif val is not None and val != "": |
| action_data[name] = val |
| return await _step_with_action(action_data)() |
|
|
| step_fn = step_form |
|
|
| with gr.Row(): |
| step_btn = gr.Button("Step", variant="primary") |
| reset_btn = gr.Button("Reset", variant="secondary") |
| state_btn = gr.Button("Get state", variant="secondary") |
| with gr.Row(): |
| status = gr.Textbox( |
| label="Status", |
| interactive=False, |
| ) |
| raw_json = gr.Code( |
| label="Raw JSON response", |
| language="json", |
| interactive=False, |
| ) |
|
|
| reset_btn.click( |
| fn=reset_env, |
| outputs=[obs_display, raw_json, status], |
| ) |
| step_btn.click( |
| fn=step_fn, |
| inputs=step_inputs, |
| outputs=[obs_display, raw_json, status], |
| ) |
| if is_chat_env: |
| action_input.submit( |
| fn=step_fn, |
| inputs=step_inputs, |
| outputs=[obs_display, raw_json, status], |
| ) |
| state_btn.click( |
| fn=get_state_sync, |
| outputs=[raw_json], |
| ) |
|
|
| return demo |
|
|
