Implement a sample Pokemon Battle Environment with server and client integration

Author: cpich3g

examples/project-pikachu/poke_env/__init__.py

@@ -0,0 +1,24 @@
+"""
+Pokemon Battle Environment for OpenEnv.
+
+This module provides OpenEnv integration for Pokemon battles via poke-env.
+
+Example:
+    >>> from envs.pokemon_env import PokemonEnv, PokemonAction
+    >>>
+    >>> # Connect to a running Pokemon Showdown server
+    >>> env = PokemonEnv(battle_format="gen8randombattle")
+    >>>
+    >>> # Reset and interact
+    >>> result = env.reset()
+    >>> result = env.step(PokemonAction(action_type="move", action_index=0))
+    >>> print(result.reward, result.done)
+    >>>
+    >>> # Cleanup
+    >>> env.close()
+"""
+
+from .client import PokemonEnv
+from .models import PokemonAction, PokemonObservation, PokemonState, PokemonData
+
+__all__ = ["PokemonEnv", "PokemonAction", "PokemonObservation", "PokemonState", "PokemonData"]

examples/project-pikachu/poke_env/client.py

@@ -0,0 +1,157 @@
+"""
+Pokemon Battle Environment HTTP Client.
+
+This module provides the client for connecting to a Pokemon Battle Environment server
+over HTTP.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, TYPE_CHECKING
+
+from core.client_types import StepResult
+from core.http_env_client import HTTPEnvClient
+
+from .models import PokemonAction, PokemonObservation, PokemonState, PokemonData
+
+if TYPE_CHECKING:
+    from core.containers.runtime import ContainerProvider
+
+
+class PokemonEnv(HTTPEnvClient[PokemonAction, PokemonObservation]):
+    """
+    HTTP client for Pokemon Battle Environment.
+
+    This client connects to a Pokemon Battle Environment HTTP server and provides
+    methods to interact with it: reset(), step(), and state access.
+
+    Example:
+        >>> # Connect to a running server
+        >>> client = PokemonEnv(base_url="http://localhost:8000")
+        >>> result = client.reset()
+        >>> print(result.observation.active_pokemon.species)
+        >>>
+        >>> # Take an action
+        >>> result = client.step(PokemonAction(action_type="move", action_index=0))
+        >>> print(result.reward, result.done)
+
+    Example with Docker:
+        >>> # Automatically start container and connect
+        >>> client = PokemonEnv.from_docker_image("pokemon-env:latest")
+        >>> result = client.reset()
+        >>> result = client.step(PokemonAction(action_type="switch", action_index=1))
+    """
+
+    def _step_payload(self, action: PokemonAction) -> Dict[str, Any]:
+        """
+        Convert PokemonAction to JSON payload for step request.
+
+        Args:
+            action: PokemonAction instance.
+
+        Returns:
+            Dictionary representation suitable for JSON encoding.
+        """
+        return {
+            "action_type": action.action_type,
+            "action_index": action.action_index,
+            "move_id": action.move_id,
+            "switch_pokemon": action.switch_pokemon,
+            "mega_evolve": action.mega_evolve,
+            "dynamax": action.dynamax,
+            "terastallize": action.terastallize,
+        }
+
+    def _parse_pokemon_data(self, data: Dict[str, Any]) -> PokemonData:
+        """Parse Pokemon data from JSON."""
+        return PokemonData(
+            species=data.get("species", "unknown"),
+            hp_percent=data.get("hp_percent", 0.0),
+            max_hp=data.get("max_hp", 100),
+            current_hp=data.get("current_hp", 0),
+            level=data.get("level", 50),
+            status=data.get("status"),
+            types=data.get("types", []),
+            ability=data.get("ability"),
+            item=data.get("item"),
+            attack=data.get("attack", 0),
+            defense=data.get("defense", 0),
+            special_attack=data.get("special_attack", 0),
+            special_defense=data.get("special_defense", 0),
+            speed=data.get("speed", 0),
+            boosts=data.get("boosts", {}),
+            moves=data.get("moves", []),
+            fainted=data.get("fainted", False),
+            active=data.get("active", False),
+        )
+
+    def _parse_result(self, payload: Dict[str, Any]) -> StepResult[PokemonObservation]:
+        """
+        Parse server response into StepResult[PokemonObservation].
+
+        Args:
+            payload: JSON response from server.
+
+        Returns:
+            StepResult with PokemonObservation.
+        """
+        obs_data = payload.get("observation", {})
+
+        active_pokemon = None
+        if obs_data.get("active_pokemon"):
+            active_pokemon = self._parse_pokemon_data(obs_data["active_pokemon"])
+
+        opponent_active = None
+        if obs_data.get("opponent_active_pokemon"):
+            opponent_active = self._parse_pokemon_data(obs_data["opponent_active_pokemon"])
+
+        team = [self._parse_pokemon_data(p) for p in obs_data.get("team", [])]
+        opponent_team = [self._parse_pokemon_data(p) for p in obs_data.get("opponent_team", [])]
+
+        observation = PokemonObservation(
+            active_pokemon=active_pokemon,
+            opponent_active_pokemon=opponent_active,
+            team=team,
+            opponent_team=opponent_team,
+            available_moves=obs_data.get("available_moves", []),
+            available_switches=obs_data.get("available_switches", []),
+            legal_actions=obs_data.get("legal_actions", []),
+            field_conditions=obs_data.get("field_conditions", {}),
+            turn=obs_data.get("turn", 0),
+            forced_switch=obs_data.get("forced_switch", False),
+            can_mega_evolve=obs_data.get("can_mega_evolve", False),
+            can_dynamax=obs_data.get("can_dynamax", False),
+            can_terastallize=obs_data.get("can_terastallize", False),
+            battle_format=obs_data.get("battle_format", "gen8randombattle"),
+            battle_id=obs_data.get("battle_id"),
+            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[str, Any]) -> PokemonState:
+        """
+        Parse server response into PokemonState object.
+
+        Args:
+            payload: JSON response from /state endpoint.
+
+        Returns:
+            PokemonState object with environment state information.
+        """
+        return PokemonState(
+            episode_id=payload.get("episode_id"),
+            step_count=payload.get("step_count", 0),
+            battle_format=payload.get("battle_format", "gen8randombattle"),
+            player_username=payload.get("player_username", "player"),
+            server_url=payload.get("server_url", "localhost:8000"),
+            battle_id=payload.get("battle_id"),
+            is_battle_finished=payload.get("is_battle_finished", False),
+            battle_winner=payload.get("battle_winner"),
+        )

examples/project-pikachu/poke_env/models.py

@@ -0,0 +1,127 @@
+"""
+Data models for Pokemon Battle Environment.
+
+This module defines the Action, Observation, and State types for Pokemon battles
+via poke-env integration.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Literal, Optional
+
+from core.env_server import Action, Observation, State
+
+
+@dataclass
+class PokemonAction(Action):
+    """
+    Action for Pokemon battles.
+
+    Attributes:
+        action_type: Type of action - "move" or "switch"
+        action_index: Index of the move (0-3) or switch target (0-5)
+        move_id: Optional move identifier (e.g., "thunderbolt")
+        switch_pokemon: Optional Pokemon to switch to (by species name or index)
+        mega_evolve: Whether to mega evolve this turn (if applicable)
+        dynamax: Whether to dynamax this turn (if applicable)
+        terastallize: Whether to terastallize this turn (if applicable)
+    """
+    action_type: Literal["move", "switch"] = "move"
+    action_index: int = 0
+    move_id: Optional[str] = None
+    switch_pokemon: Optional[str] = None
+    mega_evolve: bool = False
+    dynamax: bool = False
+    terastallize: bool = False
+
+
+@dataclass
+class PokemonData:
+    """Simplified Pokemon data for observations."""
+    species: str
+    hp_percent: float
+    max_hp: int
+    current_hp: int
+    level: int
+    status: Optional[str]
+    types: List[str]
+    ability: Optional[str]
+    item: Optional[str]
+    
+    attack: int
+    defense: int
+    special_attack: int
+    special_defense: int
+    speed: int
+    
+    boosts: Dict[str, int] = field(default_factory=dict)
+    moves: List[Dict[str, Any]] = field(default_factory=list)
+    
+    fainted: bool = False
+    active: bool = False
+
+
+@dataclass
+class PokemonObservation(Observation):
+    """
+    Observation from Pokemon battle environment.
+
+    This represents the full battle state visible to the agent.
+
+    Attributes:
+        active_pokemon: Currently active Pokemon on your side
+        opponent_active_pokemon: Currently active opponent Pokemon
+        team: Your full team of 6 Pokemon
+        opponent_team: Opponent's team (may have limited visibility)
+        available_moves: List of move indices you can use (0-3)
+        available_switches: List of Pokemon indices you can switch to (0-5)
+        legal_actions: Combined list of legal action descriptors
+        field_conditions: Dict of field effects (weather, terrain, hazards, etc.)
+        turn: Current turn number
+        forced_switch: Whether you must switch (active Pokemon fainted)
+        can_mega_evolve: Whether mega evolution is possible this turn
+        can_dynamax: Whether dynamax is possible this turn
+        can_terastallize: Whether terastallization is possible this turn
+        battle_format: Battle format (e.g., "gen8randombattle", "gen8ou")
+    """
+    active_pokemon: Optional[PokemonData] = None
+    opponent_active_pokemon: Optional[PokemonData] = None
+    team: List[PokemonData] = field(default_factory=list)
+    opponent_team: List[PokemonData] = field(default_factory=list)
+    
+    available_moves: List[int] = field(default_factory=list)
+    available_switches: List[int] = field(default_factory=list)
+    legal_actions: List[Dict[str, Any]] = field(default_factory=list)
+    
+    field_conditions: Dict[str, Any] = field(default_factory=dict)
+    turn: int = 0
+    forced_switch: bool = False
+    
+    can_mega_evolve: bool = False
+    can_dynamax: bool = False
+    can_terastallize: bool = False
+    
+    battle_format: str = "gen8randombattle"
+    battle_id: Optional[str] = None
+
+
+@dataclass
+class PokemonState(State):
+    """
+    State for Pokemon battle environment.
+
+    Attributes:
+        battle_format: Battle format being used
+        player_username: Player's username
+        server_url: Pokemon Showdown server URL
+        battle_id: Current battle ID
+        is_battle_finished: Whether the battle has concluded
+        battle_winner: Winner of the battle (if finished)
+    """
+    battle_format: str = "gen8randombattle"
+    player_username: str = "player"
+    server_url: str = "localhost:8000"
+    battle_id: Optional[str] = None
+    is_battle_finished: bool = False
+    battle_winner: Optional[str] = None