openenv/coding_env

1# Copyright (c) Meta Platforms, Inc. and affiliates.2# All rights reserved.3#4# This source code is licensed under the BSD-style license found in the5# LICENSE file in the root directory of this source tree.6 7"""8__ENV_TITLE_NAME__ Environment Implementation.9 10A simple test environment that echoes back messages sent to it.11Perfect for testing HTTP server infrastructure.12"""13 14from uuid import uuid415 16from openenv.core.env_server.interfaces import Environment17from openenv.core.env_server.types import State18 19try:20    from ..models import __ENV_CLASS_NAME__Action, __ENV_CLASS_NAME__Observation21except ImportError:22    from models import __ENV_CLASS_NAME__Action, __ENV_CLASS_NAME__Observation23 24 25class __ENV_CLASS_NAME__Environment(Environment):26    """27    A simple echo environment that echoes back messages.28 29    This environment is designed for testing the HTTP server infrastructure.30    It maintains minimal state and simply echoes back whatever message it receives.31 32    Example:33        >>> env = __ENV_CLASS_NAME__Environment()34        >>> obs = env.reset()35        >>> print(obs.echoed_message)  # "__ENV_TITLE_NAME__ environment ready!"36        >>>37        >>> obs = env.step(__ENV_CLASS_NAME__Action(message="Hello"))38        >>> print(obs.echoed_message)  # "Hello"39        >>> print(obs.message_length)  # 540    """41 42    # Enable concurrent WebSocket sessions.43    # Set to True if your environment isolates state between instances.44    # When True, multiple WebSocket clients can connect simultaneously, each45    # getting their own environment instance (when using factory mode in app.py).46    SUPPORTS_CONCURRENT_SESSIONS: bool = True47 48    def __init__(self):49        """Initialize the __ENV_NAME__ environment."""50        self._state = State(episode_id=str(uuid4()), step_count=0)51        self._reset_count = 052 53    def reset(self) -> __ENV_CLASS_NAME__Observation:54        """55        Reset the environment.56 57        Returns:58            __ENV_CLASS_NAME__Observation with a ready message59        """60        self._state = State(episode_id=str(uuid4()), step_count=0)61        self._reset_count += 162 63        return __ENV_CLASS_NAME__Observation(64            echoed_message="__ENV_TITLE_NAME__ environment ready!",65            message_length=0,66            done=False,67            reward=0.0,68        )69 70    def step(self, action: __ENV_CLASS_NAME__Action) -> __ENV_CLASS_NAME__Observation:  # type: ignore[override]71        """72        Execute a step in the environment by echoing the message.73 74        Args:75            action: __ENV_CLASS_NAME__Action containing the message to echo76 77        Returns:78            __ENV_CLASS_NAME__Observation with the echoed message and its length79        """80        self._state.step_count += 181 82        message = action.message83        length = len(message)84 85        # Simple reward: longer messages get higher rewards86        reward = length * 0.187 88        return __ENV_CLASS_NAME__Observation(89            echoed_message=message,90            message_length=length,91            done=False,92            reward=reward,93            metadata={"original_message": message, "step": self._state.step_count},94        )95 96    @property97    def state(self) -> State:98        """99        Get the current environment state.100 101        Returns:102            Current State with episode_id and step_count103        """104        return self._state105