Line too long (112 > 79 characters) (E501)

Line too long (80 > 79 characters) (E501)

Line too long (83 > 79 characters) (E501)

Line too long (119 > 79 characters) (E501)

Line too long (81 > 79 characters) (E501)

Line too long (90 > 79 characters) (E501)

Line too long (120 > 79 characters) (E501)

Line too long (93 > 79 characters) (E501)

Line too long (80 > 79 characters) (E501)

Line too long (82 > 79 characters) (E501)

Line too long (101 > 79 characters) (E501)

Line too long (84 > 79 characters) (E501)

Line too long (127 > 79 characters) (E501)

Line too long (130 > 79 characters) (E501)

Line too long (92 > 79 characters) (E501)

Line too long (95 > 79 characters) (E501)

Line too long (93 > 79 characters) (E501)

Line too long (81 > 79 characters) (E501)

Line too long (137 > 79 characters) (E501)

from ament_copyright.main import main import pytest # Remove the `skip` decorator once the source file(s) have a copyright header [email protected](reason='No copyright header has been placed in the generated source file.') [email protected]( + reason="No copyright header has been placed in the generated source file." +) @pytest.mark.copyright @pytest.mark.linter def test_copyright(): - rc = main(argv=['.', 'test']) - assert rc == 0, 'Found errors' + rc = main(argv=[".", "test"]) + assert rc == 0, "Found errors"

Line too long (146 > 79 characters) (E501)

from setuptools import find_packages, setup import os from glob import glob -package_name = 'apriltag' + +package_name = "apriltag" setup( name=package_name, - version='0.0.0', - packages=find_packages(exclude=['test']), + version="0.0.0", + packages=find_packages(exclude=["test"]), data_files=[ - ('share/ament_index/resource_index/packages', - ['resource/' + package_name]), - ('share/' + package_name, ['package.xml']), + ("share/ament_index/resource_index/packages", ["resource/" + package_name]), + ("share/" + package_name, ["package.xml"]), ( os.path.join("share", package_name), glob("launch/*launch.[pxy][yma]*", recursive=True), ), ], - install_requires=['setuptools'], + install_requires=["setuptools"], zip_safe=True, - maintainer='umn-robotics', - maintainer_email='[email protected]', - description='Package for processing apriltag detections received from ISAAC ROS Apriltag.', - license='MIT License', - tests_require=['pytest'], + maintainer="umn-robotics", + maintainer_email="[email protected]", + description="Package for processing apriltag detections received from ISAAC ROS Apriltag.", + license="MIT License", + tests_require=["pytest"], entry_points={ - 'console_scripts': ["apriltag_node = apriltag.apriltag_node:main"], + "console_scripts": ["apriltag_node = apriltag.apriltag_node:main"], }, )

Line too long (82 > 79 characters) (E501)

@pytest.mark.flake8 @pytest.mark.linter def test_flake8(): rc, errors = main_with_errors(argv=[]) - assert rc == 0, \ - 'Found %d code style errors / warnings:\n' % len(errors) + \ - '\n'.join(errors) + assert rc == 0, "Found %d code style errors / warnings:\n" % len( + errors + ) + "\n".join(errors)

Line too long (140 > 79 characters) (E501)

isaac_ros_apriltag = ComposableNode( package="isaac_ros_apriltag", plugin="nvidia::isaac_ros::apriltag::AprilTagNode", name="isaac_ros_apriltag", namespace="", - remappings=[("image", "zed2i/zed_node/left/image_rect_color_rgb"), ("camera_info", "zed2i/zed_node/left/camera_info")], + remappings=[ + ("image", "zed2i/zed_node/left/image_rect_color_rgb"), + ("camera_info", "zed2i/zed_node/left/camera_info"), + ], ) image_format_converter_node_left = ComposableNode( package="isaac_ros_image_proc", plugin="nvidia::isaac_ros::image_proc::ImageFormatConverterNode",

Line too long (94 > 79 characters) (E501)

parameters=[ { "encoding_desired": "rgb8", } ], - remappings=[("image_raw", "zed2i/zed_node/left/image_rect_color"), ("image", "zed2i/zed_node/left/image_rect_color_rgb")], + remappings=[ + ("image_raw", "zed2i/zed_node/left/image_rect_color"), + ("image", "zed2i/zed_node/left/image_rect_color_rgb"), + ], ) apriltag_container = ComposableNodeContainer( package="rclcpp_components", name="apriltag_container", namespace="", executable="component_container_mt", - composable_node_descriptions=[isaac_ros_apriltag, image_format_converter_node_left], + composable_node_descriptions=[ + isaac_ros_apriltag, + image_format_converter_node_left, + ], output="screen", ) # Add nodes and containers to LaunchDescription return launch.LaunchDescription([tag_reader, apriltag_container])

Line too long (92 > 79 characters) (E501)

@pytest.mark.linter @pytest.mark.pep257 def test_pep257(): - rc = main(argv=['.', 'test']) - assert rc == 0, 'Found code style errors / warnings' + rc = main(argv=[".", "test"]) + assert rc == 0, "Found code style errors / warnings"

Line too long (94 > 79 characters) (E501)

@pytest.mark.copyright @pytest.mark.linter def test_copyright(): - rc = main(argv=['.', 'test']) - assert rc == 0, 'Found errors' + rc = main(argv=[".", "test"]) + assert rc == 0, "Found errors"

Line too long (90 > 79 characters) (E501)

@pytest.mark.flake8 @pytest.mark.linter def test_flake8(): rc, errors = main_with_errors(argv=[]) - assert rc == 0, \ - 'Found %d code style errors / warnings:\n' % len(errors) + \ - '\n'.join(errors) + assert rc == 0, "Found %d code style errors / warnings:\n" % len( + errors + ) + "\n".join(errors)

Line too long (92 > 79 characters) (E501)

@pytest.mark.linter @pytest.mark.pep257 def test_pep257(): - rc = main(argv=['.', 'test']) - assert rc == 0, 'Found code style errors / warnings' + rc = main(argv=[".", "test"]) + assert rc == 0, "Found code style errors / warnings"

Line too long (90 > 79 characters) (E501)

def generate_launch_description(): # Setup to launch the simulator and Gazebo world gz_sim = IncludeLaunchDescription( PythonLaunchDescriptionSource( - [PathJoinSubstitution([FindPackageShare("ros_gz_sim"), "launch", "gz_sim.launch.py"])] + [ + PathJoinSubstitution( + [FindPackageShare("ros_gz_sim"), "launch", "gz_sim.launch.py"] + ) + ] ), launch_arguments={ - "gz_args": PathJoinSubstitution([FindPackageShare("gazebo_files"), "worlds", "NASA_field.sdf"]) + "gz_args": PathJoinSubstitution( + [FindPackageShare("gazebo_files"), "worlds", "NASA_field.sdf"] + ) }.items(), ) # Takes the description and joint angles as inputs and publishes the 3D poses of the robot links robot_state_publisher = IncludeLaunchDescription( PythonLaunchDescriptionSource( - [PathJoinSubstitution([FindPackageShare("robot_description"), "launch", "robot_description.launch.py"])] + [ + PathJoinSubstitution( + [ + FindPackageShare("robot_description"), + "launch", + "robot_description.launch.py", + ] + ) + ] ), launch_arguments={ "setup_for_gazebo": "True", }.items(), ) # Launch Arguments - run_rviz_arg = DeclareLaunchArgument("run_rviz", default_value="True", description="Whether to start RVIZ") + run_rviz_arg = DeclareLaunchArgument( + "run_rviz", default_value="True", description="Whether to start RVIZ" + ) # Visualize in RViz rviz = Node( package="rviz2", executable="rviz2", - arguments=["-d", PathJoinSubstitution([FindPackageShare("ros_gz_launch"), "config", "camera.rviz"])], + arguments=[ + "-d", + PathJoinSubstitution( + [FindPackageShare("ros_gz_launch"), "config", "camera.rviz"] + ), + ], condition=IfCondition(LaunchConfiguration("run_rviz")), ) # Bridge ROS topics and Gazebo messages for establishing communication bridge = Node(

Line too long (96 > 79 characters) (E501)

drivetrain = Node( package="drivetrain", executable="drivetrain_node", name="drivetrain_node", - parameters=["config/drivetrain_config.yaml", "config/motor_control.yaml", {"GAZEBO_SIMULATION": True}], + parameters=[ + "config/drivetrain_config.yaml", + "config/motor_control.yaml", + {"GAZEBO_SIMULATION": True}, + ], output="screen", emulate_tty=True, ) - return LaunchDescription([run_rviz_arg, gz_sim, bridge, drivetrain, robot_state_publisher, rviz]) + return LaunchDescription( + [run_rviz_arg, gz_sim, bridge, drivetrain, robot_state_publisher, rviz] + )

Line too long (85 > 79 characters) (E501)

