RLGym

https://rlgym.org/
Admin
RLGym is a Python API designed for creating and interacting with reinforcement learning...
Tokens:18,246
Snippets:72
Trust Score:6.1
Update:11 months ago
Show doc for...
Tokens:
Raw
### Install RLGym with RocketSim and RLGym-PPO

Source: https://rlgym.org/Getting%20Started/quickstart

Installs the RLGym library with RocketSim support and the RLGym-PPO package for training agents. Ensure you have pip installed. For faster training on NVIDIA GPUs, install PyTorch with CUDA support separately from pytorch.org.

```bash
pip install rlgym[rl-sim]
pip install git+https://github.com/AechPro/rlgym-ppo
```

--------------------------------

### Build RLGym v2 2v2 Environment with PPO Training

Source: https://rlgym.org/Getting%20Started/quickstart

Configures a 2v2 Rocket League environment with standard kickoff positions, goal and touch-based rewards, and no-touch timeout conditions. Returns a gym-wrapped environment for training. Dependencies include RLGym, RLGym PPO utilities, and NumPy. The environment uses RepeatAction for action parsing, DefaultObs for observations, and RocketSimEngine for physics simulation.

```python
def build_rlgym_v2_env():
    from rlgym.api import RLGym
    from rlgym.rocket_league.action_parsers import LookupTableAction, RepeatAction
    from rlgym.rocket_league.done_conditions import GoalCondition, NoTouchTimeoutCondition, TimeoutCondition, AnyCondition
    from rlgym.rocket_league.obs_builders import DefaultObs
    from rlgym.rocket_league.reward_functions import CombinedReward, GoalReward, TouchReward
    from rlgym.rocket_league.sim import RocketSimEngine
    from rlgym.rocket_league.state_mutators import MutatorSequence, FixedTeamSizeMutator, KickoffMutator
    from rlgym.rocket_league import common_values
    from rlgym_ppo.util import RLGymV2GymWrapper
    import numpy as np

    spawn_opponents = True
    team_size = 2
    blue_team_size = team_size
    orange_team_size = team_size if spawn_opponents else 0
    action_repeat = 8
    no_touch_timeout_seconds = 30
    game_timeout_seconds = 300

    action_parser = RepeatAction(LookupTableAction(), repeats=action_repeat)
    termination_condition = GoalCondition()
    truncation_condition = AnyCondition(NoTouchTimeoutCondition(timeout_seconds=no_touch_timeout_seconds), TimeoutCondition(timeout_seconds=game_timeout_seconds))

    reward_fn = CombinedReward((GoalReward(), 10), (TouchReward(), 0.1))

    obs_builder = DefaultObs(zero_padding=None,
                             pos_coef=np.asarray([1 / common_values.SIDE_WALL_X, 1 / common_values.BACK_NET_Y, 1 / common_values.CEILING_Z]),
                             ang_coef=1 / np.pi,
                             lin_vel_coef=1 / common_values.CAR_MAX_SPEED,
                             ang_vel_coef=1 / common_values.CAR_MAX_ANG_VEL,
                             boost_coef=1 / 100.0,)

    state_mutator = MutatorSequence(FixedTeamSizeMutator(blue_size=blue_team_size, orange_size=orange_team_size),
                                    KickoffMutator())
    rlgym_env = RLGym(
        state_mutator=state_mutator,
        obs_builder=obs_builder,
        action_parser=action_parser,
        reward_fn=reward_fn,
        termination_cond=termination_condition,
        truncation_cond=truncation_condition,
        transition_engine=RocketSimEngine())

    return RLGymV2GymWrapper(rlgym_env)
```

--------------------------------

### Initialize PPO Learner with Multi-Process Training Configuration

Source: https://rlgym.org/Getting%20Started/quickstart

Initializes a Learner object for training an agent using Proximal Policy Optimization across 8 parallel processes. Configures neural network layer sizes for both policy and critic networks, batch sizes, learning rates, entropy coefficient for exploration, and training duration (1 billion timesteps). Includes model checkpointing every 1 million steps and Weights & Biases logging integration.

```python
if __name__ == "__main__":
    from rlgym_ppo import Learner

    # 8 processes
    n_proc = 8

    # educated guess - could be slightly higher or lower
    min_inference_size = max(1, int(round(n_proc * 0.9)))

    learner = Learner(build_rlgym_v2_env,
                      n_proc=n_proc,
                      min_inference_size=min_inference_size,
                      metrics_logger=None,
                      ppo_batch_size=50000, # batch size - set this number to as large as your GPU can handle
                      policy_layer_sizes=[512, 512], # policy network
                      critic_layer_sizes=[512, 512], # value network
                      ts_per_iteration=50000, # timesteps per training iteration - set this equal to the batch size
                      exp_buffer_size=150000, # size of experience buffer - keep this 2 - 3x the batch size
                      ppo_minibatch_size=50000, # minibatch size - set this less than or equal to the batch size
                      ppo_ent_coef=0.01, # entropy coefficient - this determines the impact of exploration on the policy
                      policy_lr=5e-5, # policy learning rate
                      critic_lr=5e-5, # value function learning rate
                      ppo_epochs=1,   # number of PPO epochs
                      standardize_returns=True,
                      standardize_obs=False,
                      save_every_ts=1_000_000, # save every 1M steps
                      timestep_limit=1_000_000_000, # Train for 1B steps
                      log_to_wandb=True)
    learner.learn()
```

