### CARLA Environment Setup and Episode Run in Python

Source: https://context7.com/augcog/roar_py_rl/llms.txt

This snippet demonstrates the direct setup of the RoarRLCarlaSimEnv in Python. It connects to a CARLA server, configures simulation steps, spawns a vehicle with various sensors (collision, velocimeter, location, roll/pitch/yaw), creates the environment instance, applies action and observation wrappers, and then runs a basic episode loop with random actions. The environment implements distance-based reward and terminates on collision.

```python
from roar_py_rl_carla import RoarRLCarlaSimEnv, FlattenActionWrapper
from roar_py_carla import RoarPyCarlaInstance
import carla
import gymnasium as gym
import numpy as np

# Connect to CARLA server
carla_client = carla.Client("localhost", 2000)
carla_client.set_timeout(15.0)
roar_py_instance = RoarPyCarlaInstance(carla_client)
world = roar_py_instance.world

# Configure simulation timesteps
control_timestep = 0.04  # 25 FPS
physics_timestep = 0.008  # 5 substeps per control step
world.set_control_steps(control_timestep, physics_timestep)
world.set_asynchronous(False)

# Spawn vehicle with sensors
spawn_point = world.spawn_points[0]
vehicle = world.spawn_vehicle(
    "vehicle.tesla.model3",
    spawn_point[0] + np.array([0, 0, 2.0]),
    spawn_point[1],
    True,
    "vehicle"
)

# Attach required sensors
collision_sensor = vehicle.attach_collision_sensor(
    np.zeros(3), np.zeros(3), name="collision_sensor"
)
velocimeter_sensor = vehicle.attach_velocimeter_sensor("velocimeter")
location_sensor = vehicle.attach_location_in_world_sensor("location")
rpy_sensor = vehicle.attach_roll_pitch_yaw_sensor("roll_pitch_yaw")

# Create CARLA-specific environment
env = RoarRLCarlaSimEnv(
    vehicle,
    world.maneuverable_waypoints,
    location_sensor,
    rpy_sensor,
    velocimeter_sensor,
    collision_sensor,
    waypoint_information_distances=set([2.0, 5.0, 10.0, 20.0, 40.0, 80.0]),
    world=world,
    collision_threshold=1.0
)

# Apply action wrapper to simplify to throttle/steer only
env = FlattenActionWrapper(env)
env = gym.wrappers.FlattenObservation(env)

# Reset automatically randomizes spawn location
obs, info = env.reset()
# Vehicle is placed at random spawn point from world.spawn_points
# Vehicle is held stationary for 1 second with brakes applied
# Collision sensor is checked to ensure clean spawn

# Run episode
for _ in range(1000):
    action = env.action_space.sample()
    obs, reward, terminated, truncated, info = env.step(action)
    if terminated:
        print("Collision detected, episode ended")
        obs, info = env.reset()

```

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

### Async Environment Initialization in Python

Source: https://context7.com/augcog/roar_py_rl/llms.txt

This snippet showcases an asynchronous function `initialize_roar_env` for setting up a comprehensive CARLA training environment. It handles CARLA connection, simulation timestep configuration, and vehicle spawning with a wide array of sensors including cameras. It also applies several gym wrappers to filter observations and actions, simplifying the training interface. The function returns a ready-to-use environment object.