def generate_launch_description(): # Setup to launch the simulator and Gazebo world gz_sim = IncludeLaunchDescription( PythonLaunchDescriptionSource( - [PathJoinSubstitution([FindPackageShare("ros_gz_sim"), "launch", "gz_sim.launch.py"])] + [ + PathJoinSubstitution( + [FindPackageShare("ros_gz_sim"), "launch", "gz_sim.launch.py"] + ) + ] ), launch_arguments={ - "gz_args": PathJoinSubstitution([FindPackageShare("gazebo_files"), "worlds", "UCF_field.sdf"]) + "gz_args": PathJoinSubstitution( + [FindPackageShare("gazebo_files"), "worlds", "UCF_field.sdf"] + ) }.items(), ) # Takes the description and joint angles as inputs and publishes the 3D poses of the robot links robot_state_publisher = IncludeLaunchDescription( PythonLaunchDescriptionSource( - [PathJoinSubstitution([FindPackageShare("robot_description"), "launch", "robot_description.launch.py"])] + [ + PathJoinSubstitution( + [ + FindPackageShare("robot_description"), + "launch", + "robot_description.launch.py", + ] + ) + ] ), launch_arguments={ "setup_for_gazebo": "True", }.items(), ) # Launch Arguments - run_rviz_arg = DeclareLaunchArgument("run_rviz", default_value="True", description="Whether to start RVIZ") + run_rviz_arg = DeclareLaunchArgument( + "run_rviz", default_value="True", description="Whether to start RVIZ" + ) # Visualize in RViz rviz = Node( package="rviz2", executable="rviz2", - arguments=["-d", PathJoinSubstitution([FindPackageShare("ros_gz_launch"), "config", "camera.rviz"])], + arguments=[ + "-d", + PathJoinSubstitution( + [FindPackageShare("ros_gz_launch"), "config", "camera.rviz"] + ), + ], condition=IfCondition(LaunchConfiguration("run_rviz")), ) # Bridge ROS topics and Gazebo messages for establishing communication bridge = Node(

Line too long (89 > 79 characters) (E501)

drivetrain = Node( package="drivetrain", executable="drivetrain_node", name="drivetrain_node", - parameters=["config/drivetrain_config.yaml", "config/motor_control.yaml", {"GAZEBO_SIMULATION": True}], + parameters=[ + "config/drivetrain_config.yaml", + "config/motor_control.yaml", + {"GAZEBO_SIMULATION": True}, + ], output="screen", emulate_tty=True, ) - return LaunchDescription([run_rviz_arg, gz_sim, bridge, drivetrain, robot_state_publisher, rviz]) + return LaunchDescription( + [run_rviz_arg, gz_sim, bridge, drivetrain, robot_state_publisher, rviz] + )

Line too long (89 > 79 characters) (E501)

class ApriltagNode(Node): def __init__(self): super().__init__("apriltag_node") current_dir = os.getcwd() - self.declare_parameter("autonomous_field_type", "bottom") # The type of field ("top", "bottom", "nasa") + self.declare_parameter( + "autonomous_field_type", "bottom" + ) # The type of field ("top", "bottom", "nasa") field_type = self.get_parameter("autonomous_field_type").value paths = { "top": "src/apriltag/apriltag/apriltag_location_ucf_top.urdf.xarco", "bottom": "src/apriltag/apriltag/apriltag_location_ucf_bot.urdf.xarco", "nasa": "src/apriltag/apriltag/apriltag_location_nasa.urdf.xarco",

Line too long (89 > 79 characters) (E501)

self.map_transform.transform.rotation.x = 0.0 self.map_transform.transform.rotation.y = 0.0 self.map_transform.transform.rotation.z = 0.0 self.map_transform.transform.rotation.w = 1.0 - self.transforms = self.create_subscription(AprilTagDetectionArray, "/tag_detections", self.tagDetectionSub, 10) + self.transforms = self.create_subscription( + AprilTagDetectionArray, "/tag_detections", self.tagDetectionSub, 10 + ) self.tf_broadcaster = TransformBroadcaster(self) self.tf_buffer = Buffer() self.tf_listener = TransformListener(self.tf_buffer, self) self.create_service(ResetOdom, "resetOdom", self.reset_callback)

Line too long (89 > 79 characters) (E501)

Line too long (91 > 79 characters) (E501)

for tag in tags: id = tag.id tree = ET.parse(self.file_path) root = tree.getroot() - link = root[id - 1] # assumes tag 1 = home 1, tag 2 = home 2, 3 = berm 1 etc. + link = root[ + id - 1 + ] # assumes tag 1 = home 1, tag 2 = home 2, 3 = berm 1 etc. xyz_elements = link.findall(".//origin[@xyz]") xyz_values = [element.attrib["xyz"] for element in xyz_elements] xyz = xyz_values[0].split(" ")

Line too long (91 > 79 characters) (E501)

rpy_values = [element.attrib["rpy"] for element in rpy_elements] rpy = rpy_values[0].split(" ") # Lookup the odom to detected tag tf from the tf buffer try: - odom_to_tag_transform = self.tf_buffer.lookup_transform("odom", f"{tag.family}:{id}", rclpy.time.Time()) + odom_to_tag_transform = self.tf_buffer.lookup_transform( + "odom", f"{tag.family}:{id}", rclpy.time.Time() + ) except TransformException as ex: - self.get_logger().warn(f"Could not transform odom to the detected tag: {ex}") + self.get_logger().warn( + f"Could not transform odom to the detected tag: {ex}" + ) return odom_to_tag_transform.child_frame_id = "odom" odom_to_tag_transform.header.frame_id = "map" odom_to_tag_transform.header.stamp = self.get_clock().now().to_msg()

Line too long (91 > 79 characters) (E501)

odom_to_tag_transform.transform.rotation.y, odom_to_tag_transform.transform.rotation.z, odom_to_tag_transform.transform.rotation.w, ] ).as_quat() - rotated_quaternion = current_quaternion * rotation_quaternion # Multiply the quaternions + rotated_quaternion = ( + current_quaternion * rotation_quaternion + ) # Multiply the quaternions # Update the transform with the rotated quaternion odom_to_tag_transform.transform.rotation.x = rotated_quaternion[0] odom_to_tag_transform.transform.rotation.y = rotated_quaternion[1] odom_to_tag_transform.transform.rotation.z = rotated_quaternion[2]

Line too long (91 > 79 characters) (E501)

from rqt_gui.main import Main def main(): main = Main() - sys.exit(main.main(sys.argv, standalone='gstreamer.client.ClientPlugin')) + sys.exit(main.main(sys.argv, standalone="gstreamer.client.ClientPlugin")) -if __name__ == '__main__': - main() +if __name__ == "__main__": + main()

Line too long (88 > 79 characters) (E501)

class ClientPlugin(Plugin): def __init__(self, context): super(ClientPlugin, self).__init__(context) self.setObjectName("ClientPlugin") - assert hasattr(context, 'node'), 'Context does not have a node.' + assert hasattr(context, "node"), "Context does not have a node." self._widget = ClientWidget(context.node) if context.serial_number() > 1: self._widget.setWindowTitle( - self._widget.windowTitle() + (' (%d)' % context.serial_number())) + self._widget.windowTitle() + (" (%d)" % context.serial_number()) + ) context.add_widget(self._widget) - \ No newline at end of file

Line too long (92 > 79 characters) (E501)

from threading import Thread, Event import gi + gi.require_version("Gst", "1.0") from gi.repository import Gst # noqa: E402 + class GstreamerClient: def __init__(self): # Initialize GStreamer Gst.init(None)

Line too long (103 > 79 characters) (E501)

Line too long (86 > 79 characters) (E501)

self.pipeline.set_state(Gst.State.PLAYING) self.stop_event = Event() change_source_thread = Thread() change_source_thread.start() - + def stop(self): self.stop_event.set() self.pipeline.set_state(Gst.State.NULL) + if __name__ == "__main__": client = GstreamerClient() client.run() stop_event = Event() change_source_thread = Thread() change_source_thread.start() while True: try: - message:Gst.Message = client.pipeline.get_bus().timed_pop(Gst.SECOND) + message: Gst.Message = client.pipeline.get_bus().timed_pop(Gst.SECOND) if message is None: pass elif message.type == Gst.MessageType.EOS: break elif message.type == Gst.MessageType.ERROR: gi.error("Error", message.parse_error()) break except KeyboardInterrupt: break - client.stop() + client.stop()

Line too long (97 > 79 characters) (E501)

