### Install GPC using Conda

Source: https://github.com/vincekurtz/gpc/blob/main/README.md

This snippet shows how to clone the GPC repository, create a conda environment from the 'environment.yml' file, activate the environment, and install the package and its dependencies using pip.

```bash
git clone https://github.com/vincekurtz/gpc.git
cd gpc
conda env create -f environment.yml
conda activate gpc
pip install -e .
```

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

### Test Cart-Pole Swingup Policy Interactively

Source: https://github.com/vincekurtz/gpc/blob/main/README.md

This example shows how to run an interactive simulation using a pre-trained cart-pole swingup policy. This allows for testing the learned policy in a dynamic environment.

```python
python examples/cart_pole.py test
```

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

### Simulate Episode with Policy-Augmented Controller in Python

Source: https://context7.com/vincekurtz/gpc/llms.txt

Simulates a training episode using a sampling-based controller augmented with policy samples. This example sets up an environment, a policy-augmented controller, loads a policy, and then calls `simulate_episode` to generate observations, actions, and costs. It highlights how to use different strategies for aggregating samples and provides output shapes and average costs.

```python
import jax
from hydrax.algs import PredictiveSampling
from gpc.envs import CartPoleEnv
from gpc.augmented import PolicyAugmentedController
from gpc.policy import Policy
from gpc.training import simulate_episode

# Set up environment and controller
env = CartPoleEnv(episode_length=200)
base_ctrl = PredictiveSampling(env.task, num_samples=8, noise_level=0.1)
ctrl = PolicyAugmentedController(base_ctrl, num_policy_samples=2)

# Load or create a policy
policy = Policy.load("/tmp/cart_pole_policy.pkl")
policy.model.eval()

# Simulate an episode
rng = jax.random.key(0)
observations, actions, prev_actions, spc_cost, policy_cost, states = simulate_episode(
    env=env,
    ctrl=ctrl,
    policy=policy,
    exploration_noise_level=0.0,
    rng=rng,
    strategy="policy",  # Use "best" to aggregate all samples
)

# observations: (episode_length, obs_dim)
# actions: (episode_length, horizon, action_dim)
# prev_actions: (episode_length, horizon, action_dim) - initial guesses
# spc_cost: (episode_length,) - cost from sampling-based control
# policy_cost: (episode_length,) - cost from policy samples
print(f"Episode observations shape: {observations.shape}")
print(f"Average SPC cost: {jax.numpy.mean(spc_cost):.4f}")
print(f"Average policy cost: {jax.numpy.mean(policy_cost):.4f}")
```

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

### Command-Line Training Script (Python)

Source: https://context7.com/vincekurtz/gpc/llms.txt

This script provides a complete example for training, testing, and sampling policies using the GPC framework and Hydrax. It uses `argparse` to handle different tasks (train, test, sample) and integrates with environments like CartPoleEnv. Dependencies include argparse, mujoco, flax, hydrax, and gpc modules. The script defines how to load, train, save, test, and sample policies.