--------------------------------

### Install RLGym with RLViser Visualizer Support

Source: https://rlgym.org/Rocket%20League/training_an_agent

Installs the RLGym package with optional RLViser visualization support for monitoring agent training in Rocket League environments. This is a prerequisite for using the visualizer to watch agent learning progress.

```bash
pip install rlgym[rl-rlviser]
```

--------------------------------

### Use State Mutators with RLGym in Python

Source: https://rlgym.org/Rocket%20League/Configuration%20Objects/state_mutators

Shows how to initialize the RLGym environment with a sequence of state mutators. This example combines FixedTeamSizeMutator, KickoffMutator, and a CustomStateMutator to set up a 2v2 game with specific kickoff and custom initial states. It requires RLGym's API and rocket_league.state_mutators modules.

```python
from rlgym.api import RLGym
from rlgym.rocket_league.state_mutators import MutatorSequence, FixedTeamSizeMutator, KickoffMutator

# Create environment with a sequence of mutators
env = RLGym(
    state_mutator=MutatorSequence(
        FixedTeamSizeMutator(blue_size=2, orange_size=2),  # Set up 2v2 game
        KickoffMutator(),  # Set up kickoff positions
        CustomStateMutator()  # Apply our custom state changes
    ),
    # ... other configuration objects ...
)

```

--------------------------------

### Create Continuous Action Parser - Python

Source: https://rlgym.org/Rocket%20League/Configuration%20Objects/action_parsers

An example Python class `ContinuousAction` that inherits from `ActionParser`. It demonstrates parsing an array of 8 continuous values ([-1, 1]) into valid car controls, converting the last 3 values to binary (0 or 1) as expected by the game.

```python
from typing import Dict, Any

import numpy as np

from rlgym.api import ActionParser, AgentID
from rlgym.rocket_league.api import GameState

class ContinuousAction(ActionParser[AgentID, np.ndarray, np.ndarray, GameState, int]):
    """
    Simple continuous action space that maps an array of 8 values on the interval [-1, 1] into an array of valid car
    controls.
    """

    def __init__(self):
        super().__init__()
        # Rocket League expects 8 values per controller input.
        self._n_controller_inputs = 8
        
    def get_action_space(self) -> tuple:
        return float(self._n_controller_inputs), 'continuous'

    def reset(self, initial_state: GameState, shared_info: Dict[str, Any]) -> None:
        pass

    def parse_actions(self, actions: Dict[AgentID, np.ndarray], state: GameState, shared_info: Dict[str, Any]) -> Dict[AgentID, np.ndarray]:
        parsed_actions = {}
          
        # Loop over the agent action dictionary
        for agent, action in actions.items():
            # Copy the action into a new array
            car_controls = np.zeros(self._n_controller_inputs)
            car_controls[:] = action[:]
              
            # All the actions from our policy will be on the interval [-1, 1], but the last 3 values in the car controls
            # need to be either 0 or 1. We will shift and round the result such that any value below 0 becomes 0 and
            # any value above 0 becomes 1.
            car_controls[-3:] = np.round((car_controls[-3:] + 1) / 2)
            parsed_actions[agent] = car_controls
  
        return parsed_actions
```

--------------------------------

### Implement a Simple Ball Position Renderer (Python)

Source: https://rlgym.org/Rocket%20League/Configuration%20Objects/renderers

Example of a custom renderer that inherits from RLGym's `Renderer` and prints the ball's current position. It implements the `render` method to access and display `state.ball.position`.

```python
from typing import Dict, Any
import numpy as np
from rlgym.api import Renderer
from rlgym.rocket_league.api import GameState

class BallPositionRenderer(Renderer[GameState]):
    """A simple renderer that prints the ball's position."""
      
    def render(self, state: GameState, shared_info: Dict[str, Any]) -> Any:
        print(f"Ball position: {state.ball.position}")
      
    def close(self):
        """Called when the environment is closed."""
        pass

```

--------------------------------

### Implement Custom State Mutator in Python

Source: https://rlgym.org/Rocket%20League/Configuration%20Objects/state_mutators

Demonstrates how to create a custom StateMutator by inheriting from the base class and implementing the 'apply' method. This example sets custom positions and orientations for cars and the ball in the GameState. It requires 'typing', 'numpy', and RLGym's API and rocket_league modules.

