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

Use flake8-comprehensions (#1976)

Browse Source 
* Add flake8 to Travis

* Add flake8-comprehensions

[flake8 plugin](https://github.com/adamchainz/flake8-comprehensions) that
checks for useless constructions.

* Use generators instead of lists where appropriate

A lot of the builtins can take in generators instead of lists.

This commit applies `flake8-comprehensions` to find them.

* Fix lint error

* Fix some string formatting

The rest can be fixed in another PR

* Fix compound literals syntax

This should probably be merged after #1963.

* dict() -> {}

* Use dict literal syntax

dict(...) -> {...}

* Rewrite nested dicts

* Fix hanging indent

* Add missing import

* Add missing quote

* fmt

* Add missing whitespace

* rm duplicate pip install

This is already installed in another file.

* Fix indent

* move `merge_dicts` into utils

* Bring up to date with `master`

* Add automatic syntax upgrade

* rm pyupgrade

In case users want to still use it on their own, the upgrade-syn.sh script was
left in the `.travis` dir.
This commit is contained in:
Alok Singh
2018-05-20 16:15:06 -07:00
committed by Philipp Moritz
parent 99ae74e1d2
commit f795173b51
37 changed files with 329 additions and 272 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/actor.py
+2 -2
View File
@@ -491,8 +491,8 @@ class ActorClass(object):
# Extract the signatures of each of the methods. This will be used
# to catch some errors if the methods are called with inappropriate
# arguments.
self._method_signatures = dict()
self._actor_method_num_return_vals = dict()
self._method_signatures = {}
self._actor_method_num_return_vals = {}
for method_name, method in self._actor_methods:
# Print a warning message if the method signature is not
# supported. We don't raise an exception because if the actor
python/ray/autoscaler/aws/config.py
+4 -6
View File
@@ -145,10 +145,8 @@ def _configure_key_pair(config):
def _configure_subnet(config):
ec2 = _resource("ec2", config)
subnets = sorted(
[
s for s in ec2.subnets.all()
if s.state == "available" and s.map_public_ip_on_launch
],
(s for s in ec2.subnets.all()
if s.state == "available" and s.map_public_ip_on_launch),
reverse=True, # sort from Z-A
key=lambda subnet: subnet.availability_zone)
if not subnets:
@@ -293,11 +291,11 @@ def _get_key(key_name, config):
def _client(name, config):
boto_config = Config(retries=dict(max_attempts=BOTO_MAX_RETRIES))
boto_config = Config(retries={'max_attempts': BOTO_MAX_RETRIES})
return boto3.client(name, config["provider"]["region"], config=boto_config)
def _resource(name, config):
boto_config = Config(retries=dict(max_attempts=BOTO_MAX_RETRIES))
boto_config = Config(retries={'max_attempts': BOTO_MAX_RETRIES})
return boto3.resource(
name, config["provider"]["region"], config=boto_config)
python/ray/autoscaler/aws/node_provider.py
+1 -1
View File
@@ -13,7 +13,7 @@ from ray.ray_constants import BOTO_MAX_RETRIES
class AWSNodeProvider(NodeProvider):
def __init__(self, provider_config, cluster_name):
NodeProvider.__init__(self, provider_config, cluster_name)
config = Config(retries=dict(max_attempts=BOTO_MAX_RETRIES))
config = Config(retries={'max_attempts': BOTO_MAX_RETRIES})
self.ec2 = boto3.resource(
"ec2", region_name=provider_config["region"], config=config)
python/ray/dataframe/concat.py
+4 -4
View File
@@ -38,8 +38,8 @@ def concat(objs, axis=0, join='outer', join_axes=None, ignore_index=False,
"and ray.dataframe.DataFrame objs are "
"valid", type(type_check))
all_series = all([isinstance(obj, pandas.Series)
for obj in objs])
all_series = all(isinstance(obj, pandas.Series)
for obj in objs)
if all_series:
return pandas.concat(objs, axis, join, join_axes,
ignore_index, keys, levels, names,
@@ -47,8 +47,8 @@ def concat(objs, axis=0, join='outer', join_axes=None, ignore_index=False,
if isinstance(objs, dict):
raise NotImplementedError(
"Obj as dicts not implemented. To contribute to "
"Pandas on Ray, please visit github.com/ray-project/ray.")
"Obj as dicts not implemented. To contribute to "
"Pandas on Ray, please visit github.com/ray-project/ray.")
axis = pandas.DataFrame()._get_axis_number(axis)
python/ray/dataframe/dataframe.py
+28 -16
View File
@@ -668,7 +668,7 @@ class DataFrame(object):
mismatch = len(by) != len(self) if axis == 0 \
else len(by) != len(self.columns)
if all([obj in self for obj in by]) and mismatch:
if all(obj in self for obj in by) and mismatch:
raise NotImplementedError(
"Groupby with lists of columns not yet supported.")
elif mismatch:
@@ -2194,7 +2194,7 @@ class DataFrame(object):
A Series with the index for each maximum value for the axis
specified.
"""
if not all([d != np.dtype('O') for d in self.dtypes]):
if not all(d != np.dtype('O') for d in self.dtypes):
raise TypeError(
"reduction operation 'argmax' not allowed for this dtype")
@@ -2216,7 +2216,7 @@ class DataFrame(object):
A Series with the index for each minimum value for the axis
specified.
"""
if not all([d != np.dtype('O') for d in self.dtypes]):
if not all(d != np.dtype('O') for d in self.dtypes):
raise TypeError(
"reduction operation 'argmax' not allowed for this dtype")
@@ -3196,9 +3196,9 @@ class DataFrame(object):
"""
# This if call prevents ValueErrors with object only partitions
if (numeric_only and
all([dtype == np.dtype('O') or
is_timedelta64_dtype(dtype)
for dtype in df.dtypes])):
all(dtype == np.dtype('O') or
is_timedelta64_dtype(dtype)
for dtype in df.dtypes)):
return base_object
else:
return df.quantile(q=q, axis=axis, numeric_only=numeric_only,
@@ -4224,16 +4224,28 @@ class DataFrame(object):
tupleize_cols=None, date_format=None, doublequote=True,
escapechar=None, decimal="."):
kwargs = dict(
path_or_buf=path_or_buf, sep=sep, na_rep=na_rep,
float_format=float_format, columns=columns, header=header,
index=index, index_label=index_label, mode=mode,
encoding=encoding, compression=compression, quoting=quoting,
quotechar=quotechar, line_terminator=line_terminator,
chunksize=chunksize, tupleize_cols=tupleize_cols,
date_format=date_format, doublequote=doublequote,
escapechar=escapechar, decimal=decimal
)
kwargs = {
'path_or_buf': path_or_buf,
'sep': sep,
'na_rep': na_rep,
'float_format': float_format,
'columns': columns,
'header': header,
'index': index,
'index_label': index_label,
'mode': mode,
'encoding': encoding,
'compression': compression,
'quoting': quoting,
'quotechar': quotechar,
'line_terminator': line_terminator,
'chunksize': chunksize,
'tupleize_cols': tupleize_cols,
'date_format': date_format,
'doublequote': doublequote,
'escapechar': escapechar,
'decimal': decimal
}
if compression is not None:
warnings.warn("Defaulting to Pandas implementation",
python/ray/dataframe/io.py
+55 -54
View File
@@ -208,60 +208,61 @@ def read_csv(filepath_or_buffer,
kwargs: Keyword arguments in pandas::from_csv
"""
kwargs = dict(
sep=sep,
delimiter=delimiter,
header=header,
names=names,
index_col=index_col,
usecols=usecols,
squeeze=squeeze,
prefix=prefix,
mangle_dupe_cols=mangle_dupe_cols,
dtype=dtype,
engine=engine,
converters=converters,
true_values=true_values,
false_values=false_values,
skipinitialspace=skipinitialspace,
skiprows=skiprows,
nrows=nrows,
na_values=na_values,
keep_default_na=keep_default_na,
na_filter=na_filter,
verbose=verbose,
skip_blank_lines=skip_blank_lines,
parse_dates=parse_dates,
infer_datetime_format=infer_datetime_format,
keep_date_col=keep_date_col,
date_parser=date_parser,
dayfirst=dayfirst,
iterator=iterator,
chunksize=chunksize,
compression=compression,
thousands=thousands,
decimal=decimal,
lineterminator=lineterminator,
quotechar=quotechar,
quoting=quoting,
escapechar=escapechar,
comment=comment,
encoding=encoding,
dialect=dialect,
tupleize_cols=tupleize_cols,
error_bad_lines=error_bad_lines,
warn_bad_lines=warn_bad_lines,
skipfooter=skipfooter,
skip_footer=skip_footer,
doublequote=doublequote,
delim_whitespace=delim_whitespace,
as_recarray=as_recarray,
compact_ints=compact_ints,
use_unsigned=use_unsigned,
low_memory=low_memory,
buffer_lines=buffer_lines,
memory_map=memory_map,
float_precision=float_precision)
kwargs = {
'sep': sep,
'delimiter': delimiter,
'header': header,
'names': names,
'index_col': index_col,
'usecols': usecols,
'squeeze': squeeze,
'prefix': prefix,
'mangle_dupe_cols': mangle_dupe_cols,
'dtype': dtype,
'engine': engine,
'converters': converters,
'true_values': true_values,
'false_values': false_values,
'skipinitialspace': skipinitialspace,
'skiprows': skiprows,
'nrows': nrows,
'na_values': na_values,
'keep_default_na': keep_default_na,
'na_filter': na_filter,
'verbose': verbose,
'skip_blank_lines': skip_blank_lines,
'parse_dates': parse_dates,
'infer_datetime_format': infer_datetime_format,
'keep_date_col': keep_date_col,
'date_parser': date_parser,
'dayfirst': dayfirst,
'iterator': iterator,
'chunksize': chunksize,
'compression': compression,
'thousands': thousands,
'decimal': decimal,
'lineterminator': lineterminator,
'quotechar': quotechar,
'quoting': quoting,
'escapechar': escapechar,
'comment': comment,
'encoding': encoding,
'dialect': dialect,
'tupleize_cols': tupleize_cols,
'error_bad_lines': error_bad_lines,
'warn_bad_lines': warn_bad_lines,
'skipfooter': skipfooter,
'skip_footer': skip_footer,
'doublequote': doublequote,
'delim_whitespace': delim_whitespace,
'as_recarray': as_recarray,
'compact_ints': compact_ints,
'use_unsigned': use_unsigned,
'low_memory': low_memory,
'buffer_lines': buffer_lines,
'memory_map': memory_map,
'float_precision': float_precision,
}
# Default to Pandas read_csv for non-serializable objects
if not isinstance(filepath_or_buffer, str) or \
python/ray/dataframe/test/test_dataframe.py
+1 -1
View File
@@ -1783,7 +1783,7 @@ def test_fillna_dtype_conversion(num_partitions=2):
)
# equiv of replace
df = pd.DataFrame(dict(A=[1, np.nan], B=[1., 2.]))
df = pd.DataFrame({'A': [1, np.nan], 'B': [1., 2.]})
ray_df = from_pandas(df, num_partitions)
for v in ['', 1, np.nan, 1.0]:
assert ray_df_equals_pandas(
python/ray/dataframe/utils.py
+2 -2
View File
@@ -9,7 +9,7 @@ import ray
from . import get_npartitions
_NAN_BLOCKS = dict()
_NAN_BLOCKS = {}
def _get_nan_block_id(n_row=1, n_col=1, transpose=False):
@@ -225,7 +225,7 @@ def _map_partitions(func, partitions, *argslists):
return [_deploy_func.remote(func, part, argslists[0])
for part in partitions]
else:
assert(all([len(args) == len(partitions) for args in argslists]))
assert(all(len(args) == len(partitions) for args in argslists))
return [_deploy_func.remote(func, *args)
for args in zip(partitions, *argslists)]
python/ray/experimental/array/distributed/core.py
+1 -1
View File
@@ -241,7 +241,7 @@ def subblocks(a, *ranges):
result = DistArray(shape)
for index in np.ndindex(*result.num_blocks):
result.objectids[index] = a.objectids[tuple(
[ranges[i][index[i]] for i in range(a.ndim)])]
ranges[i][index[i]] for i in range(a.ndim))]
return result
python/ray/experimental/state.py
+7 -7
View File
@@ -360,7 +360,7 @@ class GlobalState(object):
"""
self._check_connected()
db_client_keys = self.redis_client.keys(DB_CLIENT_PREFIX + "*")
node_info = dict()
node_info = {}
for key in db_client_keys:
client_info = self.redis_client.hgetall(key)
node_ip_address = decode(client_info[b"node_ip_address"])
@@ -403,7 +403,7 @@ class GlobalState(object):
"""
relevant_files = self.redis_client.keys("LOGFILE*")
ip_filename_file = dict()
ip_filename_file = {}
for filename in relevant_files:
filename = filename.decode("ascii")
@@ -417,7 +417,7 @@ class GlobalState(object):
file_str.append(y)
if ip_addr not in ip_filename_file:
ip_filename_file[ip_addr] = dict()
ip_filename_file[ip_addr] = {}
ip_filename_file[ip_addr][filename] = file_str
@@ -445,7 +445,7 @@ class GlobalState(object):
list of profiling information for tasks where the events have
no task ID.
"""
task_info = dict()
task_info = {}
event_log_sets = self.redis_client.keys("event_log*")
# The heap is used to maintain the set of x tasks that occurred the
@@ -498,7 +498,7 @@ class GlobalState(object):
for event in event_dict:
if "task_id" in event[3]:
task_id = event[3]["task_id"]
task_info[task_id] = dict()
task_info[task_id] = {}
task_info[task_id]["score"] = score
# Add task to (min/max) heap by its start point.
# if fwd, we want to delete the largest elements, so -score
@@ -901,7 +901,7 @@ class GlobalState(object):
def workers(self):
"""Get a dictionary mapping worker ID to worker information."""
worker_keys = self.redis_client.keys("Worker*")
workers_data = dict()
workers_data = {}
for worker_key in worker_keys:
worker_info = self.redis_client.hgetall(worker_key)
@@ -927,7 +927,7 @@ class GlobalState(object):
def actors(self):
actor_keys = self.redis_client.keys("Actor:*")
actor_info = dict()
actor_info = {}
for key in actor_keys:
info = self.redis_client.hgetall(key)
actor_id = key[len("Actor:"):]
python/ray/experimental/tfutils.py
+4 -5
View File
@@ -84,8 +84,8 @@ class TensorFlowVariables(object):
for v in variable_list:
self.variables[v.op.node_def.name] = v
self.placeholders = dict()
self.assignment_nodes = dict()
self.placeholders = {}
self.assignment_nodes = {}
# Create new placeholders to put in custom weights.
for k, var in self.variables.items():
@@ -109,9 +109,8 @@ class TensorFlowVariables(object):
Returns:
The length of all flattened variables concatenated.
"""
return sum([
np.prod(v.get_shape().as_list()) for v in self.variables.values()
])
return sum(
np.prod(v.get_shape().as_list()) for v in self.variables.values())
def _check_sess(self):
"""Checks if the session is set, and if not throw an error message."""
python/ray/experimental/ui.py
+6 -3
View File
@@ -580,8 +580,11 @@ def cpu_usage():
y_range=[0, 1])
# Create the data source that the plot will pull from
time_series_source = ColumnDataSource(
data=dict(left=[], right=[], top=[]))
time_series_source = ColumnDataSource(data={
'left': [],
'right': [],
'top': []
})
# Plot the rectangles representing the distribution
time_series_fig.quad(
@@ -731,7 +734,7 @@ def cluster_usage():
earliest = time.time()
latest = 0
node_to_tasks = dict()
node_to_tasks = {}
# Determine which task has the earlest start time out of the ones
# passed into the update function
for task_id, data in tasks.items():
python/ray/global_scheduler/test/test.py
+6 -11
View File
@@ -285,13 +285,10 @@ class TestGlobalScheduler(unittest.TestCase):
for task_entry in task_entries.values()
]
self.assertTrue(
all([
status in [
state.TASK_STATUS_WAITING,
state.TASK_STATUS_SCHEDULED,
state.TASK_STATUS_QUEUED
] for status in task_statuses
]))
all(status in [
state.TASK_STATUS_WAITING, state.TASK_STATUS_SCHEDULED,
state.TASK_STATUS_QUEUED
] for status in task_statuses))
num_tasks_done = task_statuses.count(state.TASK_STATUS_QUEUED)
num_tasks_scheduled = task_statuses.count(
state.TASK_STATUS_SCHEDULED)
@@ -302,10 +299,8 @@ class TestGlobalScheduler(unittest.TestCase):
"tasks queued = {}, retries left = {}".format(
len(task_entries), num_tasks_waiting,
num_tasks_scheduled, num_tasks_done, num_retries))
if all([
status == state.TASK_STATUS_QUEUED
for status in task_statuses
]):
if all(status == state.TASK_STATUS_QUEUED
for status in task_statuses):
# We're done, so pass.
break
num_retries -= 1
python/ray/monitor.py
+1 -1
View File
@@ -97,7 +97,7 @@ class Monitor(object):
self.dead_plasma_managers = set()
# Keep a mapping from local scheduler client ID to IP address to use
# for updating the load metrics.
self.local_scheduler_id_to_ip_map = dict()
self.local_scheduler_id_to_ip_map = {}
self.load_metrics = LoadMetrics()
if autoscaling_config:
self.autoscaler = StandardAutoscaler(autoscaling_config,
python/ray/rllib/a3c/tfpolicy.py
+2 -2
View File
@@ -21,8 +21,8 @@ class TFPolicy(Policy):
with self.g.as_default(), tf.device(worker_device):
with tf.variable_scope(name):
self._setup_graph(ob_space, action_space)
assert all([hasattr(self, attr)
for attr in ["vf", "logits", "x", "var_list"]])
assert all(hasattr(self, attr)
for attr in ["vf", "logits", "x", "var_list"])
print("Setting up loss")
self.setup_loss(action_space)
self.setup_gradients()
python/ray/rllib/ddpg/apex.py
+25 -22
View File
@@ -3,29 +3,32 @@ from __future__ import division
from __future__ import print_function
from ray.rllib.ddpg.ddpg import DDPGAgent, DEFAULT_CONFIG as DDPG_CONFIG
from ray.utils import merge_dicts
APEX_DDPG_DEFAULT_CONFIG = dict(DDPG_CONFIG,
**dict(
optimizer_class="ApexOptimizer",
optimizer_config=dict(
DDPG_CONFIG["optimizer_config"],
**dict(
max_weight_sync_delay=400,
num_replay_buffer_shards=4,
debug=False,
)),
n_step=3,
num_workers=32,
buffer_size=2000000,
learning_starts=50000,
train_batch_size=512,
sample_batch_size=50,
max_weight_sync_delay=400,
target_network_update_freq=500000,
timesteps_per_iteration=25000,
per_worker_exploration=True,
worker_side_prioritization=True,
))
APEX_DDPG_DEFAULT_CONFIG = merge_dicts(
DDPG_CONFIG,
{
'optimizer_class': 'ApexOptimizer',
'optimizer_config':
merge_dicts(
DDPG_CONFIG['optimizer_config'], {
'max_weight_sync_delay': 400,
'num_replay_buffer_shards': 4,
'debug': False
}),
'n_step': 3,
'num_workers': 32,
'buffer_size': 2000000,
'learning_starts': 50000,
'train_batch_size': 512,
'sample_batch_size': 50,
'max_weight_sync_delay': 400,
'target_network_update_freq': 500000,
'timesteps_per_iteration': 25000,
'per_worker_exploration': True,
'worker_side_prioritization': True,
},
)
class ApexDDPGAgent(DDPGAgent):
python/ray/rllib/ddpg/ddpg.py
+41 -40
View File
@@ -20,89 +20,89 @@ OPTIMIZER_SHARED_CONFIGS = [
"train_batch_size", "learning_starts", "clip_rewards"
]
DEFAULT_CONFIG = dict(
DEFAULT_CONFIG = {
# === Model ===
# Hidden layer sizes of the policy networks
actor_hiddens=[64, 64],
'actor_hiddens': [64, 64],
# Hidden layer sizes of the policy networks
critic_hiddens=[64, 64],
'critic_hiddens': [64, 64],
# N-step Q learning
n_step=1,
'n_step': 1,
# Config options to pass to the model constructor
model={},
'model': {},
# Discount factor for the MDP
gamma=0.99,
'gamma': 0.99,
# Arguments to pass to the env creator
env_config={},
'env_config': {},
# === Exploration ===
# Max num timesteps for annealing schedules. Exploration is annealed from
# 1.0 to exploration_fraction over this number of timesteps scaled by
# exploration_fraction
schedule_max_timesteps=100000,
'schedule_max_timesteps': 100000,
# Number of env steps to optimize for before returning
timesteps_per_iteration=1000,
'timesteps_per_iteration': 1000,
# Fraction of entire training period over which the exploration rate is
# annealed
exploration_fraction=0.1,
'exploration_fraction': 0.1,
# Final value of random action probability
exploration_final_eps=0.02,
'exploration_final_eps': 0.02,
# OU-noise scale
noise_scale=0.1,
'noise_scale': 0.1,
# theta
exploration_theta=0.15,
'exploration_theta': 0.15,
# sigma
exploration_sigma=0.2,
'exploration_sigma': 0.2,
# Update the target network every `target_network_update_freq` steps.
target_network_update_freq=0,
'target_network_update_freq': 0,
# Update the target by \tau * policy + (1-\tau) * target_policy
tau=0.002,
'tau': 0.002,
# Whether to start with random actions instead of noops.
random_starts=True,
'random_starts': True,
# === Replay buffer ===
# Size of the replay buffer. Note that if async_updates is set, then
# each worker will have a replay buffer of this size.
buffer_size=50000,
'buffer_size': 50000,
# If True prioritized replay buffer will be used.
prioritized_replay=True,
'prioritized_replay': True,
# Alpha parameter for prioritized replay buffer.
prioritized_replay_alpha=0.6,
'prioritized_replay_alpha': 0.6,
# Beta parameter for sampling from prioritized replay buffer.
prioritized_replay_beta=0.4,
'prioritized_replay_beta': 0.4,
# Epsilon to add to the TD errors when updating priorities.
prioritized_replay_eps=1e-6,
'prioritized_replay_eps': 1e-6,
# Whether to clip rewards to [-1, 1] prior to adding to the replay buffer.
clip_rewards=True,
'clip_rewards': True,
# === Optimization ===
# Learning rate for adam optimizer
actor_lr=1e-4,
critic_lr=1e-3,
'actor_lr': 1e-4,
'critic_lr': 1e-3,
# If True, use huber loss instead of squared loss for critic network
# Conventionally, no need to clip gradients if using a huber loss
use_huber=False,
'use_huber': False,
# Threshold of a huber loss
huber_threshold=1.0,
'huber_threshold': 1.0,
# Weights for L2 regularization
l2_reg=1e-6,
'l2_reg': 1e-6,
# If not None, clip gradients during optimization at this value
grad_norm_clipping=None,
'grad_norm_clipping': None,
# How many steps of the model to sample before learning starts.
learning_starts=1500,
'learning_starts': 1500,
# Update the replay buffer with this many samples at once. Note that this
# setting applies per-worker if num_workers > 1.
sample_batch_size=1,
'sample_batch_size': 1,
# Size of a batched sampled from replay buffer for training. Note that
# if async_updates is set, then each worker returns gradients for a
# batch of this size.
train_batch_size=256,
'train_batch_size': 256,
# Smooth the current average reward over this many previous episodes.
smoothing_num_episodes=100,
'smoothing_num_episodes': 100,
# === Tensorflow ===
# Arguments to pass to tensorflow
tf_session_args={
'tf_session_args': {
"device_count": {
"CPU": 2
},
@@ -119,17 +119,18 @@ DEFAULT_CONFIG = dict(
# Number of workers for collecting samples with. This only makes sense
# to increase if your environment is particularly slow to sample, or if
# you're using the Async or Ape-X optimizers.
num_workers=0,
'num_workers': 0,
# Whether to allocate GPUs for workers (if > 0).
num_gpus_per_worker=0,
'num_gpus_per_worker': 0,
# Optimizer class to use.
optimizer_class="LocalSyncReplayOptimizer",
'optimizer_class': "LocalSyncReplayOptimizer",
# Config to pass to the optimizer.
optimizer_config=dict(),
'optimizer_config': {},
# Whether to use a distribution of epsilons across workers for exploration.
per_worker_exploration=False,
'per_worker_exploration': False,
# Whether to compute priorities on workers.
worker_side_prioritization=False)
'worker_side_prioritization': False
}
class DDPGAgent(Agent):
python/ray/rllib/dqn/apex.py
+26 -20
View File
@@ -4,27 +4,33 @@ from __future__ import print_function
from ray.rllib.dqn.dqn import DQNAgent, DEFAULT_CONFIG as DQN_CONFIG
from ray.tune.trial import Resources
from ray.utils import merge_dicts
APEX_DEFAULT_CONFIG = dict(DQN_CONFIG, **dict(
optimizer_class="ApexOptimizer",
optimizer_config=dict(DQN_CONFIG["optimizer_config"], **dict(
max_weight_sync_delay=400,
num_replay_buffer_shards=4,
debug=False,
)),
n_step=3,
gpu=True,
num_workers=32,
buffer_size=2000000,
learning_starts=50000,
train_batch_size=512,
sample_batch_size=50,
max_weight_sync_delay=400,
target_network_update_freq=500000,
timesteps_per_iteration=25000,
per_worker_exploration=True,
worker_side_prioritization=True,
))
APEX_DEFAULT_CONFIG = merge_dicts(
DQN_CONFIG,
{
'optimizer_class': 'ApexOptimizer',
'optimizer_config':
merge_dicts(
DQN_CONFIG['optimizer_config'], {
'max_weight_sync_delay': 400,
'num_replay_buffer_shards': 4,
'debug': False
}),
'n_step': 3,
'gpu': True,
'num_workers': 32,
'buffer_size': 2000000,
'learning_starts': 50000,
'train_batch_size': 512,
'sample_batch_size': 50,
'max_weight_sync_delay': 400,
'target_network_update_freq': 500000,
'timesteps_per_iteration': 25000,
'per_worker_exploration': True,
'worker_side_prioritization': True,
},
)
class ApexAgent(DQNAgent):
python/ray/rllib/dqn/dqn.py
+36 -35
View File
@@ -21,75 +21,75 @@ OPTIMIZER_SHARED_CONFIGS = [
"prioritized_replay_beta", "prioritized_replay_eps", "sample_batch_size",
"train_batch_size", "learning_starts", "clip_rewards"]
DEFAULT_CONFIG = dict(
DEFAULT_CONFIG = {
# === Model ===
# Whether to use dueling dqn
dueling=True,
'dueling': True,
# Whether to use double dqn
double_q=True,
'double_q': True,
# Hidden layer sizes of the state and action value networks
hiddens=[256],
'hiddens': [256],
# N-step Q learning
n_step=1,
'n_step': 1,
# Config options to pass to the model constructor
model={},
'model': {},
# Discount factor for the MDP
gamma=0.99,
'gamma': 0.99,
# Arguments to pass to the env creator
env_config={},
'env_config': {},
# === Exploration ===
# Max num timesteps for annealing schedules. Exploration is annealed from
# 1.0 to exploration_fraction over this number of timesteps scaled by
# exploration_fraction
schedule_max_timesteps=100000,
'schedule_max_timesteps': 100000,
# Number of env steps to optimize for before returning
timesteps_per_iteration=1000,
'timesteps_per_iteration': 1000,
# Fraction of entire training period over which the exploration rate is
# annealed
exploration_fraction=0.1,
'exploration_fraction': 0.1,
# Final value of random action probability
exploration_final_eps=0.02,
'exploration_final_eps': 0.02,
# Update the target network every `target_network_update_freq` steps.
target_network_update_freq=500,
'target_network_update_freq': 500,
# Whether to start with random actions instead of noops.
random_starts=True,
'random_starts': True,
# === Replay buffer ===
# Size of the replay buffer. Note that if async_updates is set, then
# each worker will have a replay buffer of this size.
buffer_size=50000,
'buffer_size': 50000,
# If True prioritized replay buffer will be used.
prioritized_replay=True,
'prioritized_replay': True,
# Alpha parameter for prioritized replay buffer.
prioritized_replay_alpha=0.6,
'prioritized_replay_alpha': 0.6,
# Beta parameter for sampling from prioritized replay buffer.
prioritized_replay_beta=0.4,
'prioritized_replay_beta': 0.4,
# Epsilon to add to the TD errors when updating priorities.
prioritized_replay_eps=1e-6,
'prioritized_replay_eps': 1e-6,
# Whether to clip rewards to [-1, 1] prior to adding to the replay buffer.
clip_rewards=True,
'clip_rewards': True,
# === Optimization ===
# Learning rate for adam optimizer
lr=5e-4,
'lr': 5e-4,
# If not None, clip gradients during optimization at this value
grad_norm_clipping=40,
'grad_norm_clipping': 40,
# How many steps of the model to sample before learning starts.
learning_starts=1000,
'learning_starts': 1000,
# Update the replay buffer with this many samples at once. Note that
# this setting applies per-worker if num_workers > 1.
sample_batch_size=4,
'sample_batch_size': 4,
# Size of a batched sampled from replay buffer for training. Note that
# if async_updates is set, then each worker returns gradients for a
# batch of this size.
train_batch_size=32,
'train_batch_size': 32,
# Smooth the current average reward over this many previous episodes.
smoothing_num_episodes=100,
'smoothing_num_episodes': 100,
# === Tensorflow ===
# Arguments to pass to tensorflow
tf_session_args={
'tf_session_args': {
"device_count": {"CPU": 2},
"log_device_placement": False,
"allow_soft_placement": True,
@@ -102,23 +102,24 @@ DEFAULT_CONFIG = dict(
# === Parallelism ===
# Whether to use a GPU for local optimization.
gpu=False,
'gpu': False,
# Number of workers for collecting samples with. This only makes sense
# to increase if your environment is particularly slow to sample, or if
# you're using the Async or Ape-X optimizers.
num_workers=0,
'num_workers': 0,
# Whether to allocate GPUs for workers (if > 0).
num_gpus_per_worker=0,
'num_gpus_per_worker': 0,
# Whether to allocate CPUs for workers (if > 0).
num_cpus_per_worker=1,
'num_cpus_per_worker': 1,
# Optimizer class to use.
optimizer_class="LocalSyncReplayOptimizer",
'optimizer_class': "LocalSyncReplayOptimizer",
# Config to pass to the optimizer.
optimizer_config=dict(),
'optimizer_config': {},
# Whether to use a distribution of epsilons across workers for exploration.
per_worker_exploration=False,
'per_worker_exploration': False,
# Whether to compute priorities on workers.
worker_side_prioritization=False)
'worker_side_prioritization': False
}
class DQNAgent(Agent):
python/ray/rllib/es/es.py
+17 -16
View File
@@ -27,18 +27,19 @@ Result = namedtuple("Result", [
])
DEFAULT_CONFIG = dict(
l2_coeff=0.005,
noise_stdev=0.02,
episodes_per_batch=1000,
timesteps_per_batch=10000,
eval_prob=0.003,
return_proc_mode="centered_rank",
num_workers=10,
stepsize=0.01,
observation_filter="MeanStdFilter",
noise_size=250000000,
env_config={})
DEFAULT_CONFIG = {
'l2_coeff': 0.005,
'noise_stdev': 0.02,
'episodes_per_batch': 1000,
'timesteps_per_batch': 10000,
'eval_prob': 0.003,
'return_proc_mode': "centered_rank",
'num_workers': 10,
'stepsize': 0.01,
'observation_filter': "MeanStdFilter",
'noise_size': 250000000,
'env_config': {},
}
@ray.remote
@@ -192,10 +193,10 @@ class ESAgent(agent.Agent):
# Update the number of episodes and the number of timesteps
# keeping in mind that result.noisy_lengths is a list of lists,
# where the inner lists have length 2.
num_episodes += sum([len(pair) for pair
in result.noisy_lengths])
num_timesteps += sum([sum(pair) for pair
in result.noisy_lengths])
num_episodes += sum(len(pair) for pair
in result.noisy_lengths)
num_timesteps += sum(sum(pair) for pair
in result.noisy_lengths)
return results, num_episodes, num_timesteps
def _train(self):
python/ray/rllib/es/policies.py
+3 -3
View File
@@ -59,9 +59,9 @@ class GenericPolicy(object):
self.variables = ray.experimental.TensorFlowVariables(
model.outputs, self.sess)
self.num_params = sum([np.prod(variable.shape.as_list())
for _, variable
in self.variables.variables.items()])
self.num_params = sum(np.prod(variable.shape.as_list())
for _, variable
in self.variables.variables.items())
self.sess.run(tf.global_variables_initializer())
def compute(self, observation, add_noise=False, update=True):
python/ray/rllib/models/catalog.py
+4 -4
View File
@@ -123,7 +123,7 @@ class ModelCatalog(object):
" not supported".format(action_space))
@staticmethod
def get_model(registry, inputs, num_outputs, options=dict()):
def get_model(registry, inputs, num_outputs, options={}):
"""Returns a suitable model conforming to given input and output specs.
Args:
@@ -156,7 +156,7 @@ class ModelCatalog(object):
return FullyConnectedNetwork(inputs, num_outputs, options)
@staticmethod
def get_torch_model(registry, input_shape, num_outputs, options=dict()):
def get_torch_model(registry, input_shape, num_outputs, options={}):
"""Returns a PyTorch suitable model. This is currently only supported
in A3C.
@@ -188,7 +188,7 @@ class ModelCatalog(object):
return PyTorchFCNet(input_shape[0], num_outputs, options)
@staticmethod
def get_preprocessor(registry, env, options=dict()):
def get_preprocessor(registry, env, options={}):
"""Returns a suitable processor for the given environment.
Args:
@@ -215,7 +215,7 @@ class ModelCatalog(object):
return preprocessor(env.observation_space, options)
@staticmethod
def get_preprocessor_as_wrapper(registry, env, options=dict()):
def get_preprocessor_as_wrapper(registry, env, options={}):
"""Returns a preprocessor as a gym observation wrapper.
Args:
python/ray/rllib/optimizers/replay_buffer.py
+1 -1
View File
@@ -44,7 +44,7 @@ class ReplayBuffer(object):
if self._next_idx >= len(self._storage):
self._storage.append(data)
self._est_size_bytes += sum([sys.getsizeof(d) for d in data])
self._est_size_bytes += sum(sys.getsizeof(d) for d in data)
else:
self._storage[self._next_idx] = data
if self._next_idx + 1 >= self._maxsize:
python/ray/signature.py
+2 -2
View File
@@ -54,7 +54,7 @@ def get_signature_params(func):
"__code__", "__annotations__", "__defaults__", "__kwdefaults__"
]
if all([hasattr(func, attr) for attr in attrs]):
if all(hasattr(func, attr) for attr in attrs):
original_func = func
def func():
@@ -63,7 +63,7 @@ def get_signature_params(func):
for attr in attrs:
setattr(func, attr, getattr(original_func, attr))
else:
raise TypeError("{0!r} is not a Python function we can process"
raise TypeError("{!r} is not a Python function we can process"
.format(func))
return list(funcsigs.signature(func).parameters.items())
python/ray/test/test_utils.py
+1 -1
View File
@@ -95,7 +95,7 @@ def _wait_for_event(event_name, redis_address, extra_buffer=0):
redis_client = redis.StrictRedis(host=redis_host, port=int(redis_port))
while True:
event_infos = redis_client.lrange(EVENT_KEY, 0, -1)
events = dict()
events = {}
for event_info in event_infos:
name, data = json.loads(event_info)
if name in events:
python/ray/tune/hyperband.py
+1 -1
View File
@@ -397,5 +397,5 @@ class Bracket():
])
counts = collections.Counter([t.status for t in self._all_trials])
trial_statuses = ", ".join(
sorted(["{}: {}".format(k, v) for k, v in counts.items()]))
sorted("{}: {}".format(k, v) for k, v in counts.items()))
return "Bracket({}): {{{}}} ".format(status, trial_statuses)
python/ray/tune/median_stopping_rule.py
+1 -1
View File
@@ -113,4 +113,4 @@ class MedianStoppingRule(FIFOScheduler):
def _best_result(self, trial):
results = self._results[trial]
return max([getattr(r, self._reward_attr) for r in results])
return max(getattr(r, self._reward_attr) for r in results)
python/ray/tune/test/tune_server_test.py
+4 -1
View File
@@ -63,7 +63,10 @@ class TuneServerSuite(unittest.TestCase):
"stop": {
"training_iteration": 3
},
"trial_resources": dict(cpu=1, gpu=1),
"trial_resources": {
'cpu': 1,
'gpu': 1
},
}
client.add_trial("test", spec)
runner.step()
python/ray/utils.py
+7
View File
@@ -208,3 +208,10 @@ def resources_from_resource_arguments(default_num_cpus, default_num_gpus,
resources["GPU"] = default_num_gpus
return resources
def merge_dicts(d1, d2):
"""Merge two dicts and return a new dict that's their union."""
d = d1.copy()
d.update(d2)
return d