```python
import argparse
import mujoco
from flax import nnx
from hydrax.algs import PredictiveSampling
from hydrax.simulation.deterministic import run_interactive as run_sampling
from gpc.architectures import DenoisingMLP
from gpc.envs import CartPoleEnv
from gpc.policy import Policy
from gpc.sampling import BootstrappedPredictiveSampling
from gpc.testing import test_interactive
from gpc.training import train

# Parse command-line arguments
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="task")
subparsers.add_parser("train", help="Train a generative policy")
subparsers.add_parser("test", help="Test a generative policy")
subparsers.add_parser("sample", help="Bootstrap sampling with policy")
args = parser.parse_args()

env = CartPoleEnv(episode_length=200)
save_file = "/tmp/cart_pole_policy.pkl"

if args.task == "train":
    ctrl = PredictiveSampling(env.task, num_samples=8, noise_level=0.1)
    net = DenoisingMLP(
        action_size=env.task.model.nu,
        observation_size=env.observation_size,
        horizon=env.task.planning_horizon,
        hidden_layers=[64, 64],
        rngs=nnx.Rngs(0),
    )
    policy = train(env, ctrl, net, num_policy_samples=2,
                   log_dir="/tmp/gpc", num_iters=10, num_envs=128)
    policy.save(save_file)

elif args.task == "test":
    policy = Policy.load(save_file)
    test_interactive(env, policy)

elif args.task == "sample":
    policy = Policy.load(save_file)
    ctrl = BootstrappedPredictiveSampling(
        policy, env.get_obs, num_policy_samples=2,
        task=env.task, num_samples=1, noise_level=0.1
    )
    mj_data = mujoco.MjData(env.task.mj_model)
    run_sampling(ctrl, env.task.mj_model, mj_data, frequency=50)

# Run with: python examples/cart_pole.py train
#           python examples/cart_pole.py test
#           python examples/cart_pole.py sample
```

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

### Train Cart-Pole Swingup Policy with GPC

Source: https://github.com/vincekurtz/gpc/blob/main/README.md

This example demonstrates how to train a flow-matching policy for a cart-pole swingup task using the GPC framework. The trained policy is saved to a file for later use.

```python
python examples/cart_pole.py train
```

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

### Define Custom Training Environment in Python

Source: https://github.com/vincekurtz/gpc/blob/main/README.md

This snippet illustrates how to define a custom training environment for GPC by inheriting from 'gpc.envs.base.TrainingEnv'. It includes required methods for resetting the simulator, getting observations, and specifying the observation size. Dependencies include 'mjx' and 'jax'.

```python
import jax
import mjx
from gpc.envs.base import TrainingEnv

class MyCustomEnv(TrainingEnv):
    def __init__(self):
        super().__init__(task=MyCustomHydraxTask(), episode_length=100)

    def reset(self, data: mjx.Data, rng: jax.Array) -> mjx.Data:
        """Reset the simulator to start a new episode."""
        ...
        return new_data

    def get_obs(self, data: mjx.Data) -> jax.Array:
        """Get the observation from the simulator."""
        ...
        return jax.array([obs1, obs2, ...])

    @property
    def observation_size(self) -> int:
        """Return the size of the observation vector."""
        ...

```

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

### Load and Apply GPC Policy for Action Generation in Python

Source: https://context7.com/vincekurtz/gpc/llms.txt

Loads a trained Generative Predictive Control (GPC) policy and demonstrates how to generate action sequences conditioned on observations. The policy is set to evaluation mode, and an example shows how to apply it with previous actions, current observation, and a random number generator. This is useful for applying the trained policy in a control loop.

```python
import jax
import jax.numpy as jnp
from gpc.policy import Policy

# Load the policy
policy = Policy.load("/tmp/cart_pole_policy.pkl")

# Set the policy to evaluation mode
policy.model.eval()

# Initialize an action sequence and observation
prev_actions = jnp.zeros((policy.u_max.shape[0], 1))  # horizon x action_dim
observation = jnp.array([0.0, 1.0, 0.0, 0.0, 0.0])  # current state
rng = jax.random.key(42)

# Generate a new action sequence conditioned on the observation
action_sequence = policy.apply(
    prev=prev_actions,
    y=observation,
    rng=rng,
    warm_start_level=0.5,  # 0.0 = fully random, 1.0 = fully reuse prev
)

# Extract the first action to execute
action = action_sequence[0]
print(f"Action to execute: {action}")
```

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

### Define Custom Training Environment in Python

Source: https://context7.com/vincekurtz/gpc/llms.txt

Demonstrates how to create a custom training environment by inheriting from `TrainingEnv`. This involves defining `__init__`, `reset`, `get_obs`, and `observation_size` methods to customize the simulation's behavior and state representation. It utilizes JAX and MuJoCo for efficient simulation.