```python
from typing import Dict, Any
import numpy as np
from rlgym.api import StateMutator
from rlgym.rocket_league.api import GameState
from rlgym.rocket_league import common_values

class CustomStateMutator(StateMutator[GameState]):
    """A StateMutator that sets custom positions for cars and the ball."""
      
    def apply(self, state: GameState, shared_info: Dict[str, Any]) -> None:
        # Define spawn location and orientation
        desired_car_pos = np.array([100, 100, 17], dtype=np.float32)  # x, y, z
        desired_yaw = np.pi/2

        # Iterate over all cars in the game
        for car in state.cars.values():
            if car.is_orange:
                # Orange team positions
                pos = desired_car_pos
                yaw = desired_yaw
            else:
                # Blue team positions (inverted)
                pos = -desired_car_pos
                yaw = -desired_yaw

            # Set car physics state
            car.physics.position = pos
            car.physics.euler_angles = np.array([0, 0, yaw], dtype=np.float32)
            car.boost = 33

        # Set ball physics state
        state.ball.position = np.array([0, 0, common_values.CEILING_Z/2], dtype=np.float32)
        state.ball.linear_velocity = np.zeros(3, dtype=np.float32)
        state.ball.angular_velocity = np.zeros(3, dtype=np.float32)

```

--------------------------------

### Create Custom Reward Functions for RLGym Agent

Source: https://rlgym.org/Rocket%20League/training_an_agent

Implements three custom reward functions (SpeedTowardBallReward, InAirReward, VelocityBallToGoalReward) that inherit from RewardFunction to guide agent learning. These functions calculate rewards based on agent velocity toward the ball, being airborne, and ball velocity toward the opponent's goal, providing richer feedback than basic rewards.

```python
from typing import List, Dict, Any
from rlgym.api import RewardFunction, AgentID
from rlgym.rocket_league.api import GameState
from rlgym.rocket_league import common_values
import numpy as np

class SpeedTowardBallReward(RewardFunction[AgentID, GameState, float]):
    """Rewards the agent for moving quickly toward the ball"""
    
    def reset(self, agents: List[AgentID], initial_state: GameState, shared_info: Dict[str, Any]) -> None:
        pass
    
    def get_rewards(self, agents: List[AgentID], state: GameState, is_terminated: Dict[AgentID, bool],
                    is_truncated: Dict[AgentID, bool], shared_info: Dict[str, Any]) -> Dict[AgentID, float]:
        rewards = {}
        for agent in agents:
            car = state.cars[agent]
            car_physics = car.physics if car.is_orange else car.inverted_physics
            ball_physics = state.ball if car.is_orange else state.inverted_ball
            player_vel = car_physics.linear_velocity
            pos_diff = (ball_physics.position - car_physics.position)
            dist_to_ball = np.linalg.norm(pos_diff)
            dir_to_ball = pos_diff / dist_to_ball

            speed_toward_ball = np.dot(player_vel, dir_to_ball)

            rewards[agent] = max(speed_toward_ball / common_values.CAR_MAX_SPEED, 0.0)
        return rewards

class InAirReward(RewardFunction[AgentID, GameState, float]):
    """Rewards the agent for being in the air"""
    
    def reset(self, agents: List[AgentID], initial_state: GameState, shared_info: Dict[str, Any]) -> None:
        pass
    
    def get_rewards(self, agents: List[AgentID], state: GameState, is_terminated: Dict[AgentID, bool],
                    is_truncated: Dict[AgentID, bool], shared_info: Dict[str, Any]) -> Dict[AgentID, float]:
        return {agent: float(not state.cars[agent].on_ground) for agent in agents}

class VelocityBallToGoalReward(RewardFunction[AgentID, GameState, float]):
    """Rewards the agent for hitting the ball toward the opponent's goal"""
    
    def reset(self, agents: List[AgentID], initial_state: GameState, shared_info: Dict[str, Any]) -> None:
        pass
    
    def get_rewards(self, agents: List[AgentID], state: GameState, is_terminated: Dict[AgentID, bool],
                    is_truncated: Dict[AgentID, bool], shared_info: Dict[str, Any]) -> Dict[AgentID, float]:
        rewards = {}
        for agent in agents:
            car = state.cars[agent]
            ball = state.ball
            if car.is_orange:
                goal_y = -common_values.BACK_NET_Y
            else:
                goal_y = common_values.BACK_NET_Y

            ball_vel = ball.linear_velocity
            pos_diff = np.array([0, goal_y, 0]) - ball.position
            dist = np.linalg.norm(pos_diff)
            dir_to_goal = pos_diff / dist
            
            vel_toward_goal = np.dot(ball_vel, dir_to_goal)
            rewards[agent] = max(vel_toward_goal / common_values.BALL_MAX_SPEED, 0)
        return rewards
```

--------------------------------

### RLGym-PPO Learner Initialization and Training

Source: https://rlgym.org/Rocket%20League/training_an_agent

Initializes and runs the RLGym-PPO learner. This involves setting up the environment, number of processes, inference size, and numerous PPO-specific hyperparameters for policy and critic networks, learning rates, batch sizes, and training duration.

```python
if __name__ == "__main__":
    from rlgym_ppo import Learner

    n_proc = 32
    min_inference_size = max(1, int(round(n_proc * 0.9)))

    learner = Learner(build_rlgym_v2_env,
                      n_proc=n_proc,
                      min_inference_size=min_inference_size,
                      metrics_logger=None,
                      ppo_batch_size=100_000,
                      policy_layer_sizes=[2048, 2048, 1024, 1024],
                      critic_layer_sizes=[2048, 2048, 1024, 1024],
                      ts_per_iteration=100_000,
                      exp_buffer_size=300_000,
                      ppo_minibatch_size=50_000,
                      ppo_ent_coef=0.01,
                      policy_lr=1e-4,
                      critic_lr=1e-4,
                      ppo_epochs=2,
                      standardize_returns=True,
                      standardize_obs=False,
                      save_every_ts=1_000_000,
                      timestep_limit=1_000_000_000,
                      log_to_wandb=False
                      )
    learner.learn()
```

