Davis

opencda/scenario_testing/single_2lanefree_carla.py

@@ -16,7 +16,7 @@
 from opencda.scenario_testing.evaluations.evaluate_manager import \
     EvaluationManager
 from opencda.scenario_testing.utils.yaml_utils import load_yaml
-
+from opencda.scenario_testing.utils.keyboard_listener import KeyListener
 
 def run_scenario(opt, config_yaml):
     try:
@@ -55,8 +55,20 @@ def run_scenario(opt, config_yaml):
                               current_time=scenario_params['current_time'])
 
         spectator = scenario_manager.world.get_spectator()
+
+        # initialize a key listener and start monitoring the keyboard
+        kl = KeyListener()
+        kl.start()
+
         # run steps
         while True:
+            # press esc to end the program
+            # press p to pause demonstration
+            if kl.keys['esc']:
+                exit(0)
+            if kl.keys['p']:
+                continue
+
             scenario_manager.tick()
             transform = single_cav_list[0].vehicle.get_transform()
             spectator.set_transform(carla.Transform(

opencda/scenario_testing/utils/keyboard_listener.py

@@ -0,0 +1,84 @@
+# -*- coding: utf-8 -*-
+"""
+keyboard listener: a keyboard listenr to record typeing and can be used for pasue, resume and stop the program using multi-threading.
+"""
+# Author: Wei Shao <weishao@ucdavis.edu>
+# License: TDG-Attribution-NonCommercial-NoDistrib
+
+from pynput import keyboard
+
+
+class KeyListener(object):
+    """
+    A keyboard listener class to record the states of keys.
+
+    Parameters
+    ----------
+    None
+
+    Attributes
+    ----------
+    keys : dict
+        Keyboard keys and states
+    """
+    def __init__(self):
+        """
+        Initialize a listener instance
+        """
+        self.keys = {'p': False, 'esc': False}
+
+    def start(self):
+        """
+        Start keyboard listening
+
+        Parameters
+        ----------
+        None
+
+        Returns
+        -------
+        None
+        """
+        listener = keyboard.Listener(
+            on_press=self.on_press,
+            on_release=self.on_release)
+        listener.start()
+
+    def on_press(self, key):
+        """
+        Update the dict when a key is pressed
+
+        Parameters
+        ----------
+        key: A KeyCode represents the description of a key code used by the operating system.
+
+        Returns
+        -------
+        Update the dict keys
+        """
+        try:
+            print(f'alphanumeric key {key.char} pressed')
+            if key.char in self.keys:
+                self.keys[key.char] = not self.keys[key.char]
+            else:
+                self.keys[key] = False
+        except AttributeError:
+            print(f'special key {key} pressed')
+
+    def on_release(self, key):
+        """
+        Update the dict keys when a key is released
+
+        Parameters
+        ----------
+        key: A KeyCode represents the description of a key code used by the operating system.
+
+        Returns
+        -------
+        Update the dict keys
+        """
+        if key == keyboard.Key.esc:
+            self.keys['esc'] = True
+
+
+