# Copyright (c) Meta Platforms, Inc. and affiliates. # All rights reserved. # # This source code is licensed under the BSD-style license found in the # LICENSE file in the root directory of this source tree. """ Web interface for OpenEnv environments. This module provides a web-based interface for interacting with OpenEnv environments, including a two-pane layout for HumanAgent interaction and state observation. """ from __future__ import annotations import json import time from dataclasses import asdict, dataclass from typing import Any, Dict, List, Optional, Type from datetime import datetime from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request from fastapi.responses import HTMLResponse, FileResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from .interfaces import Environment from .types import Action, Observation, State, EnvironmentMetadata def load_environment_metadata(env: Environment, env_name: Optional[str] = None) -> EnvironmentMetadata: """ Load environment metadata including README content. Args: env: The environment instance env_name: Optional environment name for README file lookup Returns: EnvironmentMetadata with loaded information """ # Try to get metadata from environment if it has a method for it if hasattr(env, 'get_metadata'): return env.get_metadata() # Default metadata metadata = EnvironmentMetadata( name=env_name or env.__class__.__name__, description=f"{env.__class__.__name__} environment", version="1.0.0" ) # Try to load README from file system readme_content = _load_readme_from_filesystem(env_name) if readme_content: metadata.readme_content = readme_content return metadata def _load_readme_from_filesystem(env_name: Optional[str]) -> Optional[str]: """ Load README content from the filesystem. Tries multiple locations: 1. Container filesystem: /app/README.md 2. Local development: src/envs/{env_name}/README.md 3. Environment variable: ENV_README_PATH """ import os from pathlib import Path # Try container filesystem first container_readme = Path("/app/README.md") if container_readme.exists(): try: return container_readme.read_text(encoding='utf-8') except Exception: pass # Try environment variable path custom_path = os.environ.get("ENV_README_PATH") if custom_path and Path(custom_path).exists(): try: return Path(custom_path).read_text(encoding='utf-8') except Exception: pass # Try local development path if env_name: local_readme = Path(f"src/envs/{env_name}/README.md") if local_readme.exists(): try: return local_readme.read_text(encoding='utf-8') except Exception: pass return None @dataclass class ActionLog: """Log entry for an action taken.""" timestamp: str action: Dict[str, Any] observation: Dict[str, Any] reward: Optional[float] done: bool step_count: int @dataclass class EpisodeState: """Current episode state for the web interface.""" episode_id: Optional[str] step_count: int current_observation: Optional[Dict[str, Any]] action_logs: List[ActionLog] is_reset: bool = True class WebInterfaceManager: """Manages the web interface for an environment.""" def __init__( self, env: Environment, action_cls: Type[Action], observation_cls: Type[Observation], metadata: Optional[EnvironmentMetadata] = None, ): self.env = env self.action_cls = action_cls self.observation_cls = observation_cls self.metadata = metadata or EnvironmentMetadata( name=env.__class__.__name__, description=f"{env.__class__.__name__} environment" ) self.episode_state = EpisodeState( episode_id=None, step_count=0, current_observation=None, action_logs=[] ) self.connected_clients: List[WebSocket] = [] async def connect_websocket(self, websocket: WebSocket): """Connect a new WebSocket client.""" await websocket.accept() self.connected_clients.append(websocket) # Send current state to the new client await self._send_state_update() async def disconnect_websocket(self, websocket: WebSocket): """Disconnect a WebSocket client.""" if websocket in self.connected_clients: self.connected_clients.remove(websocket) async def _send_state_update(self): """Send current state to all connected clients.""" if not self.connected_clients: return state_data = { "type": "state_update", "episode_state": asdict(self.episode_state) } # Send to all connected clients disconnected_clients = [] for client in self.connected_clients: try: await client.send_text(json.dumps(state_data)) except: disconnected_clients.append(client) # Remove disconnected clients for client in disconnected_clients: self.connected_clients.remove(client) async def reset_environment(self) -> Dict[str, Any]: """Reset the environment and update state.""" observation = self.env.reset() state = self.env.state # Update episode state self.episode_state.episode_id = state.episode_id self.episode_state.step_count = 0 self.episode_state.current_observation = asdict(observation) self.episode_state.action_logs = [] self.episode_state.is_reset = True # Send state update await self._send_state_update() return { "observation": asdict(observation), "reward": observation.reward, "done": observation.done, } async def step_environment(self, action_data: Dict[str, Any]) -> Dict[str, Any]: """Execute a step in the environment and update state.""" # Deserialize action action = self._deserialize_action(action_data) # Execute step observation = self.env.step(action) state = self.env.state # Create action log action_log = ActionLog( timestamp=datetime.now().isoformat(), action=asdict(action), observation=asdict(observation), reward=observation.reward, done=observation.done, step_count=state.step_count ) # Update episode state self.episode_state.episode_id = state.episode_id self.episode_state.step_count = state.step_count self.episode_state.current_observation = asdict(observation) self.episode_state.action_logs.append(action_log) self.episode_state.is_reset = False # Send state update await self._send_state_update() return { "observation": asdict(observation), "reward": observation.reward, "done": observation.done, } def get_state(self) -> Dict[str, Any]: """Get current environment state.""" state = self.env.state return asdict(state) def _deserialize_action(self, action_data: Dict[str, Any]) -> Action: """Convert JSON dict to Action instance.""" metadata = action_data.pop("metadata", {}) # Handle tensor fields that come from JSON as lists processed_data = {} for key, value in action_data.items(): if key == "tokens" and isinstance(value, (list, str)): # Convert list or string to tensor if isinstance(value, str): # If it's a string, try to parse it as a list of numbers try: import json value = json.loads(value) except: # If parsing fails, treat as empty list value = [] if isinstance(value, list): import torch processed_data[key] = torch.tensor(value, dtype=torch.long) else: processed_data[key] = value elif key == "action_id" and isinstance(value, str): # Convert action_id from string to int try: processed_data[key] = int(value) except ValueError: # If conversion fails, keep original value processed_data[key] = value else: processed_data[key] = value action = self.action_cls(**processed_data) action.metadata = metadata return action def create_web_interface_app( env: Environment, action_cls: Type[Action], observation_cls: Type[Observation], env_name: Optional[str] = None, ) -> FastAPI: """ Create a FastAPI application with web interface for the given environment. Args: env: The Environment instance to serve action_cls: The Action subclass this environment expects observation_cls: The Observation subclass this environment returns env_name: Optional environment name for README loading Returns: FastAPI application instance with web interface """ from .http_server import create_fastapi_app # Create the base environment app app = create_fastapi_app(env, action_cls, observation_cls) # Load environment metadata metadata = load_environment_metadata(env, env_name) # Create web interface manager web_manager = WebInterfaceManager(env, action_cls, observation_cls, metadata) # Add web interface routes @app.get("/web", response_class=HTMLResponse) async def web_interface(): """Serve the web interface.""" return get_web_interface_html(action_cls, web_manager.metadata) @app.get("/web/metadata") async def web_metadata(): """Get environment metadata.""" return asdict(web_manager.metadata) @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): """WebSocket endpoint for real-time updates.""" await web_manager.connect_websocket(websocket) try: while True: # Keep connection alive await websocket.receive_text() except WebSocketDisconnect: await web_manager.disconnect_websocket(websocket) @app.post("/web/reset") async def web_reset(): """Reset endpoint for web interface.""" return await web_manager.reset_environment() @app.post("/web/step") async def web_step(request: Dict[str, Any]): """Step endpoint for web interface.""" # Check if this is a message-based request (chat environment) if "message" in request: message = request["message"] # Convert message to action using the environment's message_to_action method action = web_manager.env.message_to_action(message) action_data = {"tokens": action.tokens.tolist()} else: action_data = request.get("action", {}) return await web_manager.step_environment(action_data) @app.get("/web/state") async def web_state(): """State endpoint for web interface.""" return web_manager.get_state() return app def get_web_interface_html(action_cls: Type[Action], metadata: Optional[EnvironmentMetadata] = None) -> str: """Generate the HTML for the web interface.""" # Check if this is a chat environment by looking for tokens field is_chat_env = False if hasattr(action_cls, '__dataclass_fields__'): for field_name, field_info in action_cls.__dataclass_fields__.items(): if field_name == 'tokens' and hasattr(field_info.type, '__name__') and 'Tensor' in field_info.type.__name__: is_chat_env = True break # Get action fields for dynamic form generation with enhanced metadata action_fields = _extract_action_fields(action_cls) return f""" OpenEnv Web Interface
HumanAgent Interface
{_generate_instructions_section(metadata)} {_generate_action_interface(action_fields, is_chat_env)}