--------------------------------

### RLGym Speed Reward Function Example (Python)

Source: https://rlgym.org/Rocket%20League/Configuration%20Objects/reward_functions

A sample RLGym reward function that calculates rewards based on agent speed. It utilizes the `np.linalg.norm` to determine the magnitude of the linear velocity. This function can be directly passed to an RLGym environment.

```python
from typing import List, Dict, Any, Union, Tuple  

from rlgym.api import RewardFunction, AgentID  
from rlgym.rocket_league.api import GameState  
import numpy as np  


class SpeedReward(RewardFunction):  
  def reset(self, agents: List[AgentID], initial_state: GameState, shared_info: Dict[str, Any]) -> None:  
      pass  
    
  def get_rewards(self, agents: List[AgentID], state: GameState, is_terminated: Dict[AgentID, bool],  
                  is_truncated: Dict[AgentID, bool], shared_info: Dict[str, Any]) -> Dict[AgentID, float]:  
        
    rewards = {}  
    for agent in agents:  
        car = state.cars[agent]  
        linear_velocity = car.physics.linear_velocity  
        reward = np.linalg.norm(linear_velocity)  # Reward is the car's speed  
        rewards[agent] = reward  
          
    return rewards  

```

--------------------------------

### Configure RLGym v2 Environment with Action Parsers and Done Conditions

Source: https://rlgym.org/Rocket%20League/training_an_agent

Sets up the RLGym v2 environment with configuration parameters including team sizes, action repeat, timeout conditions, action parser, and termination/truncation conditions. This function initializes the core environment for training agents with customizable game rules and action spaces.

```python
def build_rlgym_v2_env():
    import numpy as np
    from rlgym.api import RLGym
    from rlgym.rocket_league.action_parsers import LookupTableAction, RepeatAction
    from rlgym.rocket_league.done_conditions import GoalCondition, NoTouchTimeoutCondition, TimeoutCondition, AnyCondition
    from rlgym.rocket_league.obs_builders import DefaultObs
    from rlgym.rocket_league.reward_functions import CombinedReward, GoalReward
    from rlgym.rocket_league.sim import RocketSimEngine
    from rlgym.rocket_league.state_mutators import MutatorSequence, FixedTeamSizeMutator, KickoffMutator
    from rlgym.rocket_league import common_values
    from rlgym_ppo.util import RLGymV2GymWrapper

    spawn_opponents = True
    team_size = 1
    blue_team_size = team_size
    orange_team_size = team_size if spawn_opponents else 0
    action_repeat = 8
    no_touch_timeout_seconds = 30
    game_timeout_seconds = 300

    action_parser = RepeatAction(LookupTableAction(), repeats=action_repeat)
    termination_condition = GoalCondition()
    truncation_condition = AnyCondition(
```

--------------------------------

### RLGym Action Parser Methods

Source: https://rlgym.org/Rocket%20League/Configuration%20Objects/action_parsers

Defines the three core methods required for any RLGym ActionParser: `get_action_space` to define the policy's output, `reset` for environment initialization, and `parse_actions` to translate actions into game inputs.

```python
def get_action_space(self, agent: AgentID) -> SpaceType:
  
# Called every time `TransitionEngine.create_base_state()` is called.
def reset(self, initial_state: StateType, shared_info: Dict[str, Any]) -> None:
      
# Called every time `TransitionEngine.step()` is called. This function is responsible for translating actions into transition engine inputs for each agent.
def parse_actions(self, actions: Dict[AgentID, ActionType], state: StateType, shared_info: Dict[str, Any]) -> Dict[AgentID, EngineActionType]:
```

--------------------------------

### Build RLGym Environment

Source: https://rlgym.org/Rocket%20League/training_an_agent

Defines and returns a configured RLGym environment wrapped for Gym compatibility. It sets up termination and truncation conditions, reward functions, observation builders, state mutators, and the simulation engine.

```python
def build_rlgym_v2_env():
    no_touch_timeout_seconds = 10.0
    game_timeout_seconds = 300.0
    blue_team_size = 1
    orange_team_size = 1

    termination_condition = AnyTerminationCondition(
        NoTouchTimeoutCondition(timeout_seconds=no_touch_timeout_seconds),
        TimeoutCondition(timeout_seconds=game_timeout_seconds)
    )

    reward_fn = CombinedReward(
        (InAirReward(), 0.002),
        (SpeedTowardBallReward(), 0.01),
        (VelocityBallToGoalReward(), 0.1),
        (GoalReward(), 10.0)
    )

    obs_builder = DefaultObs(zero_padding=None,
                           pos_coef=np.asarray([1 / common_values.SIDE_WALL_X,
                                              1 / common_values.BACK_NET_Y,
                                              1 / common_values.CEILING_Z]),
                           ang_coef=1 / np.pi,
                           lin_vel_coef=1 / common_values.CAR_MAX_SPEED,
                           ang_vel_coef=1 / common_values.CAR_MAX_ANG_VEL,
                           boost_coef=1 / 100.0)

    state_mutator = MutatorSequence(
        FixedTeamSizeMutator(blue_size=blue_team_size, orange_size=orange_team_size),
        KickoffMutator()
    )

    rlgym_env = RLGym(
        state_mutator=state_mutator,
        obs_builder=obs_builder,
        action_parser=action_parser,
        reward_fn=reward_fn,
        termination_cond=termination_condition,
        truncation_cond=truncation_condition,
        transition_engine=RocketSimEngine()
    )

    return RLGymV2GymWrapper(rlgym_env)
```

