wassname/pyrobolearn
1
0
Fork 0
mirror of https://github.com/wassname/pyrobolearn.git synced 2026-09-10 12:21:16 +08:00
Code Issues Packages Projects Releases Wiki Activity

update rl-algo (exploration phase): add multiprocessing + correct bugs

Browse Source
This commit is contained in:
Brian Delhaisse
2019-05-15 07:21:47 +02:00
parent a2cb7f301e
commit d6bf55cd43
13 changed files with 622 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyrobolearn/algos/explorer.py
+480 -55
View File
@@ -5,16 +5,23 @@ It consists to explore and collect samples in the environment using the policy.
given memory/storage unit which will be used to evaluate the policy, and then update its parameters.
"""
import os
import copy
import inspect
import torch
import torch.multiprocessing as multiprocessing
import torch.distributed as distributed # note that you can only send/receive tensors with P2P backends
from pyrobolearn.tasks import RLTask
from pyrobolearn.envs import Env
from pyrobolearn.policies import Policy
from pyrobolearn.exploration import Exploration
from pyrobolearn.storages import DictStorage # RolloutStorage
from pyrobolearn.storages import RolloutStorage, ExperienceReplay
from pyrobolearn import logger
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
@@ -34,25 +41,108 @@ class Explorer(object):
2. Evaluate: Assess the quality of the actions/trajectories using the returns.
3. Update: Update the policy (and/or value function) parameters based on the loss
This class focuses on the first step of RL algorithms. It accepts the environment, and the exploration strategy
which wraps the policy.
This class focuses on the first step of RL algorithms. It accepts the task (environment and policy), the
exploration strategy which wraps the policy, and the rollout (for on-policy) or experience replay (for off-policy)
storage unit.
"""
def __init__(self, task, explorer, storage, num_workers=1):
def __init__(self, task, explorer, storage, num_workers=1, backend='multiprocessing'):
"""
Initialize the exploration phase.
Args:
task (Task, Env, tuple of Env and Policy): RL task or environment.
explorer (Exploration): policies.
storage (DictStorage): Rollout storage unit (=replay memory). It will save the trajectories / rollouts /
transitions in the storage while exploring.
num_workers (int): number of processes / workers to run in parallel.
storage (RolloutStorage, ExperienceReplay): Rollout (for on-policy) or experience replay (for off-policy)
storage unit (=replay memory). It will save the trajectories / rollouts / transitions in the storage
while exploring.
num_workers (int): number of processes / workers to run in parallel. This number has to be equal or smaller
than the number of CPUs on the computer. If bigger, it will automatically be clipped when using the
'multiprocessing' backend (see below). If only :attr:`num_workers=1`, it doesn't
backend: backend to be used when using multiple processes. The different possible backends are
'multiprocessing' (by default), 'gloo' (good for CPUs), 'nccl' (good for GPUs), 'mpi' (only valid if
PyTorch has been built from source with MPI support). For more information, we refer the reader to
references [2,4]. If the backend is 'mpi', you have to run the code using the following command:
`mpirun -n 4 python <code>.py`.
References:
[1] Multiprocessing best practices: https://pytorch.org/docs/stable/notes/multiprocessing.html
[2] torch multiprocessing: https://pytorch.org/docs/stable/multiprocessing.html
[3] Writing Distributed Applications with PyTorch: https://pytorch.org/tutorials/intermediate/dist_tuto.html
[4] torch distributed: https://pytorch.org/docs/stable/distributed.html
"""
self.task = task
self.explorer = explorer
self.storage = storage
self.num_workers = int(num_workers)
# check the number of workers
if not isinstance(num_workers, (int, long)):
raise TypeError("Expecting the number of workers to be an integer, instead got: {}".format(num_workers))
self.num_workers = 1 if num_workers < 1 else int(num_workers)
# make sure that the maximum number of worker/process is smaller or equal to the number of CPUs
if self.num_workers > multiprocessing.cpu_count():
self.num_workers = multiprocessing.cpu_count()
# check backend
if backend is None:
backend = 'multiprocessing'
if not isinstance(backend, str):
raise TypeError("Expecting the given 'backend' to be a string, instead got: {}".format(type(backend)))
backend = backend.lower()
if backend not in {'multiprocessing', 'mpi', 'gloo', 'nccl'}:
raise ValueError("Expecting the given 'backend' to be 'multiprocessing', 'mpi', 'gloo', or 'nccl', instead "
"got: {}".format(backend))
self.backend = backend
# create processes
self.processes = []
self.pipe = None
self.process_id = 0 # only the master should have the id set to 0, the workers have a strictly positive id
if self.num_workers > 1:
if self.backend == 'multiprocessing':
is_rendering = self.environment.is_rendering
rendering_mode = self.environment.rendering_mode
# hide the GUI of the environment
if is_rendering:
self.environment.hide()
# create processes
self.queue = multiprocessing.Queue()
self.pipes = [multiprocessing.Pipe() for _ in range(self.num_workers)]
self.processes = [multiprocessing.Process(target=self.explore_in_parallel, args=(pipe[1], self.queue,
task))
for pipe in self.pipes]
# start processes
for process in self.processes:
process.start()
# render the GUI if it was initially rendered
if is_rendering:
self.environment.render(mode=rendering_mode)
else:
raise NotImplementedError("Currently, other backends are not provided.")
# elif self.backend == 'gloo' or self.backend == 'nccl':
#
# if self.backend == 'nccl' and not distributed.is_nccl_available():
# raise ValueError("The 'nccl' backend is not available on this computer.")
#
# # create processes
# self.processes = [multiprocessing.Process(target=init_processes, args=(rank, size, function))]
#
# # start processes
# for process in self.processes:
# process.start()
#
# elif self.backend == 'mpi':
# if not distributed.is_mpi_available():
# raise ValueError("The 'mpi' backend is not available on this computer.")
#
# init_processes(0, 0, function, self.backend)
##############
# Properties #
@@ -91,6 +181,12 @@ class Explorer(object):
"""Return the environment."""
return self.task.environment
# alias
@property
def environment(self):
"""Return the environment"""
return self.task.environment
@property
def explorer(self):
"""Return the exploration strategy."""
@@ -113,89 +209,417 @@ class Explorer(object):
@storage.setter
def storage(self, storage):
"""Set the storage unit."""
if not isinstance(storage, DictStorage):
raise TypeError("Expecting the storage to be an instance of `DictStorage`, instead got: "
"{}".format(type(storage)))
if not isinstance(storage, (RolloutStorage, ExperienceReplay)): # DictStorage):
raise TypeError("Expecting the storage to be an instance of `RolloutStorage` or `ExperienceReplay`, "
"instead got: {}".format(type(storage)))
self._storage = storage
###########
# Methods #
###########
def explore(self, num_steps, rollout_idx=0, deterministic=False, verbose=False):
"""
Explore in the environment.
def close(self):
"""End the processes."""
for process in self.processes:
process.terminate()
def rollout(self, num_steps, deterministic=False, render=False, verbose=False):
"""Perform a rollout.
Args:
num_steps (int): number of steps
rollout_idx (int): trajectory/rollout index.
deterministic (bool): if deterministic is True, then it does not explore in the environment.
verbose (bool): If true, print information on the standard output.
Returns:
DictStorage: updated memory storage
num_steps (int): number of steps.
deterministic (bool): if the policy should act deterministically instead of exploring (based on the
exploration strategy).
render (bool): if we should render the environment.
verbose (bool): if we should print information about the rollout.
"""
# reset environment
observation = self.env.reset()
if verbose:
print("\n#### Starting the Exploration phase ####")
# print("Explorer - initial state: {}".format(observation))
# reset storage
self.storage.reset(init_states=observation, rollout_idx=rollout_idx)
# reset explorer
self.explorer.reset()
if verbose:
print("Start the rollout")
# run RL task for T steps
trajectory = []
for step in range(num_steps):
# if we need to render
if render:
self.env.render()
# get action and corresponding distribution from policy
action, distribution = self.explorer.act(observation, deterministic=deterministic)
# perform one step in the environment
next_observation, reward, done, info = self.env.step(action)
# append the transition tuple
trajectory.append({'states': observation, 'actions': action, 'next_states': next_observation,
'reward': reward, 'mask': (1 - done), 'distribution': distribution})
# if verbose:
# print("\nExplorer:")
# print("1. Observation data: {}".format(observation))
# print("2. Action data: {}".format(action))
# print("3. Next observation data: {}".format(next_observation))
# print("4. Reward: {}".format(reward))
# print("5. \\pi(.|s): {}".format(distribution))
# print("6. log \\pi(a|s): {}".format([d.log_prob(action) for d in distribution]))
# insert in storage
self.storage.insert(observation, action, next_observation, reward, mask=(1-done),
distributions=distribution, rollout_idx=rollout_idx)
# set current observation to current one
observation = next_observation
# print("\n\t1. Observation data: {}".format(observation))
# print("\t2. Action data: {}".format(action))
# print("\t3. Next observation data: {}".format(next_observation))
# print("\t4. Reward: {}".format(reward))
# print("\t5. \\pi(.|s): {}".format(distribution))
# print("\t6. log \\pi(a|s): {}".format([d.log_prob(action) for d in distribution]))
# if done, get out of the loop
if done:
break
# fill remaining mask values
self.storage.end(rollout_idx)
# set current observation to current one
observation = next_observation
if verbose:
print("#### End of the Exploration phase #####")
# print("states: {}".format(self.storage['states']))
# print("actions: {}".format(self.storage['actions']))
# print("rewards: {}".format(self.storage['rewards']))
# print("masks: {}".format(self.storage['masks']))
# print("distributions: {}".format(self.storage['distributions']))
print("End of the rollout")
# # clear explorer
# self.explorer.clear()
# return trajectory
return trajectory
def __send(self, msg, dst=0, rank=0):
"""
Send the given message to the specified destination.
- master: send to the specified worker process the information about the rollout (number of steps, parameters,
etc).
- worker: send its rank / process id and the trajectory to the master
Args:
msg (tuple of object): message to send.
dst (int): destination rank or process id to send the message to. The master process has a rank of 0.
rank (int): rank or process id. The master process has a rank/id of 0, while the workers have a strictly
positive integer id.
"""
if self.backend == 'multiprocessing':
# master process: send to the specified worker process the information about the rollout (number of steps,
# parameters, etc).
if rank == 0:
self.pipes[dst][0].send(msg)
# worker process: add the message (process id and trajectory) to the queue
else:
self.queue.put(msg)
else:
# TODO: the messages have to be tensors!!
# master process: send to the specified worker process the information about the rollout (number of steps,
# parameters, etc).
if rank == 0:
num_steps, render, verbose, parameters = msg
distributed.send(tensor=torch.tensor(num_steps).view(1), dst=dst)
distributed.send(tensor=torch.tensor(render).view(1), dst=dst)
distributed.send(tensor=torch.tensor(verbose).view(1), dst=dst)
for parameter in parameters:
distributed.send(tensor=parameter, dst=dst)
# worker process: send the message (process id and trajectory) to the queue
else:
process_id, trajectory = msg
distributed.send(tensor=torch.tensor(process_id).view(1), dst=0)
for transition in trajectory:
distributed.send(tensor=transition['states'], dst=0)
distributed.send(tensor=transition['actions'], dst=0)
distributed.send(tensor=transition['next_states'], dst=0)
distributed.send(tensor=transition['reward'], dst=0)
distributed.send(tensor=transition['mask'], dst=0)
# TODO: send distribution (its hyperparameters? because I can only send tensors... Or its entropy
# and the log likelihood of the action evaluated with the distribution?)
# distributed.send(tensor=transition['distribution'], dst=0)
def __recv(self, rank=0):
"""
Receive the message from the master or worker process.
- master: receive the message (i.e. the process id (or rank) and the trajectory) from the worker
- worker: receive information about the rollout (num_steps, parameters, etc) from the master
Args:
rank (int): rank or process id. The master process has a rank/id of 0, while the workers have a strictly
positive integer id.
Returns:
int: process id
list of torch.Tensor: the trajectory.
"""
if self.backend == 'multiprocessing':
# master process: receive the process id and trajectory from the workers
if rank == 0:
process_id, trajectory = self.queue.get()
return process_id, trajectory
# worker process: receive information about the rollout (num_steps, parameters, etc) from the master
msg = self.pipe.recv()
return msg
else:
# TODO: can only receive tensors and they have to be allocated in advance with the correct dimensions!!
# master process: receive the process id and trajectory from the workers
if rank == 0:
# preallocate the tensors
scalar_tensor = torch.zeros(1)
state_shapes = self.explorer.policy.states.merged_shape
action_shapes = self.explorer.policy.actions.merged_shape
# get process id / rank
distributed.recv(tensor=scalar_tensor, src=None)
src = int(scalar_tensor)
# get trajectory length
distributed.recv(tensor=scalar_tensor, src=src)
trajectory_length = int(scalar_tensor)
# copy trajectory
trajectory = []
state_tensors = [torch.zeros(shape) for shape in state_shapes]
for i in range(trajectory_length):
# copy transition tuple
transition = {}
# allocate tensors in advance
action_tensors = [torch.zeros(shape) for shape in action_shapes]
next_state_tensors = [torch.zeros(shape) for shape in state_shapes]
reward_tensor = torch.zeros(1)
mask_tensor = torch.zeros(1)
# copy states/observations
for state_tensor in state_tensors:
distributed.recv(tensor=state_tensor, src=src)
transition['states'] = state_tensors
# copy actions
for action_tensor in action_tensors:
distributed.recv(tensor=action_tensor, src=src)
transition['actions'] = action_tensors
# copy next states/observations
for state_tensor in next_state_tensors:
distributed.recv(tensor=state_tensor, src=src)
transition['next_states'] = state_tensors
# copy reward
distributed.recv(tensor=reward_tensor, src=src)
transition['reward'] = reward_tensor
# copy mask/done
distributed.recv(tensor=mask_tensor, src=src)
transition['mask'] = mask_tensor
# TODO: copy distribution
# distributed.recv(tensor=distribution_tensor, src=src)
# transition['distribution'] = distribution_tensor
# add transition tuple in the trajectory
trajectory.append(transition)
# set the state tensors to the next one (to be memory and time efficient)
state_tensors = next_state_tensors
return src, trajectory
# worker process: receive information about the rollout (num_steps, parameters, etc) from the master
# get number of steps
scalar_tensor = torch.zeros(1)
distributed.recv(tensor=scalar_tensor, src=0)
num_steps = int(scalar_tensor)
# get if we should render or not, and verbose
distributed.recv(tensor=scalar_tensor, src=0)
render = bool(scalar_tensor)
distributed.recv(tensor=scalar_tensor, src=0)
verbose = bool(scalar_tensor)
# copy parameters
parameters = []
for parameter in self.explorer.policy.parameters():
parameter = torch.zeros(parameter.shape)
distributed.recv(tensor=parameter, src=0)
parameters.append(parameter)
return num_steps, render, verbose, parameters
def _master_explore(self, num_steps, num_rollouts=1, render=False, verbose=False):
"""Master: send the add the trajectories or transition tuples in the storage unit."""
# create set of integers
pool = set(range(min(len(self.processes), num_rollouts)))
process, rollout = 0, 0
while True:
if pool and process < num_rollouts:
# send job to process
process_id = pool.pop()
# self.pipes[process_id][0].send((num_steps, self.explorer))
self.__send((num_steps, render, verbose, self.explorer.policy.parameters()), dst=process_id, rank=0)
process += 1
elif process >= num_rollouts and rollout >= num_rollouts:
# get out of the loop
break
else:
# wait for result from queue
process_id, trajectory = self.__recv(rank=0) # self.queue.get()
# put the process id in the pool to let know that it is free
pool.add(process_id)
# add the trajectory in the storage
self.storage.add_trajectory(trajectory, rollout_idx=rollout)
rollout += 1
return self.storage
def _worker_explore(self, pipe=None):
"""Worker explore in the environment."""
# get process
process = multiprocessing.current_process()
self.process_id = int(process.name.split('-')[-1])
if self.backend == 'multiprocessing':
self.pipe = pipe
else:
os.environ['MASTER_ADDR'] = '127.0.0.1'
os.environ['MASTER_PORT'] = '29500'
os.environ['WORLD_SIZE'] = str(self.num_workers)
os.environ['RANK'] = str(self.process_id)
distributed.init_process_group(self.backend, rank=self.process_id, world_size=self.num_workers)
# copy task
self.task = copy.deepcopy(self.task)
while True:
# get the message (number of steps, parameters, etc).
msg = self.__recv(rank=self.process_id)
# end the process if specified, i.e. if the number of steps is negative
if msg[0] == -1:
break
# decompose the received message
num_steps, render, verbose, parameters = msg
# set the parameters of the policy
self.explorer.policy.copy_parameters(parameters)
# perform a rollout
trajectory = self.rollout(num_steps=num_steps, deterministic=False, render=render, verbose=verbose)
# return the process id and the trajectory
self.__send((self.process_id, trajectory), dst=0, rank=self.process_id)
def _explore(self, num_steps, num_rollouts=1, deterministic=False, render=False, verbose=False):
"""
Explore in the environment.
Args:
num_steps (int): number of steps in one episode in the on-policy case. This is also the number of updates
in the off-policy case.
num_rollouts (int): number of trajectories/rollouts (only valid in the on-policy case).
deterministic (bool): if deterministic is True, then it does not explore in the environment.
render (bool): if we should render the environment.
verbose (bool): If true, print information on the standard output.
Returns:
DictStorage: updated memory storage
"""
for rollout in range(num_rollouts):
# reset environment
observation = self.env.reset()
if verbose:
print("Start rollout: {}/{}".format(rollout + 1, num_rollouts))
# print("Explorer - initial state: {}".format(observation))
# reset storage
self.storage.reset(init_states=observation, rollout_idx=rollout)
# reset explorer
self.explorer.reset()
# run RL task for T steps
for step in range(num_steps):
# if we need to render
if render:
self.env.render()
# get action and corresponding distribution from policy
action, distribution = self.explorer.act(observation, deterministic=deterministic)
# perform one step in the environment
next_observation, reward, done, info = self.env.step(action)
# if verbose:
# print("\nExplorer:")
# print("1. Observation data: {}".format(observation))
# print("2. Action data: {}".format(action))
# print("3. Next observation data: {}".format(next_observation))
# print("4. Reward: {}".format(reward))
# print("5. \\pi(.|s): {}".format(distribution))
# print("6. log \\pi(a|s): {}".format([d.log_prob(action) for d in distribution]))
# insert in storage
self.storage.insert(observation, action, next_observation, reward, mask=(1 - done),
distributions=distribution, rollout_idx=rollout)
# set current observation to current one
observation = next_observation
# if done, get out of the loop
if done:
break
# fill remaining mask values
self.storage.end(rollout)
if verbose:
print("End rollout: {}/{}".format(rollout + 1, num_rollouts))
# print("states: {}".format(self.storage['states']))
# print("actions: {}".format(self.storage['actions']))
# print("rewards: {}".format(self.storage['rewards']))
# print("masks: {}".format(self.storage['masks']))
# print("distributions: {}".format(self.storage['distributions']))
# # clear explorer
# self.explorer.clear()
# return storage unit
return self.storage
def explore(self, num_steps, num_rollouts=1, deterministic=False, render=False, verbose=False):
"""
Explore in the environment.
Args:
num_steps (int): number of steps in one episode in the on-policy case. This is also the number of updates
in the off-policy case.
num_rollouts (int): number of trajectories/rollouts (only valid in the on-policy case).
deterministic (bool): if deterministic is True, then it does not explore in the environment.
render (bool): if we should render the environment.
verbose (bool): If true, print information on the standard output.
Returns:
RolloutStorage, ExperienceReplay: updated memory storage
"""
if self.num_workers > 1: # parallel exploration
return self._master_explore(num_steps, num_rollouts=num_rollouts, render=render, verbose=verbose)
else: # simple exploration
return self._explore(num_steps, num_rollouts=num_rollouts, deterministic=deterministic, render=render,
verbose=verbose)
#############
# Operators #
#############
def __del__(self):
"""Delete the exploration phase; this will close the processes."""
self.close()
def __repr__(self):
"""Return a representation string about the class."""
return self.__class__.__name__
@@ -204,16 +628,17 @@ class Explorer(object):
"""Return a string describing the class."""
return self.__class__.__name__
def __call__(self, num_steps, rollout_idx=0, deterministic=False, verbose=True):
def __call__(self, num_steps, num_rollouts=1, deterministic=False, verbose=True): # rollout_idx
"""Explore in the environment.
Args:
num_steps (int): number of steps
rollout_idx (int): trajectory/rollout index.
num_steps (int): number of steps in one episode in the on-policy case. This is also the number of updates
in the off-policy case.
num_rollouts (int): number of trajectories/rollouts (only valid in the on-policy case).
deterministic (bool): if deterministic is True, then it does not explore in the environment.
verbose (bool): If true, print information on the standard output.
Returns:
DictStorage: updated memory storage
"""
self.explore(num_steps, rollout_idx=rollout_idx)
self.explore(num_steps, num_rollouts=num_rollouts, deterministic=deterministic, verbose=verbose)
pyrobolearn/algos/rl_algo.py
+14 -12
View File
@@ -8,7 +8,6 @@ Dependencies:
"""
import numpy as np
# from pathos.multiprocessing import Pool
from pyrobolearn.storages import RolloutStorage
@@ -149,7 +148,7 @@ class RLAlgo(object): # Algo):
[5] OpenAI - Spinning Up: https://spinningup.openai.com/
"""
def __init__(self, explorer, evaluator, updater, dynamic_model=None): # , hyperparameters={}, num_workers=1):
def __init__(self, explorer, evaluator, updater, dynamic_model=None):
"""
Initialize the reinforcement learning algorithm.
@@ -327,19 +326,22 @@ class RLAlgo(object): # Algo):
# for each episode
for episode in range(num_episodes):
# for each rollout
for rollout in range(num_rollouts):
# TODO: consider to learn the dynamic model if provided
# # for each rollout
# for rollout in range(num_rollouts):
# # TODO: consider to learn the dynamic model if provided
#
# if verbose:
# print("Episode: {}/{} - Rollout: {}/{}".format(episode+1, num_episodes, rollout+1, num_rollouts))
#
# # Explore
# self.explorer.explore(num_steps, rollout, verbose=verbose)
if verbose:
print("Episode: {}/{} - Rollout: {}/{}".format(episode+1, num_episodes, rollout+1, num_rollouts))
# Explore
self.explorer.explore(num_steps, rollout, verbose=verbose)
# evaluate and update
# 1. explore
self.explorer.explore(num_steps, num_rollouts, verbose=verbose)
# 2. evaluate
if self.evaluator is not None:
self.evaluator.evaluate(verbose=verbose)
# 3. update
losses = self.updater.update(verbose=verbose)
# add the loss in the history
pyrobolearn/approximators/approximator.py
+12
View File
@@ -339,6 +339,18 @@ class Approximator(object):
"""Set the inner model into testing mode."""
self.model.eval()
def copy_parameters(self, parameters):
"""Copy the given parameters.
Args:
parameters (Approximator, Model, torch.nn.Module, generator, iterable): the other policy's parameters to
copy.
"""
if isinstance(parameters, (self.__class__, self.model.__class__)):
self.model.copy_parameters(parameters.parameters())
else:
self.model.copy_parameters(parameters)
def parameters(self):
"""Return an iterator over the approximator parameters."""
return self.model.parameters()
pyrobolearn/envs/env.py
+7 -3
View File
@@ -88,7 +88,9 @@ class Env(object): # gym.Env): # TODO: make it inheriting the gym.Env
self.extra_info = extra_info if extra_info is not None else lambda: False
self.actions = actions
self.rendering = False # check with simulator
# check if we are rendering with the simulator
self.is_rendering = self.simulator.is_rendering()
self.rendering_mode = 'human'
# save the world state in memory
self.initial_world_state = self.world.save()
@@ -337,8 +339,7 @@ class Env(object): # gym.Env): # TODO: make it inheriting the gym.Env
# compute reward
# rewards = [reward.compute() for reward in self.rewards]
if self.rewards is not None:
rewards = self.rewards()
rewards = self.rewards() if self.rewards is not None else None
# compute terminating condition
done = any([condition() for condition in self.terminal_conditions])
@@ -355,10 +356,13 @@ class Env(object): # gym.Env): # TODO: make it inheriting the gym.Env
def render(self, mode='human'):
"""Renders the environment (show the GUI)."""
self.is_rendering = True
self.rendering_mode = mode
self.sim.render()
def hide(self):
"""hide the GUI."""
self.is_rendering = False
self.sim.hide()
def close(self):
pyrobolearn/envs/gym_wrapper.py
+5 -1
View File
@@ -101,6 +101,9 @@ class GymEnvWrapper(gym.Env):
self.state_processors = state_processors
self.reward_processors = reward_processors
# rendering
self.is_rendering = False
##############
# Properties #
##############
@@ -270,11 +273,12 @@ class GymEnvWrapper(gym.Env):
def render(self, mode='human'):
"""Render the gym environment."""
self.is_rendering = True
self.env.render(mode)
def hide(self):
"""Hide the gym environment (not used)."""
pass
self.is_rendering = False
def close(self):
"""Close the gym environment."""
pyrobolearn/models/basics/linear.py
+5 -3
View File
@@ -5,6 +5,8 @@ The linear model is a discriminative deterministic model given by: :math:`y = f(
"""
import copy
import types
import collections
try:
import cPickle as pickle
@@ -162,7 +164,7 @@ class Linear(object):
"""Copy the given parameters.
Args:
parameters (NN, torch.nn.Module, generator, iterable): the other model's parameters to copy.
parameters (Linear, torch.nn.Module, generator, iterable): the other model's parameters to copy.
"""
if isinstance(parameters, self.__class__):
self.model.load_state_dict(parameters.model.state_dict())
@@ -172,8 +174,8 @@ class Linear(object):
for model_params, other_params in zip(self.parameters(), parameters):
model_params.data.copy_(other_params.data)
else:
raise TypeError("Expecting the given parameters to be an instance of `NN`, `torch.nn.Module`, `generator`"
", or an iterable object, instead got: {}".format(type(parameters)))
raise TypeError("Expecting the given parameters to be an instance of `Linear`, `torch.nn.Module`, "
"`generator`, or an iterable object, instead got: {}".format(type(parameters)))
def parameters(self):
"""Returns an iterator over the model parameters."""
pyrobolearn/models/model.py
+21 -16
View File
@@ -219,22 +219,27 @@ class Model(object):
pass
self._models.append(model)
# def copy_parameters(self, parameters):
# """Copy the given parameters.
#
# Args:
# parameters (NN, torch.nn.Module, generator, iterable): the other model's parameters to copy.
# """
# if isinstance(parameters, self.__class__):
# self.model.load_state_dict(parameters.model.state_dict())
# elif isinstance(parameters, torch.nn.Module):
# self.model.load_state_dict(parameters.state_dict())
# elif isinstance(parameters, (types.GeneratorType, collections.Iterable)):
# for model_params, other_params in zip(self.parameters(), parameters):
# model_params.data.copy_(other_params.data)
# else:
# raise TypeError("Expecting the given parameters to be an instance of `NN`, `torch.nn.Module`, `generator`"
# ", or an iterable object, instead got: {}".format(type(parameters)))
def copy_parameters(self, parameters):
"""Copy the given parameters.
Args:
parameters (torch.nn.Module, generator, iterable): the other model's parameters to copy.
"""
if len(self.models) == 1:
self.models[0].copy_parameters(parameters)
else:
for model, parameter in zip(self.models, parameters):
model.copy_parameters(parameter)
# if isinstance(parameters, self.__class__):
# self.model.load_state_dict(parameters.model.state_dict())
# elif isinstance(parameters, torch.nn.Module):
# self.model.load_state_dict(parameters.state_dict())
# elif isinstance(parameters, (types.GeneratorType, collections.Iterable)):
# for model_params, other_params in zip(self.parameters(), parameters):
# model_params.data.copy_(other_params.data)
# else:
# raise TypeError("Expecting the given parameters to be an instance of `torch.nn.Module`, `generator`"
# ", or an iterable object, instead got: {}".format(type(parameters)))
@abstractmethod
def parameters(self):
pyrobolearn/policies/policy.py
+12
View File
@@ -326,6 +326,18 @@ class Policy(object):
"""
raise self.model.is_recurrent()
def copy_parameters(self, parameters):
"""Copy the given parameters.
Args:
parameters (Policy, Approximator, torch.nn.Module, generator, iterable): the other policy's parameters to
copy.
"""
if isinstance(parameters, (self.__class__, self.model.__class__)):
self.model.copy_parameters(parameters.parameters())
else:
self.model.copy_parameters(parameters)
def parameters(self):
"""
Return an iterator over the learning model parameters.
pyrobolearn/simulators/README.md
+13 -2
View File
@@ -1,13 +1,24 @@
## Simulators
This folder contains the APIs to the various simulators. Currently, the main simulator being supported is PyBullet. Work is under progress for Gazebo+ROS, and OpenSIM.
This folder contains the APIs to the various simulators. Currently, the main simulator being supported is PyBullet.
Work is under progress for other simulators.
```python
import pyrobolearn as prl
sim = prl.simulators.BulletSim()
sim = prl.simulators.Bullet()
sim1 = prl.simulators.Dart()
```
#### What to check next?
Check the `worlds` folder and the `robots` folder.
#### TODOs
- [x] implement Bullet interface
- [ ] implement Dart interface
- [ ] implement ROS_RBDL interface
- [ ] implement Gazebo_ROS interface
- [ ] implement Mujoco interface
- [ ] implement `simulator_randomizer` (similar to `physics_randomizer`)
pyrobolearn/simulators/__init__.py
+5 -1
View File
@@ -5,7 +5,11 @@
from .simulator import Simulator
# bullet simulator
from .bullet import Bullet as BulletSim
from .bullet import Bullet
from .bullet import Bullet as BulletSim # alias # TODO: remove that alias
# dart simulator
from .dart import Dart
# # PyBullet simulator
# import pybullet
pyrobolearn/simulators/bullet.py
+5 -7
View File
@@ -20,10 +20,9 @@ References:
# general imports
import os
import inspect
# import inspect
import time
import numpy as np
import quaternion
# import pybullet
import pybullet
@@ -69,7 +68,7 @@ class Bullet(Simulator):
In the following documentation:
* `vec3` specifies a list/tuple/np.array of 3 floats
* `quat` specifies a list/tuple/np.quaternion of 4 floats
* `quat` specifies a list/tuple/np.array of 4 floats
Examples:
sim = Bullet()
@@ -800,11 +799,10 @@ class Bullet(Simulator):
Args:
filename (str): path to file for the mesh. Currently, only Wavefront .obj. It will create convex hulls
for each object (marked as 'o') in the .obj file.
position (float[3]): position of the mesh in the Cartesian world space (in meters)
orientation (float[4], np.quaternion): orientation of the mesh using quaternion.
If np.quaternion then it uses the convention (w,x,y,z). If float[4], it uses the convention (x,y,z,w)
position (list of 3 float, np.array[3]): position of the mesh in the Cartesian world space (in meters)
orientation (list of 4 float, np.array[4]): orientation of the mesh using quaternion [x,y,z,w].
mass (float): mass of the mesh (in kg). If mass = 0, it won't move even if there is a collision.
scale (float[3]): scale the mesh in the (x,y,z) directions
scale (list of 3 float, np.array[3]): scale the mesh in the (x,y,z) directions
color (int[4], None): color of the mesh for red, green, blue, and alpha, each in range [0,1].
with_collision (bool): If True, it will also create the collision mesh, and not only a visual mesh.
flags (int, None): if flag = `sim.GEOM_FORCE_CONCAVE_TRIMESH` (=1), this will create a concave static
pyrobolearn/simulators/dart.py
+2 -2
View File
@@ -293,7 +293,7 @@ class Dart(Simulator):
Returns:
int (non-negative): unique id associated to the load model.
"""
return self.world.add_skeleton(filename)
return self.world.add_skeleton(filename).id
def load_sdf(self, filename, scaling=1., *args, **kwargs):
"""Load a SDF file in the simulator.
@@ -305,7 +305,7 @@ class Dart(Simulator):
Returns:
list(int): list of object unique id for each object loaded
"""
return self.world.add_skeleton(filename)
return self.world.add_skeleton(filename).id
def load_mjcf(self, filename, scaling=1., *args, **kwargs):
"""Load a Mujoco file in the simulator.
pyrobolearn/tasks/task.py
+41 -5
View File
@@ -64,8 +64,19 @@ class Task(object):
__metaclass__ = ABCMeta
def __init__(self, environment, policies):
"""
Initialize the task.
Args:
environment (Env, gym.Env): environment.
policies (list of Policy, Policy): the policy(ies).
"""
# check the environment
if not isinstance(environment, (Env, gym.Env)):
raise TypeError("Expecting 'environment' to be an instance of Env or gym.Env")
self.env = environment
# check the policies
if isinstance(policies, collections.Iterable):
for policy in policies:
if not isinstance(policy, Policy):
@@ -74,9 +85,8 @@ class Task(object):
policies = [policies]
else:
raise TypeError("Expecting 'policies' to be an instance of Policy, or list/tuple of policies")
self.env = environment
self.policies = policies
self._done = False
self._succeeded = False
@@ -204,6 +214,13 @@ class Task(object):
def run(self, num_steps=None, dt=0, use_terminating_condition=False, render=False):
"""
Reset and run the task until it is done, or the current time step matches num_steps.
Args:
num_steps (None, int): number of steps to run.
dt (float): time to sleep for the next step in the environment.
use_terminating_condition (bool): if we should continue or not once the terminal condition has been
fulfilled.
render (bool): if we should render the environment or not.
"""
if num_steps is None:
num_steps = np.infty
@@ -231,6 +248,10 @@ class Task(object):
def step(self, deterministic=True, render=False):
"""
Perform one step.
Args:
deterministic (bool): if policy should be deterministic or not.
render (bool): if we should render or not.
"""
# if we need to render the environment
if render:
@@ -253,11 +274,29 @@ class Task(object):
return np.array(rewards)
def get_policy(self, idx=None):
"""
Get the `idx`th policy.
Args:
idx (int, None): the `idx`th policy to return. If None, it will return the list of policies.
Returns:
(list of) Policy: policy(ies)
"""
if idx is None:
return self.policies
return self.policies[idx]
def get_learning_model(self, idx=None):
"""
Get the learning model associated to the `idx`th policy.
Args:
idx (int): the `idx`th policy. If None, it will return the list of learning models.
Returns:
(list of) Model: learning model(s).
"""
if idx is None:
return [policy.model for policy in self.policies]
return self.policies[idx].model
@@ -271,9 +310,6 @@ class Task(object):
"""Load the storage from the disk."""
return pickle.load(open(filename, 'r'))
def rollout(self): # TODO
pass
#############
# Operators #
#############