Increase code quality and fix test

manual_control.py

@@ -396,6 +396,7 @@ def on_world_tick(self, timestamp):
         self.simulation_time = timestamp.elapsed_seconds
 
     def tick(self, world, clock):
+        # type: (carla.World, pygame.time.Clock) -> None
         self._notifications.tick(world, clock)
         if not self._show_info:
             return
@@ -781,10 +782,12 @@ def __init__(self, parent_actor, hud):
         Attachment = carla.AttachmentType
 
         self._camera_transforms = [
-            (carla.Transform(carla.Location(x=-2.0*bound_x, y=+0.0*bound_y, z=2.0*bound_z), carla.Rotation(pitch=8.0)), Attachment.SpringArm),
+            (carla.Transform(carla.Location(x=-2.0*bound_x, y=+0.0*bound_y, z=2.0*bound_z), 
+                             carla.Rotation(pitch=8.0)), Attachment.SpringArm),
             (carla.Transform(carla.Location(x=+0.8*bound_x, y=+0.0*bound_y, z=1.3*bound_z)), Attachment.Rigid),
             (carla.Transform(carla.Location(x=+1.9*bound_x, y=+1.0*bound_y, z=1.2*bound_z)), Attachment.SpringArm),
-            (carla.Transform(carla.Location(x=-2.8*bound_x, y=+0.0*bound_y, z=4.6*bound_z), carla.Rotation(pitch=6.0)), Attachment.SpringArm),
+            (carla.Transform(carla.Location(x=-2.8*bound_x, y=+0.0*bound_y, z=4.6*bound_z), 
+                             carla.Rotation(pitch=6.0)), Attachment.SpringArm),
             (carla.Transform(carla.Location(x=-1.0, y=-1.0*bound_y, z=0.4*bound_z)), Attachment.Rigid)]
 
         self.transform_index = 1

metrics_manager.py