--------------------------------

### Construct RLGym Environment with Configuration Objects

Source: https://rlgym.org/Custom%20Environments/custom-environment

Instantiates the RLGym environment by passing configured instances of all component classes: state mutator, observation builder, action parser, reward function, transition engine, and terminal/truncation conditions. This creates a complete multi-agent reinforcement learning environment ready for training or simulation.

```python
# Build the environment
env = RLGym(
    state_mutator=GridWorldMutator(grid_size),
    obs_builder=GridWorldObs(),
    action_parser=GridWorldActions(),
    reward_fn=GridWorldReward(),
    transition_engine=GridWorldEngine(grid_size),
    termination_cond=GridWorldTerminalCondition(),
    truncation_cond=GridWorldTruncatedCondition()
)
```

--------------------------------

### Initialize RLGym Environment with a Custom Renderer (Python)

Source: https://rlgym.org/Rocket%20League/Configuration%20Objects/renderers

Demonstrates how to instantiate the RLGym environment and pass a custom renderer, such as `BallPositionRenderer`, to its constructor. This integrates the renderer for subsequent use.

```python
env = RLGym(
    # ... other configuration objects ...
    renderer=BallPositionRenderer()
)

```

--------------------------------

### Define GridWorldState and GridWorldEngine Transition Engine in Python

Source: https://rlgym.org/Custom%20Environments/custom-environment

Establishes the state representation using a dataclass with agent position, target position, obstacles, and grid size. Implements the TransitionEngine interface to handle core environment dynamics including movement (up/right/down/left), boundary clipping, and collision detection with obstacles. The engine supports single-agent control through step() method and state initialization through create_base_state() and reset() methods.

```python
from typing import Dict, List, Tuple, Optional
import numpy as np
from dataclasses import dataclass
from rlgym.api import TransitionEngine, StateMutator, ObsBuilder, ActionParser, RewardFunction, DoneCondition

# First, define our state type
@dataclass
class GridWorldState:
    agent_pos: np.ndarray  # [x, y]
    target_pos: np.ndarray  # [x, y]
    obstacles: List[np.ndarray]  # List of [x, y] positions
    grid_size: int
    steps: int = 0

# Now we implement our Transition Engine, which is the core of the environment.
class GridWorldEngine(TransitionEngine[int, GridWorldState, int]):
    """Handles the core game logic"""
    def __init__(self, grid_size: int):
        self.grid_size = grid_size
        self._state = None
        self._config = {}

    @property
    def agents(self) -> List[int]:
        return [0]  # Single agent environment

    @property
    def max_num_agents(self) -> int:
        return 1  # This environment only supports one agent

    @property
    def state(self) -> GridWorldState:
        return self._state

    @property
    def config(self) -> Dict[str, Any]:
        return self._config

    @config.setter
    def config(self, value: Dict[str, Any]):
        self._config = value

    def step(self, actions: Dict[int, int], shared_info: Dict[str, Any]) -> GridWorldState:
        action = actions[0]  # Get action for our single agent
        current_pos = self._state.agent_pos.copy()

        # Apply movement: 0=up, 1=right, 2=down, 3=left
        if action == 0:   current_pos[1] += 1
        elif action == 1: current_pos[0] += 1
        elif action == 2: current_pos[1] -= 1
        elif action == 3: current_pos[0] -= 1

        # Ensure we stay in bounds
        current_pos = np.clip(current_pos, 0, self.grid_size - 1)

        # Check if move is valid (not into obstacle)
        if not any(np.array_equal(current_pos, obs) for obs in self._state.obstacles):
            self._state.agent_pos = current_pos

        self._state.steps += 1
        return self._state

    def create_base_state(self) -> GridWorldState:
        # Create a minimal state for the mutator to modify
        return GridWorldState(
            agent_pos=np.zeros(2),  # Will be set by mutator
            target_pos=np.zeros(2),  # Will be set by mutator
            obstacles=[],  # Will be set by mutator
            grid_size=self.grid_size,
            steps=0
        )

    def reset(self, initial_state: Optional[GridWorldState] = None) -> None:
        """Reset the engine with an optional initial state"""
        self._state = initial_state if initial_state is not None else self.create_base_state()
```

--------------------------------

### GridWorldObs - Convert Environment State to Agent Observations

Source: https://rlgym.org/Custom%20Environments/custom-environment

Observation builder that converts grid world state into numpy arrays containing agent position, target position, and obstacle locations. Returns fixed-size observation space of 10 elements (4 for positions + 6 for up to 3 obstacles). The build_obs method concatenates state information for each agent into observation dictionaries.