from rovr_interfaces.srv import SetClientIp, SetActiveCamera, SetEncoding gi.require_version("Gst", "1.0") from gi.repository import Gst # noqa: E402 + class GstreamerServer: - def __init__(self, ip_srv: SetClientIp, camera_srv: SetActiveCamera, encod_srv: SetEncoding): + def __init__( + self, ip_srv: SetClientIp, camera_srv: SetActiveCamera, encod_srv: SetEncoding + ): Gst.init(None) self.pipeline = Gst.Pipeline() camera_src = Gst.ElementFactory.make(camera_srv.srctype, "camera_src") if camera_srv.srctype == "videotestsrc": print(f"Using videotestsrc{camera_srv.device}") camera_src.set_property("pattern", int(camera_srv.device)) else: camera_src.set_property("device", camera_srv.device) self.pipeline.add(camera_src) - + nonNVvideoconvert = Gst.ElementFactory.make("videoconvert", "videoconvert") self.pipeline.add(nonNVvideoconvert) camera_src.link(nonNVvideoconvert) caps = Gst.Caps.from_string(

Line too long (95 > 79 characters) (E501)

Line too long (99 > 79 characters) (E501)

) src_caps = Gst.ElementFactory.make("capsfilter", "src_caps") src_caps.set_property("caps", caps) self.pipeline.add(src_caps) nonNVvideoconvert.link(src_caps) - if platform.machine() == "aarch64": videoconvert = Gst.ElementFactory.make("nvvidconv", "nvvidconv") elif platform.machine() == "x86_64" or platform.machine() == "amd64": videoconvert = Gst.ElementFactory.make("nvvideoconvert", "nvvideoconvert")

Line too long (97 > 79 characters) (E501)

self.pipeline.set_state(Gst.State.PLAYING) def stop(self): self.pipeline.set_state(Gst.State.NULL) + if __name__ == "__main__": ip = SetClientIp.Request() ip.client_ip = "127.0.0.1" camera = SetActiveCamera.Request() camera.srctype = "videotestsrc"

Line too long (95 > 79 characters) (E501)

camera.height = 480 camera.framerate = 30 camera.format = "NV12" encoding = SetEncoding.Request() encoding.encoding = "h265" - server = GstreamerServer(ip,camera,encoding) + server = GstreamerServer(ip, camera, encoding) server.run() stop_event = Event() change_source_thread = Thread() change_source_thread.start() while True: try: - message:Gst.Message = server.pipeline.get_bus().timed_pop(Gst.SECOND) + message: Gst.Message = server.pipeline.get_bus().timed_pop(Gst.SECOND) if message is None: pass elif message.type == Gst.MessageType.EOS: break elif message.type == Gst.MessageType.ERROR:

Line too long (93 > 79 characters) (E501)

from rclpy.node import Node from rovr_interfaces.srv import SetClientIp, SetActiveCamera, SetEncoding from .server_gstreamer import GstreamerServer import time + class ServerNode(Node): g_server: GstreamerServer = None ip_srv: SetClientIp = None camera_srv: SetActiveCamera = None encod_srv: SetEncoding = None + def __init__(self): super().__init__("server") self.serv_set_client_ip = self.create_service( SetClientIp, "/set_client_ip", self.set_client_ip_callback )

Line too long (97 > 79 characters) (E501)

def set_active_camera(self, request: SetActiveCamera, response): print("recieved camera request") self.camera_srv = request response.success = self.restart_server() return response - + def set_encoding(self, request: SetEncoding, response): print("recieved encoding request") self.encod_srv = request response.success = self.restart_server() return response

Line too long (95 > 79 characters) (E501)

self.g_server = GstreamerServer(self.ip_srv, self.camera_srv, self.encod_srv) self.g_server.run() print("Server restarted") return 0 + def main(args=None): rclpy.init(args=args) server = ServerNode() rclpy.spin(server) # Destroy the node explicitly

# Config config_name = LaunchConfiguration("config_name", default="zed_example.rviz") config_path = PathJoinSubstitution(["config", "rviz", config_name]) global_frame = LaunchConfiguration("global_frame", default="odom") - run_rviz_arg = DeclareLaunchArgument("run_rviz_client", default_value="True", description="Whether to start RVIZ") + run_rviz_arg = DeclareLaunchArgument( + "run_rviz_client", default_value="True", description="Whether to start RVIZ" + ) joystick_node = Node( package="joy", executable="joy_node", parameters=["config/joy_node.yaml"], ) start_gStreamer_client = ExecuteProcess( - cmd=["rqt", "--force-discover", "--standalone", "CameraClient"], shell=True, output="screen" + cmd=["rqt", "--force-discover", "--standalone", "CameraClient"], + shell=True, + output="screen", ) # NOTE: RVIZ uses a lot of bandwidth, so it should be turned off during competition matches rviz_node = Node( package="rviz2",

arguments=["-d", config_path, "-f", global_frame], condition=IfCondition(LaunchConfiguration("run_rviz_client")), ) ld.add_action(run_rviz_arg) - ld.add_action(joystick_node) # TODO: The joystick node currently does not work inside the container! + ld.add_action( + joystick_node + ) # TODO: The joystick node currently does not work inside the container! ld.add_action(start_gStreamer_client) ld.add_action(rviz_node) return ld

from ament_copyright.main import main import pytest # Remove the `skip` decorator once the source file(s) have a copyright header [email protected](reason='No copyright header has been placed in the generated source file.') [email protected]( + reason="No copyright header has been placed in the generated source file." +) @pytest.mark.copyright @pytest.mark.linter def test_copyright(): - rc = main(argv=['.', 'test']) - assert rc == 0, 'Found errors' + rc = main(argv=[".", "test"]) + assert rc == 0, "Found errors"

import socket import fcntl import struct from rqt_py_common.extended_combo_box import ExtendedComboBox + class ClientWidget(QWidget): - timeout = 2e9 # 2 seconds with nano seconds as unit + timeout = 2e9 # 2 seconds with nano seconds as unit encodings = ["av1", "h265"] + def __init__(self, node: Node): super(ClientWidget, self).__init__() self.setObjectName("ClientWidget") self.node = node self.display_window = GstreamerClient()

# Call the buttons to set ip and encoding by default # self.on_ip_push_button_clicked() self.on_encoding_push_button_clicked() def add_network_interfaces(self, comboBox: QComboBox): - for _,interface in socket.if_nameindex(): + for _, interface in socket.if_nameindex(): if interface != "lo": comboBox.addItem(interface) comboBox.setCurrentIndex(0) def get_encodings(self, comboBox: QComboBox): for encoding in self.encodings: comboBox.addItem(encoding) comboBox.setCurrentIndex(0) - + def get_ip_address(self): s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) interface = str(self.network_dropdown.currentText()) return socket.inet_ntoa( fcntl.ioctl( s.fileno(), 0x8915, - struct.pack( - "256s", interface[:15].encode("utf-8") - ), # SIOCGIFADDR + struct.pack("256s", interface[:15].encode("utf-8")), # SIOCGIFADDR )[20:24] - ) - - def wait_cli(self,cli:Client,req): + ) + + def wait_cli(self, cli: Client, req): future = cli.call_async(req) start_time = self.node.get_clock().now().nanoseconds # Block while waiting for server to respond - while rclpy.ok() and not future.done() and self.node.get_clock().now().nanoseconds - start_time < self.timeout: + while ( + rclpy.ok() + and not future.done() + and self.node.get_clock().now().nanoseconds - start_time < self.timeout + ): pass if not future.done(): print("Service Call Failed") return - + print("Service Call Returned") result = future.result().success if result == -1: print("No client IP Set") elif result == -2: print("No encoding set") elif result == -3: print("No camera selected") - - # self.restart_window() + + # self.restart_window() self.node.destroy_client(cli) - + @slot() def on_camera1_push_button_clicked(self): print("Requesting Camera 1") req = SetActiveCamera.Request() req.srctype = "v4l2src"

req.width = 640 req.height = 480 req.framerate = 30 req.format = "NV12" cli = self.node.create_client(SetActiveCamera, "/set_active_camera") - self.wait_cli(cli,req) + self.wait_cli(cli, req) @slot() def on_camera2_push_button_clicked(self): print("Requesting Camera 2") req = SetActiveCamera.Request()

req.width = 640 req.height = 480 req.framerate = 30 req.format = "NV12" cli = self.node.create_client(SetActiveCamera, "/set_active_camera") - self.wait_cli(cli,req) + self.wait_cli(cli, req) @slot() def on_camera3_push_button_clicked(self): print("Requesting Camera 3") req = SetActiveCamera.Request()

req.width = 640 req.height = 480 req.framerate = 30 req.format = "NV12" cli = self.node.create_client(SetActiveCamera, "/set_active_camera") - self.wait_cli(cli,req) + self.wait_cli(cli, req) @slot() def on_camera4_push_button_clicked(self): print("Requesting Camera 4") req = SetActiveCamera.Request()

req.width = 640 req.height = 480 req.framerate = 30 req.format = "NV12" cli = self.node.create_client(SetActiveCamera, "/set_active_camera") - self.wait_cli(cli,req) - + self.wait_cli(cli, req) + @slot() def on_camera5_push_button_clicked(self): print("Requesting Camera 5") req = SetActiveCamera.Request() req.srctype = "v4l2src"