```python
import asyncio
from training.env_util import initialize_roar_env
import gymnasium as gym

async def setup_training():
    # Initialize complete environment with default configuration
    env = await initialize_roar_env(
        carla_host="localhost",
        carla_port=2000,
        control_timestep=0.04,  # 25 FPS
        physics_timestep=0.008,  # 125 FPS (5 substeps)
        waypoint_information_distances=[2.0, 5.0, 10.0, 15.0, 20.0,
                                       30.0, 40.0, 50.0, 80.0, 100.0],
        image_width=400,
        image_height=200
    )

    # Environment includes:
    # - Tesla Model 3 vehicle spawned at first spawn point
    # - Collision sensor for termination detection
    # - Velocimeter (global frame)
    # - Local velocimeter (vehicle frame)
    # - Location sensor
    # - Roll/pitch/yaw sensor
    # - Gyroscope (angular velocity)
    # - Camera (RGB, positioned behind and above vehicle)
    # - Waypoint information at specified distances

    # Applied wrappers:
    # 1. SimplifyCarlaActionFilter: Maps throttle/steer to full action dict
    # 2. FilterObservation: Keeps only ["gyroscope", "waypoints_information", "local_velocimeter"]
    # 3. Result: action_space = Dict{"throttle": Box(-1,1), "steer": Box(-1,1)}
    #            observation_space = Dict with filtered sensors

    return env

# Use in training script
env = asyncio.run(setup_training())

# Apply additional wrappers for training
env = gym.wrappers.FlattenObservation(env)  # Flatten dict to vector
env = gym.wrappers.TimeLimit(env, max_episode_steps=3000)
env = gym.wrappers.RecordEpisodeStatistics(env)

# Observation is now flat numpy array containing:
# - gyroscope (3D): [roll_rate, pitch_rate, yaw_rate] (rad/s)
# - local_velocimeter (3D): [vx, vy, vz] in vehicle frame (m/s)
# - waypoints_information (Nx4): N waypoints × [x, y, delta_yaw, width]

obs, info = env.reset()
print(f"Observation shape: {obs.shape}")  # Typically (45,) for 10 waypoints
action = env.action_space.sample()
obs, reward, terminated, truncated, info = env.step(action)

```

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

### Initialize Simulation Environment with RoarRLSimEnv

Source: https://context7.com/augcog/roar_py_rl/llms.txt

This Python code demonstrates the initialization of RoarRLSimEnv, an extended Gymnasium environment for racing simulations. It configures various sensors (location, roll/pitch/yaw, velocity, collision) attached to a vehicle and defines waypoint tracking with multiple lookahead distances. The environment supports reward shaping based on waypoint progress and penalizes deviations, terminating episodes upon significant collisions.

```python
from roar_py_rl import RoarRLSimEnv
from roar_py_interface import (
    RoarPyActor, RoarPyWaypoint, RoarPyWorld,
    RoarPyLocationInWorldSensor, RoarPyRollPitchYawSensor,
    RoarPyVelocimeterSensor, RoarPyCollisionSensor
)
import numpy as np

# Initialize sensors attached to vehicle
location_sensor = vehicle.attach_location_in_world_sensor("location")
rpy_sensor = vehicle.attach_roll_pitch_yaw_sensor("roll_pitch_yaw")
velocimeter_sensor = vehicle.attach_velocimeter_sensor("velocimeter")
collision_sensor = vehicle.attach_collision_sensor(
    np.zeros(3), np.zeros(3), name="collision_sensor"
)

# Create environment with waypoint information at multiple distances
waypoint_distances = [2.0, 5.0, 10.0, 15.0, 20.0, 30.0, 40.0, 50.0, 80.0, 100.0]
env = RoarRLSimEnv(
    actor=vehicle,
    manuverable_waypoints=world.maneuverable_waypoints,
    location_sensor=location_sensor,
    roll_pitch_yaw_sensor=rpy_sensor,
    velocimeter_sensor=velocimeter_sensor,
    collision_sensor=collision_sensor,
    collision_threshold=30.0,  # Impulse threshold for termination
    waypoint_information_distances=set(waypoint_distances),
    world=world,
    render_mode="rgb_array"
)

# Observation space now includes:
# - All vehicle sensors (velocity, gyroscope, etc.)
# - waypoints_information: Dict with keys "waypoint_{dist}" for each distance
#   Each waypoint observation is shape (4,): [local_x, local_y, delta_yaw, lane_width]
#   Waypoints are in vehicle's local coordinate frame

# Reward function:
# - Positive reward proportional to forward distance traveled along waypoints
# - Penalty based on distance from centered waypoint line
# - Episode terminates on collision exceeding threshold

obs, info = env.reset()
# obs["waypoints_information"]["waypoint_10.0"] = [x, y, yaw_diff, width]
# x, y are in vehicle local coordinates (meters)
# yaw_diff is angular difference to waypoint heading (radians)

```

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

