Merge pull request #438 from flow-project/velocity_bottleneck_tests

Velocity bottleneck tests

Author: AboudyKreidieh

flow/controllers/__init__.py

@@ -16,7 +16,7 @@
     BCMController, OVMController, LinearOVM, IDMController, \
     SimCarFollowingController
 from flow.controllers.velocity_controllers import FollowerStopper, \
-    PISaturation, HandTunedVelocityController
+    PISaturation
 
 # lane change controllers
 from flow.controllers.base_lane_changing_controller import \
@@ -33,7 +33,6 @@
     "RLController", "BaseController", "BaseLaneChangeController", "BaseRouter",
     "CFMController", "BCMController", "OVMController", "LinearOVM",
     "IDMController", "SimCarFollowingController", "FollowerStopper",
-    "PISaturation", "HandTunedVelocityController", "StaticLaneChanger",
-    "SimLaneChangeController", "ContinuousRouter", "GridRouter",
-    "BayBridgeRouter"
+    "PISaturation", "StaticLaneChanger", "SimLaneChangeController",
+    "ContinuousRouter", "GridRouter", "BayBridgeRouter"
 ]

flow/controllers/velocity_controllers.py

@@ -182,33 +182,6 @@ def get_accel(self, env):
         return min(accel, self.max_accel)
 
 
-class HandTunedVelocityController(FollowerStopper):
-    def __init__(self,
-                 veh_id,
-                 v_regions,
-                 car_following_params,
-                 danger_edges=None):
-        super().__init__(
-            veh_id, car_following_params, v_regions[0],
-            danger_edges=danger_edges)
-        self.v_regions = v_regions
-
-    def get_accel(self, env):
-        edge = env.k.vehicle.get_edge(self.veh_id)
-        if edge:
-            if edge[0] != ':' and edge in env.controlled_edges:
-                pos = env.k.vehicle.get_position(self.veh_id)
-                # find what segment we fall into
-                bucket = np.searchsorted(env.slices[edge], pos) - 1
-                action = self.v_regions[bucket +
-                                        env.action_index[int(edge) - 1]]
-                # set the desired velocity of the controller to the action
-                controller = env.k.vehicle.get_acc_controller(self.veh_id)
-                controller.v_des = action
-
-        return super().get_accel(env)
-
-
 class FeedbackController(FollowerStopper):
     def __init__(self,
                  veh_id,

flow/core/kernel/vehicle/traci.py

@@ -323,6 +323,10 @@ def test_set_speed(self, veh_id, speed):
         """Set the speed of the specified vehicle."""
         self.__sumo_obs[veh_id][tc.VAR_SPEED] = speed
 
+    def test_set_edge(self, veh_id, edge):
+        """Set the speed of the specified vehicle."""
+        self.__sumo_obs[veh_id][tc.VAR_ROAD_ID] = edge
+
     def set_follower(self, veh_id, follower):
         """Set the follower of the specified vehicle."""
         self.__vehicles[veh_id]["follower"] = follower

tests/fast_tests/test_controllers.py

@@ -8,6 +8,7 @@
 from flow.controllers.routing_controllers import ContinuousRouter
 from flow.controllers.car_following_models import IDMController, \
     OVMController, BCMController, LinearOVM, CFMController
+from flow.controllers import FollowerStopper, PISaturation
 from tests.setup_scripts import ring_road_exp_setup
 import os
 import numpy as np
@@ -23,7 +24,6 @@ class TestCFMController(unittest.TestCase):
     def setUp(self):
         # add a few vehicles to the network using the requested model
         # also make sure that the input params are what is expected
-
         contr_params = {
             "time_delay": 0,
             "k_d": 1,
@@ -456,5 +456,131 @@ def test_static_lane_changer(self):
         self.assertEqual(sum(np.array(lanes)), 0)
 
 
+class TestFollowerStopper(unittest.TestCase):
+
+    """
+    Makes sure that vehicles with a static lane-changing controller do not
+    change lanes.
+    """
+
+    def setUp(self):
+        # add a few vehicles to the network using the requested model
+        # also make sure that the input params are what is expected
+        contr_params = {"v_des": 7.5}
+
+        vehicles = VehicleParams()
+        vehicles.add(
+            veh_id="test_0",
+            acceleration_controller=(FollowerStopper, contr_params),
+            routing_controller=(ContinuousRouter, {}),
+            car_following_params=SumoCarFollowingParams(
+                accel=20, decel=5),
+            num_vehicles=5)
+
+        # create the environment and scenario classes for a ring road
+        self.env, scenario = ring_road_exp_setup(vehicles=vehicles)
+
+    def tearDown(self):
+        # terminate the traci instance
+        self.env.terminate()
+
+        # free data used by the class
+        self.env = None
+
+    def test_get_action(self):
+        self.env.reset()
+        ids = self.env.k.vehicle.get_ids()
+
+        test_headways = [5, 10, 15, 20, 25]
+        test_speeds = [5, 7.5, 7.5, 8, 7]
+        for i, veh_id in enumerate(ids):
+            self.env.k.vehicle.set_headway(veh_id, test_headways[i])
+            self.env.k.vehicle.test_set_speed(veh_id, test_speeds[i])
+
+        requested_accel = [
+            self.env.k.vehicle.get_acc_controller(veh_id).get_action(self.env)
+            for veh_id in ids
+        ]
+
+        expected_accel = [-16.666667, 0, 0., -5., 5.]
+
+        np.testing.assert_array_almost_equal(requested_accel, expected_accel)
+
+    def test_find_intersection_dist(self):
+        self.env.reset()
+        ids = self.env.k.vehicle.get_ids()
+
+        test_edges = ["", "center"]
+        for i, veh_id in enumerate(ids):
+            if i < 2:
+                self.env.k.vehicle.test_set_edge(veh_id, test_edges[i])
+
+        requested = [
+            self.env.k.vehicle.get_acc_controller(
+                veh_id).find_intersection_dist(self.env)
+            for veh_id in ids
+        ]
+
+        expected = [-10, 0, 23., 34.5, 46.]
+
+        np.testing.assert_array_almost_equal(requested, expected)
+
+        # we also check that the accel value is None when this value is
+        # negative
+        self.assertIsNone(self.env.k.vehicle.get_acc_controller(
+            ids[0]).get_action(self.env))
+
+
+class TestPISaturation(unittest.TestCase):
+
+    """
+    Makes sure that vehicles with a static lane-changing controller do not
+    change lanes.
+    """
+
+    def setUp(self):
+        # add a few vehicles to the network using the requested model
+        # also make sure that the input params are what is expected
+        contr_params = {}
+
+        vehicles = VehicleParams()
+        vehicles.add(
+            veh_id="test_0",
+            acceleration_controller=(PISaturation, contr_params),
+            routing_controller=(ContinuousRouter, {}),
+            car_following_params=SumoCarFollowingParams(
+                accel=20, decel=5),
+            num_vehicles=5)
+
+        # create the environment and scenario classes for a ring road
+        self.env, scenario = ring_road_exp_setup(vehicles=vehicles)
+
+    def tearDown(self):
+        # terminate the traci instance
+        self.env.terminate()
+
+        # free data used by the class
+        self.env = None
+
+    def test_get_action(self):
+        self.env.reset()
+        ids = self.env.k.vehicle.get_ids()
+
+        test_headways = [5, 10, 15, 20, 25]
+        test_speeds = [5, 7.5, 7.5, 8, 7]
+        for i, veh_id in enumerate(ids):
+            self.env.k.vehicle.set_headway(veh_id, test_headways[i])
+            self.env.k.vehicle.test_set_speed(veh_id, test_speeds[i])
+
+        requested_accel = [
+            self.env.k.vehicle.get_acc_controller(veh_id).get_action(self.env)
+            for veh_id in ids
+        ]
+
+        expected_accel = [20., -36.847826, -35.76087, -37.173913, -31.086957]
+
+        np.testing.assert_array_almost_equal(requested_accel, expected_accel)
+
+
 if __name__ == '__main__':
     unittest.main()