req.width = 640 req.height = 480 req.framerate = 30 req.format = "NV12" cli = self.node.create_client(SetActiveCamera, "/set_active_camera") - self.wait_cli(cli,req) + self.wait_cli(cli, req) def restart_window(self): self.display_window.stop() self.display_window = GstreamerClient() self.display_window.run()

req.client_ip = self.get_ip_address() except OSError as e: print(e) return cli = self.node.create_client(SetClientIp, "/set_client_ip") - self.wait_cli(cli,req) - + self.wait_cli(cli, req) + @slot() def on_encoding_push_button_clicked(self): req = SetEncoding.Request() req.encoding = str(self.encoding_dropdown.currentText()) cli = self.node.create_client(SetEncoding, "/set_encoding") - self.wait_cli(cli,req) + self.wait_cli(cli, req)

@pytest.mark.linter @pytest.mark.pep257 def test_pep257(): - rc = main(argv=['.', 'test']) - assert rc == 0, 'Found code style errors / warnings' + rc = main(argv=[".", "test"]) + assert rc == 0, "Found code style errors / warnings"

@pytest.mark.flake8 @pytest.mark.linter def test_flake8(): rc, errors = main_with_errors(argv=[]) - assert rc == 0, \ - 'Found %d code style errors / warnings:\n' % len(errors) + \ - '\n'.join(errors) + assert rc == 0, "Found %d code style errors / warnings:\n" % len( + errors + ) + "\n".join(errors)

