leaderboard/leaderboard/autoagents/ros_agent.py
def run_step(self, input_data, timestamp):
        """
        Execute one step of navigation.
        """
        self.vehicle_control_event.clear()
        self.timestamp = timestamp
        self.clock_publisher.publish(Clock(rospy.Time.from_sec(timestamp)))

        # check if stack is still running
        if self.stack_process and self.stack_process.poll() is not None:
            raise RuntimeError("Stack exited with: {} {}".format(
                self.stack_process.returncode, self.stack_process.communicate()[0]))

        # publish global plan to ROS once
        if self._global_plan_world_coord and not self.global_plan_published:
            self.global_plan_published = True
            self.publish_plan()

        new_data_available = False

        # publish data of all sensors
        for key, val in input_data.items():
            new_data_available = True
            sensor_type = self.id_to_sensor_type_map[key]
            if sensor_type == 'sensor.camera.rgb':
                self.publish_camera(key, val[1])
            elif sensor_type == 'sensor.lidar.ray_cast':
                self.publish_lidar(key, val[1])
            elif sensor_type == 'sensor.other.gnss':
                self.publish_gnss(key, val[1])
            elif sensor_type == 'sensor.can_bus':
                self.publish_can(key, val[1])
            elif sensor_type == 'sensor.hd_map':
                self.publish_hd_map(key, val[1])
            else:
                raise TypeError("Invalid sensor type: {}".format(sensor_type))

        if self.use_stepping_mode():
            if self.step_mode_possible and new_data_available:
                self.vehicle_control_event.wait()
        # if the stepping mode is not used or active, there is no need to wait here

        return self.current_control

        Execute one step of navigation.
        
Execute one step of navigation.
Execute one step of navigation .
def run_step(self, input_data, timestamp):
        self.vehicle_control_event.clear()
        self.timestamp = timestamp
        self.clock_publisher.publish(Clock(rospy.Time.from_sec(timestamp)))
        if self.stack_process and self.stack_process.poll() is not None:
            raise RuntimeError("Stack exited with: {} {}".format(
                self.stack_process.returncode, self.stack_process.communicate()[0]))
        if self._global_plan_world_coord and not self.global_plan_published:
            self.global_plan_published = True
            self.publish_plan()
        new_data_available = False
        for key, val in input_data.items():
            new_data_available = True
            sensor_type = self.id_to_sensor_type_map[key]
            if sensor_type == 'sensor.camera.rgb':
                self.publish_camera(key, val[1])
            elif sensor_type == 'sensor.lidar.ray_cast':
                self.publish_lidar(key, val[1])
            elif sensor_type == 'sensor.other.gnss':
                self.publish_gnss(key, val[1])
            elif sensor_type == 'sensor.can_bus':
                self.publish_can(key, val[1])
            elif sensor_type == 'sensor.hd_map':
                self.publish_hd_map(key, val[1])
            else:
                raise TypeError("Invalid sensor type: {}".format(sensor_type))
        if self.use_stepping_mode():
            if self.step_mode_possible and new_data_available:
                self.vehicle_control_event.wait()
        return self.current_control
def run_step ( self , input_data , timestamp ) : self . vehicle_control_event . clear ( ) self . timestamp = timestamp self . clock_publisher . publish ( Clock ( rospy . Time . from_sec ( timestamp ) ) ) if self . stack_process and self . stack_process . poll ( ) is not None : raise RuntimeError ( "Stack exited with: {} {}" . format ( self . stack_process . returncode , self . stack_process . communicate ( ) [ 0 ] ) ) if self . _global_plan_world_coord and not self . global_plan_published : self . global_plan_published = True self . publish_plan ( ) new_data_available = False for key , val in input_data . items ( ) : new_data_available = True sensor_type = self . id_to_sensor_type_map [ key ] if sensor_type == 'sensor.camera.rgb' : self . publish_camera ( key , val [ 1 ] ) elif sensor_type == 'sensor.lidar.ray_cast' : self . publish_lidar ( key , val [ 1 ] ) elif sensor_type == 'sensor.other.gnss' : self . publish_gnss ( key , val [ 1 ] ) elif sensor_type == 'sensor.can_bus' : self . publish_can ( key , val [ 1 ] ) elif sensor_type == 'sensor.hd_map' : self . publish_hd_map ( key , val [ 1 ] ) else : raise TypeError ( "Invalid sensor type: {}" . format ( sensor_type ) ) if self . use_stepping_mode ( ) : if self . step_mode_possible and new_data_available : self . vehicle_control_event . wait ( ) return self . current_control
Execute one step of navigation.
Execute one step of navigation .
"""
        Execute one step of navigation.
        """ # check if stack is still running # publish global plan to ROS once # publish data of all sensors # if the stepping mode is not used or active, there is no need to wait here
self
input_data
timestamp