### Train SAC Agent with Checkpointing and WandB Logging

Source: https://context7.com/augcog/roar_py_rl/llms.txt

This Python script demonstrates a full training pipeline for a Soft Actor-Critic (SAC) agent using Stable-Baselines3. It includes environment initialization with CARLA, observation and action space wrapping, and integrates advanced features like Weights & Biases (WandB) for experiment tracking, automatic checkpointing, and video recording of episodes.

```python
import gymnasium as gym
from stable_baselines3 import SAC
from stable_baselines3.common.monitor import Monitor
from stable_baselines3.common.callbacks import CheckpointCallback
from wandb.integration.sb3 import WandbCallback
import wandb
import asyncio
import nest_asyncio
from pathlib import Path
from training.env_util import initialize_roar_env
from roar_py_rl_carla import FlattenActionWrapper

nest_asyncio.apply()  # Required for async in Jupyter/IPython

# Initialize wandb logging
wandb_run = wandb.init(
    project="ROAR_PY_RL",
    entity="your_entity",
    name="my_training_run",
    sync_tensorboard=True,
    monitor_gym=True,
    save_code=True
)

# Setup environment
async def get_env():
    env = await initialize_roar_env(
        control_timestep=1.0/25,
        physics_timestep=1.0/125
    )
    env = gym.wrappers.FlattenObservation(env)
    env = FlattenActionWrapper(env)
    env = gym.wrappers.TimeLimit(env, max_episode_steps=3000)
    env = gym.wrappers.RecordEpisodeStatistics(env)
    env = gym.wrappers.RecordVideo(
        env, f"videos/{wandb_run.name}",
        step_trigger=lambda x: x % 20000 == 0
    )
    env = Monitor(env, f"logs/{wandb_run.name}_{wandb_run.id}")
    return env

env = asyncio.run(get_env())

# Configure SAC hyperparameters
model = SAC(
    "MlpPolicy",
    env,
    learning_rate=1e-5,
    batch_size=256,
    gamma=0.97,
    ent_coef="auto",  # Automatic entropy tuning
    target_entropy=-10.0,  # Target entropy for auto-tuning
    use_sde=True,  # State-dependent exploration
    sde_sample_freq=50,  # Resample exploration noise every 50 steps
    replay_buffer_kwargs={"handle_timeout_termination": True},
    verbose=1,
    device="cuda",
    seed=1
)

# Setup callbacks
wandb_callback = WandbCallback(
    gradient_save_freq=50000,
    model_save_path=f"models/{wandb_run.name}",
    verbose=2
)
checkpoint_callback = CheckpointCallback(
    save_freq=50000,
    save_path=f"models/{wandb_run.name}/logs",
    verbose=2
)

# Train agent
model.learn(
    total_timesteps=10_000_000,
    callback=[wandb_callback, checkpoint_callback],
    progress_bar=True
)

# Model checkpoints saved at:
# models/{run_name}/logs/rl_model_{timesteps}_steps.zip

```

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

### Evaluate Trained SAC Agent with Video Recording

Source: https://context7.com/augcog/roar_py_rl/llms.txt

Loads a pre-trained SAC model from a specified path, sets up an evaluation environment with video recording capabilities, and runs an episode to assess performance. It finds the latest trained model based on step count. The environment is wrapped for observation flattening, action processing, episode statistics, and video recording. The evaluation runs for a maximum of one hour at 25 FPS.