@@ -130,21 +130,32 @@ def main():
     """
 
     # pylint: disable=line-too-long
-    description = ("Scenario Runner's metrics module. Evaluate the execution of a specific scenario by developing your own metric.\n")
+    description = "Scenario Runner's metrics module. Evaluate the execution of a specific scenario by developing your own metric.\n"
+    # pylint: enable=line-too-long
 
     parser = argparse.ArgumentParser(description=description,
                                     formatter_class=RawTextHelpFormatter)
     parser.add_argument('--host', default='127.0.0.1',
                         help='IP of the host server (default: localhost)')
     parser.add_argument('--port', '-p', default=2000,
                         help='TCP port to listen to (default: 2000)')
-    parser.add_argument('--log', required=True,
-                        help='Path to the CARLA recorder .log file (relative to SCENARIO_RUNNER_ROOT).\nThis file is created by the record functionality at ScenarioRunner')
-    parser.add_argument('--metric', required=True,
-                        help='Path to the .py file defining the used metric.\nSome examples at srunner/metrics')
-    parser.add_argument('--criteria', default="",
-                        help='Path to the .json file with the criteria information.\nThis file is created by the record functionality at ScenarioRunner')
-    # pylint: enable=line-too-long
+    parser.add_argument(
+        "--log",
+        required=True,
+        help="Path to the CARLA recorder .log file (relative to SCENARIO_RUNNER_ROOT)."
+             "\nThis file is created by the record functionality at ScenarioRunner",
+    )
+    parser.add_argument(
+        "--metric",
+        required=True,
+        help="Path to the .py file defining the used metric.\nSome examples at srunner/metrics",
+    )
+    parser.add_argument(
+        "--criteria",
+        default="",
+        help="Path to the .json file with the criteria information."
+             "\nThis file is created by the record functionality at ScenarioRunner",
+    )
 
     args = parser.parse_args()
 

no_rendering_mode.py

@@ -490,22 +490,55 @@ def draw_broken_line(surface, color, closed, points, width):
 
         def get_lane_markings(lane_marking_type, lane_marking_color, waypoints, sign):
             margin = 0.20
-            if lane_marking_type == carla.LaneMarkingType.Broken or (lane_marking_type == carla.LaneMarkingType.Solid):
-                marking_1 = [world_to_pixel(lateral_shift(w.transform, sign * w.lane_width * 0.5)) for w in waypoints]
+            if lane_marking_type == carla.LaneMarkingType.Broken or (
+                lane_marking_type == carla.LaneMarkingType.Solid
+            ):
+                marking_1 = [
+                    world_to_pixel(
+                        lateral_shift(w.transform, sign * w.lane_width * 0.5)
+                    )
+                    for w in waypoints
+                ]
                 return [(lane_marking_type, lane_marking_color, marking_1)]
-            elif lane_marking_type == carla.LaneMarkingType.SolidBroken or lane_marking_type == carla.LaneMarkingType.BrokenSolid:
-                marking_1 = [world_to_pixel(lateral_shift(w.transform, sign * w.lane_width * 0.5)) for w in waypoints]
-                marking_2 = [world_to_pixel(lateral_shift(w.transform,
-                                                          sign * (w.lane_width * 0.5 + margin * 2))) for w in waypoints]
-                return [(carla.LaneMarkingType.Solid, lane_marking_color, marking_1),
-                        (carla.LaneMarkingType.Broken, lane_marking_color, marking_2)]
+            elif (
+                lane_marking_type == carla.LaneMarkingType.SolidBroken
+                or lane_marking_type == carla.LaneMarkingType.BrokenSolid
+            ):
+                marking_1 = [
+                    world_to_pixel(
+                        lateral_shift(w.transform, sign * w.lane_width * 0.5)
+                    )
+                    for w in waypoints
+                ]
+                marking_2 = [
+                    world_to_pixel(
+                        lateral_shift(
+                            w.transform, sign * (w.lane_width * 0.5 + margin * 2)
+                        )
+                    )
+                    for w in waypoints
+                ]
+                return [
+                    (carla.LaneMarkingType.Solid, lane_marking_color, marking_1),
+                    (carla.LaneMarkingType.Broken, lane_marking_color, marking_2),
+                ]
             elif lane_marking_type == carla.LaneMarkingType.BrokenBroken:
-                marking = [world_to_pixel(lateral_shift(w.transform,
-                                                        sign * (w.lane_width * 0.5 - margin))) for w in waypoints]
+                marking = [
+                    world_to_pixel(
+                        lateral_shift(w.transform, sign * (w.lane_width * 0.5 - margin))
+                    )
+                    for w in waypoints
+                ]
                 return [(carla.LaneMarkingType.Broken, lane_marking_color, marking)]
             elif lane_marking_type == carla.LaneMarkingType.SolidSolid:
-                marking = [world_to_pixel(lateral_shift(w.transform,
-                                                        sign * ((w.lane_width * 0.5) - margin))) for w in waypoints]
+                marking = [
+                    world_to_pixel(
+                        lateral_shift(
+                            w.transform, sign * ((w.lane_width * 0.5) - margin)
+                        )
+                    )
+                    for w in waypoints
+                ]
                 return [(carla.LaneMarkingType.Solid, lane_marking_color, marking)]
 
             return [(carla.LaneMarkingType.NONE, carla.LaneMarkingColor.Other, [])]

scenario_runner.py

@@ -106,7 +106,11 @@ def __init__(self, args):
         self.client.set_timeout(self.client_timeout)
         carla_version = get_carla_version()
         if carla_version < Version(MIN_CARLA_VERSION):
-            raise ImportError("CARLA version {} or newer required. CARLA version found: {}".format(MIN_CARLA_VERSION, carla_version))
+            raise ImportError(
+                "CARLA version {} or newer required. CARLA version found: {}".format(
+                    MIN_CARLA_VERSION, carla_version
+                )
+            )
 
         # Load agent if requested via command line args
         # If something goes wrong an exception will be thrown by importlib (ok here)
@@ -582,8 +586,11 @@ def main():
                         help='Frame rate (Hz) to use in \'sync\' mode (default: 20)')
 
     parser.add_argument(
-        '--scenario', help='Name of the scenario to be executed. Use the preposition \'group:\' to run all scenarios of one class, e.g. ControlLoss or FollowLeadingVehicle')
-    parser.add_argument('--openscenario', help='Provide an OpenSCENARIO definition')
+        "--scenario",
+        help="Name of the scenario to be executed. Use the preposition 'group:' to run all scenarios of one class"
+             " e.g. ControlLoss or FollowLeadingVehicle",
+    )
+    parser.add_argument("--openscenario", help="Provide an OpenSCENARIO definition")
     parser.add_argument('--openscenarioparams', help='Overwrited for OpenSCENARIO ParameterDeclaration')
     parser.add_argument('--openscenario2', help='Provide an openscenario2 definition')
     parser.add_argument('--route', help='Run a route as a scenario', type=str)
@@ -604,11 +611,20 @@ def main():
     parser.add_argument('--debug', action="store_true", help='Run with debug output')
     parser.add_argument('--reloadWorld', action="store_true",
                         help='Reload the CARLA world before starting a scenario (default=True)')
-    parser.add_argument('--record', type=str, default='',
-                        help='Path were the files will be saved, relative to SCENARIO_RUNNER_ROOT.\nActivates the CARLA recording feature and saves to file all the criteria information.')
-    parser.add_argument('--randomize', action="store_true", help='Scenario parameters are randomized')
-    parser.add_argument('--repetitions', default=1, type=int, help='Number of scenario executions')
-    parser.add_argument('--waitForEgo', action="store_true", help='Connect the scenario to an existing ego vehicle')
+    parser.add_argument(
+        "--record",
+        type=str,
+        default="",
+        help="Path were the files will be saved, relative to SCENARIO_RUNNER_ROOT."
+             "\nActivates the CARLA recording feature and saves to file all the criteria information.",
+    )
+    parser.add_argument("--randomize", action="store_true", help="Scenario parameters are randomized")
+    parser.add_argument("--repetitions", default=1, type=int, help="Number of scenario executions")
+    parser.add_argument(
+        "--waitForEgo",
+        action="store_true",
+        help="Connect the scenario to an existing ego vehicle",
+    )
 
     arguments = parser.parse_args()
     # pylint: enable=line-too-long

srunner/autoagents/ros_agent.py

@@ -30,9 +30,13 @@
 from sensor_msgs.point_cloud2 import create_cloud_xyz32
 from std_msgs.msg import Header, String
 import tf
-# pylint: disable=line-too-long
-from carla_msgs.msg import CarlaEgoVehicleStatus, CarlaEgoVehicleInfo, CarlaEgoVehicleInfoWheel, CarlaEgoVehicleControl, CarlaWorldInfo
-# pylint: enable=line-too-long
+from carla_msgs.msg import (
+    CarlaEgoVehicleStatus,
+    CarlaEgoVehicleInfo,
+    CarlaEgoVehicleInfoWheel,
+    CarlaEgoVehicleControl,
+    CarlaWorldInfo,
+)
 
 from srunner.autoagents.autonomous_agent import AutonomousAgent
 
@@ -122,42 +126,73 @@ def setup(self, path_to_conf_file):
         self.cv_bridge = CvBridge()
 
         # setup ros publishers for sensors
-        # pylint: disable=line-too-long
         for sensor in self.sensors():
-            self.id_to_sensor_type_map[sensor['id']] = sensor['type']
-            if sensor['type'] == 'sensor.camera.rgb':
-                self.publisher_map[sensor['id']] = rospy.Publisher(
-                    '/carla/ego_vehicle/camera/rgb/' + sensor['id'] + "/image_color", Image, queue_size=1, latch=True)
-                self.id_to_camera_info_map[sensor['id']] = self.build_camera_info(sensor)
-                self.publisher_map[sensor['id'] + '_info'] = rospy.Publisher(
-                    '/carla/ego_vehicle/camera/rgb/' + sensor['id'] + "/camera_info", CameraInfo, queue_size=1, latch=True)
-            elif sensor['type'] == 'sensor.lidar.ray_cast':
-                self.publisher_map[sensor['id']] = rospy.Publisher(
-                    '/carla/ego_vehicle/lidar/' + sensor['id'] + "/point_cloud", PointCloud2, queue_size=1, latch=True)
-            elif sensor['type'] == 'sensor.other.gnss':
-                self.publisher_map[sensor['id']] = rospy.Publisher(
-                    '/carla/ego_vehicle/gnss/' + sensor['id'] + "/fix", NavSatFix, queue_size=1, latch=True)
-            elif sensor['type'] == 'sensor.can_bus':
+            self.id_to_sensor_type_map[sensor["id"]] = sensor["type"]
+            if sensor["type"] == "sensor.camera.rgb":
+                self.publisher_map[sensor["id"]] = rospy.Publisher(
+                    "/carla/ego_vehicle/camera/rgb/" + sensor["id"] + "/image_color",
+                    Image,
+                    queue_size=1,
+                    latch=True,
+                )
+                self.id_to_camera_info_map[sensor["id"]] = self.build_camera_info(
+                    sensor
+                )
+                self.publisher_map[sensor["id"] + "_info"] = rospy.Publisher(
+                    "/carla/ego_vehicle/camera/rgb/" + sensor["id"] + "/camera_info",
+                    CameraInfo,
+                    queue_size=1,
+                    latch=True,
+                )
+            elif sensor["type"] == "sensor.lidar.ray_cast":
+                self.publisher_map[sensor["id"]] = rospy.Publisher(
+                    "/carla/ego_vehicle/lidar/" + sensor["id"] + "/point_cloud",
+                    PointCloud2,
+                    queue_size=1,
+                    latch=True,
+                )
+            elif sensor["type"] == "sensor.other.gnss":
+                self.publisher_map[sensor["id"]] = rospy.Publisher(
+                    "/carla/ego_vehicle/gnss/" + sensor["id"] + "/fix",
+                    NavSatFix,
+                    queue_size=1,
+                    latch=True,
+                )
+            elif sensor["type"] == "sensor.can_bus":
                 if not self.vehicle_info_publisher:
                     self.vehicle_info_publisher = rospy.Publisher(
-                        '/carla/ego_vehicle/vehicle_info', CarlaEgoVehicleInfo, queue_size=1, latch=True)
+                        "/carla/ego_vehicle/vehicle_info",
+                        CarlaEgoVehicleInfo,
+                        queue_size=1,
+                        latch=True,
+                    )
                 if not self.vehicle_status_publisher:
                     self.vehicle_status_publisher = rospy.Publisher(
-                        '/carla/ego_vehicle/vehicle_status', CarlaEgoVehicleStatus, queue_size=1, latch=True)
-            elif sensor['type'] == 'sensor.hd_map':
+                        "/carla/ego_vehicle/vehicle_status",
+                        CarlaEgoVehicleStatus,
+                        queue_size=1,
+                        latch=True,
+                    )
+            elif sensor["type"] == "sensor.hd_map":
                 if not self.odometry_publisher:
                     self.odometry_publisher = rospy.Publisher(
-                        '/carla/ego_vehicle/odometry', Odometry, queue_size=1, latch=True)
+                        "/carla/ego_vehicle/odometry",
+                        Odometry,
+                        queue_size=1,
+                        latch=True,
+                    )
                 if not self.world_info_publisher:
                     self.world_info_publisher = rospy.Publisher(
-                        '/carla/world_info', CarlaWorldInfo, queue_size=1, latch=True)
+                        "/carla/world_info", CarlaWorldInfo, queue_size=1, latch=True
+                    )
                 if not self.map_file_publisher:
-                    self.map_file_publisher = rospy.Publisher('/carla/map_file', String, queue_size=1, latch=True)
+                    self.map_file_publisher = rospy.Publisher(
+                        "/carla/map_file", String, queue_size=1, latch=True
+                    )
                 if not self.tf_broadcaster:
                     self.tf_broadcaster = tf.TransformBroadcaster()
             else:
-                raise TypeError("Invalid sensor type: {}".format(sensor['type']))
-        # pylint: enable=line-too-long
+                raise TypeError("Invalid sensor type: {}".format(sensor["type"]))
 
     def destroy(self):
         """
