Merge ab1459e into bd49a78

environment.yml

@@ -9,7 +9,7 @@ dependencies:
     - path.py
     - python-dateutil==2.7.3
     - tensorflow==1.9.0
-    - cloudpickle==0.5.3
+    - cloudpickle==1.2.1
     - setuptools==39.1.0
     - pip:
         - gym==0.14.0
@@ -20,7 +20,7 @@ dependencies:
         - matplotlib==3.0.0
         - dill
         - lz4
-        - ray==0.6.1
+        - ray==0.7.3
         - setproctitle
         - psutil
         - opencv-python

examples/rllib/figure_eight.py

@@ -135,7 +135,7 @@ def setup_exps():
 
 if __name__ == '__main__':
     alg_run, gym_name, config = setup_exps()
-    ray.init(num_cpus=N_CPUS + 1, redirect_output=False)
+    ray.init(num_cpus=N_CPUS + 1)
     trials = run_experiments({
         flow_params['exp_tag']: {
             'run': alg_run,

examples/rllib/stabilizing_highway.py

@@ -181,7 +181,7 @@ def setup_exps():
 
 if __name__ == "__main__":
     alg_run, gym_name, config = setup_exps()
-    ray.init(num_cpus=N_CPUS + 1, redirect_output=False)
+    ray.init(num_cpus=N_CPUS + 1)
     trials = run_experiments({
         flow_params["exp_tag"]: {
             "run": alg_run,

examples/rllib/stabilizing_the_ring.py

@@ -138,7 +138,7 @@ def setup_exps():
 
 if __name__ == "__main__":
     alg_run, gym_name, config = setup_exps()
-    ray.init(num_cpus=N_CPUS + 1, redirect_output=False)
+    ray.init(num_cpus=N_CPUS + 1)
     trials = run_experiments({
         flow_params["exp_tag"]: {
             "run": alg_run,

examples/rllib/traffic_light_grid.py

@@ -281,7 +281,7 @@ def setup_exps(use_inflows=False):
 
 if __name__ == '__main__':
     alg_run, gym_name, config = setup_exps()
-    ray.init(num_cpus=N_CPUS + 1, redirect_output=False)
+    ray.init(num_cpus=N_CPUS + 1)
     trials = run_experiments({
         flow_params['exp_tag']: {
             'run': alg_run,

examples/rllib/velocity_bottleneck.py

@@ -203,7 +203,7 @@ def setup_exps():
 
 if __name__ == "__main__":
     alg_run, gym_name, config = setup_exps()
-    ray.init(num_cpus=N_CPUS + 1, redirect_output=False)
+    ray.init(num_cpus=N_CPUS + 1)
     trials = run_experiments({
         flow_params["exp_tag"]: {
             "run": alg_run,

flow/benchmarks/rllib/ars_runner.py

@@ -79,7 +79,7 @@
     alg_run = "ARS"
 
     # initialize a ray instance
-    ray.init(redirect_output=True)
+    ray.init()
 
     agent_cls = get_agent_class(alg_run)
     config = agent_cls._default_config.copy()

flow/benchmarks/rllib/es_runner.py

@@ -78,7 +78,7 @@
     create_env, env_name = make_create_env(params=flow_params, version=0)
 
     # initialize a ray instance
-    ray.init(redirect_output=True)
+    ray.init()
 
     alg_run = "ES"
 

flow/benchmarks/rllib/ppo_runner.py

@@ -76,7 +76,7 @@
     create_env, env_name = make_create_env(params=flow_params, version=0)
 
     # initialize a ray instance
-    ray.init(redirect_output=True)
+    ray.init()
 
     alg_run = "PPO"
 

flow/core/kernel/network/base.py

@@ -344,7 +344,7 @@ def gen_even_start_pos(self, initial_config, num_vehicles):
 
             # ensures that you are in an acceptable edge
             while pos[0] not in available_edges:
-                x = (x + self.edge_length(pos[0])) % self.length()
+                x = (x + self.edge_length(pos[0])) % self.non_internal_length()
                 pos = self.get_edge(x)
 
             # ensure that in variable lane settings vehicles always start a
@@ -366,7 +366,7 @@ def gen_even_start_pos(self, initial_config, num_vehicles):
                 if car_count == num_vehicles:
                     break
 
-            x = (x + increment + VEHICLE_LENGTH + min_gap) % self.length()
+            x = (x + increment + VEHICLE_LENGTH + min_gap) % self.non_internal_length()
 
         # add a perturbation to each vehicle, while not letting the vehicle
         # leave its current edge

flow/core/kernel/network/traci.py

@@ -79,7 +79,8 @@ def __init__(self, master_kernel, sim_params):
         self._edge_list = None
         self._junction_list = None
         self.__max_speed = None
-        self.__length = None
+        self.__length = None  # total length
+        self.__non_internal_length = None  # total length of non-internal edges
         self.rts = None
         self.cfg = None
 
@@ -158,7 +159,7 @@ def generate_network(self, network):
 
         # length of the network, or the portion of the network in
         # which cars are meant to be distributed
-        self.__length = sum(
+        self.__non_internal_length = sum(
             self.edge_length(edge_id) for edge_id in self.get_edge_list()
         )
 
@@ -190,6 +191,10 @@ def generate_network(self, network):
 
         self.total_edgestarts_dict = dict(self.total_edgestarts)
 
+        self.__length = sum(
+            self._edges[edge_id]['length'] for edge_id in self._edges
+        )
+
         if self.network.routes is None:
             print("No routes specified, defaulting to single edge routes.")
             self.network.routes = {edge: [edge] for edge in self._edge_list}
@@ -278,6 +283,10 @@ def length(self):
         """See parent class."""
         return self.__length
 
+    def non_internal_length(self):
+        """See parent class."""
+        return self.__non_internal_length
+
     def speed_limit(self, edge_id):
         """See parent class."""
         try:

scripts/benchmark_autoscale.yaml

@@ -40,7 +40,7 @@ auth:
 # http://boto3.readthedocs.io/en/latest/reference/services/ec2.html#EC2.ServiceResource.create_instances
 head_node:
     InstanceType: c4.4xlarge
-    ImageId: ami-060b3311dd2b675ee # Flow AMI (Ubuntu)
+    ImageId: ami-09544298704576518 # Flow AMI (Ubuntu)
     InstanceMarketOptions:
         MarketType: spot
          #Additional options can be found in the boto docs, e.g.
@@ -55,7 +55,7 @@ head_node:
 # http://boto3.readthedocs.io/en/latest/reference/services/ec2.html#EC2.ServiceResource.create_instances
 worker_nodes:
     InstanceType: c4.4xlarge
-    ImageId: ami-060b3311dd2b675ee # Flow AMI (Ubuntu)
+    ImageId: ami-09544298704576518 # Flow AMI (Ubuntu)
 
     #Run workers on spot by default. Comment this out to use on-demand.
     InstanceMarketOptions:
@@ -66,17 +66,10 @@ worker_nodes:
 
     # Additional options in the boto docs.
 
-setup_commands:
-    # checkout your desired branch on all worker nodes
-    - cd flow && git fetch && git checkout origin/new_sumo
-#    - cd ray && git fetch flow_upstream && git checkout flow_upstream/ray_merge && git pull flow_upstream master
-    - echo 'export PATH="/home/ubuntu/sumo/bin:$PATH"' >> ~/.bashrc
-    - echo 'export SUMO_HOME="/home/ubuntu/sumo"' >> ~/.bashrc
-    - echo 'export PYTHONPATH="/home/ubuntu/sumo/tools:$PYTHONPATH"' >> ~/.bashrc
-    - echo 'export PATH="/home/ubuntu/anaconda3/bin:$PATH"' >> ~/.bashrc
+setup_commands: []
 # Custom commands that will be run on the head node after common setup.
 head_setup_commands:
-    - PATH="/home/ubuntu/anaconda3/bin:$PATH" yes | ~/anaconda3/bin/conda install boto3=1.4.8  # 1.4.8 adds InstanceMarketOptions
+    - pip install boto3==1.4.8  # 1.4.8 adds InstanceMarketOptions
 
 # Custom commands that will be run on worker nodes after common setup.
 worker_setup_commands: []

scripts/ray_autoscale.yaml

@@ -40,8 +40,7 @@ auth:
 # http://boto3.readthedocs.io/en/latest/reference/services/ec2.html#EC2.ServiceResource.create_instances
 head_node:
     InstanceType: c4.4xlarge
-    #ImageId: ami-060b3311dd2b675ee # Flow AMI (Ubuntu)
-    ImageId: ami-0067051ba5c8f0b64 # Flow AMI (Ubuntu)
+    ImageId: ami-09544298704576518 # Flow AMI (Ubuntu)
     InstanceMarketOptions:
         MarketType: spot
          #Additional options can be found in the boto docs, e.g.
@@ -56,8 +55,7 @@ head_node:
 # http://boto3.readthedocs.io/en/latest/reference/services/ec2.html#EC2.ServiceResource.create_instances
 worker_nodes:
     InstanceType: c4.4xlarge
-    #ImageId: ami-060b3311dd2b675ee # Flow AMI (Ubuntu)
-    ImageId: ami-0067051ba5c8f0b64 # Flow AMI (Ubuntu)
+    ImageId: ami-09544298704576518 # Flow AMI (Ubuntu)
 
     #Run workers on spot by default. Comment this out to use on-demand.
     InstanceMarketOptions:
@@ -68,27 +66,9 @@ worker_nodes:
 
     # Additional options in the boto docs.
 
-setup_commands:
-    # checkout your desired branch on all worker nodes
-    - cd flow && git fetch && git checkout itsc_bottleneck
-    # set up ray by pip installing and copying over rllib and tune folder
-    - pip install --upgrade pyzmq notebook
-    - pip install ray==0.6.1
-    - pip install lz4
-    - git clone https://github.com/flow-project/ray.git
-    - cd ray && git fetch && git checkout master
-    - rm -rf /home/ubuntu/anaconda3/lib/python3.5/site-packages/ray/rllib
-    - cd ray && cp -r python/ray/rllib /home/ubuntu/anaconda3/lib/python3.5/site-packages/ray/rllib
-    - rm -rf /home/ubuntu/anaconda3/lib/python3.5/site-packages/ray/tune
-    - cd ray && cp -r python/ray/tune /home/ubuntu/anaconda3/lib/python3.5/site-packages/ray/tune
-    # set up sumo and anaconda
-    - echo 'export PATH="/home/ubuntu/sumo/bin:$PATH"' >> ~/.bashrc
-    - echo 'export SUMO_HOME="/home/ubuntu/sumo"' >> ~/.bashrc
-    - echo 'export PYTHONPATH="/home/ubuntu/sumo/tools:$PYTHONPATH"' >> ~/.bashrc
-    - echo 'export PATH="/home/ubuntu/anaconda3/bin:$PATH"' >> ~/.bashrc
-# Custom commands that will be run on the head node after common setup.
+setup_commands: []
 head_setup_commands:
-    - PATH="/home/ubuntu/anaconda3/bin:$PATH" yes | ~/anaconda3/bin/conda install boto3=1.4.8  # 1.4.8 adds InstanceMarketOptions
+    - pip install boto3==1.4.8  # 1.4.8 adds InstanceMarketOptions
 
 # Custom commands that will be run on worker nodes after common setup.
 worker_setup_commands: []

tests/fast_tests/test_environments.py

@@ -423,11 +423,11 @@ def test_reset(self):
         )
 
         # reset the network several times and check its length
-        self.assertEqual(env.k.network.length(), 230)
+        self.assertEqual(env.k.network.non_internal_length(), 230)
         env.reset()
-        self.assertEqual(env.k.network.length(), 239)
+        self.assertEqual(env.k.network.non_internal_length(), 239)
         env.reset()
-        self.assertEqual(env.k.network.length(), 256)
+        self.assertEqual(env.k.network.non_internal_length(), 256)
 
     def test_v_eq_max_function(self):
         """
@@ -460,11 +460,11 @@ def test_reset_no_same_length(self):
         )
 
         # reset the network several times and check its length
-        self.assertEqual(env.k.network.length(), RING_PARAMS["length"])
+        self.assertEqual(env.k.network.non_internal_length(), RING_PARAMS["length"])
         env.reset()
-        self.assertEqual(env.k.network.length(), RING_PARAMS["length"])
+        self.assertEqual(env.k.network.non_internal_length(), RING_PARAMS["length"])
         env.reset()
-        self.assertEqual(env.k.network.length(), RING_PARAMS["length"])
+        self.assertEqual(env.k.network.non_internal_length(), RING_PARAMS["length"])
 
 
 class TestWaveAttenuationPOEnv(unittest.TestCase):

tests/fast_tests/test_scenario_base_class.py

@@ -161,7 +161,7 @@ def test_base(self):
         # of it
         nth_headway = np.mod(
             np.append(veh_pos[1:], veh_pos[0]) - veh_pos,
-            self.env.k.network.length())
+            self.env.k.network.non_internal_length())
 
         # check that the position of the first vehicle is at 0
         self.assertEqual(veh_pos[0], 0)
@@ -194,10 +194,10 @@ def test_x0(self):
         # of it
         nth_headway = np.mod(
             np.append(veh_pos[1:], veh_pos[0]) - veh_pos,
-            self.env.k.network.length())
+            self.env.k.network.non_internal_length())
 
         # check that the position of the first vehicle is at 0
-        self.assertEqual(veh_pos[0], x0 % self.env.k.network.length())
+        self.assertEqual(veh_pos[0], x0 % self.env.k.network.non_internal_length())
 
         # if all element are equal, there should only be one unique value
         self.assertEqual(np.unique(np.around(nth_headway, 2)).size, 1)
@@ -225,7 +225,7 @@ def test_bunching(self):
         # of it
         nth_headway = np.mod(
             np.append(veh_pos[1:], veh_pos[0]) - veh_pos,
-            self.env.k.network.length())
+            self.env.k.network.non_internal_length())
 
         # check that all vehicles except the last vehicle have the same spacing
         self.assertEqual(np.unique(np.around(nth_headway[:-1], 2)).size, 1)
@@ -260,7 +260,7 @@ def test_bunching_too_small(self):
         # of it
         nth_headway = np.mod(
             np.append(veh_pos[1:], veh_pos[0]) - veh_pos,
-            self.env.k.network.length())
+            self.env.k.network.non_internal_length())
 
         # check that all vehicles, including the last vehicle, have the same
         # spacing
@@ -300,7 +300,7 @@ def test_lanes_distribution(self):
             # ahead of it
             nth_headway = \
                 np.mod(np.append(veh_pos[i][1:], veh_pos[i][0]) - veh_pos[i],
-                       self.env.k.network.length())
+                       self.env.k.network.non_internal_length())
 
             self.assertEqual(np.unique(np.around(nth_headway[:-1], 2)).size, 1)
 
@@ -361,7 +361,7 @@ def test_lanes_distribution_too_large(self):
             # ahead of it
             nth_headway = \
                 np.mod(np.append(veh_pos[i][1:], veh_pos[i][0]) - veh_pos[i],
-                       self.env.k.network.length())
+                       self.env.k.network.non_internal_length())
 
             self.assertEqual(np.unique(np.around(nth_headway[:-1], 2)).size, 1)
 
@@ -473,7 +473,7 @@ def test_even_start_pos_internal(self):
         # of it
         nth_headway = np.mod(
             np.append(veh_pos[1:], veh_pos[0]) - veh_pos,
-            self.env.k.network.length())
+            self.env.k.network.non_internal_length())
 
         try:
             # if all element are equal, there should only be one unique value

tests/slow_tests/test_benchmarks.py

@@ -35,7 +35,7 @@
 from flow.benchmarks.stable_baselines.trpo_runner import save_model
 
 N_CPUS = 1
-ray.init(num_cpus=N_CPUS, redirect_output=True)
+ray.init(num_cpus=N_CPUS)
 
 os.environ['TEST_FLAG'] = 'True'
 

tests/stress_tests/stress_test_rl.py

@@ -40,7 +40,7 @@
 
     start = time.time()
     print("stress test starting")
-    ray.init(redirect_output=False)
+    ray.init()
     flow_params["env"].horizon = 1
     horizon = flow_params["env"].horizon
 

tutorials/tutorial03_rllib.ipynb

@@ -335,7 +335,7 @@
    "metadata": {},
    "source": [
     "### 4.2 Initializing Ray\n",
-    "Here, we initialize Ray and experiment-based constant variables specifying parallelism in the experiment as well as experiment batch size in terms of number of rollouts. `redirect_output` sends stdout and stderr for non-worker processes to files if True. "
+    "Here, we initialize Ray and experiment-based constant variables specifying parallelism in the experiment as well as experiment batch size in terms of number of rollouts."
    ]
   },
   {
@@ -349,7 +349,7 @@
     "# number of rollouts per training iteration\n",
     "N_ROLLOUTS = 1\n",
     "\n",
-    "ray.init(redirect_output=True, num_cpus=N_CPUS)"
+    "ray.init(num_cpus=N_CPUS)"
    ]
   },
   {

tutorials/tutorial13_rllib_ec2.ipynb

@@ -89,7 +89,7 @@
     "```\n",
     "if __name__ == \"__main__\":\n",
     "    alg_run, gym_name, config = setup_exps()\n",
-    "    ray.init(num_cpus=N_CPUS + 1, redirect_output=False)\n",
+    "    ray.init(num_cpus=N_CPUS + 1)\n",
     "    trials = run_experiments({\n",
     "        flow_params[\"exp_tag\"]: {\n",
     "            \"run\": alg_run,\n",