```python
import gymnasium as gym
from stable_baselines3 import SAC
from stable_baselines3.common.monitor import Monitor
import asyncio
import nest_asyncio
from pathlib import Path
import os
import tqdm
from training.env_util import initialize_roar_env
from roar_py_rl_carla import FlattenActionWrapper

nest_asyncio.apply()

# Find latest checkpoint
def find_latest_model(models_path: Path):
    logs_path = models_path / "logs"
    if not logs_path.exists():
        return None
    files = os.listdir(logs_path)
    # Files named: rl_model_{steps}_steps.zip
    paths_dict = {
        int(f.split("_")[2]): logs_path / f
        for f in files if f.endswith(".zip")
    }
    if not paths_dict:
        return None
    return paths_dict[max(paths_dict.keys())]

# Setup evaluation environment
async def setup_eval_env():
    env = await initialize_roar_env(
        control_timestep=1.0/25,
        physics_timestep=1.0/125,
        image_width=1920,  # Higher resolution for video
        image_height=1080
    )
    env = gym.wrappers.FlattenObservation(env)
    env = FlattenActionWrapper(env)
    env = gym.wrappers.RecordEpisodeStatistics(env)
    env = gym.wrappers.RecordVideo(env, "videos/eval")
    env = Monitor(env, "logs/eval")
    return env

env = asyncio.run(setup_eval_env())

# Load trained model
models_path = Path("models/my_training_run")
model_path = find_latest_model(models_path)
print(f"Loading model from {model_path}")

model = SAC.load(model_path, env=env, device="cuda")

# Run evaluation episode
obs, info = env.reset()
episode_reward = 0
try:
    for i in tqdm.trange(3600 * 25):  # 1 hour max at 25 FPS
        action, _state = model.predict(obs, deterministic=True)
        obs, reward, terminated, truncated, info = env.step(action)
        episode_reward += reward

        if terminated or truncated:
            print(f"Episode ended after {i} steps")
            print(f"Total reward: {episode_reward:.2f}")
            if terminated:
                print("Reason: Collision")
            break
finally:
    env.close()

# Video saved to videos/eval/
# Episode statistics in info dict:
# - info["episode"]["r"]: total episode reward
# - info["episode"]["l"]: episode length
# - info["episode"]["t"]: episode time

```

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

### SimplifyCarlaActionFilter: Wrap CARLA Actions

Source: https://context7.com/augcog/roar_py_rl/llms.txt

This wrapper simplifies CARLA's complex action space (throttle, brake, steer, etc.) into a more manageable dictionary containing only 'throttle' and 'steer'. Positive 'throttle' values control acceleration, while negative values trigger braking. This reduces the complexity for RL agents learning control.

```python
from training.env_util import SimplifyCarlaActionFilter
import gymnasium as gym
import numpy as np

# Wrap environment to use simplified actions
env = SimplifyCarlaActionFilter(base_env)

# Original action space:
# Dict{"throttle": Box(0,1), "brake": Box(0,1), "steer": Box(-1,1),
#      "hand_brake": Box(0,1), "reverse": Box(0,1)}

# Wrapped action space:
# Dict{"throttle": Box(-1,1), "steer": Box(-1,1)}

# Action mapping:
action = {"throttle": 0.5, "steer": -0.3}
# Internally converts to:
# {
#   "throttle": max(0, 0.5) = 0.5,      # Clip to [0, 1]
#   "brake": max(0, -0.5) = 0.0,        # Negative throttle becomes brake
#   "steer": -0.3,                       # Pass through
#   "hand_brake": 0.0,                   # Always off
#   "reverse": 0.0                       # Always off
# }

action = {"throttle": -0.8, "steer": 0.2}
# Converts to:
# {
#   "throttle": 0.0,                     # Clip negative to 0
#   "brake": 0.8,                        # Negative throttle becomes brake
#   "steer": 0.2,
#   "hand_brake": 0.0,
#   "reverse": 0.0
# }

# This allows RL agent to learn single "throttle" value:
# - Positive values accelerate
# - Negative values brake
# - No need to learn separate brake control

```

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