@@ -293,9 +328,12 @@ def publish_gnss(self, sensor_id, data):
         msg.longitude = data[1]
         msg.altitude = data[2]
         msg.status.status = NavSatStatus.STATUS_SBAS_FIX
-        # pylint: disable=line-too-long
-        msg.status.service = NavSatStatus.SERVICE_GPS | NavSatStatus.SERVICE_GLONASS | NavSatStatus.SERVICE_COMPASS | NavSatStatus.SERVICE_GALILEO
-        # pylint: enable=line-too-long
+        msg.status.service = (
+            NavSatStatus.SERVICE_GPS
+            | NavSatStatus.SERVICE_GLONASS
+            | NavSatStatus.SERVICE_COMPASS
+            | NavSatStatus.SERVICE_GALILEO
+        )
         self.publisher_map[sensor_id].publish(msg)
 
     def publish_camera(self, sensor_id, data):

srunner/metrics/tools/metrics_log.py

@@ -407,10 +407,7 @@ def is_vehicle_light_active(self, light, vehicle_id, frame):
         """
         lights = self.get_vehicle_lights(vehicle_id, frame)
 
-        if light in lights:
-            return True
-
-        return False
+        return light in lights
 
     # Scene lights
     def get_scene_light_state(self, light_id, frame):

srunner/metrics/tools/metrics_parser.py

@@ -507,7 +507,7 @@ def parse_recorder_info(self):
 
                             elif name == "use_gear_auto_box":
                                 name = "use_gear_autobox"
-                                value = True if elements[1] == "true" else False
+                                value = elements[1] == "true"
                                 setattr(physics_control, name, value)
 
                             elif "forward_gears" in name or "wheels" in name:

srunner/metrics/tools/osc2_log.py

@@ -427,10 +427,7 @@ def is_vehicle_light_active(self, light, vehicle_id, frame):
         """
         lights = self.get_vehicle_lights(vehicle_id, frame)
 