```python
import jax
import jax.numpy as jnp
from mujoco import mjx
from hydrax.task_base import Task
from gpc.envs import TrainingEnv

class MyCustomEnv(TrainingEnv):
    """Custom training environment for a specific task."""

    def __init__(self, task: Task, episode_length: int):
        super().__init__(task=task, episode_length=episode_length)

    def reset(self, data: mjx.Data, rng: jax.Array) -> mjx.Data:
        """Reset the simulator to start a new episode."""
        rng, pos_rng, vel_rng = jax.random.split(rng, 3)

        # Randomize initial positions and velocities
        qpos = jax.random.uniform(
            pos_rng,
            (data.qpos.shape[0],),
            minval=-1.0,
            maxval=1.0
        )
        qvel = jax.random.uniform(
            vel_rng,
            (data.qvel.shape[0],),
            minval=-0.5,
            maxval=0.5
        )

        return data.replace(qpos=qpos, qvel=qvel)

    def get_obs(self, data: mjx.Data) -> jax.Array:
        """Get the observation from the simulator state."""
        # Extract relevant state information
        pos = data.qpos[0]
        vel = data.qvel[0]
        return jnp.array([pos, vel])

    @property
    def observation_size(self) -> int:
        """Return the size of the observation vector."""
        return 2
```

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

### Fit Flow Matching Policy to Data (Python)

Source: https://context7.com/vincekurtz/gpc/llms.txt

This snippet demonstrates how to fit a flow matching policy to synthetic training data using the GPC library. It involves setting up synthetic data, initializing a DenoisingMLP model and an AdamW optimizer, and then calling the `fit_policy` function. The loss from the training process is printed. Dependencies include jax, jax.numpy, optax, and specific modules from gpc.

```python
import jax
import jax.numpy as jnp
import optax
from flax import nnx
from gpc.architectures import DenoisingMLP
from gpc.training import fit_policy

# Create synthetic training data
num_samples = 1000
observations = jax.random.normal(jax.random.key(0), (num_samples, 5))
action_sequences = jax.random.normal(jax.random.key(1), (num_samples, 10, 2))
old_action_sequences = jax.random.normal(jax.random.key(2), (num_samples, 10, 2))

# Set up the model and optimizer
net = DenoisingMLP(
    action_size=2,
    observation_size=5,
    horizon=10,
    hidden_layers=[64, 64],
    rngs=nnx.Rngs(0),
)
optimizer = nnx.Optimizer(net, optax.adamw(1e-3))

# Fit the model
loss = fit_policy(
    observations=observations,
    action_sequences=action_sequences,
    old_action_sequences=old_action_sequences,
    model=net,
    optimizer=optimizer,
    batch_size=128,
    num_epochs=10,
    rng=jax.random.key(3),
    sigma_min=1e-2,
)

print(f"Final training loss: {loss:.6f}")
```

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

### Train GPC Flow-Matching Policy with Python

Source: https://context7.com/vincekurtz/gpc/llms.txt

Trains a Generative Predictive Control (GPC) policy using sampling-based control and flow matching. This script sets up the environment, configures the controller, defines the neural network architecture, and initiates the training process. It saves the trained policy to a specified directory.

```python
from flax import nnx
from hydrax.algs import PredictiveSampling
from gpc.architectures import DenoisingMLP
from gpc.envs import CartPoleEnv
from gpc.training import train

# Set up the training environment
env = CartPoleEnv(episode_length=200)

# Configure sampling-based controller
ctrl = PredictiveSampling(
    env.task,
    num_samples=8,
    noise_level=0.1
)

# Define the neural network architecture
net = DenoisingMLP(
    action_size=env.task.model.nu,
    observation_size=env.observation_size,
    horizon=env.task.planning_horizon,
    hidden_layers=[64, 64],
    rngs=nnx.Rngs(0),
)

# Train the policy
policy = train(
    env=env,
    ctrl=ctrl,
    net=net,
    num_policy_samples=2,
    log_dir="/tmp/gpc_cart_pole",
    num_iters=10,
    num_envs=128,
    learning_rate=1e-3,
    batch_size=128,
    num_epochs=100,
    exploration_noise_level=0.0,
    checkpoint_every=10,
)

# Save the trained policy
policy.save("/tmp/cart_pole_policy.pkl")
```

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