### Create Custom Racing Environment with RoarRLEnv

Source: https://context7.com/augcog/roar_py_rl/llms.txt

This Python code defines a custom reinforcement learning environment by subclassing RoarRLEnv. It allows for custom reward calculation, termination conditions, truncation logic, and vehicle reset procedures within the ROAR_PY framework. The environment manages asynchronous execution, observation/action spaces, and rendering.

```python
from roar_py_interface import RoarPyActor, RoarPyWaypoint, RoarPyWorld
from roar_py_rl import RoarRLEnv
import gymnasium as gym
import numpy as np

# Create a base environment (typically subclassed)
class CustomRacingEnv(RoarRLEnv):
    def get_reward(self, observation, action, info_dict):
        # Custom reward calculation
        return 1.0

    def is_terminated(self, observation, action, info_dict):
        # Check if episode should end
        return False

    def is_truncated(self, observation, action, info_dict):
        # Check if episode should be truncated
        return False

    def reset_vehicle(self):
        # Reset vehicle to starting position
        pass

# The environment automatically:
# - Manages async step/reset cycles with ROAR_PY
# - Combines actor sensors and additional sensors into observation space
# - Provides rgb_array rendering through RoarPyVisualizer
# - Executes actions and world ticks in proper sequence

# Action space matches RoarPyActor.get_action_spec():
# {"throttle": Box(-1,1), "steer": Box(-1,1), "brake": Box(0,1),
#  "hand_brake": Box(0,1), "reverse": Box(0,1)}

# Observation space is Dict combining all sensor observations

```

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

### FlattenActionWrapper: Flatten Dictionary Actions to Array

Source: https://context7.com/augcog/roar_py_rl/llms.txt

This wrapper converts a dictionary-based action space into a flattened NumPy array. This is crucial for compatibility with standard RL algorithms (like SAC, PPO) that expect flat action spaces rather than structured dictionaries. It correctly handles the conversion between the array and dictionary formats during environment steps.

```python
from roar_py_rl_carla import FlattenActionWrapper
import gymnasium as gym
import numpy as np

# Wrap environment to flatten action space
env = FlattenActionWrapper(base_env)

# Original action space (after SimplifyCarlaActionFilter):
# Dict{"throttle": Box(-1,1, (1,)), "steer": Box(-1,1, (1,))}

# Wrapped action space:
# Box(-1,1, (2,), float32)  # Flattened to single array

# Action usage:
action = np.array([0.5, -0.3])  # [throttle, steer] obs, reward, terminated, truncated, info = env.step(action)

# The wrapper automatically:
# 1. Converts flat array to dictionary for base environment
# 2. Uses gym.spaces.flatten_space and gym.spaces.unflatten
# 3. Maintains correct action space bounds

# This wrapper is essential for algorithms like SAC, PPO that expect
# flat action spaces instead of dictionary spaces

```

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

### ROAR_PY_RL: Speed-Based Reward with Asymmetric Margins

Source: https://context7.com/augcog/roar_py_rl/llms.txt

Implements a speed-based reward function using near_quadratic_bound. It allows for asymmetric margins, meaning the acceptable deviation above and below the target speed can be different. This encourages the agent to maintain a speed close to the target while allowing more tolerance on one side.

```python
# Speed-based reward with asymmetric margins
target_speed = 20.0  # m/s
current_speed = 15.0  # m/s
speed_reward = near_quadratic_bound(
    value=current_speed,
    target=target_speed,
    left_margin=10.0,   # Allow 10 m/s below target
    right_margin=5.0,   # Allow 5 m/s above target
    power=2.0
)
# Encourages speeds near 20 m/s with asymmetric tolerance
```

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