description="Whether to run nvblox", ) ld = IncludeLaunchDescription( PythonLaunchDescriptionSource( - PathJoinSubstitution([FindPackageShare("isaac_ros_launch"), "isaac_launch.py"]) + PathJoinSubstitution( + [FindPackageShare("isaac_ros_launch"), "isaac_launch.py"] + ) ), launch_arguments={ "setup_for_zed": "False", "setup_for_gazebo": "True", "use_nvblox": LaunchConfiguration("use_nvblox"),

def generate_launch_description(): ld = LaunchDescription() main_launch = IncludeLaunchDescription( PythonLaunchDescriptionSource( - PathJoinSubstitution([FindPackageShare("rovr_control"), "main_no_joysticks_launch.py"]) + PathJoinSubstitution( + [FindPackageShare("rovr_control"), "main_no_joysticks_launch.py"] + ) ) ) isaac_launch = IncludeLaunchDescription( - PythonLaunchDescriptionSource(PathJoinSubstitution([FindPackageShare("isaac_ros_launch"), "isaac_launch.py"])), + PythonLaunchDescriptionSource( + PathJoinSubstitution( + [FindPackageShare("isaac_ros_launch"), "isaac_launch.py"] + ) + ), launch_arguments={ "setup_for_zed": "True", "setup_for_gazebo": "False", "use_nvblox": "True", "run_rviz_robot": "False", # We don't need to run RViz during matches

# Rviz node rviz = Node( package="rviz2", executable="rviz2", - arguments=["-d", config_path, "-f", global_frame], # set the config # overwrite the global frame + arguments=[ + "-d", + config_path, + "-f", + global_frame, + ], # set the config # overwrite the global frame output="screen", ) return LaunchDescription([rviz])

specialization_dir = PathJoinSubstitution([base_config_dir, "specializations"]) # Config files base_config = PathJoinSubstitution([base_config_dir, "nvblox_base.yaml"]) zed_config = PathJoinSubstitution([specialization_dir, "nvblox_zed.yaml"]) - gazebo_simulation_config = PathJoinSubstitution([specialization_dir, "nvblox_gazebo_sim.yaml"]) + gazebo_simulation_config = PathJoinSubstitution( + [specialization_dir, "nvblox_gazebo_sim.yaml"] + ) # Conditionals for setup setup_for_zed = IfCondition(LaunchConfiguration("setup_for_zed", default="False")) - setup_for_gazebo = IfCondition(LaunchConfiguration("setup_for_gazebo", default="False")) + setup_for_gazebo = IfCondition( + LaunchConfiguration("setup_for_gazebo", default="False") + ) # Option to attach the nodes to a shared component container for speed ups through intra process communication. # Make sure to set the 'component_container_name' to the name of the component container you want to attach to. attach_to_shared_component_container_arg = LaunchConfiguration( "attach_to_shared_component_container", default=False ) - component_container_name_arg = LaunchConfiguration("component_container_name", default="nvblox_container") + component_container_name_arg = LaunchConfiguration( + "component_container_name", default="nvblox_container" + ) # If we do not attach to a shared component container we have to create our own container. nvblox_container = Node( name=component_container_name_arg, package="rclcpp_components",

) load_composable_nodes = LoadComposableNodes( target_container=component_container_name_arg, composable_node_descriptions=[ - ComposableNode(name="nvblox_node", package="nvblox_ros", plugin="nvblox::NvbloxNode") + ComposableNode( + name="nvblox_node", package="nvblox_ros", plugin="nvblox::NvbloxNode" + ) ], ) group_action = GroupAction( [ # Set parameters with specializations SetParametersFromFile(base_config), SetParametersFromFile(zed_config, condition=setup_for_zed), SetParametersFromFile(gazebo_simulation_config, condition=setup_for_gazebo), - SetParameter(name="global_frame", value=LaunchConfiguration("global_frame", default="odom")), + SetParameter( + name="global_frame", + value=LaunchConfiguration("global_frame", default="odom"), + ), # Remappings for zed data - SetRemap(src=["depth/image"], dst=["/zed2i/zed_node/depth/depth_registered"], condition=setup_for_zed), - SetRemap(src=["depth/camera_info"], dst=["/zed2i/zed_node/depth/camera_info"], condition=setup_for_zed), - SetRemap(src=["color/image"], dst=["/zed2i/zed_node/rgb/image_rect_color"], condition=setup_for_zed), - SetRemap(src=["color/camera_info"], dst=["/zed2i/zed_node/rgb/camera_info"], condition=setup_for_zed), - SetRemap(src=["pose"], dst=["/zed2i/zed_node/pose"], condition=setup_for_zed), + SetRemap( + src=["depth/image"], + dst=["/zed2i/zed_node/depth/depth_registered"], + condition=setup_for_zed, + ), + SetRemap( + src=["depth/camera_info"], + dst=["/zed2i/zed_node/depth/camera_info"], + condition=setup_for_zed, + ), + SetRemap( + src=["color/image"], + dst=["/zed2i/zed_node/rgb/image_rect_color"], + condition=setup_for_zed, + ), + SetRemap( + src=["color/camera_info"], + dst=["/zed2i/zed_node/rgb/camera_info"], + condition=setup_for_zed, + ), + SetRemap( + src=["pose"], dst=["/zed2i/zed_node/pose"], condition=setup_for_zed + ), # Include the node container load_composable_nodes, ] )

def generate_launch_description(): # Launch Arguments - run_rviz_arg = DeclareLaunchArgument("run_rviz_robot", default_value="True", description="Whether to start RVIZ") + run_rviz_arg = DeclareLaunchArgument( + "run_rviz_robot", default_value="True", description="Whether to start RVIZ" + ) setup_for_zed_arg = DeclareLaunchArgument( "setup_for_zed", default_value="True", description="Whether to run from live zed data", )

) # ZED zed_launch = IncludeLaunchDescription( PythonLaunchDescriptionSource( - [PathJoinSubstitution([FindPackageShare("isaac_ros_launch"), "zed2i.launch.py"])] + [ + PathJoinSubstitution( + [FindPackageShare("isaac_ros_launch"), "zed2i.launch.py"] + ) + ] ), launch_arguments={ "attach_to_shared_component_container": "True", "component_container_name": shared_container_name, "record_svo": LaunchConfiguration("record_svo"),

condition=IfCondition(LaunchConfiguration("setup_for_zed")), ) # Gazebo gazebo_launch = IncludeLaunchDescription( PythonLaunchDescriptionSource( - [PathJoinSubstitution([FindPackageShare("ros_gz_launch"), "launch", "UCF_field.launch.py"])] + [ + PathJoinSubstitution( + [FindPackageShare("ros_gz_launch"), "launch", "UCF_field.launch.py"] + ) + ] ), condition=IfCondition(LaunchConfiguration("setup_for_gazebo")), ) # Nvblox nvblox_launch = IncludeLaunchDescription( PythonLaunchDescriptionSource( - [PathJoinSubstitution([FindPackageShare("isaac_ros_launch"), "nvblox.launch.py"])] + [ + PathJoinSubstitution( + [FindPackageShare("isaac_ros_launch"), "nvblox.launch.py"] + ) + ] ), launch_arguments={ "global_frame": global_frame, "setup_for_zed": LaunchConfiguration("setup_for_zed"), "setup_for_gazebo": LaunchConfiguration("setup_for_gazebo"),

Line too long (112 > 79 characters) (E501)

Line too long (80 > 79 characters) (E501)

Line too long (83 > 79 characters) (E501)

Line too long (119 > 79 characters) (E501)

Line too long (81 > 79 characters) (E501)

Line too long (90 > 79 characters) (E501)

Line too long (120 > 79 characters) (E501)

Line too long (93 > 79 characters) (E501)

Line too long (80 > 79 characters) (E501)

Line too long (82 > 79 characters) (E501)

Line too long (101 > 79 characters) (E501)

Line too long (84 > 79 characters) (E501)

Line too long (127 > 79 characters) (E501)

from ament_copyright.main import main import pytest # Remove the `skip` decorator once the source file(s) have a copyright header [email protected](reason='No copyright header has been placed in the generated source file.') [email protected]( + reason="No copyright header has been placed in the generated source file." +) @pytest.mark.copyright @pytest.mark.linter def test_copyright(): - rc = main(argv=['.', 'test']) - assert rc == 0, 'Found errors' + rc = main(argv=[".", "test"]) + assert rc == 0, "Found errors"

Line too long (130 > 79 characters) (E501)

from setuptools import find_packages, setup import os from glob import glob -package_name = 'apriltag' + +package_name = "apriltag" setup( name=package_name, - version='0.0.0', - packages=find_packages(exclude=['test']), + version="0.0.0", + packages=find_packages(exclude=["test"]), data_files=[ - ('share/ament_index/resource_index/packages', - ['resource/' + package_name]), - ('share/' + package_name, ['package.xml']), + ("share/ament_index/resource_index/packages", ["resource/" + package_name]), + ("share/" + package_name, ["package.xml"]), ( os.path.join("share", package_name), glob("launch/*launch.[pxy][yma]*", recursive=True), ), ], - install_requires=['setuptools'], + install_requires=["setuptools"], zip_safe=True, - maintainer='umn-robotics', - maintainer_email='[email protected]', - description='Package for processing apriltag detections received from ISAAC ROS Apriltag.', - license='MIT License', - tests_require=['pytest'], + maintainer="umn-robotics", + maintainer_email="[email protected]", + description="Package for processing apriltag detections received from ISAAC ROS Apriltag.", + license="MIT License", + tests_require=["pytest"], entry_points={ - 'console_scripts': ["apriltag_node = apriltag.apriltag_node:main"], + "console_scripts": ["apriltag_node = apriltag.apriltag_node:main"], }, )

Line too long (92 > 79 characters) (E501)

@pytest.mark.flake8 @pytest.mark.linter def test_flake8(): rc, errors = main_with_errors(argv=[]) - assert rc == 0, \ - 'Found %d code style errors / warnings:\n' % len(errors) + \ - '\n'.join(errors) + assert rc == 0, "Found %d code style errors / warnings:\n" % len( + errors + ) + "\n".join(errors)

Line too long (95 > 79 characters) (E501)

@pytest.mark.linter @pytest.mark.pep257 def test_pep257(): - rc = main(argv=['.', 'test']) - assert rc == 0, 'Found code style errors / warnings' + rc = main(argv=[".", "test"]) + assert rc == 0, "Found code style errors / warnings"

Line too long (93 > 79 characters) (E501)

isaac_ros_apriltag = ComposableNode( package="isaac_ros_apriltag", plugin="nvidia::isaac_ros::apriltag::AprilTagNode", name="isaac_ros_apriltag", namespace="", - remappings=[("image", "zed2i/zed_node/left/image_rect_color_rgb"), ("camera_info", "zed2i/zed_node/left/camera_info")], + remappings=[ + ("image", "zed2i/zed_node/left/image_rect_color_rgb"), + ("camera_info", "zed2i/zed_node/left/camera_info"), + ], ) image_format_converter_node_left = ComposableNode( package="isaac_ros_image_proc", plugin="nvidia::isaac_ros::image_proc::ImageFormatConverterNode",

Line too long (81 > 79 characters) (E501)

parameters=[ { "encoding_desired": "rgb8", } ], - remappings=[("image_raw", "zed2i/zed_node/left/image_rect_color"), ("image", "zed2i/zed_node/left/image_rect_color_rgb")], + remappings=[ + ("image_raw", "zed2i/zed_node/left/image_rect_color"), + ("image", "zed2i/zed_node/left/image_rect_color_rgb"), + ], ) apriltag_container = ComposableNodeContainer( package="rclcpp_components", name="apriltag_container", namespace="", executable="component_container_mt", - composable_node_descriptions=[isaac_ros_apriltag, image_format_converter_node_left], + composable_node_descriptions=[ + isaac_ros_apriltag, + image_format_converter_node_left, + ], output="screen", ) # Add nodes and containers to LaunchDescription return launch.LaunchDescription([tag_reader, apriltag_container])

@pytest.mark.copyright @pytest.mark.linter def test_copyright(): - rc = main(argv=['.', 'test']) - assert rc == 0, 'Found errors' + rc = main(argv=[".", "test"]) + assert rc == 0, "Found errors"

Line too long (137 > 79 characters) (E501)

@pytest.mark.flake8 @pytest.mark.linter def test_flake8(): rc, errors = main_with_errors(argv=[]) - assert rc == 0, \ - 'Found %d code style errors / warnings:\n' % len(errors) + \ - '\n'.join(errors) + assert rc == 0, "Found %d code style errors / warnings:\n" % len( + errors + ) + "\n".join(errors)

Line too long (146 > 79 characters) (E501)

@pytest.mark.linter @pytest.mark.pep257 def test_pep257(): - rc = main(argv=['.', 'test']) - assert rc == 0, 'Found code style errors / warnings' + rc = main(argv=[".", "test"]) + assert rc == 0, "Found code style errors / warnings"

Line too long (82 > 79 characters) (E501)

def generate_launch_description(): # Setup to launch the simulator and Gazebo world gz_sim = IncludeLaunchDescription( PythonLaunchDescriptionSource( - [PathJoinSubstitution([FindPackageShare("ros_gz_sim"), "launch", "gz_sim.launch.py"])] + [ + PathJoinSubstitution( + [FindPackageShare("ros_gz_sim"), "launch", "gz_sim.launch.py"] + ) + ] ), launch_arguments={ - "gz_args": PathJoinSubstitution([FindPackageShare("gazebo_files"), "worlds", "NASA_field.sdf"]) + "gz_args": PathJoinSubstitution( + [FindPackageShare("gazebo_files"), "worlds", "NASA_field.sdf"] + ) }.items(), ) # Takes the description and joint angles as inputs and publishes the 3D poses of the robot links robot_state_publisher = IncludeLaunchDescription( PythonLaunchDescriptionSource( - [PathJoinSubstitution([FindPackageShare("robot_description"), "launch", "robot_description.launch.py"])] + [ + PathJoinSubstitution( + [ + FindPackageShare("robot_description"), + "launch", + "robot_description.launch.py", + ] + ) + ] ), launch_arguments={ "setup_for_gazebo": "True", }.items(), ) # Launch Arguments - run_rviz_arg = DeclareLaunchArgument("run_rviz", default_value="True", description="Whether to start RVIZ") + run_rviz_arg = DeclareLaunchArgument( + "run_rviz", default_value="True", description="Whether to start RVIZ" + ) # Visualize in RViz rviz = Node( package="rviz2", executable="rviz2", - arguments=["-d", PathJoinSubstitution([FindPackageShare("ros_gz_launch"), "config", "camera.rviz"])], + arguments=[ + "-d", + PathJoinSubstitution( + [FindPackageShare("ros_gz_launch"), "config", "camera.rviz"] + ), + ], condition=IfCondition(LaunchConfiguration("run_rviz")), ) # Bridge ROS topics and Gazebo messages for establishing communication bridge = Node(

Line too long (140 > 79 characters) (E501)

drivetrain = Node( package="drivetrain", executable="drivetrain_node", name="drivetrain_node", - parameters=["config/drivetrain_config.yaml", "config/motor_control.yaml", {"GAZEBO_SIMULATION": True}], + parameters=[ + "config/drivetrain_config.yaml", + "config/motor_control.yaml", + {"GAZEBO_SIMULATION": True}, + ], output="screen", emulate_tty=True, ) - return LaunchDescription([run_rviz_arg, gz_sim, bridge, drivetrain, robot_state_publisher, rviz]) + return LaunchDescription( + [run_rviz_arg, gz_sim, bridge, drivetrain, robot_state_publisher, rviz] + )

Line too long (94 > 79 characters) (E501)

def generate_launch_description(): # Setup to launch the simulator and Gazebo world gz_sim = IncludeLaunchDescription( PythonLaunchDescriptionSource( - [PathJoinSubstitution([FindPackageShare("ros_gz_sim"), "launch", "gz_sim.launch.py"])] + [ + PathJoinSubstitution( + [FindPackageShare("ros_gz_sim"), "launch", "gz_sim.launch.py"] + ) + ] ), launch_arguments={ - "gz_args": PathJoinSubstitution([FindPackageShare("gazebo_files"), "worlds", "UCF_field.sdf"]) + "gz_args": PathJoinSubstitution( + [FindPackageShare("gazebo_files"), "worlds", "UCF_field.sdf"] + ) }.items(), ) # Takes the description and joint angles as inputs and publishes the 3D poses of the robot links robot_state_publisher = IncludeLaunchDescription( PythonLaunchDescriptionSource( - [PathJoinSubstitution([FindPackageShare("robot_description"), "launch", "robot_description.launch.py"])] + [ + PathJoinSubstitution( + [ + FindPackageShare("robot_description"), + "launch", + "robot_description.launch.py", + ] + ) + ] ), launch_arguments={ "setup_for_gazebo": "True", }.items(), ) # Launch Arguments - run_rviz_arg = DeclareLaunchArgument("run_rviz", default_value="True", description="Whether to start RVIZ") + run_rviz_arg = DeclareLaunchArgument( + "run_rviz", default_value="True", description="Whether to start RVIZ" + ) # Visualize in RViz rviz = Node( package="rviz2", executable="rviz2", - arguments=["-d", PathJoinSubstitution([FindPackageShare("ros_gz_launch"), "config", "camera.rviz"])], + arguments=[ + "-d", + PathJoinSubstitution( + [FindPackageShare("ros_gz_launch"), "config", "camera.rviz"] + ), + ], condition=IfCondition(LaunchConfiguration("run_rviz")), ) # Bridge ROS topics and Gazebo messages for establishing communication bridge = Node(

Line too long (92 > 79 characters) (E501)

drivetrain = Node( package="drivetrain", executable="drivetrain_node", name="drivetrain_node", - parameters=["config/drivetrain_config.yaml", "config/motor_control.yaml", {"GAZEBO_SIMULATION": True}], + parameters=[ + "config/drivetrain_config.yaml", + "config/motor_control.yaml", + {"GAZEBO_SIMULATION": True}, + ], output="screen", emulate_tty=True, ) - return LaunchDescription([run_rviz_arg, gz_sim, bridge, drivetrain, robot_state_publisher, rviz]) + return LaunchDescription( + [run_rviz_arg, gz_sim, bridge, drivetrain, robot_state_publisher, rviz] + )

Line too long (94 > 79 characters) (E501)

class ApriltagNode(Node): def __init__(self): super().__init__("apriltag_node") current_dir = os.getcwd() - self.declare_parameter("autonomous_field_type", "bottom") # The type of field ("top", "bottom", "nasa") + self.declare_parameter( + "autonomous_field_type", "bottom" + ) # The type of field ("top", "bottom", "nasa") field_type = self.get_parameter("autonomous_field_type").value paths = { "top": "src/apriltag/apriltag/apriltag_location_ucf_top.urdf.xarco", "bottom": "src/apriltag/apriltag/apriltag_location_ucf_bot.urdf.xarco", "nasa": "src/apriltag/apriltag/apriltag_location_nasa.urdf.xarco",

Line too long (90 > 79 characters) (E501)

self.map_transform.transform.rotation.x = 0.0 self.map_transform.transform.rotation.y = 0.0 self.map_transform.transform.rotation.z = 0.0 self.map_transform.transform.rotation.w = 1.0 - self.transforms = self.create_subscription(AprilTagDetectionArray, "/tag_detections", self.tagDetectionSub, 10) + self.transforms = self.create_subscription( + AprilTagDetectionArray, "/tag_detections", self.tagDetectionSub, 10 + ) self.tf_broadcaster = TransformBroadcaster(self) self.tf_buffer = Buffer() self.tf_listener = TransformListener(self.tf_buffer, self) self.create_service(ResetOdom, "resetOdom", self.reset_callback)

for tag in tags: id = tag.id tree = ET.parse(self.file_path) root = tree.getroot() - link = root[id - 1] # assumes tag 1 = home 1, tag 2 = home 2, 3 = berm 1 etc. + link = root[ + id - 1 + ] # assumes tag 1 = home 1, tag 2 = home 2, 3 = berm 1 etc. xyz_elements = link.findall(".//origin[@xyz]") xyz_values = [element.attrib["xyz"] for element in xyz_elements] xyz = xyz_values[0].split(" ")

Line too long (92 > 79 characters) (E501)

Line too long (90 > 79 characters) (E501)

rpy_values = [element.attrib["rpy"] for element in rpy_elements] rpy = rpy_values[0].split(" ") # Lookup the odom to detected tag tf from the tf buffer try: - odom_to_tag_transform = self.tf_buffer.lookup_transform("odom", f"{tag.family}:{id}", rclpy.time.Time()) + odom_to_tag_transform = self.tf_buffer.lookup_transform( + "odom", f"{tag.family}:{id}", rclpy.time.Time() + ) except TransformException as ex: - self.get_logger().warn(f"Could not transform odom to the detected tag: {ex}") + self.get_logger().warn( + f"Could not transform odom to the detected tag: {ex}" + ) return odom_to_tag_transform.child_frame_id = "odom" odom_to_tag_transform.header.frame_id = "map" odom_to_tag_transform.header.stamp = self.get_clock().now().to_msg()

Line too long (96 > 79 characters) (E501)

odom_to_tag_transform.transform.rotation.y, odom_to_tag_transform.transform.rotation.z, odom_to_tag_transform.transform.rotation.w, ] ).as_quat() - rotated_quaternion = current_quaternion * rotation_quaternion # Multiply the quaternions + rotated_quaternion = ( + current_quaternion * rotation_quaternion + ) # Multiply the quaternions # Update the transform with the rotated quaternion odom_to_tag_transform.transform.rotation.x = rotated_quaternion[0] odom_to_tag_transform.transform.rotation.y = rotated_quaternion[1] odom_to_tag_transform.transform.rotation.z = rotated_quaternion[2]

class ClientPlugin(Plugin): def __init__(self, context): super(ClientPlugin, self).__init__(context) self.setObjectName("ClientPlugin") - assert hasattr(context, 'node'), 'Context does not have a node.' + assert hasattr(context, "node"), "Context does not have a node." self._widget = ClientWidget(context.node) if context.serial_number() > 1: self._widget.setWindowTitle( - self._widget.windowTitle() + (' (%d)' % context.serial_number())) + self._widget.windowTitle() + (" (%d)" % context.serial_number()) + ) context.add_widget(self._widget) - \ No newline at end of file