```python
class GridWorldObs(ObsBuilder[int, np.ndarray, GridWorldState, np.ndarray]):
    """Converts state into agent observations"""
    def get_obs_space(self, agent: int) -> np.ndarray:
        # [agent_x, agent_y, target_x, target_y, obstacles]
        return np.zeros(4 + 2*3, dtype=np.float32)  # Assuming max 3 obstacles
    
    def reset(self, agents: List[int], initial_state: GridWorldState, shared_info: Dict[str, Any]) -> None:
        pass
    
    def build_obs(self, agents: List[int], state: GridWorldState, shared_info: Dict[str, Any]) -> Dict[int, np.ndarray]:
        # Build observation for each agent
        observations = {}
        for agent in agents:
            # [agent_x, agent_y, target_x, target_y, obstacle positions]
            obs = np.concatenate([
                state.agent_pos,
                state.target_pos,
                np.concatenate(state.obstacles)
            ])
            observations[agent] = obs
        return observations
```

--------------------------------

### Interact with RLGym Gridworld Environment

Source: https://rlgym.org/Custom%20Environments/custom-environment

This snippet shows the basic interaction loop with an RLGym environment. It includes resetting the environment, sampling random actions, stepping through the environment, accumulating rewards, and handling episode completion (done or truncated). The environment is prepared for agent training.

```python
obs = env.reset()
ep_rew = 0

for _ in range(1000):
    action = {0: env.action_space.sample()}  # Random action
    obs, reward, done, truncated, info = env.step(action)
    ep_rew += reward[0] # Reward for agent 0
      
    if done or truncated:
        obs = env.reset()
        print(f"Episode reward: {ep_rew}")
        ep_rew = 0
```

--------------------------------

### Define Renderer Interface Methods (Python)

Source: https://rlgym.org/Rocket%20League/Configuration%20Objects/renderers

The Renderer interface requires two methods: `render` for displaying the game state, and `close` for cleanup when the environment is terminated. Arguments for `render` are automatically provided by the RLGym environment.

```python
# Must be manually called by the user via RLGym.render(). Arguments are provided automatically by the RLGym environment.  
def render(self, state: StateType, shared_info: Dict[str, Any]) -> Any:  
  
# Called once when the environment is closed.  
def close(self) -> None:  


```

--------------------------------

### GridWorldActions - Define Discrete Action Space Parser

Source: https://rlgym.org/Custom%20Environments/custom-environment

Action parser that defines a discrete action space with 4 possible actions (Up, Right, Down, Left). The parse_actions method passes agent actions directly through without modification since they are already in the correct integer format. Implements action space definition and reset methods required by ActionParser interface.

```python
class GridWorldActions(ActionParser[int, int, int, GridWorldState, int]):
    """Defines the action space and parsing"""
    def get_action_space(self, agent: int) -> int:
        return 4  # Up, Right, Down, Left
    
    def reset(self, agents: List[int], initial_state: GridWorldState, shared_info: Dict[str, Any]) -> None:
        pass  # No state to reset
    
    def parse_actions(self, actions: Dict[int, int], state: GridWorldState, shared_info: Dict[str, Any]) -> Dict[int, int]:
        # Actions are already in the correct format
        return actions
```

--------------------------------

### Configure RLGym Environment with Scoreboard and Match Termination

Source: https://rlgym.org/RLGym%20Tools/introduction

Integrates ScoreboardProvider with RLGym to track in-game scoreboard through the shared_info interface and enforce standard Rocket League match termination rules. Combines GameMutator and GameCondition objects to create an environment that behaves like a standard Rocket League match, including overtime logic.

```python
from rlgym.api import RLGym
from rlgym_tools.rocket_league.state_mutators.game_mutator import GameMutator
from rlgym_tools.rocket_league.shared_info_providers.scoreboard_provider import ScoreboardProvider
from rlgym_tools.rocket_league.done_conditions.game_condition import GameCondition

# Configure an environment to behave like a standard Rocket League match
env = RLGym(
    state_mutator=GameMutator(),
    shared_info_provider=ScoreboardProvider(),
    termination_cond=GameCondition(),
    ...
)
```

--------------------------------

### Define Observation Builder Interface Methods (Python)

Source: https://rlgym.org/Rocket%20League/Configuration%20Objects/observation_builders

This snippet outlines the three essential methods required for creating a custom observation builder in RLGym: `get_obs_space` to define the observation space, `reset` to initialize the builder, and `build_obs` to generate observations for each agent at each timestep. These methods are fundamental for integrating custom observation logic into the RLGym environment.

```python
def get_obs_space(self, agent: AgentID) -> ObsSpaceType:
      
# Called every time `TransitionEngine.create_base_state()` is called.  
def reset(self, agents: List[AgentID], initial_state: StateType, shared_info: Dict[str, Any]) -> None:  
  
# Called every time `TransitionEngine.step()` or `TransitionEngine.create_base_state()` is called.  
def build_obs(self, agents: List[AgentID], state: StateType, shared_info: Dict[str, Any]) -> Dict[AgentID, ObsType]:  
```

--------------------------------