Current State

Status: Not initialized
Episode ID: -
Step Count: 0
State Observer

Current Observation

No observation yet

Action History

No actions taken yet
""".replace('{_generate_action_form_fields(action_fields)}', _generate_action_form_fields(action_fields)) def _generate_instructions_section(metadata: Optional[EnvironmentMetadata]) -> str: """Generate the instructions section with environment documentation.""" if not metadata or not metadata.readme_content: return '' # Convert markdown to HTML (basic conversion) import re html_content = _markdown_to_html(metadata.readme_content) return f'''

{metadata.name}

{html_content}
''' def _extract_action_fields(action_cls: Type[Action]) -> List[Dict[str, Any]]: """Extract enhanced field metadata from Action class for form generation.""" import typing from typing import get_origin, get_args action_fields = [] if not hasattr(action_cls, '__dataclass_fields__'): return action_fields for field_name, field_info in action_cls.__dataclass_fields__.items(): if field_name == 'metadata': continue field_type = field_info.type field_metadata = _extract_field_metadata(field_name, field_info) # Determine input type based on field type input_type = _determine_input_type(field_type) # Check if field is required is_required = field_info.default is field_info.default_factory action_fields.append({ 'name': field_name, 'type': input_type, 'required': is_required, 'description': field_metadata.get('description', ''), 'default_value': field_metadata.get('default_value'), 'choices': field_metadata.get('choices', []), 'min_value': field_metadata.get('min_value'), 'max_value': field_metadata.get('max_value'), 'placeholder': field_metadata.get('placeholder', ''), 'help_text': field_metadata.get('help_text', ''), }) return action_fields def _extract_field_metadata(field_name: str, field_info) -> Dict[str, Any]: """Extract metadata from dataclass field including docstring and type hints.""" import typing from typing import get_origin, get_args, Literal, Union, Optional metadata = {} # Extract description from field docstring or annotation if hasattr(field_info, 'metadata') and field_info.metadata: # Check for custom metadata for meta in field_info.metadata: if isinstance(meta, dict): metadata.update(meta) # Extract type information field_type = field_info.type origin = get_origin(field_type) # Handle Literal types for dropdown choices if origin is Literal: args = get_args(field_type) metadata['choices'] = list(args) # Handle Optional types if origin is Union: args = get_args(field_type) if len(args) == 2 and type(None) in args: # This is Optional[SomeType] non_none_type = args[0] if args[1] is type(None) else args[1] metadata['optional'] = True # Recursively check the non-None type for choices if get_origin(non_none_type) is Literal: metadata['choices'] = list(get_args(non_none_type)) else: # Regular Union type metadata['choices'] = [str(arg) for arg in args if arg is not type(None)] # Handle numeric constraints if field_type in (int, float): # Check for common constraint patterns in field name if 'count' in field_name.lower() or 'num' in field_name.lower(): metadata['min_value'] = 0 if 'id' in field_name.lower(): metadata['min_value'] = 0 # Generate placeholder text if 'message' in field_name.lower(): metadata['placeholder'] = f'Enter {field_name.replace("_", " ")}...' elif 'code' in field_name.lower(): metadata['placeholder'] = 'Enter Python code here...' elif 'tokens' in field_name.lower(): metadata['placeholder'] = 'Enter comma-separated token IDs (e.g., 1,2,3,4,5)' else: metadata['placeholder'] = f'Enter {field_name.replace("_", " ")}...' # Generate help text based on field name and type if 'action_id' in field_name.lower(): metadata['help_text'] = 'The action ID to execute in the environment' elif 'game_name' in field_name.lower(): metadata['help_text'] = 'Name of the game or environment' elif 'tokens' in field_name.lower(): metadata['help_text'] = 'Token IDs as a comma-separated list of integers' elif 'code' in field_name.lower(): metadata['help_text'] = 'Python code to execute in the environment' elif 'message' in field_name.lower(): metadata['help_text'] = 'Text message to send' return metadata def _determine_input_type(field_type) -> str: """Determine the appropriate HTML input type for a field type.""" import typing from typing import get_origin, get_args, Literal, Union # Handle direct types if field_type == str: return "text" elif field_type == int: return "number" elif field_type == float: return "number" elif field_type == bool: return "checkbox" # Handle complex types origin = get_origin(field_type) if origin is Literal: return "select" elif origin is Union: args = get_args(field_type) if len(args) == 2 and type(None) in args: # Optional type - use the non-None type non_none_type = args[0] if args[1] is type(None) else args[1] return _determine_input_type(non_none_type) elif all(isinstance(arg, str) for arg in args if arg is not type(None)): return "select" else: return "text" elif hasattr(field_type, '__name__') and 'Tensor' in field_type.__name__: return "tensor" else: return "text" def _markdown_to_html(markdown: str) -> str: """Convert basic markdown to HTML for README display.""" import html import re # Escape HTML first html_content = html.escape(markdown) # Convert headers html_content = re.sub(r'^# (.*?)$', r'

\1

', html_content, flags=re.MULTILINE) html_content = re.sub(r'^## (.*?)$', r'

\1

', html_content, flags=re.MULTILINE) html_content = re.sub(r'^### (.*?)$', r'

\1

', html_content, flags=re.MULTILINE) # Convert code blocks html_content = re.sub(r'```(.*?)\n(.*?)\n```', r'
\2
', html_content, flags=re.DOTALL) html_content = re.sub(r'`([^`]+)`', r'\1', html_content) # Convert bold and italic html_content = re.sub(r'\*\*(.*?)\*\*', r'\1', html_content) html_content = re.sub(r'\*(.*?)\*', r'\1', html_content) # Convert lists html_content = re.sub(r'^- (.*?)$', r'
  • \1
    • ', html_content, flags=re.MULTILINE) html_content = re.sub(r'(
  • .*
    • )', r'', html_content, flags=re.DOTALL) # Convert line breaks html_content = html_content.replace('\n', '
    ') return html_content def _generate_action_interface(action_fields: List[Dict[str, Any]], is_chat_env: bool) -> str: """Generate either a chat interface or action form based on environment type.""" if is_chat_env: return _generate_chat_interface() else: return _generate_action_form(action_fields) def _generate_chat_interface() -> str: """Generate a chat-style interface for chat environments.""" return '''

    Chat Interface

    System
    Chat environment ready. Send a message to start the conversation.
    ''' def _generate_action_form(action_fields: List[Dict[str, Any]]) -> str: """Generate a traditional action form for non-chat environments.""" return f'''

    Take Action

    {_generate_action_form_fields(action_fields)}
    ''' def _generate_action_form_fields(action_fields: List[Dict[str, Any]]) -> str: """Generate HTML form fields for action input with enhanced metadata.""" if not action_fields: return '

    No action fields available

    ' fields_html = [] for field in action_fields: field_html = _generate_single_field(field) fields_html.append(field_html) return '\n'.join(fields_html) def _generate_single_field(field: Dict[str, Any]) -> str: """Generate HTML for a single form field with enhanced metadata.""" field_name = field['name'] field_type = field['type'] required = field['required'] placeholder = field.get('placeholder', '') help_text = field.get('help_text', '') choices = field.get('choices', []) min_value = field.get('min_value') max_value = field.get('max_value') default_value = field.get('default_value') # Build label with required indicator label_text = field_name.replace('_', ' ').title() if required: label_text += ' *' # Build input attributes input_attrs = [] if required: input_attrs.append('required') if placeholder: input_attrs.append(f'placeholder="{placeholder}"') if min_value is not None: input_attrs.append(f'min="{min_value}"') if max_value is not None: input_attrs.append(f'max="{max_value}"') if default_value is not None: input_attrs.append(f'value="{default_value}"') attrs_str = ' '.join(input_attrs) if field_type == 'checkbox': return f'''
    {f'{help_text}' if help_text else ''}
    ''' elif field_type == 'select': options_html = [] if not required: options_html.append(f'') for choice in choices: selected = 'selected' if str(choice) == str(default_value) else '' options_html.append(f'') return f'''
    {f'{help_text}' if help_text else ''}
    ''' elif field_type == 'tensor': return f'''
    {help_text or 'Enter token IDs as comma-separated integers (e.g., 1,2,3,4,5)'}
    ''' elif field_type == 'text' and ('message' in field_name.lower() or 'code' in field_name.lower()): return f'''
    {f'{help_text}' if help_text else ''}
    ''' else: return f'''
    {f'{help_text}' if help_text else ''}
    '''