TrafficManager is a class that manages the traffic simulation, including vehicle behavior updates, decision making, and planning. It uses the provided roadgraph and vehicle information to generate trajectories for each vehicle in the simulation. Attributes: sumo_model: The
| 36 | KEY_INPUT = "" |
| 37 | |
| 38 | class TrafficManager: |
| 39 | """ |
| 40 | TrafficManager is a class that manages the traffic simulation, including vehicle behavior updates, |
| 41 | decision making, and planning. It uses the provided roadgraph and vehicle information to generate |
| 42 | trajectories for each vehicle in the simulation. |
| 43 | |
| 44 | Attributes: |
| 45 | sumo_model: The SUMO traffic simulation model. |
| 46 | T: The current simulation time. |
| 47 | lastseen_vehicles: A dictionary containing the last seen state of each vehicle. |
| 48 | config: The configuration dictionary. |
| 49 | predictor: An instance of the UncontrolledPredictor class. |
| 50 | ego_decision: An instance of the EgoDecisionMaker class. |
| 51 | ego_planner: An instance of the EgoPlanner class. |
| 52 | multi_veh_planner: An instance of the MultiVehiclePlanner class. |
| 53 | """ |
| 54 | |
| 55 | def __init__(self, |
| 56 | model: Model, |
| 57 | predictor: AbstractPredictor = None, |
| 58 | ego_decision: AbstractEgoDecisionMaker = None, |
| 59 | ego_planner: AbstractEgoPlanner = None, |
| 60 | multi_decision=None, |
| 61 | multi_veh_planner: AbstractMultiPlanner = None, |
| 62 | config_file_path="{}/config.yaml".format(os.path.dirname(__file__))): |
| 63 | self.sumo_model = model |
| 64 | self.time_step = 0 |
| 65 | self.lastseen_vehicles = {} |
| 66 | self.config = load_config(config_file_path) |
| 67 | self.last_decision_time = -self.config["DECISION_INTERVAL"] |
| 68 | self.mul_decisions =None |
| 69 | self._set_up_keyboard_listener() |
| 70 | |
| 71 | self.predictor = predictor if predictor is not None else UncontrolledPredictor() |
| 72 | self.ego_decision = ego_decision if ego_decision is not None else EgoDecisionMaker() |
| 73 | self.ego_planner = ego_planner if ego_planner is not None else EgoPlanner() |
| 74 | self.multi_decision = multi_decision if multi_decision is not None else MultiDecisionMaker() |
| 75 | self.multi_veh_planner = multi_veh_planner if multi_veh_planner is not None else MultiVehiclePlanner() |
| 76 | |
| 77 | def _set_up_keyboard_listener(self): |
| 78 | |
| 79 | def on_press(key): |
| 80 | """ |
| 81 | This function is used to detect the key press from the keyboard. |
| 82 | When the left arrow key or 'a' is pressed, the global variable KEY_INPUT is set to 'Left'. |
| 83 | When the right arrow key or 'd' is pressed, the global variable KEY_INPUT is set to 'Right'. |
| 84 | """ |
| 85 | global KEY_INPUT |
| 86 | if key == keyboard.Key.left or key == keyboard.KeyCode.from_char( |
| 87 | 'a'): |
| 88 | KEY_INPUT = 'Left' |
| 89 | elif key == keyboard.Key.right or key == keyboard.KeyCode.from_char( |
| 90 | 'd'): |
| 91 | KEY_INPUT = 'Right' |
| 92 | |
| 93 | listener = keyboard.Listener(on_press=on_press) |
| 94 | listener.start() # start to listen on a separate thread |
| 95 |
no outgoing calls
no test coverage detected