### Create Custom Observation Builder (Python)

Source: https://rlgym.org/Rocket%20League/Configuration%20Objects/observation_builders

This Python code defines a `CustomObsBuilder` class that inherits from RLGym's `ObsBuilder`. It demonstrates how to implement `get_obs_space`, `reset`, and `build_obs` to create observations containing the physics state of the car and the ball. The `_build_obs` helper method is used to construct the observation vector for a single agent.

```python
from rlgym.api import ObsBuilder, AgentID  
from rlgym.rocket_league.api import Car, GameState  
from typing import List, Dict, Any, Tuple  
import numpy as np  
  
  
class CustomObsBuilder(ObsBuilder):  
    def get_obs_space(self, agent: AgentID) -> Tuple[str, int]:  
        # Each observation will have 24 numbers (physics data for car and ball)  
        return 'real', 24  
      
    def reset(self, agents: List[AgentID], initial_state: GameState, shared_info: Dict[str, Any]) -> None:  
        pass  
  
    def build_obs(self, agents: List[AgentID], state: GameState, shared_info: Dict[str, Any]) -> Dict[AgentID, np.ndarray]:  
        obs = {}  
          
        # Build an observation for each agent  
        for agent in agents:  
            obs[agent] = self._build_obs(agent, state, shared_info)  
          
        return obs  
      
    def _build_obs(self, agent: AgentID, state: GameState, shared_info: Dict[str, Any]) -> np.ndarray:  
        # Here we build an observation for a single agent.  
        agent_physics_state = state.cars[agent].physics  
        ball = state.ball  
          
        agent_obs = [  
            # Ball physics data.  
            ball.position,  
            ball.linear_velocity,  
            ball.angular_velocity,  
              
            # Agent physics data. Forward and up vectors are used because they fully specify the orientation of the car.  
            agent_physics_state.position,  
            agent_physics_state.forward,  
            agent_physics_state.up,  
            agent_physics_state.linear_velocity,  
            agent_physics_state.angular_velocity  
        ]  
          
        return np.concatenate(agent_obs)  
```

--------------------------------

### Parse Rocket League Replay Files to RLGym GameState Objects

Source: https://rlgym.org/RLGym%20Tools/introduction

Converts Rocket League replay files (.replay) into sequences of RLGym GameState objects for replay analysis and imitation learning. The ParsedReplay class loads the replay file, and replay_to_rlgym() converts it into ReplayFrame objects containing game state, player actions, scoreboard info, and match outcome data.

```python
from rlgym_tools.rocket_league.replays.parsed_replay import ParsedReplay
from rlgym_tools.rocket_league.replays.convert import replay_to_rlgym

# Parse replay file
replay = ParsedReplay.load("my_replay.replay")

# Convert to ReplayFrame sequence
replay_frames = replay_to_rlgym(replay)

# Now we can iterate over the replay frames.
for replay_frame in replay_frames:
    # ReplayFrame structure:
    #     state: GameState                  # Game state representation
    #     actions: Dict[int, np.ndarray]    # Actions for each player
    #     update_age: Dict[int, float]      # Time delta since last update for each player
    #     scoreboard: ScoreboardInfo        # Scoreline and game timer
    #     episode_seconds_remaining: float  # Time remaining until goal or ball hitting ground at 0s
    #     next_scoring_team: Optional[int]  # Team that scores next goal, or None if ball hits ground at 0s
    #     winning_team: Optional[int]       # Team that wins this match, or None if there is no winner
```

--------------------------------

### GridWorldReward - Calculate Agent Rewards with Goal and Step Penalties

Source: https://rlgym.org/Custom%20Environments/custom-environment

Reward function that provides positive reward (default 10.0) when agent reaches target position and negative step penalty (default -0.1) for each step taken. Compares agent position to target position using numpy array equality to determine reward value. Supports customizable goal reward and step penalty parameters through constructor.

```python
class GridWorldReward(RewardFunction[int, GridWorldState, float]):
    """Calculates rewards for actions"""
    def __init__(self, goal_reward: float = 10.0, step_penalty: float = -0.1):
        self.goal_reward = goal_reward
        self.step_penalty = step_penalty
    
    def reset(self, agents: List[int], initial_state: GridWorldState, shared_info: Dict[str, Any]) -> None:
        pass  # No state to reset
    
    def get_rewards(self, agents: List[int], state: GridWorldState,
                   is_terminated: Dict[int, bool], is_truncated: Dict[int, bool],
                   shared_info: Dict[str, Any]) -> Dict[int, float]:
        rewards = {}
        for agent in agents:
            if np.array_equal(state.agent_pos, state.target_pos):
                # If we reached the target, provide the goal reward.
                rewards[agent] = self.goal_reward
            else:
                # If we haven't reached the target, apply a step penalty.
                rewards[agent] = self.step_penalty
        return rewards
```

--------------------------------

### Implement GridWorldMutator StateMutator in Python

Source: https://rlgym.org/Custom%20Environments/custom-environment

Implements the StateMutator interface to handle environment state modifications during reset. Randomly initializes agent position, target position, and obstacles while preventing overlaps between these elements. The apply() method modifies the state in-place, ensuring valid configurations within the grid boundaries before each episode.

