A basic traffic mode for PGTrafficManager (#827)

* Implemeted a basic traffic mode

* Format

Author: olek-osikowicz

documentation/source/rl_environments.ipynb

@@ -18,10 +18,13 @@
 
 
 class TrafficMode:
+    # Traffic vehicles will be spawned once
+    Basic = "basic"
+
     # Traffic vehicles will be respawned, once they arrive at the destinations
     Respawn = "respawn"
 
-    # Traffic vehicles will be triggered only once
+    # Traffic vehicles will be triggered only once, and will be triggered when agent comes close
     Trigger = "trigger"
 
     # Hybrid, some vehicles are triggered once on map and disappear when arriving at destination, others exist all time
@@ -63,13 +66,15 @@ def reset(self):
         if abs(traffic_density) < 1e-2:
             return
         self.respawn_lanes = self._get_available_respawn_lanes(map)
-        if self.mode == TrafficMode.Respawn:
-            # add respawn vehicle
-            self._create_respawn_vehicles(map, traffic_density)
-        elif self.mode == TrafficMode.Trigger or self.mode == TrafficMode.Hybrid:
-            self._create_vehicles_once(map, traffic_density)
+
+        logging.debug(f"Resetting Traffic Manager with mode {self.mode} and density {traffic_density}")
+
+        if self.mode in {TrafficMode.Basic, TrafficMode.Respawn}:
+            self._create_basic_vehicles(map, traffic_density)
+        elif self.mode in {TrafficMode.Trigger, TrafficMode.Hybrid}:
+            self._create_trigger_vehicles(map, traffic_density)
         else:
-            raise ValueError("No such mode named {}".format(self.mode))
+            raise ValueError(f"No such mode named {self.mode}")
 
     def before_step(self):
         """
@@ -78,14 +83,15 @@ def before_step(self):
         """
         # trigger vehicles
         engine = self.engine
-        if self.mode != TrafficMode.Respawn:
+        if self.mode in {TrafficMode.Trigger, TrafficMode.Hybrid}:
             for v in engine.agent_manager.active_agents.values():
                 if len(self.block_triggered_vehicles) > 0:
                     ego_lane_idx = v.lane_index[:-1]
                     ego_road = Road(ego_lane_idx[0], ego_lane_idx[1])
                     if ego_road == self.block_triggered_vehicles[-1].trigger_road:
                         block_vehicles = self.block_triggered_vehicles.pop()
                         self._traffic_vehicles += list(self.get_objects(block_vehicles.vehicles).values())
+
         for v in self._traffic_vehicles:
             p = self.engine.get_policy(v.name)
             v.before_step(p.act())
@@ -99,17 +105,15 @@ def after_step(self, *args, **kwargs):
         for v in self._traffic_vehicles:
             v.after_step()
             if not v.on_lane:
-                if self.mode == TrafficMode.Trigger:
-                    v_to_remove.append(v)
-                elif self.mode == TrafficMode.Respawn or self.mode == TrafficMode.Hybrid:
-                    v_to_remove.append(v)
-                else:
-                    raise ValueError("Traffic mode error: {}".format(self.mode))
+                v_to_remove.append(v)
+
         for v in v_to_remove:
             vehicle_type = type(v)
             self.clear_objects([v.id])
             self._traffic_vehicles.remove(v)
-            if self.mode == TrafficMode.Respawn or self.mode == TrafficMode.Hybrid:
+
+            # Spawn new vehicles to replace the removed one
+            if self.mode in {TrafficMode.Respawn, TrafficMode.Hybrid}:
                 lane = self.respawn_lanes[self.np_random.randint(0, len(self.respawn_lanes))]
                 lane_idx = lane.index
                 long = self.np_random.rand() * lane.length / 2
@@ -136,7 +140,7 @@ def get_vehicle_num(self):
         Get the vehicles on road
         :return:
         """
-        if self.mode == TrafficMode.Respawn:
+        if self.mode in {TrafficMode.Basic, TrafficMode.Respawn}:
             return len(self._traffic_vehicles)
         return sum(len(block_vehicle_set.vehicles) for block_vehicle_set in self.block_triggered_vehicles)
 
@@ -151,7 +155,7 @@ def get_global_states(self) -> Dict:
             traffic_states[vehicle.index] = vehicle.get_state()
 
         # collect other vehicles
-        if self.mode != TrafficMode.Respawn:
+        if self.mode in {TrafficMode.Trigger, TrafficMode.Hybrid}:
             for v_b in self.block_triggered_vehicles:
                 for vehicle in v_b.vehicles:
                     traffic_states[vehicle.index] = vehicle.get_state()
@@ -188,7 +192,7 @@ def get_global_init_states(self) -> Dict:
             vehicles[vehicle.index] = init_state
 
         # collect other vehicles
-        if self.mode != TrafficMode.Respawn:
+        if self.mode in {TrafficMode.Trigger, TrafficMode.Hybrid}:
             for v_b in self.block_triggered_vehicles:
                 for vehicle in v_b.vehicles:
                     init_state = vehicle.get_state()
@@ -208,7 +212,7 @@ def _propose_vehicle_configs(self, lane: AbstractLane):
             potential_vehicle_configs.append(random_vehicle_config)
         return potential_vehicle_configs
 
-    def _create_respawn_vehicles(self, map: BaseMap, traffic_density: float):
+    def _create_basic_vehicles(self, map: BaseMap, traffic_density: float):
         total_num = len(self.respawn_lanes)
         for lane in self.respawn_lanes:
             _traffic_vehicles = []
@@ -227,7 +231,7 @@ def _create_respawn_vehicles(self, map: BaseMap, traffic_density: float):
                 self.add_policy(random_v.id, IDMPolicy, random_v, self.generate_seed())
                 self._traffic_vehicles.append(random_v)
 
-    def _create_vehicles_once(self, map: BaseMap, traffic_density: float) -> None:
+    def _create_trigger_vehicles(self, map: BaseMap, traffic_density: float) -> None:
         """
         Trigger mode, vehicles will be triggered only once, and disappear when arriving destination
         :param map: Map
@@ -365,10 +369,10 @@ def set_state(self, state: dict, old_name_to_current=None):
 
 
 class MixedPGTrafficManager(PGTrafficManager):
-    def _create_respawn_vehicles(self, *args, **kwargs):
+    def _create_basic_vehicles(self, *args, **kwargs):
         raise NotImplementedError()
 
-    def _create_vehicles_once(self, map: BaseMap, traffic_density: float) -> None:
+    def _create_trigger_vehicles(self, map: BaseMap, traffic_density: float) -> None:
         vehicle_num = 0
         for block in map.blocks[1:]: