openenv/browsergym_env

1"""Client for the BrowserGym environment."""2 3from typing import Any, Dict4 5from openenv.core.client_types import StepResult6from openenv.core.env_client import EnvClient7 8from .models import BrowserGymAction, BrowserGymObservation, BrowserGymState9 10 11class BrowserGymEnv(12    EnvClient[BrowserGymAction, BrowserGymObservation, BrowserGymState]13):14    """Client for interacting with the BrowserGym environment.15 16    BrowserGym provides unified access to multiple web navigation benchmarks:17    - MiniWoB++: 100+ training tasks (no external infrastructure needed!)18    - WebArena: 812 evaluation tasks (requires backend setup)19    - VisualWebArena: Visual navigation tasks20    - WorkArena: Enterprise automation tasks21 22    Example usage for TRAINING (MiniWoB - works out of the box):23        ```python24        from envs.browsergym_env import BrowserGymEnv, BrowserGymAction25 26        # Create environment for MiniWoB training task27        env = BrowserGymEnv.from_docker_image(28            "browsergym-env:latest",29            environment={30                "BROWSERGYM_BENCHMARK": "miniwob",31                "BROWSERGYM_TASK_NAME": "click-test",32            }33        )34 35        # Reset and get initial observation36        result = env.reset()37        print(f"Task: {result.observation.goal}")38        print(f"Page: {result.observation.text[:200]}")39 40        # Take actions41        action = BrowserGymAction(action_str="click('Submit button')")42        result = env.step(action)43        print(f"Reward: {result.reward}")44        print(f"Done: {result.done}")45 46        env.close()47        ```48 49    Example usage for EVALUATION (WebArena - requires backend):50        ```python51        from envs.browsergym_env import BrowserGymEnv, BrowserGymAction52 53        # Create environment for WebArena evaluation54        env = BrowserGymEnv.from_docker_image(55            "browsergym-env:latest",56            environment={57                "BROWSERGYM_BENCHMARK": "webarena",58                "BROWSERGYM_TASK_NAME": "0",  # Task 059                # WebArena backend URLs60                "SHOPPING": "http://your-server:7770",61                "GITLAB": "http://your-server:8023",62                # ... other URLs63            }64        )65 66        result = env.reset()67        # ... interact with environment68        env.close()69        ```70 71    Available benchmarks:72        - miniwob: MiniWoB++ tasks (training, no setup required)73        - webarena: WebArena tasks (evaluation, requires backend)74        - visualwebarena: Visual WebArena tasks (evaluation, requires backend)75        - workarena: WorkArena tasks (evaluation, requires backend)76    """77 78    def _step_payload(self, action: BrowserGymAction) -> Dict[str, Any]:79        """Convert a BrowserGymAction to the JSON payload for the server."""80        return {81            "action_str": action.action_str,82            "metadata": action.metadata,83        }84 85    def _parse_result(86        self, payload: Dict[str, Any]87    ) -> StepResult[BrowserGymObservation]:88        """Parse the server response into a StepResult."""89        obs_data = payload.get("observation", {})90 91        observation = BrowserGymObservation(92            text=obs_data.get("text", ""),93            url=obs_data.get("url", ""),94            screenshot=obs_data.get("screenshot"),95            goal=obs_data.get("goal", ""),96            axtree_txt=obs_data.get("axtree_txt", ""),97            pruned_html=obs_data.get("pruned_html", ""),98            error=obs_data.get("error", ""),99            last_action_error=obs_data.get("last_action_error", False),100            done=payload.get("done", False),101            reward=payload.get("reward"),102            metadata=payload.get("metadata", obs_data.get("metadata", {})),103        )104 105        return StepResult(106            observation=observation,107            reward=payload.get("reward"),108            done=payload.get("done", False),109            metadata=payload.get("metadata"),110        )111 112    def _parse_state(self, payload: Dict[str, Any]) -> BrowserGymState:113        """Parse the server state response into a BrowserGymState object."""114        return BrowserGymState(115            episode_id=payload.get("episode_id"),116            step_count=payload.get("step_count", 0),117            benchmark=payload.get("benchmark", ""),118            task_name=payload.get("task_name", ""),119            task_id=payload.get("task_id"),120            goal=payload.get("goal", ""),121            current_url=payload.get("current_url", ""),122            max_steps=payload.get("max_steps"),123            cum_reward=payload.get("cum_reward", 0.0),124        )125