```python
# We need to define a state mutator, which is responsible for modifying the environment state.
class GridWorldMutator(StateMutator[GridWorldState]):
    """Controls environment reset and state modifications"""
    def __init__(self, grid_size: int, num_obstacles: int = 3):
        self.grid_size = grid_size
        self.num_obstacles = num_obstacles

    def apply(self, state: GridWorldState, shared_info: Dict[str, Any]) -> None:
        # Random agent and target positions
        state.agent_pos = np.random.randint(0, self.grid_size, size=2)
        state.target_pos = np.random.randint(0, self.grid_size, size=2)

        # Random obstacle positions (ensuring they don't overlap)
        state.obstacles = []
        while len(state.obstacles) < self.num_obstacles:
            obs = np.random.randint(0, self.grid_size, size=2)
            if not (np.array_equal(obs, state.agent_pos) or
                   np.array_equal(obs, state.target_pos) or
                   any(np.array_equal(obs, o) for o in state.obstacles)):
                state.obstacles.append(obs)
```

--------------------------------

### Render Game State During Simulation Loop (Python)

Source: https://rlgym.org/Rocket%20League/Configuration%20Objects/renderers

Shows how to call `env.render()` within the main simulation loop after `env.step()`. This action triggers the `render` method of the registered custom renderer to visualize the current game state.

```python
obs = env.reset()
while True:
    actions = {agent_id: env.action_space.sample() for agent_id in env.agents}
    obs, reward, terminated, truncated, info = env.step(actions)
    env.render()  # This will call your renderer's `render` method
      
    if terminated or truncated:
        break

```

--------------------------------

### Custom RLGym Observation Builder with Inverted Physics

Source: https://rlgym.org/Rocket%20League/Configuration%20Objects/observation_builders

This Python code defines a custom `ObsBuilder` that incorporates `inverted_physics` to present observations from the blue team's perspective. It handles team-specific physics data for cars and the ball, outputting a standardized observation vector.

```python
from rlgym.api import ObsBuilder, AgentID
from rlgym.rocket_league.api import Car, GameState
from rlgym.rocket_league.common_values import BLUE_TEAM
from typing import List, Dict, Any, Tuple
import numpy as np


class CustomObsBuilder(ObsBuilder):
    def get_obs_space(self, agent: AgentID) -> Tuple[str, int]:
        # Each observation will have 24 numbers (physics data for car and ball)
        return 'real', 24
      
    def reset(self, agents: List[AgentID], initial_state: GameState, shared_info: Dict[str, Any]) -> None:
        pass

    def build_obs(self, agents: List[AgentID], state: GameState, shared_info: Dict[str, Any]) -> Dict[AgentID, np.ndarray]:
        obs = {}
          
        # Build an observation for each agent
        for agent in agents:
            obs[agent] = self._build_obs(agent, state, shared_info)
          
        return obs
      
    def _build_obs(self, agent: AgentID, state: GameState, shared_info: Dict[str, Any]) -> np.ndarray:
        # We will first grab the car that is being controlled by this agent.
        car = state.cars[agent]
          
        # Then we'll check if this agent is on the blue team already.
        if car.team_num == BLUE_TEAM:
            agent_physics_state = state.cars[agent].physics
            ball = state.ball
        else:
            # If this agent isn't on the blue team we'll use the inverted physics information for both the car and the ball.
            agent_physics_state = state.cars[agent].inverted_physics
            ball = state.inverted_ball
          
        # The rest of the code can remain unchanged!
        agent_obs = [
            # Ball physics data.
            ball.position,
            ball.linear_velocity,
            ball.angular_velocity,
              
            # Agent physics data. Forward and up vectors are used because they fully specify the orientation of the car.
            agent_physics_state.position,
            agent_physics_state.forward,
            agent_physics_state.up,
            agent_physics_state.linear_velocity,
            agent_physics_state.angular_velocity
        ]
          
        return np.concatenate(agent_obs)

```

--------------------------------

### GridWorldTruncatedCondition - Detect Episode Timeout

Source: https://rlgym.org/Custom%20Environments/custom-environment

Truncation condition that cuts episodes short when maximum step count is exceeded (default 100 steps). The is_done method compares current step count to max_steps threshold. Prevents infinite episodes and ensures training progresses with reasonable episode lengths.

```python
class GridWorldTruncatedCondition(DoneCondition[int, GridWorldState]):
    """Determines when episodes are cut short (timeout)"""
    def __init__(self, max_steps: int = 100):
        self.max_steps = max_steps
    
    def reset(self, agents: List[int], initial_state: GridWorldState, shared_info: Dict[str, Any]) -> None:
        pass
    
    def is_done(self, agent: int, state: GridWorldState) -> bool:
        # Episode is truncated if we exceed max steps
        return state.steps >= self.max_steps
```

--------------------------------

### Close RLGym Environment and Resources (Python)

Source: https://rlgym.org/Rocket%20League/Configuration%20Objects/renderers

Illustrates the final step of calling `env.close()` after the simulation loop has finished. This method ensures that any resources managed by the RLGym environment and its associated renderers are properly cleaned up.

```python
env.close()

```
Tokens:
Raw