-        if light in lights:
-            return True
-
-        return False
+        return light in lights
 
     # Scene lights
     def get_scene_light_state(self, light_id, frame):

srunner/metrics/tools/osc2_trace_parser.py

@@ -512,7 +512,7 @@ def parse_recorder_info(self):
 
                             elif name == "use_gear_auto_box":
                                 name = "use_gear_autobox"
-                                value = True if elements[1] == "true" else False
+                                value = elements[1] == "true"
                                 setattr(physics_control, name, value)
 
                             elif "forward_gears" in name or "wheels" in name:

srunner/osc2/ast_manager/ast_builder.py

@@ -1648,12 +1648,6 @@ def enterBehaviorInvocation(
         # Find the scope of type_name
         scope = self.__current_scope.resolve(name)
 
-        if scope:
-            pass
-        else:
-            msg = "behavior name: " + name + " is not defined!"
-            pass
-
         node = ast_node.BehaviorInvocation(actor, behavior_name)
         node.set_loc(ctx.start.line, ctx.start.column)
         node.set_scope(scope)
@@ -2083,7 +2077,7 @@ def exitConjunction(self, ctx: OpenSCENARIO2Parser.ConjunctionContext):
     # Enter a parse tree produced by OpenSCENARIO2Parser#inversion.
     def enterInversion(self, ctx: OpenSCENARIO2Parser.InversionContext):
         self.__node_stack.append(self.__cur_node)
-        if ctx.relation() == None:
+        if ctx.relation() is None:
             operator = "not"
             node = ast_node.LogicalExpression(operator)
             node.set_loc(ctx.start.line, ctx.start.column)

srunner/osc2/ast_manager/ast_listener.py

@@ -211,12 +211,6 @@ def enter_positional_argument(self, node: ast_node.PositionalArgument):
     def exit_positional_argument(self, node: ast_node.PositionalArgument):
         pass
 
-    def enter_variable_declaration(self, node: ast_node.VariableDeclaration):
-        pass
-
-    def exit_variable_declaration(self, node: ast_node.VariableDeclaration):
-        pass
-
     def enter_keep_constraint_declaration(
         self, node: ast_node.KeepConstraintDeclaration
     ):