### Test Trained GPC Policy Interactively with Python

Source: https://context7.com/vincekurtz/gpc/llms.txt

Tests a saved Generative Predictive Control (GPC) policy using an interactive MuJoCo simulation viewer. This script loads a pre-trained policy and then runs an interactive test session, allowing for real-time policy evaluation. It requires a saved policy file and the MuJoCo environment.

```python
import mujoco
from gpc.envs import CartPoleEnv
from gpc.policy import Policy
from gpc.testing import test_interactive

# Set up the environment
env = CartPoleEnv(episode_length=200)

# Load a trained policy
policy = Policy.load("/tmp/cart_pole_policy.pkl")

# Run interactive simulation with the policy
# This opens a MuJoCo viewer window
test_interactive(
    env=env,
    policy=policy,
    mj_data=None,  # Uses default initial state
    inference_timestep=0.1,
    warm_start_level=1.0,
)
```

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

### Bootstrapped Predictive Sampling Controller in Python

Source: https://context7.com/vincekurtz/gpc/llms.txt

Implements bootstrapped predictive sampling for policy-driven model predictive control. It loads a pre-trained policy and uses it to bootstrap sampling-based control within an interactive MuJoCo simulation. Key parameters include the number of policy samples, warm-start level, and inference timestep.

```python
import mujoco
from hydrax.simulation.deterministic import run_interactive
from gpc.envs import CartPoleEnv
from gpc.policy import Policy
from gpc.sampling import BootstrappedPredictiveSampling

# Set up environment
env = CartPoleEnv(episode_length=200)

# Load trained policy
policy = Policy.load("/tmp/cart_pole_policy.pkl")

# Create bootstrapped controller
ctrl = BootstrappedPredictiveSampling(
    policy=policy,
    observation_fn=env.get_obs,
    num_policy_samples=2,
    warm_start_level=0.0,
    inference_timestep=0.1,
    task=env.task,
    num_samples=1,
    noise_level=0.1,
)

# Set up MuJoCo simulation
mj_model = env.task.mj_model
mj_data = mujoco.MjData(mj_model)

# Run interactive simulation at 50Hz
run_interactive(ctrl, mj_model, mj_data, frequency=50)
```

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

### Build CNN Denoising Network in Python

Source: https://context7.com/vincekurtz/gpc/llms.txt

Constructs a Convolutional Neural Network (CNN) for denoising temporal action sequences using Flax (nnx). The network takes noisy actions, observations, and a timestep as input and outputs denoised actions. It is configured with specific feature dimensions, kernel size, and timestep embedding dimensions.

```python
from flax import nnx
from gpc.architectures import DenoisingCNN

# Define a CNN-based denoising network
net = DenoisingCNN(
    action_size=2,  # Dimension of action space
    observation_size=5,  # Dimension of observation space
    horizon=10,  # Planning horizon length
    feature_dims=[128, 256, 128],  # Feature dimensions for conv layers
    kernel_size=3,
    timestep_embedding_dim=32,
    rngs=nnx.Rngs(0),
)

# The network takes noisy action sequences and outputs denoised versions
# Input: (batch_size, horizon, action_size)
# Output: (batch_size, horizon, action_size)
import jax.numpy as jnp
noisy_actions = jnp.zeros((1, 10, 2))
observation = jnp.zeros((1, 5))
timestep = jnp.array([[0.5]])  # Flow matching timestep in [0, 1]

denois ed = net(noisy_actions, observation, timestep)
print(f"Denoised action shape: {denoised.shape}")  # (1, 10, 2)
```

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