openenv/echo_env
6
1# SPDX-License-Identifier: BSD-3-Clause2 3"""4Gradio-based web UI for OpenEnv environments.5 6Replaces the legacy HTML/JavaScript interface when ENABLE_WEB_INTERFACE is set.7Mount at /web via gr.mount_gradio_app() from create_web_interface_app().8"""9 10from __future__ import annotations11 12import json13import re14from typing import Any, Dict, List, Optional15 16import gradio as gr17 18from .types import EnvironmentMetadata19 20 21def _escape_md(text: str) -> str:22 """Escape Markdown special characters in user-controlled content."""23 return re.sub(r"([\\`*_\{\}\[\]()#+\-.!|~>])", r"\\\1", str(text))24 25 26def _format_observation(data: Dict[str, Any]) -> str:27 """Format reset/step response for Markdown display."""28 lines: List[str] = []29 obs = data.get("observation", {})30 if isinstance(obs, dict):31 if obs.get("prompt"):32 lines.append(f"**Prompt:**\n\n{_escape_md(obs['prompt'])}\n")33 messages = obs.get("messages", [])34 if messages:35 lines.append("**Messages:**\n")36 for msg in messages:37 sender = _escape_md(str(msg.get("sender_id", "?")))38 content = _escape_md(str(msg.get("content", "")))39 cat = _escape_md(str(msg.get("category", "")))40 lines.append(f"- `[{cat}]` Player {sender}: {content}")41 lines.append("")42 reward = data.get("reward")43 done = data.get("done")44 if reward is not None:45 lines.append(f"**Reward:** `{reward}`")46 if done is not None:47 lines.append(f"**Done:** `{done}`")48 return "\n".join(lines) if lines else "*No observation data*"49 50 51def _readme_section(metadata: Optional[EnvironmentMetadata]) -> str:52 """README content for the left panel."""53 if not metadata or not metadata.readme_content:54 return "*No README available.*"55 return metadata.readme_content56 57 58def get_gradio_display_title(59 metadata: Optional[EnvironmentMetadata],60 fallback: str = "OpenEnv Environment",61) -> str:62 """Return the title used for the Gradio app (browser tab and Blocks)."""63 name = metadata.name if metadata else fallback64 return f"OpenEnv Agentic Environment: {name}"65 66 67def build_gradio_app(68 web_manager: Any,69 action_fields: List[Dict[str, Any]],70 metadata: Optional[EnvironmentMetadata],71 is_chat_env: bool,72 title: str = "OpenEnv Environment",73 quick_start_md: Optional[str] = None,74) -> gr.Blocks:75 """76 Build a Gradio Blocks app for the OpenEnv web interface.77 78 Args:79 web_manager: WebInterfaceManager (reset/step_environment, get_state).80 action_fields: Field dicts from _extract_action_fields(action_cls).81 metadata: Environment metadata for README/name.82 is_chat_env: If True, single message textbox; else form from action_fields.83 title: App title (overridden by metadata.name when present; see get_gradio_display_title).84 quick_start_md: Optional Quick Start markdown (class names already replaced).85 86 Returns:87 gr.Blocks to mount with gr.mount_gradio_app(app, blocks, path="/web").88 """89 readme_content = _readme_section(metadata)90 display_title = get_gradio_display_title(metadata, fallback=title)91 92 async def reset_env():93 try:94 data = await web_manager.reset_environment()95 obs_md = _format_observation(data)96 return (97 obs_md,98 json.dumps(data, indent=2),99 "Environment reset successfully.",100 )101 except Exception as e:102 return ("", "", f"Error: {e}")103 104 def _step_with_action(action_data: Dict[str, Any]):105 async def _run():106 try:107 data = await web_manager.step_environment(action_data)108 obs_md = _format_observation(data)109 return (110 obs_md,111 json.dumps(data, indent=2),112 "Step complete.",113 )114 except Exception as e:115 return ("", "", f"Error: {e}")116 117 return _run118 119 async def step_chat(message: str):120 if not (message or str(message).strip()):121 return ("", "", "Please enter an action message.")122 action = {"message": str(message).strip()}123 return await _step_with_action(action)()124 125 def get_state_sync():126 try:127 data = web_manager.get_state()128 return json.dumps(data, indent=2)129 except Exception as e:130 return f"Error: {e}"131 132 with gr.Blocks(title=display_title) as demo:133 with gr.Row():134 with gr.Column(scale=1, elem_classes="col-left"):135 if quick_start_md:136 with gr.Accordion("Quick Start", open=True):137 gr.Markdown(quick_start_md)138 with gr.Accordion("README", open=False):139 gr.Markdown(readme_content)140 141 with gr.Column(scale=2, elem_classes="col-right"):142 obs_display = gr.Markdown(143 value=("# Playground\n\nClick **Reset** to start a new episode."),144 )145 with gr.Group():146 if is_chat_env:147 action_input = gr.Textbox(148 label="Action message",149 placeholder="e.g. Enter your message...",150 )151 step_inputs = [action_input]152 step_fn = step_chat153 else:154 step_inputs = []155 for field in action_fields:156 name = field["name"]157 field_type = field.get("type", "text")158 label = name.replace("_", " ").title()159 placeholder = field.get("placeholder", "")160 if field_type == "checkbox":161 inp = gr.Checkbox(label=label)162 elif field_type == "number":163 inp = gr.Number(label=label)164 elif field_type == "select":165 choices = field.get("choices") or []166 inp = gr.Dropdown(167 choices=choices,168 label=label,169 allow_custom_value=False,170 )171 elif field_type in ("textarea", "tensor"):172 inp = gr.Textbox(173 label=label,174 placeholder=placeholder,175 lines=3,176 )177 else:178 inp = gr.Textbox(179 label=label,180 placeholder=placeholder,181 )182 step_inputs.append(inp)183 184 async def step_form(*values):185 if not action_fields:186 return await _step_with_action({})()187 action_data = {}188 for i, field in enumerate(action_fields):189 if i >= len(values):190 break191 name = field["name"]192 val = values[i]193 if field.get("type") == "checkbox":194 action_data[name] = bool(val)195 elif val is not None and val != "":196 action_data[name] = val197 return await _step_with_action(action_data)()198 199 step_fn = step_form200 201 with gr.Row():202 step_btn = gr.Button("Step", variant="primary")203 reset_btn = gr.Button("Reset", variant="secondary")204 state_btn = gr.Button("Get state", variant="secondary")205 with gr.Row():206 status = gr.Textbox(207 label="Status",208 interactive=False,209 )210 raw_json = gr.Code(211 label="Raw JSON response",212 language="json",213 interactive=False,214 )215 216 reset_btn.click(217 fn=reset_env,218 outputs=[obs_display, raw_json, status],219 )220 step_btn.click(221 fn=step_fn,222 inputs=step_inputs,223 outputs=[obs_display, raw_json, status],224 )225 if is_chat_env:226 action_input.submit(227 fn=step_fn,228 inputs=step_inputs,229 outputs=[obs_display, raw_json, status],230 )231 state_btn.click(232 fn=get_state_sync,233 outputs=[raw_json],234 )235 236 return demo237