### 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() ```