### Near-Quadratic Bounded Reward Function

Source: https://context7.com/augcog/roar_py_rl/llms.txt

Implements a near-quadratic reward function that provides a bounded penalty for deviations from a target value, such as staying within track margins. The function ensures a smooth reward shaping by using a quadratic falloff within specified margins and offers a linear penalty outside these boundaries. This is useful for penalizing lateral drift.

```python
from roar_py_rl.environment.reward_util import near_quadratic_bound
import numpy as np

# Reward function that is 1.0 at target and smoothly decreases
# Use case: penalize lateral deviation from center line
lateral_offset = -2.5  # meters (negative = left of center)
track_center = 0.0     # target position
left_margin = 3.0      # left boundary (3m left of center)
right_margin = 3.0     # right boundary (3m right of center)

reward = near_quadratic_bound(
    value=lateral_offset,
    target=track_center,
    left_margin=left_margin,
    right_margin=right_margin,
    out_of_margin_activation="linear",  # Penalty type outside margins
    power=2.0,  # Quadratic falloff
    value_at_margin=0.0  # Reward at margin boundaries
)
# Returns: ~0.31 (quadratic penalty for 2.5m / 3m deviation)

```

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

### Convert Global to Local Coordinates

Source: https://context7.com/augcog/roar_py_rl/llms.txt

Provides functions to transform coordinates from a global frame of reference to a vehicle's local frame. This is crucial for representing waypoint information relative to the vehicle's current position and orientation. It includes functions for both point transformation and angle normalization.

```python
import numpy as np
from roar_py_rl.environment.sim_env import global_to_local, normalize_rad

# Vehicle state
vehicle_location = np.array([100.0, 50.0])  # Global x, y (meters)
vehicle_yaw = np.pi / 4  # 45 degrees (radians)

# Waypoint in global frame
waypoint_global = np.array([110.0, 60.0])

# Transform to vehicle local frame
waypoint_local = global_to_local(
    global_point=waypoint_global,
    local_origin=vehicle_location,
    local_yaw=vehicle_yaw
)
# Returns: [local_x, local_y] where:
# - local_x: forward direction (vehicle front)
# - local_y: left direction (vehicle left)
# Example result: [14.14, 0.0] (10√2 meters ahead)

# Normalize angle difference
waypoint_yaw = np.pi / 3
delta_yaw = normalize_rad(waypoint_yaw - vehicle_yaw)
# Returns: angle in range [-π, π]
# Example: normalize_rad(7*np.pi/4) = -π/4

# Complete waypoint observation construction
def create_waypoint_obs(waypoint, vehicle_loc, vehicle_yaw):
    local_pos = global_to_local(
        waypoint.location,
        vehicle_loc,
        vehicle_yaw
    )
    delta_yaw = normalize_rad(
        waypoint.roll_pitch_yaw[2] - vehicle_yaw
    )
    return np.array([
        local_pos[0],  # Forward distance
        local_pos[1],  # Lateral offset
        delta_yaw,     # Heading difference
        waypoint.lane_width  # Track width
    ])

```

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

### ROAR_PY_RL: Linear, Quadratic, Gaussian Penalty Modes

Source: https://context7.com/augcog/roar_py_rl/llms.txt

Configures a reward function with different out-of-margin activation modes: linear, quadratic, and gaussian. This is useful for penalizing deviations from a target path with varying severity.

```python
reward_linear = near_quadratic_bound(
    lateral_offset, 0, 3, 3,
    out_of_margin_activation="linear"  # Linear penalty beyond margins
)
reward_quadratic = near_quadratic_bound(
    lateral_offset, 0, 3, 3,
    out_of_margin_activation="quadratic"  # Quadratic penalty beyond
)
reward_gaussian = near_quadratic_bound(
    lateral_offset, 0, 3, 3,
    out_of_margin_activation="gaussian"  # Gaussian falloff beyond
)
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.