Getting Started¶
The easiest way to get familiar with the framework is to run through the tutorials below.
Model Predictive Control with True Model¶
"""
- instantiate an env for a pendulum
- instantiate an MPC controller using the true known analytical model
- define cost/reward functions as used in the openAI gym env.
- render the resulting MPC afterwards
"""
from blackbox_mpc.policies.mpc_policy import \
MPCPolicy
from blackbox_mpc.utils.pendulum import PendulumTrueModel, \
pendulum_reward_function
import gym
env = gym.make("Pendulum-v0")
mpc_policy = MPCPolicy(reward_function=pendulum_reward_function,
env_action_space=env.action_space,
env_observation_space=env.observation_space,
true_model=True,
dynamics_function=PendulumTrueModel(),
optimizer_name='RandomSearch',
num_agents=1)
current_obs = env.reset()
for t in range(200):
action_to_execute, expected_obs, expected_reward = mpc_policy.act(
current_obs, t)
current_obs, reward, _, info = env.step(action_to_execute)
env.render()
"""
- instantiate an env for a pendulum
- instantiate an MPC controller using the true known analytical model
- define cost/reward functions as used in the openAI gym env.
- render the resulting MPC afterwards
- switch optimizer (try CEM, CMA-ES, PSO, RandomSearch, SPSA or PI2)
- render the resulting MPC afterwards and compare
"""
from blackbox_mpc.policies.mpc_policy import \
MPCPolicy
from blackbox_mpc.utils.pendulum import PendulumTrueModel, \
pendulum_reward_function
import gym
env = gym.make("Pendulum-v0")
mpc_policy = MPCPolicy(reward_function=pendulum_reward_function,
env_action_space=env.action_space,
env_observation_space=env.observation_space,
true_model=True,
dynamics_function=PendulumTrueModel(),
optimizer_name='CEM',
num_agents=1)
current_obs = env.reset()
for t in range(200):
action_to_execute, expected_obs, expected_reward = mpc_policy.act(
current_obs, t)
current_obs, reward, _, info = env.step(action_to_execute)
env.render()
mpc_policy.switch_optimizer(optimizer_name='RandomSearch')
current_obs = env.reset()
for t in range(200):
action_to_execute, expected_obs, expected_reward = mpc_policy.act(
current_obs, t)
current_obs, reward, _, info = env.step(action_to_execute)
env.render()
"""
- instantiate an env for a pendulum
- instantiate an MPC controller using the true known analytical model
- define cost/reward functions as used in the openAI gym env.
- collect some rollouts and then use them as u like afterwards.
- record everything in tensorboard.
"""
from blackbox_mpc.policies.mpc_policy import \
MPCPolicy
from blackbox_mpc.utils.pendulum import PendulumTrueModel, \
pendulum_reward_function
from blackbox_mpc.utils.rollouts import perform_rollouts
import gym
from blackbox_mpc.environment_utils import EnvironmentWrapper
import tensorflow as tf
env = gym.make("Pendulum-v0")
log_dir = './'
tf_writer = tf.summary.create_file_writer(log_dir)
mpc_policy = MPCPolicy(reward_function=pendulum_reward_function,
env_action_space=env.action_space,
env_observation_space=env.observation_space,
true_model=True,
dynamics_function=PendulumTrueModel(),
optimizer_name='CEM',
tf_writer=tf_writer,
log_dir=log_dir,
num_agents=1)
traj_obs, traj_acs, traj_rews =\
perform_rollouts(env=EnvironmentWrapper.make_standard_gym_env("Pendulum-v0",
num_of_agents=1),
number_of_rollouts=4,
task_horizon=200,
policy=mpc_policy,
exploration_noise=False,
tf_writer=tf_writer)
Learn the Dynamics Model¶
"""
- instantiate an env for a pendulum
- Define an MLP to learn a dynamics model
- instantiate a random policy to collect rollouts
- learn dynamics after collecting rollouts randomly
"""
from blackbox_mpc.dynamics_functions.deterministic_mlp import \
DeterministicMLP
from blackbox_mpc.policies.random_policy import RandomPolicy
from blackbox_mpc.utils.dynamics_learning import learn_dynamics_from_policy
from blackbox_mpc.environment_utils import EnvironmentWrapper
import gym
import tensorflow as tf
log_path = './tutorial_2'
env = gym.make("Pendulum-v0")
dynamics_function = DeterministicMLP(layers=[env.action_space.shape[0]+env.observation_space.shape[0],
32,
32,
32,
env.observation_space.shape[0]],
activation_functions=[tf.math.tanh,
tf.math.tanh,
tf.math.tanh,
None])
policy = RandomPolicy(number_of_agents=10,
env_action_space=env.action_space)
dynamics_handler = learn_dynamics_from_policy(env=EnvironmentWrapper.make_standard_gym_env("Pendulum-v0",
num_of_agents=10),
policy=policy,
number_of_rollouts=5,
task_horizon=200,
dynamics_function=dynamics_function)
ModelBased RL¶
"""
- instantiate an env for a pendulum
- Define an MLP to learn a dynamics model
- instantiate a random policy to collect rollouts
- learn dynamics after collecting rollouts randomly
- plug in the dynamics model in an MPC
- render the result
"""
from blackbox_mpc.dynamics_functions.deterministic_mlp import \
DeterministicMLP
from blackbox_mpc.policies.random_policy import RandomPolicy
from blackbox_mpc.policies.mpc_policy import \
MPCPolicy
from blackbox_mpc.utils.dynamics_learning import learn_dynamics_from_policy
from blackbox_mpc.environment_utils import EnvironmentWrapper
from blackbox_mpc.utils.pendulum import pendulum_reward_function
import gym
import tensorflow as tf
env = gym.make("Pendulum-v0")
dynamics_function = DeterministicMLP(layers=[env.action_space.shape[0]+env.observation_space.shape[0],
32,
32,
32,
env.observation_space.shape[0]],
activation_functions=[tf.math.tanh,
tf.math.tanh,
tf.math.tanh,
None])
policy = RandomPolicy(number_of_agents=10,
env_action_space=env.action_space)
dynamics_handler = learn_dynamics_from_policy(env=EnvironmentWrapper.make_standard_gym_env("Pendulum-v0",
num_of_agents=10),
policy=policy,
number_of_rollouts=5,
task_horizon=200,
dynamics_function=dynamics_function)
mpc_policy = MPCPolicy(reward_function=pendulum_reward_function,
env_action_space=env.action_space,
env_observation_space=env.observation_space,
dynamics_handler=dynamics_handler,
optimizer_name='CEM')
current_obs = env.reset()
for t in range(200):
action_to_execute, expected_obs, expected_reward = mpc_policy.act(
current_obs, t)
current_obs, reward, _, info = env.step(action_to_execute)
env.render()
"""
- instantiate an env for a pendulum
- Define an MLP to learn a dynamics model
- instantiate a random policy to collect rollouts
- learn dynamics in an iterative mpc fashion
(collect -> learn model -> collect using mpc with learned model -> repeat)
- record everything in tensorboard
"""
from blackbox_mpc.dynamics_functions.deterministic_mlp import \
DeterministicMLP
from blackbox_mpc.policies.random_policy import RandomPolicy
from blackbox_mpc.utils.iterative_mpc import learn_dynamics_iteratively_w_mpc
from blackbox_mpc.environment_utils import EnvironmentWrapper
from blackbox_mpc.utils.pendulum import pendulum_reward_function
import gym
import tensorflow as tf
log_dir = './'
tf_writer = tf.summary.create_file_writer(log_dir)
env = gym.make("Pendulum-v0")
dynamics_function = DeterministicMLP(layers=[env.action_space.shape[0]+
env.observation_space.shape[0],
32,
32,
32,
env.observation_space.shape[0]],
activation_functions=[tf.math.tanh,
tf.math.tanh,
tf.math.tanh,
None])
initial_policy = RandomPolicy(number_of_agents=10,
env_action_space=env.action_space)
learn_dynamics_iteratively_w_mpc(env=EnvironmentWrapper.make_standard_gym_env(
"Pendulum-v0", num_of_agents=10),
env_action_space=env.action_space,
env_observation_space=env.observation_space,
number_of_initial_rollouts=5,
number_of_rollouts_for_refinement=2,
number_of_refinement_steps=3,
task_horizon=200,
planning_horizon=50,
initial_policy=initial_policy,
dynamics_function=dynamics_function,
num_agents=10,
reward_function=pendulum_reward_function,
log_dir=log_dir,
tf_writer=tf_writer)
"""
- instantiate an env for a pendulum
- Define an MLP to learn a dynamics model
- instantiate a random policy to collect rollouts
- learn dynamics model from random rollouts
- render the result using mpc to control
- learn dynamics model from mpc rollouts
- render the result using mpc to control
"""
from blackbox_mpc.dynamics_functions.deterministic_mlp import \
DeterministicMLP
from blackbox_mpc.policies.random_policy import RandomPolicy
from blackbox_mpc.policies.mpc_policy import \
MPCPolicy
from blackbox_mpc.utils.dynamics_learning import learn_dynamics_from_policy
from blackbox_mpc.environment_utils import EnvironmentWrapper
from blackbox_mpc.utils.pendulum import pendulum_reward_function
import gym
import tensorflow as tf
env = gym.make("Pendulum-v0")
dynamics_function = DeterministicMLP(layers=[env.action_space.shape[0]+env.observation_space.shape[0],
32,
32,
32,
env.observation_space.shape[0]],
activation_functions=[tf.math.tanh,
tf.math.tanh,
tf.math.tanh,
None])
policy = RandomPolicy(number_of_agents=10,
env_action_space=env.action_space)
dynamics_handler = learn_dynamics_from_policy(env=EnvironmentWrapper.make_standard_gym_env("Pendulum-v0",
num_of_agents=10),
policy=policy,
number_of_rollouts=2,
task_horizon=50,
dynamics_function=dynamics_function)
mpc_policy = MPCPolicy(reward_function=pendulum_reward_function,
env_action_space=env.action_space,
env_observation_space=env.observation_space,
dynamics_handler=dynamics_handler,
optimizer_name='CEM',
num_agents=10)
current_obs = env.reset()
for t in range(10):
action_to_execute, expected_obs, expected_reward = mpc_policy.act(
current_obs, t)
current_obs, reward, _, info = env.step(action_to_execute)
env.render()
dynamics_handler = learn_dynamics_from_policy(env=EnvironmentWrapper.make_standard_gym_env("Pendulum-v0",
num_of_agents=10),
policy=mpc_policy,
number_of_rollouts=2,
task_horizon=50,
system_dynamics_handler=dynamics_handler)
current_obs = env.reset()
for t in range(200):
action_to_execute, expected_obs, expected_reward = mpc_policy.act(
current_obs, t)
current_obs, reward, _, info = env.step(action_to_execute)
env.render()
Load/ Save Models¶
"""
- instantiate an env for a pendulum
- Define an MLP to learn a dynamics model
- instantiate a random policy to collect rollouts
- learn dynamics after collecting rollouts randomly
- save the model in the log dir with frequency 1.
- log everything in tensorboard.
"""
from blackbox_mpc.dynamics_functions.deterministic_mlp import \
DeterministicMLP
from blackbox_mpc.policies.random_policy import RandomPolicy
from blackbox_mpc.utils.dynamics_learning import learn_dynamics_from_policy
from blackbox_mpc.environment_utils import EnvironmentWrapper
import gym
import tensorflow as tf
env = gym.make("Pendulum-v0")
log_dir = './'
tf_writer = tf.summary.create_file_writer(log_dir)
dynamics_function = DeterministicMLP(layers=[env.action_space.shape[0]+env.observation_space.shape[0],
32,
32,
32,
env.observation_space.shape[0]],
activation_functions=[tf.math.tanh,
tf.math.tanh,
tf.math.tanh,
None])
policy = RandomPolicy(number_of_agents=10,
env_action_space=env.action_space)
dynamics_handler = learn_dynamics_from_policy(env=EnvironmentWrapper.make_standard_gym_env("Pendulum-v0",
num_of_agents=10),
policy=policy,
number_of_rollouts=5,
task_horizon=200,
dynamics_function=dynamics_function,
log_dir=log_dir,
tf_writer=tf_writer)
"""
- instantiate an env for a pendulum
- instantiate an MPC by loading a previosuluy saved dynamics model.
- render the result.
"""
from blackbox_mpc.policies.mpc_policy import \
MPCPolicy
from blackbox_mpc.utils.pendulum import pendulum_reward_function
import gym
env = gym.make("Pendulum-v0")
mpc_policy = MPCPolicy(reward_function=pendulum_reward_function,
env_action_space=env.action_space,
env_observation_space=env.observation_space,
optimizer_name='CEM',
saved_model_dir='./saved_model',
num_agents=1)
current_obs = env.reset()
for t in range(200):
action_to_execute, expected_obs, expected_reward = mpc_policy.act(
current_obs, t)
current_obs, reward, _, info = env.step(action_to_execute)
env.render()
Record Videos¶
"""
- instantiate an env for a pendulum
- instantiate an MPC controller using the true known analytical model
- record a rollout in a video.
"""
from blackbox_mpc.policies.mpc_policy import \
MPCPolicy
from blackbox_mpc.utils.pendulum import PendulumTrueModel, \
pendulum_reward_function
from blackbox_mpc.utils.recording import record_rollout
import gym
env = gym.make("Pendulum-v0")
mpc_policy = MPCPolicy(reward_function=pendulum_reward_function,
env_action_space=env.action_space,
env_observation_space=env.observation_space,
true_model=True,
dynamics_function=PendulumTrueModel(),
optimizer_name='CMA-ES',
num_agents=1)
record_rollout(env, horizon=200, policy=mpc_policy,
record_file_path='./cma')
Using low-level API¶
"""
This tutorial is meant to show the modular structure of the package,
and the possibility of extending the functionality of each block further
if needed in your research.(such as new optimizer or
a new trajectory evaluator method..etc)
- instantiate an env for a pendulum.
- Define an MLP to learn a dynamics model
- Define the system handler that takes care of training the model
and processing the rollouts..etc.
- Define a trajectory evaluator that evaluates the rewards of trajectories.
- Define an optimizer.
- instantiate a random policy to collect rollouts
- instantiate an mpc policy using the previous blocks.
- learn dynamics from random policy.
- use the learned dynamics with mpc and render the result
- record everything in tensorboard
"""
from blackbox_mpc.dynamics_functions.deterministic_mlp import \
DeterministicMLP
from blackbox_mpc.policies.random_policy import RandomPolicy
from blackbox_mpc.utils.dynamics_learning import learn_dynamics_from_policy
from blackbox_mpc.environment_utils import EnvironmentWrapper
from blackbox_mpc.policies.mpc_policy import \
MPCPolicy
from blackbox_mpc.trajectory_evaluators.deterministic import \
DeterministicTrajectoryEvaluator
from blackbox_mpc.optimizers.cem import CEMOptimizer
from blackbox_mpc.dynamics_handlers.system_dynamics_handler import \
SystemDynamicsHandler
from blackbox_mpc.utils.pendulum import pendulum_reward_function
import gym
import tensorflow as tf
log_dir = './'
env = gym.make("Pendulum-v0")
tf_writer = tf.summary.create_file_writer(log_dir)
dynamics_function = DeterministicMLP(layers=[env.action_space.shape[0]+env.observation_space.shape[0],
32,
32,
32,
env.observation_space.shape[0]],
activation_functions=[tf.math.tanh,
tf.math.tanh,
tf.math.tanh,
None])
system_dynamics_handler = SystemDynamicsHandler(env_action_space=env.action_space,
env_observation_space=env.observation_space,
true_model=False,
dynamics_function=dynamics_function,
tf_writer=tf_writer,
is_normalized=True,
log_dir=log_dir,
save_model_frequency=2)
trajectory_evaluator = \
DeterministicTrajectoryEvaluator(reward_function=pendulum_reward_function,
system_dynamics_handler=system_dynamics_handler)
optimizer = CEMOptimizer(env_action_space=env.action_space,
env_observation_space=env.observation_space,
num_agents=10,
planning_horizon=30,
max_iterations=6)
policy = MPCPolicy(trajectory_evaluator=trajectory_evaluator,
optimizer=optimizer,
tf_writer=tf_writer)
random_policy = RandomPolicy(number_of_agents=10,
env_action_space=env.action_space)
dynamics_handler = learn_dynamics_from_policy(env=EnvironmentWrapper.make_standard_gym_env("Pendulum-v0",
num_of_agents=10),
policy=random_policy,
number_of_rollouts=5,
task_horizon=200,
system_dynamics_handler=system_dynamics_handler)
current_obs = env.reset()
for t in range(200):
action_to_execute, expected_obs, expected_reward = policy.act(
current_obs, t)
current_obs, reward, _, info = env.step(action_to_execute)
env.render()
HalfCheetah¶
"""
- instantiate an env for a modified version of halfcheetah mujoco env.
- Define an MLP to learn a dynamics model
- instantiate a random policy to collect rollouts
- learn dynamics in an iterative mpc fashion
(collect -> learn model -> collect using mpc with learned model -> repeat)
- record everything in tensorboard
"""
from blackbox_mpc.dynamics_functions.deterministic_mlp import \
DeterministicMLP
from blackbox_mpc.policies.random_policy import RandomPolicy
from blackbox_mpc.utils.iterative_mpc import learn_dynamics_iteratively_w_mpc
from blackbox_mpc.environment_utils import EnvironmentWrapper
from blackbox_mpc.utils.pendulum import pendulum_reward_function
import tensorflow as tf
from env_modified import HalfCheetahEnvModified
from cost_func import reward_function
log_dir = './'
tf_writer = tf.summary.create_file_writer(log_dir)
env = HalfCheetahEnvModified()
dynamics_function = DeterministicMLP(layers=[env.action_space.shape[0]+
env.observation_space.shape[0],
32,
32,
32,
env.observation_space.shape[0]],
activation_functions=[tf.math.tanh,
tf.math.tanh,
tf.math.tanh,
None])
initial_policy = RandomPolicy(number_of_agents=10,
env_action_space=env.action_space)
learn_dynamics_iteratively_w_mpc(env=EnvironmentWrapper.make_custom_gym_env(
HalfCheetahEnvModified, num_of_agents=10),
env_action_space=env.action_space,
env_observation_space=env.observation_space,
number_of_initial_rollouts=5,
number_of_rollouts_for_refinement=2,
number_of_refinement_steps=3,
task_horizon=200,
planning_horizon=50,
initial_policy=initial_policy,
dynamics_function=dynamics_function,
num_agents=10,
reward_function=pendulum_reward_function,
log_dir=log_dir,
tf_writer=tf_writer)
"""
This tutorial is meant to show how to train a NN dynamics with MPC for the
cheetah environment.
"""
from blackbox_mpc.dynamics_functions.deterministic_mlp import \
DeterministicMLP
from blackbox_mpc.policies.random_policy import RandomPolicy
from env_modified import HalfCheetahEnvModified
from cost_func import reward_function
from blackbox_mpc.utils.iterative_mpc import learn_dynamics_iteratively_w_mpc
from blackbox_mpc.environment_utils import EnvironmentWrapper
from blackbox_mpc.utils.recording import record_rollout
import tensorflow as tf
log_dir = './'
tf_writer = tf.summary.create_file_writer(log_dir)
env = HalfCheetahEnvModified()
num_of_agents = 10
parallel_env = EnvironmentWrapper.make_custom_gym_env(
HalfCheetahEnvModified,
num_of_agents=num_of_agents)
dynamics_function = DeterministicMLP(layers=[env.action_space.shape[0]+
env.observation_space.shape[0],
500,
500,
500,
env.observation_space.shape[0]],
activation_functions=[tf.math.tanh,
tf.math.tanh,
tf.math.tanh,
None])
initial_policy = RandomPolicy(number_of_agents=num_of_agents,
env_action_space=env.action_space)
system_dynamics_handler, mpc_policy = learn_dynamics_iteratively_w_mpc(
env=parallel_env,
env_action_space=env.action_space,
env_observation_space=env.observation_space,
number_of_initial_rollouts=5,
number_of_rollouts_for_refinement=3,
number_of_refinement_steps=1,
task_horizon=1000,
planning_horizon=15,
initial_policy=initial_policy,
dynamics_function=dynamics_function,
num_agents=num_of_agents,
reward_function=reward_function,
log_dir=log_dir,
tf_writer=tf_writer,
optimizer_name='RandomSearch',
population_size=4048,
save_model_frequency=2,
batch_size=512,
epochs=100)
record_rollout(env, horizon=1000, policy=mpc_policy,
record_file_path='./current_policy_0')
for i in range(9):
system_dynamics_handler, mpc_policy = learn_dynamics_iteratively_w_mpc(
env=parallel_env,
number_of_initial_rollouts=0,
number_of_rollouts_for_refinement=3,
number_of_refinement_steps=5,
refinement_policy=mpc_policy,
task_horizon=1000,
system_dynamics_handler=system_dynamics_handler,
batch_size=512,
epochs=100,
tf_writer=tf_writer,
start_episode=3+(i*5*3))
record_rollout(env, horizon=1000, policy=mpc_policy,
record_file_path='./current_policy_'+str(i+1))