# 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. """ Number Guessing Game HTTP Client. This module provides the client for connecting to a NumberGuessEnvironment server over HTTP. """ from typing import Dict # Support both in-repo and standalone imports try: # In-repo imports (when running from OpenEnv repository) from core.client_types import StepResult from core.env_server.types import State from core.http_env_client import HTTPEnvClient from .models import GuessAction, GuessObservation except ImportError: # Standalone imports (when environment is standalone with openenv-core from pip) from openenv_core.client_types import StepResult from openenv_core.env_server.types import State from openenv_core.http_env_client import HTTPEnvClient from models import GuessAction, GuessObservation class NumberGuessEnv(HTTPEnvClient[GuessAction, GuessObservation]): """ HTTP client for the Number Guessing Game Environment. This client connects to a NumberGuessEnvironment HTTP server and provides methods to interact with it: reset(), step(), and state access. Example: >>> # Connect to a running server >>> client = NumberGuessEnv(base_url="http://localhost:8000") >>> result = client.reset() >>> print(result.observation.hint) >>> >>> # Make guesses >>> result = client.step(GuessAction(guess=50)) >>> print(result.observation.hint) # "higher" or "lower" >>> print(result.reward) Example with Docker: >>> # Automatically start container and connect >>> client = NumberGuessEnv.from_docker_image("number-guess-env:latest") >>> result = client.reset() >>> >>> # Play the game! >>> while not result.done: ... guess = int(input("Enter your guess: ")) ... result = client.step(GuessAction(guess=guess)) ... print(f"Hint: {result.observation.hint}") ... print(f"Attempts remaining: {result.observation.attempts_remaining}") >>> >>> client.close() """ def _step_payload(self, action: GuessAction) -> Dict: """ Convert GuessAction to JSON payload for step request. Args: action: GuessAction instance Returns: Dictionary representation suitable for JSON encoding """ return { "guess": action.guess, } def _parse_result(self, payload: Dict) -> StepResult[GuessObservation]: """ Parse server response into StepResult[GuessObservation]. Args: payload: JSON response from server Returns: StepResult with GuessObservation """ obs_data = payload.get("observation", {}) observation = GuessObservation( hint=obs_data.get("hint", ""), attempts_remaining=obs_data.get("attempts_remaining", 0), guess_history=obs_data.get("guess_history", []), done=payload.get("done", False), reward=payload.get("reward"), metadata=obs_data.get("metadata", {}), ) return StepResult( observation=observation, reward=payload.get("reward"), done=payload.get("done", False), ) def _parse_state(self, payload: Dict) -> State: """ Parse server response into State object. Args: payload: JSON response from /state endpoint Returns: State object with episode_id and step_count """ return State( episode_id=payload.get("episode_id"), step_count=payload.get("step_count", 0), )