Line too long (85 > 79 characters) (E501)

from rqt_gui.main import Main def main(): main = Main() - sys.exit(main.main(sys.argv, standalone='gstreamer.client.ClientPlugin')) + sys.exit(main.main(sys.argv, standalone="gstreamer.client.ClientPlugin")) -if __name__ == '__main__': - main() +if __name__ == "__main__": + main()

Line too long (89 > 79 characters) (E501)

from threading import Thread, Event import gi + gi.require_version("Gst", "1.0") from gi.repository import Gst # noqa: E402 + class GstreamerClient: def __init__(self): # Initialize GStreamer Gst.init(None)

Line too long (89 > 79 characters) (E501)

self.pipeline.set_state(Gst.State.PLAYING) self.stop_event = Event() change_source_thread = Thread() change_source_thread.start() - + def stop(self): self.stop_event.set() self.pipeline.set_state(Gst.State.NULL) + if __name__ == "__main__": client = GstreamerClient() client.run() stop_event = Event() change_source_thread = Thread() change_source_thread.start() while True: try: - message:Gst.Message = client.pipeline.get_bus().timed_pop(Gst.SECOND) + message: Gst.Message = client.pipeline.get_bus().timed_pop(Gst.SECOND) if message is None: pass elif message.type == Gst.MessageType.EOS: break elif message.type == Gst.MessageType.ERROR: gi.error("Error", message.parse_error()) break except KeyboardInterrupt: break - client.stop() + client.stop()

Line too long (89 > 79 characters) (E501)

from rovr_interfaces.srv import SetClientIp, SetActiveCamera, SetEncoding gi.require_version("Gst", "1.0") from gi.repository import Gst # noqa: E402 + class GstreamerServer: - def __init__(self, ip_srv: SetClientIp, camera_srv: SetActiveCamera, encod_srv: SetEncoding): + def __init__( + self, ip_srv: SetClientIp, camera_srv: SetActiveCamera, encod_srv: SetEncoding + ): Gst.init(None) self.pipeline = Gst.Pipeline() camera_src = Gst.ElementFactory.make(camera_srv.srctype, "camera_src") if camera_srv.srctype == "videotestsrc": print(f"Using videotestsrc{camera_srv.device}") camera_src.set_property("pattern", int(camera_srv.device)) else: camera_src.set_property("device", camera_srv.device) self.pipeline.add(camera_src) - + nonNVvideoconvert = Gst.ElementFactory.make("videoconvert", "videoconvert") self.pipeline.add(nonNVvideoconvert) camera_src.link(nonNVvideoconvert) caps = Gst.Caps.from_string(

Line too long (89 > 79 characters) (E501)

) src_caps = Gst.ElementFactory.make("capsfilter", "src_caps") src_caps.set_property("caps", caps) self.pipeline.add(src_caps) nonNVvideoconvert.link(src_caps) - if platform.machine() == "aarch64": videoconvert = Gst.ElementFactory.make("nvvidconv", "nvvidconv") elif platform.machine() == "x86_64" or platform.machine() == "amd64": videoconvert = Gst.ElementFactory.make("nvvideoconvert", "nvvideoconvert")

Line too long (91 > 79 characters) (E501)

self.pipeline.set_state(Gst.State.PLAYING) def stop(self): self.pipeline.set_state(Gst.State.NULL) + if __name__ == "__main__": ip = SetClientIp.Request() ip.client_ip = "127.0.0.1" camera = SetActiveCamera.Request() camera.srctype = "videotestsrc"

Line too long (91 > 79 characters) (E501)

camera.height = 480 camera.framerate = 30 camera.format = "NV12" encoding = SetEncoding.Request() encoding.encoding = "h265" - server = GstreamerServer(ip,camera,encoding) + server = GstreamerServer(ip, camera, encoding) server.run() stop_event = Event() change_source_thread = Thread() change_source_thread.start() while True: try: - message:Gst.Message = server.pipeline.get_bus().timed_pop(Gst.SECOND) + message: Gst.Message = server.pipeline.get_bus().timed_pop(Gst.SECOND) if message is None: pass elif message.type == Gst.MessageType.EOS: break elif message.type == Gst.MessageType.ERROR:

Line too long (91 > 79 characters) (E501)

from rclpy.node import Node from rovr_interfaces.srv import SetClientIp, SetActiveCamera, SetEncoding from .server_gstreamer import GstreamerServer import time + class ServerNode(Node): g_server: GstreamerServer = None ip_srv: SetClientIp = None camera_srv: SetActiveCamera = None encod_srv: SetEncoding = None + def __init__(self): super().__init__("server") self.serv_set_client_ip = self.create_service( SetClientIp, "/set_client_ip", self.set_client_ip_callback )

Line too long (91 > 79 characters) (E501)

def set_active_camera(self, request: SetActiveCamera, response): print("recieved camera request") self.camera_srv = request response.success = self.restart_server() return response - + def set_encoding(self, request: SetEncoding, response): print("recieved encoding request") self.encod_srv = request response.success = self.restart_server() return response

Line too long (88 > 79 characters) (E501)

self.g_server = GstreamerServer(self.ip_srv, self.camera_srv, self.encod_srv) self.g_server.run() print("Server restarted") return 0 + def main(args=None): rclpy.init(args=args) server = ServerNode() rclpy.spin(server) # Destroy the node explicitly

Line too long (92 > 79 characters) (E501)

# Config config_name = LaunchConfiguration("config_name", default="zed_example.rviz") config_path = PathJoinSubstitution(["config", "rviz", config_name]) global_frame = LaunchConfiguration("global_frame", default="odom") - run_rviz_arg = DeclareLaunchArgument("run_rviz_client", default_value="True", description="Whether to start RVIZ") + run_rviz_arg = DeclareLaunchArgument( + "run_rviz_client", default_value="True", description="Whether to start RVIZ" + ) joystick_node = Node( package="joy", executable="joy_node", parameters=["config/joy_node.yaml"], ) start_gStreamer_client = ExecuteProcess( - cmd=["rqt", "--force-discover", "--standalone", "CameraClient"], shell=True, output="screen" + cmd=["rqt", "--force-discover", "--standalone", "CameraClient"], + shell=True, + output="screen", ) # NOTE: RVIZ uses a lot of bandwidth, so it should be turned off during competition matches rviz_node = Node( package="rviz2",

Line too long (103 > 79 characters) (E501)

arguments=["-d", config_path, "-f", global_frame], condition=IfCondition(LaunchConfiguration("run_rviz_client")), ) ld.add_action(run_rviz_arg) - ld.add_action(joystick_node) # TODO: The joystick node currently does not work inside the container! + ld.add_action( + joystick_node + ) # TODO: The joystick node currently does not work inside the container! ld.add_action(start_gStreamer_client) ld.add_action(rviz_node) return ld

Line too long (86 > 79 characters) (E501)

from ament_copyright.main import main import pytest # Remove the `skip` decorator once the source file(s) have a copyright header [email protected](reason='No copyright header has been placed in the generated source file.') [email protected]( + reason="No copyright header has been placed in the generated source file." +) @pytest.mark.copyright @pytest.mark.linter def test_copyright(): - rc = main(argv=['.', 'test']) - assert rc == 0, 'Found errors' + rc = main(argv=[".", "test"]) + assert rc == 0, "Found errors"

Line too long (97 > 79 characters) (E501)

import socket import fcntl import struct from rqt_py_common.extended_combo_box import ExtendedComboBox + class ClientWidget(QWidget): - timeout = 2e9 # 2 seconds with nano seconds as unit + timeout = 2e9 # 2 seconds with nano seconds as unit encodings = ["av1", "h265"] + def __init__(self, node: Node): super(ClientWidget, self).__init__() self.setObjectName("ClientWidget") self.node = node self.display_window = GstreamerClient()

Line too long (95 > 79 characters) (E501)

# Call the buttons to set ip and encoding by default # self.on_ip_push_button_clicked() self.on_encoding_push_button_clicked() def add_network_interfaces(self, comboBox: QComboBox): - for _,interface in socket.if_nameindex(): + for _, interface in socket.if_nameindex(): if interface != "lo": comboBox.addItem(interface) comboBox.setCurrentIndex(0) def get_encodings(self, comboBox: QComboBox): for encoding in self.encodings: comboBox.addItem(encoding) comboBox.setCurrentIndex(0) - + def get_ip_address(self): s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) interface = str(self.network_dropdown.currentText()) return socket.inet_ntoa( fcntl.ioctl( s.fileno(), 0x8915, - struct.pack( - "256s", interface[:15].encode("utf-8") - ), # SIOCGIFADDR + struct.pack("256s", interface[:15].encode("utf-8")), # SIOCGIFADDR )[20:24] - ) - - def wait_cli(self,cli:Client,req): + ) + + def wait_cli(self, cli: Client, req): future = cli.call_async(req) start_time = self.node.get_clock().now().nanoseconds # Block while waiting for server to respond - while rclpy.ok() and not future.done() and self.node.get_clock().now().nanoseconds - start_time < self.timeout: + while ( + rclpy.ok() + and not future.done() + and self.node.get_clock().now().nanoseconds - start_time < self.timeout + ): pass if not future.done(): print("Service Call Failed") return - + print("Service Call Returned") result = future.result().success if result == -1: print("No client IP Set") elif result == -2: print("No encoding set") elif result == -3: print("No camera selected") - - # self.restart_window() + + # self.restart_window() self.node.destroy_client(cli) - + @slot() def on_camera1_push_button_clicked(self): print("Requesting Camera 1") req = SetActiveCamera.Request() req.srctype = "v4l2src"

Line too long (99 > 79 characters) (E501)

req.width = 640 req.height = 480 req.framerate = 30 req.format = "NV12" cli = self.node.create_client(SetActiveCamera, "/set_active_camera") - self.wait_cli(cli,req) + self.wait_cli(cli, req) @slot() def on_camera2_push_button_clicked(self): print("Requesting Camera 2") req = SetActiveCamera.Request()

Line too long (97 > 79 characters) (E501)

req.width = 640 req.height = 480 req.framerate = 30 req.format = "NV12" cli = self.node.create_client(SetActiveCamera, "/set_active_camera") - self.wait_cli(cli,req) + self.wait_cli(cli, req) @slot() def on_camera3_push_button_clicked(self): print("Requesting Camera 3") req = SetActiveCamera.Request()

Line too long (95 > 79 characters) (E501)

req.width = 640 req.height = 480 req.framerate = 30 req.format = "NV12" cli = self.node.create_client(SetActiveCamera, "/set_active_camera") - self.wait_cli(cli,req) + self.wait_cli(cli, req) @slot() def on_camera4_push_button_clicked(self): print("Requesting Camera 4") req = SetActiveCamera.Request()

req.width = 640 req.height = 480 req.framerate = 30 req.format = "NV12" cli = self.node.create_client(SetActiveCamera, "/set_active_camera") - self.wait_cli(cli,req) - + self.wait_cli(cli, req) + @slot() def on_camera5_push_button_clicked(self): print("Requesting Camera 5") req = SetActiveCamera.Request() req.srctype = "v4l2src"

Line too long (93 > 79 characters) (E501)

req.width = 640 req.height = 480 req.framerate = 30 req.format = "NV12" cli = self.node.create_client(SetActiveCamera, "/set_active_camera") - self.wait_cli(cli,req) + self.wait_cli(cli, req) def restart_window(self): self.display_window.stop() self.display_window = GstreamerClient() self.display_window.run()

Line too long (97 > 79 characters) (E501)

req.client_ip = self.get_ip_address() except OSError as e: print(e) return cli = self.node.create_client(SetClientIp, "/set_client_ip") - self.wait_cli(cli,req) - + self.wait_cli(cli, req) + @slot() def on_encoding_push_button_clicked(self): req = SetEncoding.Request() req.encoding = str(self.encoding_dropdown.currentText()) cli = self.node.create_client(SetEncoding, "/set_encoding") - self.wait_cli(cli,req) + self.wait_cli(cli, req)

Line too long (95 > 79 characters) (E501)

@pytest.mark.flake8 @pytest.mark.linter def test_flake8(): rc, errors = main_with_errors(argv=[]) - assert rc == 0, \ - 'Found %d code style errors / warnings:\n' % len(errors) + \ - '\n'.join(errors) + assert rc == 0, "Found %d code style errors / warnings:\n" % len( + errors + ) + "\n".join(errors)

@pytest.mark.linter @pytest.mark.pep257 def test_pep257(): - rc = main(argv=['.', 'test']) - assert rc == 0, 'Found code style errors / warnings' + rc = main(argv=[".", "test"]) + assert rc == 0, "Found code style errors / warnings"

description="Whether to run nvblox", ) ld = IncludeLaunchDescription( PythonLaunchDescriptionSource( - PathJoinSubstitution([FindPackageShare("isaac_ros_launch"), "isaac_launch.py"]) + PathJoinSubstitution( + [FindPackageShare("isaac_ros_launch"), "isaac_launch.py"] + ) ), launch_arguments={ "setup_for_zed": "False", "setup_for_gazebo": "True", "use_nvblox": LaunchConfiguration("use_nvblox"),

def generate_launch_description(): ld = LaunchDescription() main_launch = IncludeLaunchDescription( PythonLaunchDescriptionSource( - PathJoinSubstitution([FindPackageShare("rovr_control"), "main_no_joysticks_launch.py"]) + PathJoinSubstitution( + [FindPackageShare("rovr_control"), "main_no_joysticks_launch.py"] + ) ) ) isaac_launch = IncludeLaunchDescription( - PythonLaunchDescriptionSource(PathJoinSubstitution([FindPackageShare("isaac_ros_launch"), "isaac_launch.py"])), + PythonLaunchDescriptionSource( + PathJoinSubstitution( + [FindPackageShare("isaac_ros_launch"), "isaac_launch.py"] + ) + ), launch_arguments={ "setup_for_zed": "True", "setup_for_gazebo": "False", "use_nvblox": "True", "run_rviz_robot": "False", # We don't need to run RViz during matches

# Rviz node rviz = Node( package="rviz2", executable="rviz2", - arguments=["-d", config_path, "-f", global_frame], # set the config # overwrite the global frame + arguments=[ + "-d", + config_path, + "-f", + global_frame, + ], # set the config # overwrite the global frame output="screen", ) return LaunchDescription([rviz])

specialization_dir = PathJoinSubstitution([base_config_dir, "specializations"]) # Config files base_config = PathJoinSubstitution([base_config_dir, "nvblox_base.yaml"]) zed_config = PathJoinSubstitution([specialization_dir, "nvblox_zed.yaml"]) - gazebo_simulation_config = PathJoinSubstitution([specialization_dir, "nvblox_gazebo_sim.yaml"]) + gazebo_simulation_config = PathJoinSubstitution( + [specialization_dir, "nvblox_gazebo_sim.yaml"] + ) # Conditionals for setup setup_for_zed = IfCondition(LaunchConfiguration("setup_for_zed", default="False")) - setup_for_gazebo = IfCondition(LaunchConfiguration("setup_for_gazebo", default="False")) + setup_for_gazebo = IfCondition( + LaunchConfiguration("setup_for_gazebo", default="False") + ) # Option to attach the nodes to a shared component container for speed ups through intra process communication. # Make sure to set the 'component_container_name' to the name of the component container you want to attach to. attach_to_shared_component_container_arg = LaunchConfiguration( "attach_to_shared_component_container", default=False ) - component_container_name_arg = LaunchConfiguration("component_container_name", default="nvblox_container") + component_container_name_arg = LaunchConfiguration( + "component_container_name", default="nvblox_container" + ) # If we do not attach to a shared component container we have to create our own container. nvblox_container = Node( name=component_container_name_arg, package="rclcpp_components",

) load_composable_nodes = LoadComposableNodes( target_container=component_container_name_arg, composable_node_descriptions=[ - ComposableNode(name="nvblox_node", package="nvblox_ros", plugin="nvblox::NvbloxNode") + ComposableNode( + name="nvblox_node", package="nvblox_ros", plugin="nvblox::NvbloxNode" + ) ], ) group_action = GroupAction( [ # Set parameters with specializations SetParametersFromFile(base_config), SetParametersFromFile(zed_config, condition=setup_for_zed), SetParametersFromFile(gazebo_simulation_config, condition=setup_for_gazebo), - SetParameter(name="global_frame", value=LaunchConfiguration("global_frame", default="odom")), + SetParameter( + name="global_frame", + value=LaunchConfiguration("global_frame", default="odom"), + ), # Remappings for zed data - SetRemap(src=["depth/image"], dst=["/zed2i/zed_node/depth/depth_registered"], condition=setup_for_zed), - SetRemap(src=["depth/camera_info"], dst=["/zed2i/zed_node/depth/camera_info"], condition=setup_for_zed), - SetRemap(src=["color/image"], dst=["/zed2i/zed_node/rgb/image_rect_color"], condition=setup_for_zed), - SetRemap(src=["color/camera_info"], dst=["/zed2i/zed_node/rgb/camera_info"], condition=setup_for_zed), - SetRemap(src=["pose"], dst=["/zed2i/zed_node/pose"], condition=setup_for_zed), + SetRemap( + src=["depth/image"], + dst=["/zed2i/zed_node/depth/depth_registered"], + condition=setup_for_zed, + ), + SetRemap( + src=["depth/camera_info"], + dst=["/zed2i/zed_node/depth/camera_info"], + condition=setup_for_zed, + ), + SetRemap( + src=["color/image"], + dst=["/zed2i/zed_node/rgb/image_rect_color"], + condition=setup_for_zed, + ), + SetRemap( + src=["color/camera_info"], + dst=["/zed2i/zed_node/rgb/camera_info"], + condition=setup_for_zed, + ), + SetRemap( + src=["pose"], dst=["/zed2i/zed_node/pose"], condition=setup_for_zed + ), # Include the node container load_composable_nodes, ] )

def generate_launch_description(): # Launch Arguments - run_rviz_arg = DeclareLaunchArgument("run_rviz_robot", default_value="True", description="Whether to start RVIZ") + run_rviz_arg = DeclareLaunchArgument( + "run_rviz_robot", default_value="True", description="Whether to start RVIZ" + ) setup_for_zed_arg = DeclareLaunchArgument( "setup_for_zed", default_value="True", description="Whether to run from live zed data", )

) # ZED zed_launch = IncludeLaunchDescription( PythonLaunchDescriptionSource( - [PathJoinSubstitution([FindPackageShare("isaac_ros_launch"), "zed2i.launch.py"])] + [ + PathJoinSubstitution( + [FindPackageShare("isaac_ros_launch"), "zed2i.launch.py"] + ) + ] ), launch_arguments={ "attach_to_shared_component_container": "True", "component_container_name": shared_container_name, "record_svo": LaunchConfiguration("record_svo"),

condition=IfCondition(LaunchConfiguration("setup_for_zed")), ) # Gazebo gazebo_launch = IncludeLaunchDescription( PythonLaunchDescriptionSource( - [PathJoinSubstitution([FindPackageShare("ros_gz_launch"), "launch", "UCF_field.launch.py"])] + [ + PathJoinSubstitution( + [FindPackageShare("ros_gz_launch"), "launch", "UCF_field.launch.py"] + ) + ] ), condition=IfCondition(LaunchConfiguration("setup_for_gazebo")), ) # Nvblox nvblox_launch = IncludeLaunchDescription( PythonLaunchDescriptionSource( - [PathJoinSubstitution([FindPackageShare("isaac_ros_launch"), "nvblox.launch.py"])] + [ + PathJoinSubstitution( + [FindPackageShare("isaac_ros_launch"), "nvblox.launch.py"] + ) + ] ), launch_arguments={ "global_frame": global_frame, "setup_for_zed": LaunchConfiguration("setup_for_zed"), "setup_for_gazebo": LaunchConfiguration("setup_for_gazebo"),

Node.js 16 actions are deprecated. Please update the following actions to use Node.js 20: actions/setup-python@v1, wearerequired/lint-action@v2. For more information see: https://github.blog/changelog/2023-09-22-github-actions-transitioning-from-node-16-to-node-20/.